rx-rust 1.0.0

mod tests_utils;

use crate::tests_utils::DURATION_10_MS;
use crate::tests_utils::DURATION_30_MS;
use crate::tests_utils::DURATION_100_MS;
use crate::tests_utils::test_runtime::block_on;
use futures::StreamExt;
use rx_rust::disposable::Disposable;
use rx_rust::scheduler::RecursionAction;
use rx_rust::scheduler::Scheduler;
use std::time::{Duration, Instant};

const RECURSION_EXECUTION_TIMES: usize = 200;

#[test]
fn test_schedule_without_delay() {
    block_on(|runtime| async move {
        let (tx, rx) = futures::channel::oneshot::channel();
        let task = || {
            tx.send(()).unwrap();
        };
        let start_time = Instant::now();
        let disposal = runtime.schedule(task, None);
        assert!(rx.await.is_ok());
        let elapsed_time = start_time.elapsed();
        assert!(elapsed_time < DURATION_30_MS);
        disposal.dispose();
    });
}

#[test]
fn test_schedule_with_delay() {
    block_on(|runtime| async move {
        let (tx, rx) = futures::channel::oneshot::channel();
        let task = || {
            tx.send(()).unwrap();
        };
        let start_time = Instant::now();
        let disposal = runtime.schedule(task, Some(DURATION_100_MS));
        assert!(rx.await.is_ok());
        let elapsed_time = start_time.elapsed();
        assert!(elapsed_time >= DURATION_100_MS);
        assert!(elapsed_time < DURATION_100_MS + DURATION_30_MS);
        disposal.dispose();
    });
}

#[test]
fn test_schedule_with_abort() {
    block_on(|runtime| async move {
        let (tx, rx) = futures::channel::oneshot::channel();
        let task = || {
            tx.send(()).unwrap();
        };
        let start_time = Instant::now();
        let disposal = runtime.schedule(task, Some(DURATION_100_MS));
        disposal.dispose();
        assert!(rx.await.is_err());
        let elapsed_time = start_time.elapsed();
        assert!(elapsed_time < DURATION_30_MS);
    });
}

#[test]
fn test_schedule_with_late_abort() {
    block_on(|runtime| async move {
        let (tx, rx) = futures::channel::oneshot::channel();
        let task = || {
            tx.send(()).unwrap();
        };
        let disposal = runtime.schedule(task, None);
        runtime.sleep(DURATION_10_MS).await;
        disposal.dispose();
        assert!(rx.await.is_ok());
    });
}

#[test]
fn test_schedule_recursively_without_delay() {
    block_on(|runtime| async move {
        let (tx, mut rx) = futures::channel::mpsc::unbounded();
        let mut tx = Some(tx);
        let first = Instant::now();
        let start_instant = Instant::now();
        let disposal = runtime.schedule_recursively(
            move |index| {
                if index == RECURSION_EXECUTION_TIMES {
                    tx.take().unwrap();
                    RecursionAction::Stop
                } else {
                    tx.as_ref().unwrap().unbounded_send(Instant::now()).unwrap();
                    RecursionAction::ContinueAt(first + DURATION_10_MS * (index as u32 + 1))
                }
            },
            None,
        );
        let mut count = 0;
        while let Some(call_instant) = rx.next().await {
            let duration = call_instant - start_instant;
            let diff = duration - (count as u32 * DURATION_10_MS);
            assert!(diff < DURATION_30_MS, "diff: {diff:?}, count: {count}");
            count += 1;
        }
        assert_eq!(count, RECURSION_EXECUTION_TIMES);
        disposal.dispose();
    });
}

#[test]
fn test_schedule_recursively_with_delay() {
    block_on(|runtime| async move {
        let (tx, mut rx) = futures::channel::mpsc::unbounded();
        let mut tx = Some(tx);
        let first = Instant::now() + DURATION_10_MS;
        let start_instant = Instant::now();
        let disposal = runtime.schedule_recursively(
            move |index| {
                if index == RECURSION_EXECUTION_TIMES {
                    tx.take().unwrap();
                    RecursionAction::Stop
                } else {
                    tx.as_ref().unwrap().unbounded_send(Instant::now()).unwrap();
                    RecursionAction::ContinueAt(first + DURATION_10_MS * (index as u32 + 1))
                }
            },
            Some(DURATION_10_MS),
        );
        let mut count = 0;
        while let Some(call_instant) = rx.next().await {
            let duration = call_instant - start_instant;
            let diff = duration - (count as u32 * DURATION_10_MS);
            assert!(diff < DURATION_30_MS, "diff: {diff:?}, count: {count}");
            count += 1;
        }
        assert_eq!(count, RECURSION_EXECUTION_TIMES);
        disposal.dispose();
    });
}

// For panic `overflow when subtracting durations` in `let delay = delay - diff`.
#[test]
fn test_schedule_recursively_small_delay() {
    block_on(|runtime| async move {
        let (tx, mut rx) = futures::channel::mpsc::unbounded();
        let mut tx = Some(tx);
        let small = DURATION_10_MS;
        let first = Instant::now() + small;
        let disposal = runtime.schedule_recursively(
            move |index| {
                if index == RECURSION_EXECUTION_TIMES {
                    tx.take().unwrap();
                    RecursionAction::Stop
                } else {
                    tx.as_ref().unwrap().unbounded_send(()).unwrap();
                    RecursionAction::ContinueAt(first + small * (index as u32 + 1))
                }
            },
            Some(small),
        );
        let mut count = 0;
        while (rx.next().await).is_some() {
            count += 1;
        }
        assert_eq!(count, RECURSION_EXECUTION_TIMES);
        disposal.dispose();
    });
}

