oc2_hlapi_v2 0.2.0

//! Device event streams: `EventStream`, `EventStreamGroup`, `parse_event`.

mod common;

use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::{Duration, Instant};

use common::*;
use oc2_hlapi_v2::client::{
    Device, DeviceBase, EventStream, EventStreamGroup, HlapiClient, HlapiClientError,
};
use oc2_hlapi_v2::devices::GenericDevice;
use oc2_hlapi_v2::uuid::Uuid;
use serde_json::{Value, json};

fn ms(n: u64) -> Option<Duration> {
    Some(Duration::from_millis(n))
}

#[derive(Debug, PartialEq, Eq)]
struct Tick(u32);

/// A wrapper with a typed event, counting how often `parse_event` runs.
#[derive(Debug, Clone)]
struct Ticker {
    client: HlapiClient,
    uuid: Uuid,
}

static TICKER_PARSES: AtomicUsize = AtomicUsize::new(0);

impl DeviceBase for Ticker {
    const TYPE: &'static str = "ticker";
    type Event = Tick;
    fn new(client: HlapiClient, uuid: Uuid) -> Result<Self, HlapiClientError> {
        Ok(Self { client, uuid })
    }
    fn uuid_base(&self) -> Uuid {
        self.uuid
    }
    fn client(&self) -> &HlapiClient {
        &self.client
    }
    fn parse_event(
        typ: &str,
        data: &Value,
    ) -> Result<Tick, Box<dyn std::error::Error + Send + Sync>> {
        TICKER_PARSES.fetch_add(1, Ordering::SeqCst);
        if typ != "tick" {
            return Err(format!("unknown event {typ}").into());
        }
        Ok(Tick(serde_json::from_value(data["n"].clone())?))
    }
}

/// Same, but with its own counter so tests can run in parallel.
#[derive(Debug, Clone)]
struct Counted {
    client: HlapiClient,
    uuid: Uuid,
}

static COUNTED_PARSES: AtomicUsize = AtomicUsize::new(0);

impl DeviceBase for Counted {
    const TYPE: &'static str = "counted";
    type Event = ();
    fn new(client: HlapiClient, uuid: Uuid) -> Result<Self, HlapiClientError> {
        Ok(Self { client, uuid })
    }
    fn uuid_base(&self) -> Uuid {
        self.uuid
    }
    fn client(&self) -> &HlapiClient {
        &self.client
    }
    fn parse_event(_: &str, _: &Value) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
        COUNTED_PARSES.fetch_add(1, Ordering::SeqCst);
        Ok(())
    }
}

fn tick(dev: u128, n: u32) -> Value {
    device_event(1, "tick", id(dev), json!({"n": n}))
}

fn ticker(client: &HlapiClient, n: u128) -> Ticker {
    Ticker::new(client.clone(), id(n)).unwrap()
}

#[test]
fn events_are_parsed_into_the_devices_event_type_in_order() {
    let (client, s) = client_silent();
    let stream = ticker(&client, 1).event_stream();
    for n in 0..3 {
        s.host.emit(&tick(1, n));
    }
    for n in 0..3 {
        assert_eq!(stream.get_event(ms(2000)).unwrap(), Some(Tick(n)));
    }
}

#[test]
fn get_event_times_out_with_none() {
    let (client, _s) = client_silent();
    let stream = ticker(&client, 1).event_stream();
    let start = Instant::now();
    assert_eq!(stream.get_event(ms(80)).unwrap(), None);
    let took = start.elapsed();
    assert!(took >= Duration::from_millis(70), "{took:?}");
    assert!(took < Duration::from_secs(5), "{took:?}");
}

#[test]
fn a_zero_timeout_is_a_non_blocking_get() {
    let (client, s) = client_silent();
    let stream = ticker(&client, 1).event_stream();
    let r = within({
        let stream = stream.clone();
        move || stream.get_event(Some(Duration::ZERO))
    });
    assert_eq!(r.unwrap(), None);
    s.host.emit(&tick(1, 1));
    std::thread::sleep(SHORT);
    assert_eq!(
        stream.get_event(Some(Duration::ZERO)).unwrap(),
        Some(Tick(1))
    );
}

#[test]
fn no_timeout_waits_for_an_event() {
    let (client, s) = client_silent();
    let stream = ticker(&client, 1).event_stream();
    let waiter = std::thread::spawn(move || stream.get_event(None));
    std::thread::sleep(SHORT);
    assert!(!waiter.is_finished());
    s.host.emit(&tick(1, 2));
    assert_eq!(waiter.join().unwrap().unwrap(), Some(Tick(2)));
}

