ISseEventStreamStore and test implementation

Author: MackinnonBuck

src/ModelContextProtocol.Core/Server/ISseEventStreamReader.cs

@@ -0,0 +1,36 @@
+using ModelContextProtocol.Protocol;
+using System.Net.ServerSentEvents;
+
+namespace ModelContextProtocol.Server;
+
+/// <summary>
+/// Provides read access to an SSE event stream, allowing events to be consumed asynchronously.
+/// </summary>
+public interface ISseEventStreamReader
+{
+    /// <summary>
+    /// Gets the session ID associated with the stream being read.
+    /// </summary>
+    string SessionId { get; }
+
+    /// <summary>
+    /// Gets the ID of the stream.
+    /// </summary>
+    /// <remarks>
+    /// This value is guaranteed to be unique on a per-session basis.
+    /// </remarks>
+    string StreamId { get; }
+
+    /// <summary>
+    /// Gets the messages from the stream as an <see cref="IAsyncEnumerable{T}"/>.
+    /// </summary>
+    /// <param name="cancellationToken">A token to cancel the operation.</param>
+    /// <returns>An <see cref="IAsyncEnumerable{T}"/> of <see cref="SseItem{T}"/> containing JSON-RPC messages.</returns>
+    /// <remarks>
+    /// If the stream's mode is set to <see cref="SseEventStreamMode.Polling"/>, the returned
+    /// messages will only include the currently-available events starting at the last event ID specified
+    /// when the reader was created. Otherwise, the returned messages will continue until the associated
+    /// <see cref="ISseEventStreamWriter"/> is disposed.
+    /// </remarks>
+    IAsyncEnumerable<SseItem<JsonRpcMessage?>> ReadEventsAsync(CancellationToken cancellationToken = default);
+}

src/ModelContextProtocol.Core/Server/ISseEventStreamStore.cs

