feat: add Default and constructors to ServerSseMessage

Author: erenatas

crates/rmcp/src/transport/common/server_side_http.rs

@@ -57,7 +57,7 @@ impl sse_stream::Timer for TokioTimer {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Default)]
 #[non_exhaustive]
 pub struct ServerSseMessage {
     /// The event ID for this message. When set, clients can use this ID
@@ -71,6 +71,28 @@ pub struct ServerSseMessage {
     pub retry: Option<Duration>,
 }
 
+impl ServerSseMessage {
+    /// Create a message carrying a JSON-RPC response/notification with an event ID.
+    pub fn new(event_id: impl Into<String>, message: ServerJsonRpcMessage) -> Self {
+        Self {
+            event_id: Some(event_id.into()),
+            message: Some(Arc::new(message)),
+            retry: None,
+        }
+    }
+
+    /// Create a priming event that tells the client to reconnect after `retry`
+    /// if the connection drops.
+    /// See [SEP-1699](https://github.com/modelcontextprotocol/modelcontextprotocol/issues/1699).
+    pub fn priming(event_id: impl Into<String>, retry: Duration) -> Self {
+        Self {
+            event_id: Some(event_id.into()),
+            message: None,
+            retry: Some(retry),
+        }
+    }
+}
+
 pub(crate) fn sse_stream_response(
     stream: impl futures::Stream<Item = ServerSseMessage> + Send + Sync + 'static,
     keep_alive: Option<Duration>,

crates/rmcp/src/transport/streamable_http_server/session/local.rs

@@ -1,7 +1,6 @@
 use std::{
     collections::{HashMap, HashSet, VecDeque},
     num::ParseIntError,
-    sync::Arc,
     time::Duration,
 };
 
@@ -222,21 +221,13 @@ impl CachedTx {
 
     async fn send(&mut self, message: ServerJsonRpcMessage) {
         let event_id = self.next_event_id();
-        let message = ServerSseMessage {
-            event_id: Some(event_id.to_string()),
-            message: Some(Arc::new(message)),
-            retry: None,
-        };
+        let message = ServerSseMessage::new(event_id.to_string(), message);
         self.cache_and_send(message).await;
     }
 
     async fn send_priming(&mut self, retry: Duration) {
         let event_id = self.next_event_id();
-        let message = ServerSseMessage {
-            event_id: Some(event_id.to_string()),
-            message: None,
-            retry: Some(retry),
-        };
+        let message = ServerSseMessage::priming(event_id.to_string(), retry);
         self.cache_and_send(message).await;
     }
 

crates/rmcp/src/transport/streamable_http_server/tower.rs

@@ -499,11 +499,7 @@ where
                 .map_err(internal_error_response("create standalone stream"))?;
             // Prepend priming event if sse_retry configured
             let stream = if let Some(retry) = self.config.sse_retry {
-                let priming = ServerSseMessage {
-                    event_id: Some("0".into()),
-                    message: None,
-                    retry: Some(retry),
-                };
+                let priming = ServerSseMessage::priming("0", retry);
                 futures::stream::once(async move { priming })
                     .chain(stream)
                     .left_stream()
@@ -609,11 +605,7 @@ where
                             .map_err(internal_error_response("get session"))?;
                         // Prepend priming event if sse_retry configured
                         let stream = if let Some(retry) = self.config.sse_retry {
-                            let priming = ServerSseMessage {
-                                event_id: Some("0".into()),
-                                message: None,
-                                retry: Some(retry),
-                            };
+                            let priming = ServerSseMessage::priming("0", retry);
                             futures::stream::once(async move { priming })
                                 .chain(stream)
                                 .left_stream()
@@ -688,19 +680,13 @@ where
                     .await
                     .map_err(internal_error_response("create stream"))?;
                 let stream = futures::stream::once(async move {
-                    ServerSseMessage {
-                        event_id: None,
-                        message: Some(Arc::new(response)),
-                        retry: None,
-                    }
+                    let mut msg = ServerSseMessage::default();
+                    msg.message = Some(Arc::new(response));
+                    msg
                 });
                 // Prepend priming event if sse_retry configured
                 let stream = if let Some(retry) = self.config.sse_retry {
-                    let priming = ServerSseMessage {
-                        event_id: Some("0".into()),
-                        message: None,
-                        retry: Some(retry),
-                    };
+                    let priming = ServerSseMessage::priming("0", retry);
                     futures::stream::once(async move { priming })
                         .chain(stream)
                         .left_stream()
@@ -774,11 +760,9 @@ where
                         // SSE mode (default): original behaviour preserved unchanged
                         let stream = ReceiverStream::new(receiver).map(|message| {
                             tracing::trace!(?message);
-                            ServerSseMessage {
-                                event_id: None,
-                                message: Some(Arc::new(message)),
-                                retry: None,
-                            }
+                            let mut msg = ServerSseMessage::default();
+                            msg.message = Some(Arc::new(message));
+                            msg
                         });
                         Ok(sse_stream_response(
                             stream,