/// An event that a method call already pulled off the wire is still there afterwards.
#[test]
fn get_event_returns_events_that_a_call_already_received() {
    let (client, s) = client_serving(|req, _, host| host.reply(&empty_result_reply(req, 1)));
    let dev = ticker(&client, 1);
    let stream = dev.event_stream();
    s.host.emit(&tick(1, 7));
    // Give the event time to arrive, then let `call` drain it into the stream.
    std::thread::sleep(SHORT);
    dev.call::<()>("noop", &[], &[]).unwrap();
    let start = Instant::now();
    assert_eq!(stream.get_event(ms(3000)).unwrap(), Some(Tick(7)));
    assert!(start.elapsed() < Duration::from_secs(2), "had to wait");
}

#[test]
fn get_queued_event_never_reads_the_wire() {
    let (client, s) = client_silent();
    let stream = ticker(&client, 1).event_stream();
    s.host.emit(&tick(1, 1));
    std::thread::sleep(SHORT);
    assert_eq!(stream.get_queued_event(), None);
    assert_eq!(stream.get_event(ms(2000)).unwrap(), Some(Tick(1)));
    assert_eq!(stream.get_queued_event(), None);
}

#[test]
fn clear_discards_queued_events() {
    let (client, s) = client_serving(|req, _, host| host.reply(&empty_result_reply(req, 1)));
    let dev = ticker(&client, 1);
    let stream = dev.event_stream();
    s.host.emit(&tick(1, 1));
    std::thread::sleep(SHORT);
    dev.call::<()>("noop", &[], &[]).unwrap();
    stream.clear();
    assert_eq!(stream.get_event(ms(50)).unwrap(), None);
}

#[test]
fn only_events_for_the_streams_device_are_delivered() {
    let (client, s) = client_silent();
    let a = ticker(&client, 1).event_stream();
    let b = ticker(&client, 2).event_stream();
    s.host.emit(&tick(2, 20));
    s.host.emit(&tick(1, 10));
    assert_eq!(a.get_event(ms(2000)).unwrap(), Some(Tick(10)));
    assert_eq!(a.get_event(ms(50)).unwrap(), None);
    assert_eq!(b.get_queued_event(), Some(Tick(20)));
}

#[test]
fn two_streams_on_one_device_each_get_every_event() {
    let (client, s) = client_silent();
    let dev = ticker(&client, 1);
    let (a, b) = (dev.event_stream(), dev.event_stream());
    s.host.emit(&tick(1, 5));
    assert_eq!(a.get_event(ms(2000)).unwrap(), Some(Tick(5)));
    assert_eq!(b.get_queued_event(), Some(Tick(5)));
}

#[test]
fn clones_share_one_queue() {
    let (client, s) = client_silent();
    let a = ticker(&client, 1).event_stream();
    let b = a.clone();
    s.host.emit(&tick(1, 1));
    s.host.emit(&tick(1, 2));
    assert_eq!(a.get_event(ms(2000)).unwrap(), Some(Tick(1)));
    assert_eq!(b.get_event(ms(2000)).unwrap(), Some(Tick(2)));
}

#[test]
fn events_that_fail_to_parse_are_skipped_and_do_not_poison_the_stream() {
    let (client, s) = client_silent();
    let stream = ticker(&client, 1).event_stream();
    s.host.emit(&device_event(1, "bogus", id(1), json!(null)));
    s.host.emit(&device_event(
        1,
        "tick",
        id(1),
        json!({"n": "not a number"}),
    ));
    s.host.emit(&device_event(1, "tick", id(1), json!(null)));
    s.host.emit(&tick(1, 9));
    assert_eq!(stream.get_event(ms(2000)).unwrap(), Some(Tick(9)));
    assert_eq!(stream.get_event(ms(50)).unwrap(), None);
}

#[test]
fn non_device_events_do_not_reach_streams_but_are_processed() {
    let (client, s) = client_silent();
    let stream = ticker(&client, 1).event_stream();
    s.host.emit(&devices_changed(2));
    s.host.emit(&events_dropped(2, 3));
    s.host.emit(&tick(1, 4));
    assert_eq!(stream.get_event(ms(2000)).unwrap(), Some(Tick(4)));
}

