android/session/
events.rs

1//! The event stream: what the connection reports and, on an emulator with
2//! its controller, frames, in arrival order on one channel.
3
4use std::sync::mpsc::{Receiver, RecvTimeoutError};
5use std::time::Instant;
6
7use mirror::backend::{EventStream, StreamItem, Timed};
8use protocol::Error;
9use protocol::session::HostEvent;
10
11/// The stream item for a host session event, or `None` for events the
12/// mirror has no use for: an oversized frame that answered no request, and
13/// a failed resync, which the session retries on its next read (the gap
14/// that caused it has already reached the mirror).
15pub fn stream_item(event: HostEvent) -> Option<StreamItem> {
16    match event {
17        HostEvent::Event(event) => Some(StreamItem::Event(event)),
18        HostEvent::OutOfSync {
19            stream,
20            expected,
21            received,
22        } => Some(StreamItem::OutOfSync {
23            stream,
24            expected,
25            received,
26        }),
27        HostEvent::Resynced { stream, seq } => Some(StreamItem::Resynced { stream, seq }),
28        HostEvent::Closed(reason) => Some(StreamItem::Disconnected(
29            Error::disconnected(&reason).message,
30        )),
31        HostEvent::Ready(_)
32        | HostEvent::Response { .. }
33        | HostEvent::ResyncFailed { .. }
34        | HostEvent::FrameTooLarge(_) => None,
35    }
36}
37
38/// The mirror's end of the channel.
39pub(crate) struct AgentEvents {
40    pub(crate) items: Receiver<Timed<StreamItem>>,
41}
42
43impl EventStream for AgentEvents {
44    fn next(&mut self, deadline: Instant) -> Option<Timed<StreamItem>> {
45        let Some(wait) = deadline.checked_duration_since(Instant::now()) else {
46            return self.items.try_recv().ok();
47        };
48        match self.items.recv_timeout(wait) {
49            Ok(item) => Some(item),
50            Err(RecvTimeoutError::Timeout) => None,
51            Err(RecvTimeoutError::Disconnected) => {
52                // Nothing will arrive; waiting out the deadline keeps the
53                // mirror's loops from spinning.
54                if let Some(rest) = deadline.checked_duration_since(Instant::now()) {
55                    std::thread::sleep(rest);
56                }
57                None
58            }
59        }
60    }
61}
62
63#[cfg(test)]
64mod tests {
65    use super::*;
66    use protocol::code;
67    use protocol::event::{
68        Event, EventPayload, Lifecycle, LifecycleKind, Stream, TreeChange, TreeChangeKind,
69    };
70    use protocol::frame::Oversized;
71    use protocol::session::DisconnectReason;
72    use std::sync::mpsc;
73    use std::time::Duration;
74
75    fn tree_event(seq: u64) -> Event {
76        Event {
77            seq,
78            time_us: 5,
79            payload: EventPayload::Tree(TreeChange {
80                change: TreeChangeKind::TEXT,
81                window: 3,
82                path: vec![0, 1],
83            }),
84        }
85    }
86
87    #[test]
88    fn protocol_events_and_sequence_notices_pass_through() {
89        assert_eq!(
90            stream_item(HostEvent::Event(tree_event(4))),
91            Some(StreamItem::Event(tree_event(4)))
92        );
93        assert_eq!(
94            stream_item(HostEvent::OutOfSync {
95                stream: Stream::TREE,
96                expected: 4,
97                received: 6
98            }),
99            Some(StreamItem::OutOfSync {
100                stream: Stream::TREE,
101                expected: 4,
102                received: 6
103            })
104        );
105        assert_eq!(
106            stream_item(HostEvent::Resynced {
107                stream: Stream::TREE,
108                seq: 9
109            }),
110            Some(StreamItem::Resynced {
111                stream: Stream::TREE,
112                seq: 9
113            })
114        );
115        let warning = Event {
116            seq: 1,
117            time_us: 0,
118            payload: EventPayload::Lifecycle(Lifecycle {
119                change: LifecycleKind::WARNING,
120                message: "input injection is unavailable".into(),
121            }),
122        };
123        assert_eq!(
124            stream_item(HostEvent::Event(warning.clone())),
125            Some(StreamItem::Event(warning))
126        );
127    }
128
129    #[test]
130    fn the_close_is_a_disconnection_with_a_reconnect_hint() {
131        let Some(StreamItem::Disconnected(message)) =
132            stream_item(HostEvent::Closed(DisconnectReason::Eof))
133        else {
134            panic!("the close is no disconnection");
135        };
136        assert!(message.contains("closed the connection"), "{message}");
137        assert!(message.contains("reconnect"), "{message}");
138        let Some(StreamItem::Disconnected(silent)) = stream_item(HostEvent::Closed(
139            DisconnectReason::Silent(Duration::from_secs(3)),
140        )) else {
141            panic!("silence is no disconnection");
142        };
143        assert!(silent.contains("3.0 s"), "{silent}");
144    }
145
146    #[test]
147    fn stray_oversized_frames_and_failed_resyncs_are_not_items() {
148        let oversized = Oversized {
149            size: 20 << 20,
150            limit: 16 << 20,
151            kind: None,
152            id: None,
153            method: None,
154        };
155        assert_eq!(stream_item(HostEvent::FrameTooLarge(oversized)), None);
156        let failed = HostEvent::ResyncFailed {
157            stream: Stream::TREE,
158            error: protocol::Error::new(code::INTERNAL, "no"),
159        };
160        assert_eq!(stream_item(failed), None);
161    }
162
163    #[test]
164    fn next_returns_what_is_queued_and_waits_no_longer_than_the_deadline() {
165        let (tx, rx) = mpsc::channel();
166        let mut events = AgentEvents { items: rx };
167        let past = Instant::now() - Duration::from_millis(10);
168        assert_eq!(events.next(past), None, "nothing queued");
169        let item = Timed {
170            at: Instant::now(),
171            item: StreamItem::Event(tree_event(1)),
172        };
173        tx.send(item.clone()).unwrap();
174        assert_eq!(events.next(past), Some(item), "queued items need no wait");
175        let started = Instant::now();
176        assert_eq!(events.next(started + Duration::from_millis(80)), None);
177        let waited = started.elapsed();
178        assert!(
179            waited >= Duration::from_millis(75) && waited < Duration::from_millis(500),
180            "waited {waited:?}"
181        );
182        let later = Timed {
183            at: Instant::now(),
184            item: StreamItem::Reconnected,
185        };
186        let sender = tx.clone();
187        let sent = later.clone();
188        std::thread::spawn(move || {
189            std::thread::sleep(Duration::from_millis(30));
190            sender.send(sent).unwrap();
191        });
192        let started = Instant::now();
193        assert_eq!(events.next(started + Duration::from_secs(5)), Some(later));
194        assert!(started.elapsed() < Duration::from_secs(1));
195    }
196
197    #[test]
198    fn a_closed_channel_ends_the_wait() {
199        let (tx, rx) = mpsc::channel::<Timed<StreamItem>>();
200        drop(tx);
201        let mut events = AgentEvents { items: rx };
202        let started = Instant::now();
203        assert_eq!(events.next(started + Duration::from_millis(200)), None);
204    }
205}
android/session/events.rs

android/session/
events.rs