// `ContinueImmediately` must yield between iterations: otherwise the first
// receive below would starve on a single-threaded pool, and disposal could
// never take effect (abort/cancel only happens at await points).
#[test]
fn test_schedule_recursively_continue_immediately_yields_and_disposes() {
    block_on(|runtime| async move {
        let (tx, mut rx) = futures::channel::mpsc::unbounded();
        let disposal = runtime.schedule_recursively(
            move |index| {
                // The receiver may be gone after disposal races; ignore errors.
                let _ = tx.unbounded_send(index);
                RecursionAction::ContinueImmediately
            },
            None,
        );
        // Requires the recursive loop to yield to this task.
        assert!(rx.next().await.is_some());
        disposal.dispose();
        // After disposal the task is dropped, dropping `tx` and closing the
        // channel. If disposal didn't stop the loop, this would hang forever.
        while rx.next().await.is_some() {}
    });
}

// Same guarantee for `ContinueAt` with an instant that has already passed.
#[test]
fn test_schedule_recursively_continue_at_past_instant_yields_and_disposes() {
    block_on(|runtime| async move {
        let (tx, mut rx) = futures::channel::mpsc::unbounded();
        let past = Instant::now();
        let disposal = runtime.schedule_recursively(
            move |index| {
                let _ = tx.unbounded_send(index);
                // Always in the past by the time it is evaluated.
                RecursionAction::ContinueAt(past)
            },
            None,
        );
        assert!(rx.next().await.is_some());
        disposal.dispose();
        while rx.next().await.is_some() {}
    });
}

#[test]
fn test_schedule_periodically_without_delay() {
    block_on(|runtime| async move {
        let (tx, mut rx) = futures::channel::mpsc::unbounded();
        let mut tx = Some(tx);
        let start_instant = Instant::now();
        let disposal = runtime.schedule_periodically(
            move |index| {
                if index == RECURSION_EXECUTION_TIMES {
                    tx.take().unwrap();
                    false
                } else {
                    tx.as_ref().unwrap().unbounded_send(Instant::now()).unwrap();
                    true
                }
            },
            DURATION_10_MS,
            None,
        );
        let mut count = 0;
        while let Some(call_instant) = rx.next().await {
            let duration = call_instant - start_instant;
            let diff = duration - (count as u32 * DURATION_10_MS);
            assert!(diff < DURATION_30_MS, "diff: {diff:?}, count: {count}");
            count += 1;
        }
        assert_eq!(count, RECURSION_EXECUTION_TIMES);
        disposal.dispose();
    });
}

#[test]
fn test_schedule_periodically_with_delay() {
    block_on(|runtime| async move {
        let (tx, mut rx) = futures::channel::mpsc::unbounded();
        let mut tx = Some(tx);
        let start_instant = Instant::now();
        let disposal = runtime.schedule_periodically(
            move |index| {
                if index == RECURSION_EXECUTION_TIMES {
                    tx.take().unwrap();
                    false
                } else {
                    tx.as_ref().unwrap().unbounded_send(Instant::now()).unwrap();
                    true
                }
            },
            DURATION_10_MS,
            Some(DURATION_10_MS),
        );
        let mut count = 0;
        while let Some(call_instant) = rx.next().await {
            let duration = call_instant - start_instant;
            let diff = duration - (count as u32 * DURATION_10_MS);
            assert!(diff < DURATION_30_MS, "diff: {diff:?}, count: {count}");
            count += 1;
        }
        assert_eq!(count, RECURSION_EXECUTION_TIMES);
        disposal.dispose();
    });
}

#[test]
#[should_panic(expected = "period must be non-zero")]
fn test_schedule_periodically_rejects_zero_period() {
    block_on(|runtime| async move {
        let _disposal = runtime.schedule_periodically(|_| false, Duration::ZERO, None);
    });
}

#[test]
fn test_schedule_stream() {
    block_on(|runtime| async move {
        let (tx, mut rx) = futures::channel::mpsc::unbounded();
        let _disposal = runtime.schedule_stream(futures::stream::iter([1, 2, 3]), move |item| {
            tx.unbounded_send(item).unwrap();
            true
        });
        let mut results = Vec::new();
        while let Some(result) = rx.next().await {
            results.push(result);
        }
        // Each element arrives, and the end of the stream is announced with a `None`.
        assert_eq!(results, [Some(1), Some(2), Some(3), None]);
    });
}

#[test]
fn test_schedule_stream_stops_on_false() {
    block_on(|runtime| async move {
        let (tx, mut rx) = futures::channel::mpsc::unbounded();
        let stream = futures::stream::iter(1..);
        let _disposal = runtime.schedule_stream(stream, move |item| {
            let item = item.unwrap();
            tx.unbounded_send(item).unwrap();
            item < 2
        });
        // A `false` answer ends the task, which drops `tx` and closes the channel. If the loop
        // kept polling the infinite stream, this would hang forever.
        let mut results = Vec::new();
        while let Some(result) = rx.next().await {
            results.push(result);
        }
        // The final `None` is not delivered after a stop: it would `unwrap` above.
        assert_eq!(results, [1, 2]);
    });
}

#[cfg(feature = "tokio-scheduler")]
#[test]
fn test_tokio_handle_schedules_outside_runtime_context() {
    let runtime = tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .expect("Failed building the Runtime");
    let handle = runtime.handle().clone();
    let (tx, rx) = futures::channel::oneshot::channel();

    let disposal = handle.schedule(
        move || tx.send(()).expect("receiver should remain alive"),
        Some(DURATION_10_MS),
    );

    assert!(runtime.block_on(rx).is_ok());
    disposal.dispose();
}