@@ -0,0 +1,23 @@
+namespace ModelContextProtocol.Server;
+
+/// <summary>
+/// Provides storage and retrieval of SSE event streams, enabling resumability and redelivery of events.
+/// </summary>
+public interface ISseEventStreamStore
+{
+    /// <summary>
+    /// Creates a new SSE event stream with the specified options.
+    /// </summary>
+    /// <param name="options">The configuration options for the new stream.</param>
+    /// <param name="cancellationToken">A token to cancel the operation.</param>
+    /// <returns>A writer for the newly created event stream.</returns>
+    ValueTask<ISseEventStreamWriter> CreateStreamAsync(SseEventStreamOptions options, CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Gets a reader for an existing event stream based on the last event ID.
+    /// </summary>
+    /// <param name="lastEventId">The ID of the last event received by the client, used to resume from that point.</param>
+    /// <param name="cancellationToken">A token to cancel the operation.</param>
+    /// <returns>A reader for the event stream, or <c>null</c> if no matching stream is found.</returns>
+    ValueTask<ISseEventStreamReader?> GetStreamReaderAsync(string lastEventId, CancellationToken cancellationToken);
+}

src/ModelContextProtocol.Core/Server/ISseEventStreamWriter.cs

@@ -0,0 +1,43 @@
+using ModelContextProtocol.Protocol;
+using System.Net.ServerSentEvents;
+
+namespace ModelContextProtocol.Server;
+
+/// <summary>
+/// Provides write access to an SSE event stream, allowing events to be written and tracked with unique IDs.
+/// </summary>
+public interface ISseEventStreamWriter : IAsyncDisposable
+{
+    /// <summary>
+    /// Gets the ID of the stream.
+    /// </summary>
+    /// <remarks>
+    /// This value is guaranteed to be unique on a per-session basis.
+    /// </remarks>
+    string StreamId { get; }
+
+    /// <summary>
+    /// Gets the current mode of the event stream.
+    /// </summary>
+    SseEventStreamMode Mode { get; }
+
+    /// <summary>
+    /// Sets the mode of the event stream.
+    /// </summary>
+    /// <param name="mode">The new mode to set for the event stream.</param>
+    /// <param name="cancellationToken">A token to cancel the operation.</param>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    ValueTask SetModeAsync(SseEventStreamMode mode, CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Writes an event to the stream.
+    /// </summary>
+    /// <param name="sseItem">The original <see cref="SseItem{T}"/>.</param>
+    /// <param name="cancellationToken">A token to cancel the operation.</param>
+    /// <returns>A new <see cref="SseItem{T}"/> with a populated event ID.</returns>
+    /// <remarks>
+    /// If the provided <paramref name="sseItem"/> already has an event ID, this method skips writing the event.
+    /// Otherwise, an event ID unique to all sessions and streams is generated and assigned to the event.
+    /// </remarks>
+    ValueTask<SseItem<JsonRpcMessage?>> WriteEventAsync(SseItem<JsonRpcMessage?> sseItem, CancellationToken cancellationToken = default);
+}

src/ModelContextProtocol.Core/Server/SseEventStreamMode.cs

@@ -0,0 +1,20 @@
+namespace ModelContextProtocol.Server;
+
+/// <summary>
+/// Represents the mode of an SSE event stream.
+/// </summary>
+public enum SseEventStreamMode
+{
+    /// <summary>
+    /// Causes the event stream returned by <see cref="ISseEventStreamReader.ReadEventsAsync(System.Threading.CancellationToken)"/> to only end when
+    /// the associated <see cref="ISseEventStreamWriter"/> gets disposed.
+    /// </summary>
+    Default = 0,
+
+    /// <summary>
+    /// Causes the event stream returned by <see cref="ISseEventStreamReader.ReadEventsAsync(System.Threading.CancellationToken)"/> to end
+    /// after the most recent event has been consumed. This forces clients to keep making new requests in order to receive
+    /// the latest messages.
+    /// </summary>
+    Polling = 1,
+}

src/ModelContextProtocol.Core/Server/SseEventStreamOptions.cs

@@ -0,0 +1,22 @@
+namespace ModelContextProtocol.Server;
+
+/// <summary>
+/// Configuration options for creating an SSE event stream.
+/// </summary>
+public sealed class SseEventStreamOptions
+{
+    /// <summary>
+    /// Gets or sets the session ID associated with the event stream.
+    /// </summary>
+    public required string SessionId { get; set; }
+
+    /// <summary>
+    /// Gets or sets the stream ID that uniquely identifies this stream within a session.
+    /// </summary>
+    public required string StreamId { get; set; }
+
+    /// <summary>
+    /// Gets or sets the mode of the event stream. Defaults to <see cref="SseEventStreamMode.Default"/>.
+    /// </summary>
+    public SseEventStreamMode Mode { get; set; }
+}

src/ModelContextProtocol.Core/Server/SseEventStreamReaderExtensions.cs

@@ -0,0 +1,50 @@
+using ModelContextProtocol.Protocol;
+using System.Buffers;
+using System.Net.ServerSentEvents;
+using System.Text.Json;
+
+namespace ModelContextProtocol.Server;
+
+/// <summary>
+/// Provides extension methods for <see cref="ISseEventStreamReader"/>.
+/// </summary>
+public static class SseEventStreamReaderExtensions
+{
+    /// <summary>
+    /// Copies all events from the reader to the destination stream in SSE format.
+    /// </summary>
+    /// <param name="reader">The event stream reader to copy events from.</param>
+    /// <param name="destination">The destination stream to write SSE-formatted events to.</param>
+    /// <param name="cancellationToken">A token to cancel the operation.</param>
+    /// <returns>A task that represents the asynchronous copy operation.</returns>
+    /// <exception cref="ArgumentNullException">Thrown when <paramref name="reader"/> or <paramref name="destination"/> is null.</exception>
+    public static async Task CopyToAsync(this ISseEventStreamReader reader, Stream destination, CancellationToken cancellationToken = default)
+    {
+        Throw.IfNull(reader);
+        Throw.IfNull(destination);
+
+        Utf8JsonWriter? jsonWriter = null;
+
+        var events = reader.ReadEventsAsync(cancellationToken);
+        await SseFormatter.WriteAsync(events, destination, FormatEvent, cancellationToken);
+
+        void FormatEvent(SseItem<JsonRpcMessage?> item, IBufferWriter<byte> writer)
+        {
+            if (item.Data is null)
+            {
+                return;
+            }
+
+            if (jsonWriter is null)
+            {
+                jsonWriter = new Utf8JsonWriter(writer);
+            }
+            else
+            {
+                jsonWriter.Reset(writer);
+            }
+
+            JsonSerializer.Serialize(jsonWriter, item.Data, McpJsonUtilities.JsonContext.Default.JsonRpcMessage!);
+        }
+    }
+}