fn stream_to_client(
    initial_events: mpsc::UnboundedReceiver<ServerSentEvent>,
    ongoing_events: broadcast::Receiver<BroadcastChannelMessage>,
    event_filter: &'static [EventFilter],
    deploy_getter: DeployGetter,
) -> impl Stream<Item = Result<WarpServerSentEvent, RecvError>> + 'static {
    // Keep a record of the IDs of the events delivered via the `initial_events` receiver.
    let initial_stream_ids = Arc::new(RwLock::new(HashSet::new()));
    let cloned_initial_ids = Arc::clone(&initial_stream_ids);

    // Map the events arriving after the initial stream to the correct error type, filtering out any
    // that have already been sent in the initial stream.
    let ongoing_stream = BroadcastStream::new(ongoing_events)
        .filter_map(move |result| {
            let cloned_initial_ids = Arc::clone(&cloned_initial_ids);
            async move {
                match result {
                    Ok(BroadcastChannelMessage::ServerSentEvent(event)) => {
                        if let Some(id) = event.id {
                            if cloned_initial_ids.read().unwrap().contains(&id) {
                                debug!(event_id=%id, "skipped duplicate event");
                                return None;
                            }
                        }
                        Some(Ok(event))
                    }
                    Ok(BroadcastChannelMessage::Shutdown) => Some(Err(RecvError::Closed)),
                    Err(BroadcastStreamRecvError::Lagged(amount)) => {
                        info!(
                            "client lagged by {} events - dropping event stream connection to client",
                            amount
                        );
                        Some(Err(RecvError::Lagged(amount)))
                    }
                }
            }
        })
        .take_while(|result| future::ready(!matches!(result, Err(RecvError::Closed))));

    // Serve the initial events followed by the ongoing ones, filtering as dictated by the
    // `event_filter`.
    UnboundedReceiverStream::new(initial_events)
        .map(move |event| {
            if let Some(id) = event.id {
                let _ = initial_stream_ids.write().unwrap().insert(id);
            }
            Ok(event)
        })
        .chain(ongoing_stream)
        .filter_map(move |result| {
            let cloned_deploy_getter = deploy_getter.clone();
            async move {
                match result {
                    Ok(event) => {
                        filter_map_server_sent_event(&event, event_filter, cloned_deploy_getter)
                            .await
                    }
                    Err(error) => Some(Err(error)),
                }
            }
        })
}
/// This takes the two channel receivers and turns them into a stream of SSEs to the subscribed
/// client.
///
/// The initial events receiver (an mpsc receiver) is exhausted first, and contains an initial
/// `ApiVersion` message, followed by any historical events the client requested using the query
/// string.
///
/// The ongoing events channel (a broadcast receiver) is then consumed, and will remain in use until
/// either the client disconnects, or the server shuts down (indicated by sending a `Shutdown`
/// variant via the channel).  This channel will receive all SSEs created from the moment the client
/// subscribed to the server's event stream.
///
/// It also takes an `EventFilter` which causes events to which the client didn't subscribe to be
/// skipped.
This takes the two channel receivers and turns them into a stream of SSEs to the subscribed
client.

The initial events receiver (an mpsc receiver) is exhausted first, and contains an initial
`ApiVersion` message, followed by any historical events the client requested using the query
string.

The ongoing events channel (a broadcast receiver) is then consumed, and will remain in use until
either the client disconnects, or the server shuts down (indicated by sending a `Shutdown`
variant via the channel).  This channel will receive all SSEs created from the moment the client
subscribed to the server's event stream.

