oino_client/
follow.rs

1//! Event streams that follow the runtime's current session.
2//!
3//! Each stream subscribes to the *session* hub, not the runtime hub: session events are
4//! queued synchronously when published, so once a command returns every event it caused is
5//! already in the stream (a frontend can drain it before writing the reply; ordering rule).
6//! When the backend replaces the session it hands every live stream a subscription to the
7//! new session plus the `session_replaced` event; a stream switches after draining the old
8//! session.
9
10use std::pin::Pin;
11use std::sync::{Arc, Mutex, MutexGuard, PoisonError, Weak};
12use std::task::{Context, Poll, Waker};
13
14use futures::Stream;
15use oino_core::events::ClientEvent;
16use oino_core::hub::EventStream;
17
18fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
19    mutex.lock().unwrap_or_else(PoisonError::into_inner)
20}
21
22#[derive(Default)]
23struct Slot {
24    /// Subscription to the replacement session and the event announcing it.
25    next: Option<(EventStream, ClientEvent)>,
26    /// The backend shut down: end after draining.
27    closed: bool,
28    waker: Option<Waker>,
29}
30
31/// Live streams of one backend.
32#[derive(Default)]
33pub(crate) struct Followers {
34    slots: Mutex<Vec<Weak<Mutex<Slot>>>>,
35}
36
37impl Followers {
38    /// A stream starting with `current`.
39    pub(crate) fn stream(&self, current: EventStream) -> FollowStream {
40        let slot = Arc::new(Mutex::new(Slot::default()));
41        lock(&self.slots).push(Arc::downgrade(&slot));
42        FollowStream {
43            current,
44            slot,
45            announce: None,
46        }
47    }
48
49    /// Ends every live stream once it is drained.
50    pub(crate) fn close(&self) {
51        for weak in lock(&self.slots).drain(..) {
52            if let Some(slot) = weak.upgrade() {
53                let mut slot = lock(&slot);
54                slot.closed = true;
55                if let Some(waker) = slot.waker.take() {
56                    waker.wake();
57                }
58            }
59        }
60    }
61
62    /// Moves every live stream to the next session; `subscribe` is called once per stream.
63    pub(crate) fn rebind(&self, subscribe: impl Fn() -> EventStream, announce: &ClientEvent) {
64        lock(&self.slots).retain(|weak| {
65            let Some(slot) = weak.upgrade() else {
66                return false;
67            };
68            let mut slot = lock(&slot);
69            slot.next = Some((subscribe(), announce.clone()));
70            if let Some(waker) = slot.waker.take() {
71                waker.wake();
72            }
73            true
74        });
75    }
76}
77
78/// A [`Stream`] of the current session's client events.
79pub struct FollowStream {
80    current: EventStream,
81    slot: Arc<Mutex<Slot>>,
82    announce: Option<ClientEvent>,
83}
84
85impl std::fmt::Debug for FollowStream {
86    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
87        f.debug_struct("FollowStream").finish_non_exhaustive()
88    }
89}
90
91impl Stream for FollowStream {
92    type Item = ClientEvent;
93
94    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<ClientEvent>> {
95        loop {
96            if let Some(event) = self.announce.take() {
97                return Poll::Ready(Some(event));
98            }
99            if let Poll::Ready(Some(event)) = Pin::new(&mut self.current).poll_next(cx) {
100                return Poll::Ready(Some(event));
101            }
102            // The old session is drained (or closed): switch if a replacement is waiting.
103            let next = {
104                let mut slot = lock(&self.slot);
105                let next = slot.next.take();
106                if next.is_none() && slot.closed {
107                    return Poll::Ready(None);
108                }
109                if next.is_none() {
110                    slot.waker = Some(cx.waker().clone());
111                }
112                next
113            };
114            match next {
115                Some((stream, announce)) => {
116                    self.current = stream;
117                    self.announce = Some(announce);
118                }
119                None => return Poll::Pending,
120            }
121        }
122    }
123}
oino_client/follow.rs

oino_client/
follow.rs
