nstreams 0.1.0

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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

use axum::extract::{Query, State};
use axum::http::StatusCode;
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::routing::{get, post};
use axum::{Json, Router};
use nstreams_core::event::DeliveredEvent;
use nstreams_core::namespace::{Namespace, StreamNamespace};
use serde::Deserialize;

use crate::service::NStreams;
use crate::transport::{delivered_event_json, registered_event_json};

#[derive(Debug, Deserialize)]
pub struct PublishRequest {
    pub namespace: String,
    pub payload: serde_json::Value,
}

#[derive(Debug, Deserialize)]
pub struct SubscribeQuery {
    pub namespace: String,
    #[serde(default = "default_history")]
    pub history: u64,
}

fn default_history() -> u64 {
    1_000
}

/// Route definitions without state. Mount with `.nest(prefix, routes().with_state(nstreams))`.
pub fn routes() -> Router<NStreams> {
    Router::new()
        .route("/subscribe", get(subscribe))
        .route("/publish", post(publish))
}

/// Convenience helper when `NStreams` is the only application state.
pub fn router(nstreams: NStreams) -> Router<NStreams> {
    routes().with_state(nstreams)
}

/// `GET /subscribe?namespace=...&history=N` — SSE stream of versioned events.
pub async fn subscribe(
    State(nstreams): State<NStreams>,
    Query(query): Query<SubscribeQuery>,
) -> Result<Sse<impl futures::Stream<Item = Result<Event, std::convert::Infallible>>>, StatusCode> {
    let namespace = StreamNamespace::new(query.namespace);
    StreamNamespace::validate(namespace.as_str()).map_err(|_| StatusCode::BAD_REQUEST)?;

    let mut subscription = nstreams
        .subscribe(&namespace, query.history)
        .await
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    let request_id = subscription.request_id().to_string();
    let stream = async_stream::stream! {
        yield Ok(Event::default()
            .event("registered")
            .data(registered_event_json(&request_id).to_string()));

        while let Some(item) = subscription.next().await {
            match item {
                Ok(event) => match encode_message_event(&event) {
                    Ok(event) => yield Ok(event),
                    Err(error) => yield Ok(Event::default().event("error").data(error.to_string())),
                },
                Err(error) => yield Ok(Event::default().event("error").data(error.to_string())),
            }
        }
    };

    Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
}

/// `POST /publish` — enqueue an event on the write path.
pub async fn publish(
    State(nstreams): State<NStreams>,
    Json(body): Json<PublishRequest>,
) -> Result<StatusCode, StatusCode> {
    let namespace = StreamNamespace::new(body.namespace);
    StreamNamespace::validate(namespace.as_str()).map_err(|_| StatusCode::BAD_REQUEST)?;

    let payload = serde_json::to_vec(&body.payload).map_err(|_| StatusCode::BAD_REQUEST)?;
    nstreams
        .publish(&namespace, &payload)
        .await
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;

    Ok(StatusCode::ACCEPTED)
}

fn encode_message_event(event: &DeliveredEvent) -> Result<Event, nstreams_core::Error> {
    let data = delivered_event_json(event)?.to_string();
    Ok(Event::default().event("message").data(data))
}