sim-lib-stream-host 0.1.2

Host-device stream backend substrate for SIM streams.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

use sim_kernel::{Expr, Symbol};
use sim_lib_stream_core::{StreamPacket, TransportProfile};

use crate::{FakeBackend, HostBackendRegistry, HostCallbackCassette, HostCallbackReplayReport};

use super::support::authorized_cx;

#[test]
fn host_callback_cassette_replays_callback_timeline() {
    let mut registry = HostBackendRegistry::new();
    registry.register(FakeBackend::new()).unwrap();
    let mut cx = authorized_cx();
    let opened = registry
        .open_checked(&mut cx, FakeBackend::data_request(4).unwrap())
        .unwrap();
    let mut cassette = HostCallbackCassette::new();
    cassette.record_packet(StreamPacket::data(
        Symbol::qualified("stream/data", "expr"),
        Expr::String("first".to_owned()),
    ));
    cassette.record_packet(StreamPacket::data(
        Symbol::qualified("stream/data", "expr"),
        Expr::String("second".to_owned()),
    ));

    let shared = cassette
        .to_stream_cassette(
            opened.config().metadata(),
            TransportProfile::remote_stream_fabric(),
        )
        .unwrap();
    let cassette = HostCallbackCassette::from_stream_cassette(&shared).unwrap();

    let report = cassette.replay(opened.queue()).unwrap();

    let replayed = opened.queue().drain(4).unwrap();
    assert_eq!(
        report,
        HostCallbackReplayReport {
            accepted: 2,
            ..HostCallbackReplayReport::default()
        }
    );
    assert_eq!(replayed.len(), 2);
    assert_eq!(
        replayed[0].packet(),
        &StreamPacket::data(
            Symbol::qualified("stream/data", "expr"),
            Expr::String("first".to_owned()),
        )
    );
}

#[test]
fn host_callback_cassette_replay_reports_dropped_newest() {
    let mut registry = HostBackendRegistry::new();
    registry.register(FakeBackend::new()).unwrap();
    let mut cx = authorized_cx();
    let opened = registry
        .open_checked(&mut cx, FakeBackend::data_request(1).unwrap())
        .unwrap();
    let mut cassette = HostCallbackCassette::new();
    cassette.record_packet(StreamPacket::data(
        Symbol::qualified("stream/data", "expr"),
        Expr::String("first".to_owned()),
    ));
    cassette.record_packet(StreamPacket::data(
        Symbol::qualified("stream/data", "expr"),
        Expr::String("second".to_owned()),
    ));

    let report = cassette.replay(opened.queue()).unwrap();

    assert_eq!(
        report,
        HostCallbackReplayReport {
            accepted: 1,
            dropped_newest: 1,
            ..HostCallbackReplayReport::default()
        }
    );
    assert_eq!(opened.queue().drain(4).unwrap().len(), 1);
}

#[test]
fn host_callback_cassette_replay_reports_closed_queue() {
    let mut registry = HostBackendRegistry::new();
    registry.register(FakeBackend::new()).unwrap();
    let mut cx = authorized_cx();
    let opened = registry
        .open_checked(&mut cx, FakeBackend::data_request(2).unwrap())
        .unwrap();
    let mut cassette = HostCallbackCassette::new();
    cassette.record_packet(StreamPacket::data(
        Symbol::qualified("stream/data", "expr"),
        Expr::String("first".to_owned()),
    ));
    cassette.record_packet(StreamPacket::data(
        Symbol::qualified("stream/data", "expr"),
        Expr::String("second".to_owned()),
    ));

    opened.close().unwrap();
    let report = cassette.replay(opened.queue()).unwrap();

    assert_eq!(
        report,
        HostCallbackReplayReport {
            closed: 2,
            ..HostCallbackReplayReport::default()
        }
    );
}