#[test]
fn generic_devices_get_the_raw_event() {
    let (client, s) = client_silent();
    let dev = GenericDevice::new(client.clone(), id(3)).unwrap();
    let stream = dev.event_stream();
    s.host
        .emit(&device_event(1, "whatever", id(3), json!({"a": [1, 2]})));
    s.host.emit(&device_event(1, "nodata", id(3), Value::Null));
    assert_eq!(
        stream.get_event(ms(2000)).unwrap(),
        Some(("whatever".to_string(), json!({"a": [1, 2]})))
    );
    assert_eq!(
        stream.get_event(ms(2000)).unwrap(),
        Some(("nodata".to_string(), Value::Null))
    );
}

#[test]
fn devices_without_event_support_never_yield_events() {
    let (client, s) = client_silent();
    let stream = Redstone::new(client.clone(), id(1)).unwrap().event_stream();
    s.host.emit(&device_event(1, "tick", id(1), json!(null)));
    assert_eq!(stream.get_event(ms(100)).unwrap(), None);
}

#[test]
fn events_with_a_new_generation_still_invalidate_the_device_list() {
    let (client, s) = client_serving(host_for_list);
    client.list_devices().unwrap();
    let stream = ticker(&client, 1).event_stream();
    s.host
        .emit(&device_event(2, "tick", id(1), json!({"n": 1})));
    assert_eq!(stream.get_event(ms(2000)).unwrap(), Some(Tick(1)));
    client.list_devices().unwrap();
    assert_eq!(s.count("list"), 2);
}

fn host_for_list(req: &Value, _: &[u8], host: &Host) {
    host.reply(&list_reply(req, 1, &[(id(1), &["ticker"])]));
}

// ---------------------------------------------------------------------------------------------
// Dropping streams
// ---------------------------------------------------------------------------------------------

/// Once the last handle of a stream is gone nobody reads its events, so the client must stop
/// parsing and queueing them.
#[test]
fn dropping_a_stream_unregisters_its_listener() {
    let (client, s) = client_silent();
    let dev = Counted::new(client.clone(), id(1)).unwrap();
    let keep = dev.event_stream();
    let dropped = dev.event_stream();
    let before = COUNTED_PARSES.load(Ordering::SeqCst);
    s.host.emit(&device_event(1, "e", id(1), json!(null)));
    keep.get_event(ms(2000)).unwrap().unwrap();
    assert_eq!(COUNTED_PARSES.load(Ordering::SeqCst) - before, 2);

    drop(dropped);
    let before = COUNTED_PARSES.load(Ordering::SeqCst);
    s.host.emit(&device_event(1, "e", id(1), json!(null)));
    keep.get_event(ms(2000)).unwrap().unwrap();
    assert_eq!(
        COUNTED_PARSES.load(Ordering::SeqCst) - before,
        1,
        "the dropped stream's listener is still parsing events"
    );
}

#[test]
fn a_stream_with_live_clones_stays_registered() {
    let (client, s) = client_silent();
    let dev = ticker(&client, 1);
    let a = dev.event_stream();
    let b = a.clone();
    drop(a);
    s.host.emit(&tick(1, 3));
    assert_eq!(b.get_event(ms(2000)).unwrap(), Some(Tick(3)));
}

#[test]
fn dropping_every_stream_leaves_the_client_usable() {
    let (client, s) = client_serving(|req, _, host| host.reply(&empty_result_reply(req, 1)));
    let dev = ticker(&client, 1);
    drop(dev.event_stream());
    s.host.emit(&tick(1, 1));
    dev.call::<()>("noop", &[], &[]).unwrap();
    // A new stream starts empty
    assert_eq!(dev.event_stream().get_event(ms(30)).unwrap(), None);
}

#[test]
fn a_stream_created_later_does_not_see_earlier_events() {
    let (client, s) = client_serving(|req, _, host| host.reply(&empty_result_reply(req, 1)));
    let dev = ticker(&client, 1);
    s.host.emit(&tick(1, 1));
    std::thread::sleep(SHORT);
    dev.call::<()>("noop", &[], &[]).unwrap();
    assert_eq!(dev.event_stream().get_event(ms(30)).unwrap(), None);
}

// ---------------------------------------------------------------------------------------------
// EventStreamGroup
// ---------------------------------------------------------------------------------------------

#[test]
fn a_group_wakes_for_any_of_its_streams() {
    let (client, s) = client_silent();
    let a = ticker(&client, 1).event_stream();
    let b = ticker(&client, 2).event_stream();
    let mut group = EventStreamGroup::new(&client);
    group.add(&a);
    group.add(&b);
    s.host.emit(&tick(2, 1));
    assert!(group.wait_for_event(ms(2000)).unwrap());
    assert_eq!(a.get_queued_event(), None);
    assert_eq!(b.get_queued_event(), Some(Tick(1)));
}