It also takes an `EventFilter` which causes events to which the client didn't subscribe to be
skipped.
This takes the two channel receivers and turns them into a stream of SSEs to the subscribed client . The initial events receiver ( an mpsc receiver ) is exhausted first and contains an initial ` ApiVersion ` message followed by any historical events the client requested using the query string . The ongoing events channel ( a broadcast receiver ) is then consumed and will remain in use until either the client disconnects or the server shuts down ( indicated by sending a ` Shutdown ` variant via the channel ) . This channel will receive all SSEs created from the moment the client subscribed to the server ' s event stream . It also takes an ` EventFilter ` which causes events to which the client didn ' t subscribe to be skipped .
fn stream_to_client(
    initial_events: mpsc::UnboundedReceiver<ServerSentEvent>,
    ongoing_events: broadcast::Receiver<BroadcastChannelMessage>,
    event_filter: &'static [EventFilter],
    deploy_getter: DeployGetter,
) -> impl Stream<Item = Result<WarpServerSentEvent, RecvError>> + 'static {
    // Keep a record of the IDs of the events delivered via the `initial_events` receiver.
    let initial_stream_ids = Arc::new(RwLock::new(HashSet::new()));
    let cloned_initial_ids = Arc::clone(&initial_stream_ids);
    // Map the events arriving after the initial stream to the correct error type, filtering out any
    // that have already been sent in the initial stream.
    let ongoing_stream = BroadcastStream::new(ongoing_events)
        .filter_map(move |result| {
            let cloned_initial_ids = Arc::clone(&cloned_initial_ids);
            async move {
                match result {
                    Ok(BroadcastChannelMessage::ServerSentEvent(event)) => {
                        if let Some(id) = event.id {
                            if cloned_initial_ids.read().unwrap().contains(&id) {
                                debug!(event_id=%id, "skipped duplicate event");
                                return None;
                            }
                        }
                        Some(Ok(event))
                    }
                    Ok(BroadcastChannelMessage::Shutdown) => Some(Err(RecvError::Closed)),
                    Err(BroadcastStreamRecvError::Lagged(amount)) => {
                        info!(
                            "client lagged by {} events - dropping event stream connection to client",
                            amount
                        );
                        Some(Err(RecvError::Lagged(amount)))
                    }
                }
            }
        })
        .take_while(|result| future::ready(!matches!(result, Err(RecvError::Closed))));
    // Serve the initial events followed by the ongoing ones, filtering as dictated by the
    // `event_filter`.
    UnboundedReceiverStream::new(initial_events)
        .map(move |event| {
            if let Some(id) = event.id {
                let _ = initial_stream_ids.write().unwrap().insert(id);
            }
            Ok(event)
        })
        .chain(ongoing_stream)
        .filter_map(move |result| {
            let cloned_deploy_getter = deploy_getter.clone();
            async move {
                match result {
                    Ok(event) => {
                        filter_map_server_sent_event(&event, event_filter, cloned_deploy_getter)
                            .await
                    }
                    Err(error) => Some(Err(error)),
                }
            }
        })
}
fn stream_to_client ( initial_events : mpsc :: UnboundedReceiver < ServerSentEvent > , ongoing_events : broadcast :: Receiver < BroadcastChannelMessage > , event_filter : & ' static [ EventFilter ] , deploy_getter : DeployGetter , ) -> impl Stream < Item = Result < WarpServerSentEvent , RecvError > > + ' static { // Keep a record of the IDs of the events delivered via the `initial_events` receiver. let initial_stream_ids = Arc :: new ( RwLock :: new ( HashSet :: new ( ) ) ) ; let cloned_initial_ids = Arc :: clone ( & initial_stream_ids ) ; // Map the events arriving after the initial stream to the correct error type, filtering out any // that have already been sent in the initial stream. let ongoing_stream = BroadcastStream :: new ( ongoing_events ) . filter_map ( move | result | { let cloned_initial_ids = Arc :: clone ( & cloned_initial_ids ) ; async move { match result { Ok ( BroadcastChannelMessage :: ServerSentEvent ( event ) ) => { if let Some ( id ) = event . id { if cloned_initial_ids . read ( ) . unwrap ( ) . contains ( & id ) { debug ! ( event_id id " " ) ; return None ; } } Some ( Ok ( event ) ) } Ok ( BroadcastChannelMessage :: Shutdown ) => Some ( Err ( RecvError :: Closed ) ) , Err ( BroadcastStreamRecvError :: Lagged ( amount ) ) => { info ! ( " " amount ) ; Some ( Err ( RecvError :: Lagged ( amount ) ) ) } } } } ) . take_while ( | result | future :: ready ( ! matches ! ( result Err ( RecvError Closed ) ) ) ) ; // Serve the initial events followed by the ongoing ones, filtering as dictated by the // `event_filter`. UnboundedReceiverStream :: new ( initial_events ) . map ( move | event | { if let Some ( id ) = event . id { let _ = initial_stream_ids . write ( ) . unwrap ( ) . insert ( id ) ; } Ok ( event ) } ) . chain ( ongoing_stream ) . filter_map ( move | result | { let cloned_deploy_getter = deploy_getter . clone ( ) ; async move { match result { Ok ( event ) => { filter_map_server_sent_event ( & event , event_filter , cloned_deploy_getter ) . await } Err ( error ) => Some ( Err ( error ) ) , } } } ) }
This takes the two channel receivers and turns them into a stream of SSEs to the subscribed
client.
This takes the two channel receivers and turns them into a stream of SSEs to the subscribed client .
initial_events
mpsc::UnboundedReceiver<ServerSentEvent>
ongoing_events
broadcast::Receiver<BroadcastChannelMessage>
event_filter
&'static [EventFilter]
deploy_getter
DeployGetter
initial_events
mpsc::UnboundedReceiver<ServerSentEvent>
ongoing_events
broadcast::Receiver<BroadcastChannelMessage>
event_filter
&'static [EventFilter]
deploy_getter
DeployGetter