cdp_lite/
event_filter.rs

1use crate::error::CdpResult;
2use crate::protocol::WsResponse;
3use std::pin::Pin;
4use std::task::{Context, Poll};
5use tokio_stream::{Stream, wrappers::BroadcastStream};
6
7pub struct EventFilter {
8    inner: BroadcastStream<WsResponse>,
9    prefix: String,
10}
11
12impl EventFilter {
13    pub fn new(
14        receiver: tokio::sync::broadcast::Receiver<WsResponse>,
15        domain: &'static str,
16    ) -> Self {
17        Self {
18            inner: BroadcastStream::new(receiver),
19            prefix: format!("{}.", domain),
20        }
21    }
22}
23
24impl Stream for EventFilter {
25    type Item = CdpResult<WsResponse>;
26
27    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
28        loop {
29            match Pin::new(&mut self.inner).poll_next(cx) {
30                Poll::Ready(Some(Ok(response))) => {
31                    if let Some(ref method) = response.method
32                        && method.starts_with(&self.prefix)
33                    {
34                        return Poll::Ready(Some(Ok(response)));
35                    }
36                    // Not the domain we want, loop again to poll next
37                    continue;
38                }
39                Poll::Ready(Some(Err(e))) => {
40                    // This 'Err' is specifically a BroadcastStreamRecvError::Lagged
41                    return Poll::Ready(Some(Err(crate::error::CdpError::InternalError(format!(
42                        "Event stream lagged: {}",
43                        e
44                    )))));
45                }
46                Poll::Ready(None) => return Poll::Ready(None), // Channel closed
47                Poll::Pending => return Poll::Pending,
48            }
49        }
50    }
51}
cdp_lite/event_filter.rs

cdp_lite/
event_filter.rs
