contextvm_sdk/transport/open_stream/
receiver.rs

1//! Thin transport-facing adapter over an [`OpenStreamRegistry`].
2//!
3//! Ports `sdk/src/transport/open-stream/receiver.ts`. The client transport owns
4//! one receiver (single peer); the server owns one per peer. It recognizes
5//! inbound CEP-41 frames
6//! ([`is_open_stream_frame`](OpenStreamReceiver::is_open_stream_frame)) and feeds
7//! them to the registry with the real clock
8//! ([`process_frame`](OpenStreamReceiver::process_frame)). The registry's own
9//! `process_frame` takes an explicit `now` so the engine stays unit-testable
10//! with an injected clock; this adapter supplies `Instant::now()` at the
11//! transport boundary.
12
13use std::time::Instant;
14
15use crate::core::types::JsonRpcNotification;
16
17use super::errors::OpenStreamError;
18use super::registry::{OpenStreamRegistry, OpenStreamRegistryPolicy};
19use super::session::{FrameOutcome, OpenStreamSession};
20
21/// Transport-facing CEP-41 receiver wrapping a per-peer registry.
22pub struct OpenStreamReceiver {
23    registry: OpenStreamRegistry,
24}
25
26impl Default for OpenStreamReceiver {
27    fn default() -> Self {
28        Self::new()
29    }
30}
31
32impl OpenStreamReceiver {
33    /// Create a receiver with the default policy.
34    pub fn new() -> Self {
35        Self {
36            registry: OpenStreamRegistry::new(),
37        }
38    }
39
40    /// Create a receiver with an explicit registry policy.
41    pub fn with_policy(policy: OpenStreamRegistryPolicy) -> Self {
42        Self {
43            registry: OpenStreamRegistry::with_policy(policy),
44        }
45    }
46
47    /// Returns `true` when `notification` carries a CEP-41 frame in `params.cvm`.
48    pub fn is_open_stream_frame(notification: &JsonRpcNotification) -> bool {
49        OpenStreamRegistry::is_open_stream_progress(notification)
50    }
51
52    /// Feed one inbound `notifications/progress` frame to the registry, stamping
53    /// liveness with the real clock.
54    pub async fn process_frame(
55        &mut self,
56        notification: &JsonRpcNotification,
57    ) -> Result<FrameOutcome, OpenStreamError> {
58        self.registry
59            .process_frame(Instant::now(), notification)
60            .await
61    }
62
63    /// Look up an active reader session by token.
64    pub fn get_session(&self, progress_token: &str) -> Option<OpenStreamSession> {
65        self.registry.get_session(progress_token)
66    }
67
68    /// Number of active streams.
69    pub fn active_stream_count(&self) -> usize {
70        self.registry.size()
71    }
72
73    /// Borrow the underlying registry (e.g. to create an outbound reader session
74    /// or drive the keepalive sweep).
75    pub fn registry(&self) -> &OpenStreamRegistry {
76        &self.registry
77    }
78
79    /// Mutably borrow the underlying registry.
80    pub fn registry_mut(&mut self) -> &mut OpenStreamRegistry {
81        &mut self.registry
82    }
83
84    /// Dispose all sessions and clear the registry.
85    pub fn clear(&mut self) {
86        self.registry.clear();
87    }
88}
89
90#[cfg(test)]
91mod tests {
92    use super::*;
93    use crate::transport::open_stream::frame::OpenStreamFrame;
94    use serde_json::json;
95
96    fn start_notification(token: &str, progress: u64) -> JsonRpcNotification {
97        OpenStreamFrame::Start { content_type: None }
98            .into_progress_notification(token, progress, None)
99            .unwrap()
100    }
101
102    #[test]
103    fn is_open_stream_frame_detects_cvm_payload() {
104        let frame = start_notification("tok", 1);
105        assert!(OpenStreamReceiver::is_open_stream_frame(&frame));
106
107        let plain = JsonRpcNotification {
108            jsonrpc: "2.0".to_string(),
109            method: "notifications/progress".to_string(),
110            params: Some(json!({ "progressToken": "tok", "progress": 1 })),
111        };
112        assert!(!OpenStreamReceiver::is_open_stream_frame(&plain));
113
114        // An oversized-transfer frame is not an open-stream frame.
115        let oversized = JsonRpcNotification {
116            jsonrpc: "2.0".to_string(),
117            method: "notifications/progress".to_string(),
118            params: Some(json!({
119                "progressToken": "tok",
120                "progress": 1,
121                "cvm": { "type": "oversized-transfer", "frameType": "end" }
122            })),
123        };
124        assert!(!OpenStreamReceiver::is_open_stream_frame(&oversized));
125    }
126
127    #[tokio::test]
128    async fn process_frame_delegates_to_registry_and_tracks_session() {
129        let mut receiver = OpenStreamReceiver::new();
130        receiver
131            .process_frame(&start_notification("tok", 1))
132            .await
133            .unwrap();
134        assert_eq!(receiver.active_stream_count(), 1);
135        assert!(receiver.get_session("tok").is_some());
136
137        // Driving a graceful close through the adapter removes the session.
138        let close = OpenStreamFrame::Close {
139            last_chunk_index: None,
140        }
141        .into_progress_notification("tok", 2, None)
142        .unwrap();
143        receiver.process_frame(&close).await.unwrap();
144        assert_eq!(receiver.active_stream_count(), 0);
145    }
146
147    #[tokio::test]
148    async fn non_open_stream_progress_notification_is_not_intercepted() {
149        let mut receiver = OpenStreamReceiver::new();
150
151        // A CEP-22 oversized-transfer frame and a plain progress notification are
152        // both NOT open-stream frames — the two receivers are type-disjoint.
153        let oversized = JsonRpcNotification {
154            jsonrpc: "2.0".to_string(),
155            method: "notifications/progress".to_string(),
156            params: Some(json!({
157                "progressToken": "tok",
158                "progress": 1,
159                "cvm": { "type": "oversized-transfer", "frameType": "end" }
160            })),
161        };
162        let plain = JsonRpcNotification {
163            jsonrpc: "2.0".to_string(),
164            method: "notifications/progress".to_string(),
165            params: Some(json!({ "progressToken": "tok", "progress": 1 })),
166        };
167
168        for notification in [&oversized, &plain] {
169            assert!(!OpenStreamReceiver::is_open_stream_frame(notification));
170            // The dispatcher only feeds the registry when the predicate is true.
171            // Mirror that gate: a non-open-stream frame must never be processed.
172            if OpenStreamReceiver::is_open_stream_frame(notification) {
173                receiver.process_frame(notification).await.unwrap();
174            }
175        }
176
177        // Neither frame reached the registry — no session was created/tracked.
178        assert_eq!(receiver.active_stream_count(), 0);
179        assert!(receiver.get_session("tok").is_none());
180    }
181}
contextvm_sdk/transport/open_stream/receiver.rs

contextvm_sdk/transport/open_stream/
receiver.rs
