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
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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

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 futures::Stream;
use nstreams_core::event::DeliveredEvent;
use nstreams_core::filter::SubscribeRequest;
use nstreams_core::namespace::{Namespace, StreamNamespace};
use nstreams_core::Error;
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,
    /// Bloom filter value (`x-stream-filter-value`) attached to the read stream message.
    #[serde(default)]
    pub filter_value: Option<String>,
}

#[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_get).post(subscribe_post))
        .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_get(
    State(nstreams): State<NStreams>,
    Query(query): Query<SubscribeQuery>,
) -> Result<Sse<impl Stream<Item = Result<Event, std::convert::Infallible>>>, StatusCode> {
    let request = SubscribeRequest {
        namespace: query.namespace,
        history: query.history,
        filter: None,
    };
    subscribe_sse(nstreams, request).await
}

/// `POST /subscribe` — JSON subscribe request with optional RabbitMQ stream filters.
///
/// Request body: [`SubscribeRequest`]. Returns an SSE stream of `registered` and
/// `message` events. See the README *Stream filtering* section for details.
pub async fn subscribe_post(
    State(nstreams): State<NStreams>,
    Json(request): Json<SubscribeRequest>,
) -> Result<Sse<impl Stream<Item = Result<Event, std::convert::Infallible>>>, StatusCode> {
    subscribe_sse(nstreams, request).await
}

async fn subscribe_sse(
    nstreams: NStreams,
    request: SubscribeRequest,
) -> Result<Sse<impl Stream<Item = Result<Event, std::convert::Infallible>>>, StatusCode> {
    StreamNamespace::validate(request.namespace.as_str()).map_err(|_| StatusCode::BAD_REQUEST)?;

    let mut subscription = nstreams
        .subscribe_request(request)
        .await
        .map_err(map_subscribe_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,
            body.filter_value.as_deref(),
        )
        .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))
}

fn map_subscribe_error(error: Error) -> StatusCode {
    match error {
        Error::InvalidNamespace(_, _) | Error::InvalidFilter(_) => StatusCode::BAD_REQUEST,
        _ => StatusCode::INTERNAL_SERVER_ERROR,
    }
}