smarm/tests/selective_recv.rs

//! Selective-receive tests: `Receiver::recv_match` and `try_recv_match`.
//!
//! `recv_match(pred)` scans the queued messages, removes and returns the first
//! one for which `pred` holds, and leaves the rest in arrival order. With no
//! match it parks and re-scans on every send (a selective receiver may park on
//! a *non-empty* queue), returning `Err` only once the channel is closed and no
//! queued message matches. `try_recv_match` is the non-blocking variant.

use smarm::{channel, run, spawn};
use std::sync::{Arc, Mutex};

// ---------------------------------------------------------------------------
// recv_match — same-actor scan semantics (no parking)
// ---------------------------------------------------------------------------

// First matching message is pulled out of arrival order; non-matches stay
// queued in their original order for later plain recv().
#[test]
fn match_pulled_first_non_matches_remain_in_order() {
    let got = Arc::new(Mutex::new(Vec::<i64>::new()));
    let got2 = got.clone();

    run(move || {
        let (tx, rx) = channel::<i64>();
        for v in [1, 2, 3] {
            tx.send(v).unwrap();
        }
        // 2 is the only even; it's pulled despite arriving second.
        let m = rx.recv_match(|&v| v % 2 == 0).unwrap();
        // The non-matches remain, in order: 1 then 3.
        let a = rx.recv().unwrap();
        let b = rx.recv().unwrap();
        let mut g = got2.lock().unwrap();
        g.push(m);
        g.push(a);
        g.push(b);
    });

    assert_eq!(*got.lock().unwrap(), vec![2, 1, 3]);
}

// A read-only captured value (request-id style) is a plain `Fn` predicate.
#[test]
fn matches_on_captured_value() {
    let got = Arc::new(Mutex::new(0i64));
    let got2 = got.clone();

    run(move || {
        let (tx, rx) = channel::<i64>();
        for v in [10, 20, 30] {
            tx.send(v).unwrap();
        }
        let target = 20;
        let m = rx.recv_match(move |&v| v == target).unwrap();
        *got2.lock().unwrap() = m;
    });

    assert_eq!(*got.lock().unwrap(), 20);
}

// ---------------------------------------------------------------------------
// recv_match — parking on a non-empty, no-match queue
// ---------------------------------------------------------------------------

// The receiver parks while the queue holds only non-matching messages, must
// wake on a send that does NOT empty->fill the queue, re-scan, re-park on a
// still-non-matching send, and finally return when a match arrives.
#[test]
fn parks_on_non_matching_queue_then_wakes_on_match() {
    let got = Arc::new(Mutex::new(0i64));
    let got2 = got.clone();

    run(move || {
        let (tx, rx) = channel::<i64>();
        tx.send(1).unwrap(); // odd: queued, will not match
        let h = spawn(move || {
            // Scans [1], no even -> parks on a non-empty queue.
            let v = rx.recv_match(|&v| v % 2 == 0).unwrap();
            *got2.lock().unwrap() = v;
        });
        smarm::yield_now(); // let the child park
        tx.send(3).unwrap(); // odd: child wakes, re-scans [1,3], re-parks
        tx.send(4).unwrap(); // even: child wakes, scans [1,3,4], returns 4
        h.join().unwrap();
    });

    assert_eq!(*got.lock().unwrap(), 4);
}

// ---------------------------------------------------------------------------
// recv_match — close semantics
// ---------------------------------------------------------------------------

// Channel closes (all senders dropped) while only non-matching messages are
// queued: recv_match must return Err rather than block forever.
#[test]
fn closed_with_only_non_matches_returns_err() {
    let saw_err = Arc::new(Mutex::new(false));
    let saw_err2 = saw_err.clone();

    run(move || {
        let (tx, rx) = channel::<i64>();
        let h = spawn(move || {
            // Wants an even; only odds will ever exist.
            let r = rx.recv_match(|&v| v % 2 == 0);
            *saw_err2.lock().unwrap() = r.is_err();
        });
        smarm::yield_now(); // child parks on empty queue
        tx.send(1).unwrap(); // odd: wakes, re-scans, re-parks (no match)
        drop(tx); // last sender gone, still no match -> Err
        h.join().unwrap();
    });

    assert!(*saw_err.lock().unwrap());
}

// A match present on a closed channel is still returned (closure doesn't
// pre-empt an available match).
#[test]
fn closed_but_match_present_returns_match() {
    let got = Arc::new(Mutex::new(0i64));
    let got2 = got.clone();

    run(move || {
        let (tx, rx) = channel::<i64>();
        tx.send(1).unwrap();
        tx.send(2).unwrap();
        drop(tx); // closed, but a match (2) is queued
        let m = rx.recv_match(|&v| v % 2 == 0).unwrap();
        *got2.lock().unwrap() = m;
    });

    assert_eq!(*got.lock().unwrap(), 2);
}

// ---------------------------------------------------------------------------
// try_recv_match — non-blocking
// ---------------------------------------------------------------------------

#[test]
fn try_recv_match_states() {
    let out = Arc::new(Mutex::new(Vec::<String>::new()));
    let out2 = out.clone();

    run(move || {
        let (tx, rx) = channel::<i64>();
        let log = |s: &str| out2.lock().unwrap().push(s.to_string());

        // Empty + open -> Ok(None).
        log(match rx.try_recv_match(|&v| v % 2 == 0) {
            Ok(None) => "open_empty_none",
            _ => "WRONG_1",
        });

        // Non-matching present + open -> Ok(None), message left in place.
        tx.send(1).unwrap();
        log(match rx.try_recv_match(|&v| v % 2 == 0) {
            Ok(None) => "open_nomatch_none",
            _ => "WRONG_2",
        });

        // Match present -> Ok(Some(v)).
        tx.send(2).unwrap();
        log(match rx.try_recv_match(|&v| v % 2 == 0) {
            Ok(Some(2)) => "got_match",
            _ => "WRONG_3",
        });

        // Drain the leftover non-match with a plain recv to confirm it stayed.
        log(match rx.recv() {
            Ok(1) => "leftover_one",
            _ => "WRONG_4",
        });

        // Closed + no match -> Err.
        drop(tx);
        log(match rx.try_recv_match(|&v| v % 2 == 0) {
            Err(_) => "closed_err",
            _ => "WRONG_5",
        });
    });

    assert_eq!(
        *out.lock().unwrap(),
        vec![
            "open_empty_none",
            "open_nomatch_none",
            "got_match",
            "leftover_one",
            "closed_err",
        ]
    );
}