wingfoil 8.0.0

graph based stream processing framework
Documentation 
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

//! `web_sub` — receive values from connected WebSocket clients.
//!
//! Each call to [`web_sub`] registers a bounded mpsc listener on the
//! [`WebServer`] for a topic. When a client sends an envelope on that
//! topic, its payload is decoded with the server's codec and yielded to
//! the graph as a `Burst<T>`.
//!
//! The listener mpsc is bounded and the server uses `try_send` with a
//! drop-newest policy so a misbehaving browser cannot back-pressure the
//! graph.

use std::rc::Rc;

use serde::de::DeserializeOwned;
use tokio::sync::mpsc;

use super::server::{SUBSCRIBE_MPSC_CAPACITY, WebServer};
use crate::nodes::{RunParams, produce_async};
use crate::types::*;

/// Subscribe to frames that clients send on `topic`.
///
/// Returns a source [`Stream`] of `Burst<T>`, decoded with the server's
/// configured codec (bincode or JSON). Decoding errors surface as graph
/// errors.
///
/// The stream never ticks in historical mode — whether the server is a
/// [`start_historical`](super::WebServerBuilder::start_historical) no-op
/// *or* a live server driving a graph in
/// [`RunMode::HistoricalFrom`]. Live browser input has no place in a
/// deterministic replay, and an open listener would block the historical
/// run waiting for frames that never come; so `web_sub` yields an empty,
/// immediately-ending stream. A historical `web_pub` still streams its
/// output to the browser as usual.
#[must_use]
pub fn web_sub<T: Element + Send + DeserializeOwned>(
    server: &WebServer,
    topic: impl Into<String>,
) -> Rc<dyn Stream<Burst<T>>> {
    let topic = topic.into();
    let codec = server.codec();
    let historical = server.is_historical_noop();

    // Register the mpsc listener at construction time so frames arriving
    // before the graph starts are buffered up to the mpsc capacity rather
    // than dropped. Skipped for a no-op server, which has no live socket.
    let rx_opt = if historical {
        None
    } else {
        let (tx, rx) = mpsc::channel(SUBSCRIBE_MPSC_CAPACITY);
        server.inner.register_sub_sender(&topic, tx);
        Some(rx)
    };

    produce_async(
        move |ctx: RunParams| async move {
            Ok(async_stream::stream! {
                // A historical run replays deterministically from its sources;
                // an open client listener would never yield and would block the
                // run from completing, so treat web_sub as an empty source here.
                if matches!(ctx.run_mode, crate::RunMode::HistoricalFrom(_)) {
                    return;
                }
                let Some(mut rx) = rx_opt else { return; };
                while let Some(payload) = rx.recv().await {
                    match codec.decode::<T>(&payload) {
                        Ok(v) => yield Ok((NanoTime::now(), v)),
                        Err(e) => yield Err(anyhow::anyhow!("web_sub '{topic}': {e}")),
                    }
                }
            })
        },
        None,
    )
}