nstreams 0.2.2

Embeddable versioned event stream handler for existing services
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

use futures::Stream;
use nstreams_core::event::DeliveredEvent;
use nstreams_core::Result;

/// Framework-neutral helpers for delivering versioned events to clients.
pub fn delivered_event_json(event: &DeliveredEvent) -> Result<serde_json::Value> {
    Ok(serde_json::json!({
        "request_id": event.request_id,
        "version": event.version,
        "payload": serde_json::from_slice::<serde_json::Value>(&event.payload)
            .map_err(|error| nstreams_core::Error::Serialization(error.to_string()))?,
    }))
}

pub fn registered_event_json(request_id: &str) -> serde_json::Value {
    serde_json::json!({ "request_id": request_id })
}

/// Convert a core subscription into a stream of delivered events.
pub fn subscription_stream(
    mut subscription: nstreams_core::Subscription,
) -> impl Stream<Item = Result<DeliveredEvent>> + Send {
    async_stream::stream! {
        while let Some(item) = subscription.next().await {
            yield item;
        }
    }
}