#[test]
fn a_group_times_out_with_false() {
    let (client, _s) = client_silent();
    let a = ticker(&client, 1).event_stream();
    let mut group = EventStreamGroup::new(&client);
    group.add(&a);
    assert!(!group.wait_for_event(ms(60)).unwrap());
}

#[test]
fn a_group_can_mix_event_types() {
    let (client, s) = client_silent();
    let a = ticker(&client, 1).event_stream();
    let b = GenericDevice::new(client.clone(), id(2))
        .unwrap()
        .event_stream();
    let mut group = EventStreamGroup::new(&client);
    group.add(&a);
    group.add(&b);
    s.host.emit(&device_event(1, "x", id(2), json!(1)));
    assert!(group.wait_for_event(ms(2000)).unwrap());
    assert!(b.get_queued_event().is_some());
}

#[test]
fn events_for_streams_that_are_not_in_the_group_do_not_wake_it() {
    let (client, s) = client_silent();
    let a = ticker(&client, 1).event_stream();
    let b = ticker(&client, 2).event_stream();
    let mut group = EventStreamGroup::new(&client);
    group.add(&a);
    s.host.emit(&tick(2, 1));
    assert!(!group.wait_for_event(ms(80)).unwrap());
    assert_eq!(b.get_queued_event(), Some(Tick(1)));
}

#[test]
fn removed_streams_no_longer_wake_the_group() {
    let (client, s) = client_silent();
    let a = ticker(&client, 1).event_stream();
    let mut group = EventStreamGroup::new(&client);
    group.add(&a);
    group.remove(&a);
    s.host.emit(&tick(1, 1));
    assert!(!group.wait_for_event(ms(80)).unwrap());
}

#[test]
fn a_group_reports_events_that_are_already_queued() {
    let (client, s) = client_serving(|req, _, host| host.reply(&empty_result_reply(req, 1)));
    let dev = ticker(&client, 1);
    let a = dev.event_stream();
    let mut group = EventStreamGroup::new(&client);
    group.add(&a);
    s.host.emit(&tick(1, 1));
    std::thread::sleep(SHORT);
    dev.call::<()>("noop", &[], &[]).unwrap();
    let start = Instant::now();
    assert!(group.wait_for_event(ms(3000)).unwrap());
    assert!(start.elapsed() < Duration::from_secs(2), "had to wait");
}

#[test]
fn group_zero_timeout_polls_and_no_timeout_waits() {
    let (client, s) = client_silent();
    let a = ticker(&client, 1).event_stream();
    let mut group = EventStreamGroup::new(&client);
    group.add(&a);
    assert!(!within(move || group.wait_for_event(Some(Duration::ZERO))).unwrap());

    let mut group = EventStreamGroup::new(&client);
    group.add(&a);
    let waiter = std::thread::spawn(move || group.wait_for_event(None));
    std::thread::sleep(SHORT);
    assert!(!waiter.is_finished());
    s.host.emit(&tick(1, 2));
    assert!(waiter.join().unwrap().unwrap());
}

#[test]
fn stream_type_is_nameable() {
    let (client, _s) = client_silent();
    let _: EventStream<Tick> = ticker(&client, 1).event_stream();
}

/// Many clones of one stream dropped at the same moment: the listener must still be
/// unregistered exactly when the last one goes.
#[test]
fn concurrently_dropping_clones_still_unregisters_the_listener() {
    use std::sync::{Arc, Barrier};
    let (client, s) = client_silent();
    let dev = Counted::new(client.clone(), id(1)).unwrap();
    let keep = dev.event_stream();
    for _ in 0..200 {
        let stream = dev.event_stream();
        let clones: Vec<_> = (0..4).map(|_| stream.clone()).collect();
        drop(stream);
        let barrier = Arc::new(Barrier::new(clones.len()));
        let threads: Vec<_> = clones
            .into_iter()
            .map(|c| {
                let b = barrier.clone();
                std::thread::spawn(move || {
                    b.wait();
                    drop(c);
                })
            })
            .collect();
        for t in threads {
            t.join().unwrap();
        }
    }
    let before = COUNTED_PARSES.load(Ordering::SeqCst);
    s.host.emit(&device_event(1, "e", id(1), json!(null)));
    keep.get_event(ms(2000)).unwrap().unwrap();
    assert_eq!(
        COUNTED_PARSES.load(Ordering::SeqCst) - before,
        1,
        "listeners leaked by racing drops"
    );
}