cratestack_axum/rpc/
procedure_subscription.rs

1//! A `subscription procedure`'s input and items
2//! (`docs/design/subscriptions-websocket.md` §3.2, §3.3, slice S4).
3//!
4//! The input is the procedure's `Args` object, keyed by argument name, the
5//! same object a unary procedure takes over RPC. A binding that carries a
6//! JSON input (WebTransport, `POST /rpc/subscribe/{op_id}`) hands it over
7//! as is; `GET` carries each argument as a query parameter whose value is
8//! that argument's JSON (`?args={"roomId":"r1"}`), so an `EventSource` can
9//! open one. Both meet as the query string, the form the generated
10//! dispatcher's `open_query` takes.
11//!
12//! A handler's stream ends the subscription when it ends: unlike a model
13//! feed, it is not a backplane that can drop events, so a natural end is
14//! not an error. An `Err` item is the stream's terminal error.
15
16use cratestack_core::{CratestackError, SubscriptionStream};
17use futures_util::stream::{self, StreamExt};
18use serde::Serialize;
19use serde::de::DeserializeOwned;
20
21/// A subscription procedure's JSON input as the query string `open_query`
22/// takes: `null` or `{}` is none.
23pub fn procedure_input_query(input: serde_json::Value) -> Result<Option<String>, CratestackError> {
24    let object = match input {
25        serde_json::Value::Null => return Ok(None),
26        serde_json::Value::Object(object) => object,
27        _ => {
28            return Err(CratestackError::BadRequest(
29                "a subscription procedure's input is an object of its arguments".to_owned(),
30            ));
31        }
32    };
33    if object.is_empty() {
34        return Ok(None);
35    }
36    let mut query = url::form_urlencoded::Serializer::new(String::new());
37    for (name, value) in &object {
38        query.append_pair(name, &value.to_string());
39    }
40    Ok(Some(query.finish()))
41}
42
43/// The procedure's `Args` from the query string: each parameter is one
44/// argument, its value that argument's JSON.
45pub fn procedure_query_args<A: DeserializeOwned>(
46    query: Option<&str>,
47) -> Result<A, CratestackError> {
48    let mut object = serde_json::Map::new();
49    for (name, value) in crate::parse_query_pairs(query)? {
50        let value = serde_json::from_str(&value).map_err(|_| {
51            CratestackError::BadRequest(format!(
52                "argument `{name}` must be JSON, such as `{name}={{\"id\":1}}` or `{name}=\"text\"`"
53            ))
54        })?;
55        if object.insert(name.clone(), value).is_some() {
56            return Err(CratestackError::BadRequest(format!(
57                "argument `{name}` is given twice"
58            )));
59        }
60    }
61    serde_json::from_value(serde_json::Value::Object(object)).map_err(|error| {
62        CratestackError::BadRequest(format!("invalid subscription arguments: {error}"))
63    })
64}
65
66/// A handler's items encoded to JSON values, ending after the first error.
67pub fn procedure_feed<T>(items: SubscriptionStream<T>) -> SubscriptionStream<serde_json::Value>
68where
69    T: Serialize + Send + 'static,
70{
71    Box::pin(stream::unfold(Some(items), |items| async move {
72        let mut items = items?;
73        let item = match items.next().await? {
74            Ok(item) => serde_json::to_value(&item).map_err(|error| {
75                CratestackError::Codec(format!("failed to encode a subscription item: {error}"))
76            }),
77            Err(error) => Err(error),
78        };
79        let rest = item.is_ok().then_some(items);
80        Some((item, rest))
81    }))
82}
83
84#[cfg(test)]
85mod tests;
cratestack_axum/rpc/procedure_subscription.rs

cratestack_axum/rpc/
procedure_subscription.rs
