Function make_sse_stream

src/http.rs:594–621

Create a stream of real-time changelog events matching `filter`, optionally also including historical events occuring after `synced-tip`

(
    filter: ChangelogFilter,
    listeners: Listeners,
    query: &Query,
)

// Create a stream of real-time changelog events matching `filter`, optionally also including
// historical events occuring after `synced-tip`
fn make_sse_stream(
    filter: ChangelogFilter,
    listeners: Listeners,
    query: &Query,
) -> Result<impl Stream<Item = Result<Event, warp::Error>>, Error> {
    debug!("subscribing sse client with {:?}", filter);

    let (tx, rx) = tmpsc::unbounded_channel();
    let rx = UnboundedReceiverStream::new(rx);
    listeners.lock().unwrap().push(Listener {
        tx,
        filter: filter.clone(),
    });

    // fetch historical changelog since the requested start point (if requesed)
    let changelog = match &filter.synced_tip {
        Some(synced_tip) => query.get_changelog_after(synced_tip)?,
        None => vec![],
    }
    .into_iter()
    .filter(move |change| filter.matches(change));
    // TODO don't produce unwanted events to begin with instead of filtering them

    Ok(tokio_stream::iter(changelog)
        .chain(rx)
        .map(make_sse_msg)
        .map(Ok))
}

fn make_sse_msg(change: IndexChange) -> Event {
    match &change {