Struct StreamManager 

pub struct StreamManager { /* private fields */ }

Enhanced stream manager with resumability support (MCP spec compliant)

Implementations

impl StreamManager

pub fn new(storage: Arc<BoxedSessionStorage>) -> Self

Create new stream manager with session storage backend

pub fn with_config( storage: Arc<BoxedSessionStorage>, config: StreamConfig, ) -> Self

Create stream manager with custom configuration

pub async fn handle_sse_connection( &self, session_id: String, connection_id: ConnectionId, last_event_id: Option<u64>, ) -> Result<Response<UnsyncBoxBody<Bytes, Error>>, StreamError>

Handle SSE connection request with proper resumability

pub async fn register_streaming_connection( &self, session_id: &str, connection_id: ConnectionId, sender: Sender<SseEvent>, ) -> Result<(), StreamError>

Register a streaming connection to receive events for a session (public API for POST streaming)

pub async fn unregister_connection( &self, session_id: &str, connection_id: &ConnectionId, )

Remove a connection when it’s closed

pub async fn close_session_connections(&self, session_id: &str) -> usize

Close all SSE connections for a session (useful for session termination)

pub async fn has_connections(&self, session_id: &str) -> bool

Check if a session has any active SSE connections

pub async fn broadcast_to_session( &self, session_id: &str, event_type: String, data: Value, ) -> Result<u64, StreamError>

Send event to specific session (MCP compliant - ONE connection only)

pub async fn broadcast_to_session_with_options( &self, session_id: &str, event_type: String, data: Value, store_when_no_connections: bool, ) -> Result<u64, StreamError>

Send event to specific session with option to suppress when no connections exist

pub async fn broadcast_to_all_sessions( &self, event_type: String, data: Value, ) -> Result<Vec<String>, StreamError>

Broadcast to all sessions (for server-wide notifications)

pub async fn cleanup_connections(&self) -> usize

Clean up closed connections

pub async fn create_post_sse_stream( &self, session_id: String, response: JsonRpcResponse, ) -> Result<Response<BoxBody<Bytes, Infallible>>, StreamError>

Create SSE stream for POST requests (MCP Streamable HTTP)

pub async fn subscribe_to_notifications( &self, session_id: &str, notification_types: Vec<String>, )

Subscribe a session to specific notification types

pub async fn unsubscribe_from_notifications( &self, session_id: &str, notification_types: Vec<String>, )

Unsubscribe a session from specific notification types

pub async fn is_subscribed( &self, session_id: &str, notification_type: &str, ) -> bool

Check if a session is subscribed to a specific notification type

pub async fn get_subscriptions(&self, session_id: &str) -> HashSet<String>

Get all subscriptions for a session

pub async fn clear_subscriptions(&self, session_id: &str)

Clear all subscriptions for a session (used during session cleanup)

pub fn get_config(&self) -> &StreamConfig

Get the stream configuration (for testing and debugging)

pub async fn get_stats(&self) -> StreamStats

Get statistics about active streams

Trait Implementations

impl Drop for StreamManager

fn drop(&mut self)

Executes the destructor for this type.