cratestack_client_rust/
subscription.rs

1//! Subscriptions (`docs/design/subscriptions-websocket.md` §8): one
2//! contract whatever the binding — the call resolves once the server says
3//! `Ready`, then the stream yields items and ends, or ends with an error.
4//!
5//! The binding is client configuration ([`SubscriptionTransport`]), never
6//! part of a generated method's name. `Auto` tries WebTransport when the
7//! `webtransport` feature is on and the config names the listener, then a
8//! WebSocket when the `ws` feature is on, then SSE (native targets for the
9//! first two). It moves on only when a binding cannot be opened; a refusal
10//! is never a reason to fall back.
11
12mod failure;
13mod frames;
14mod open;
15mod sse;
16#[cfg(all(feature = "ws", not(target_arch = "wasm32")))]
17mod ws;
18#[cfg(all(feature = "webtransport", not(target_arch = "wasm32")))]
19mod wt;
20
21pub use frames::rpc_error_to_client;
22
23/// A live subscription: items until the server ends it, or a terminal `Err`.
24/// `None` after an `Err`, or when the stream ends normally (a subscription
25/// procedure whose source ran out). Dropping it cancels the subscription.
26pub type ClientSubscription<T> = tokio::sync::mpsc::Receiver<Result<T, ClientError>>;
27
28use crate::error::ClientError;
29
30/// Which binding a subscription uses.
31#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
32pub enum SubscriptionTransport {
33    /// WebTransport, then a WebSocket, then SSE: each binding this build
34    /// has and the config allows, moving on only when one cannot be opened.
35    #[default]
36    Auto,
37    /// One WebTransport session per subscription (`webtransport` feature,
38    /// native targets, `ClientConfig::with_webtransport`).
39    WebTransport,
40    /// One WebSocket per subscription (`ws` feature, native targets).
41    WebSocket,
42    /// `text/event-stream` over HTTP.
43    Sse,
44}
45
46/// Where a subscription is served. The generated client fills it in; the
47/// binding decides which half it reads.
48#[derive(Debug, Clone)]
49pub struct SubscriptionTarget {
50    /// `model.<Model>.subscribe` or `procedure.<name>`.
51    pub op_id: String,
52    /// The input a socket's `Request` carries: a model feed's
53    /// `RpcSubscribeInput`, a subscription procedure's `Args`.
54    pub input: serde_json::Value,
55    /// The SSE route: `/rpc/subscribe/<op_id>` on `transport rpc`,
56    /// `/<plural>/$subscribe` or `/$procs/<name>` on REST.
57    pub sse_path: String,
58    /// The input as that route's query string.
59    pub sse_query: Option<String>,
60    /// The socket path: `/rpc/ws`, or `/$ws` on REST.
61    pub socket_path: &'static str,
62}
63
64impl SubscriptionTarget {
65    /// A subscription procedure's `Args` as its SSE query string: each
66    /// argument a parameter whose value is that argument's JSON.
67    pub fn procedure_query(args: &serde_json::Value) -> Option<String> {
68        let object = args.as_object().filter(|object| !object.is_empty())?;
69        let mut query = url::form_urlencoded::Serializer::new(String::new());
70        for (name, value) in object {
71            query.append_pair(name, &value.to_string());
72        }
73        Some(query.finish())
74    }
75}
cratestack_client_rust/subscription.rs

cratestack_client_rust/
subscription.rs
