feat(sse): add IAsyncDisposable support with graceful shutdown

- Add IAsyncDisposable to ISseStream interface
- Implement disposal in SseStream that completes all client channels
- Update FakeCompletableSseStream with matching disposal behavior
- Add disposal behavior tests (Subscribe after dispose throws,
  Publish after dispose is no-op, DisposeAsync completes channels)
- Add simplified MapSse overload without payload selector

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: ANcpLua

SWEN3.Paperless.RabbitMq.Tests/Helpers/FakeCompletableSseStream.cs

@@ -9,6 +9,7 @@ namespace SWEN3.Paperless.RabbitMq.Tests.Helpers;
 /// </summary>
 internal sealed class FakeCompletableSseStream<T> : ISseStream<T> where T : class
 {
+    private volatile bool _disposed;
     private readonly ConcurrentDictionary<Guid, Channel<T>> _channels = new();
 
     /// <summary>
@@ -21,6 +22,9 @@ internal sealed class FakeCompletableSseStream<T> : ISseStream<T> where T : clas
     /// </summary>
     public void Publish(T message)
     {
+        if (_disposed)
+            return;
+
         foreach (var channel in _channels.Values)
         {
             channel.Writer.TryWrite(message);
@@ -30,8 +34,11 @@ public void Publish(T message)
     /// <summary>
     ///     Subscribes a client and returns a channel reader for receiving events.
     /// </summary>
+    /// <exception cref="ObjectDisposedException">Thrown if the stream has been disposed.</exception>
     public ChannelReader<T> Subscribe(Guid clientId)
     {
+        ObjectDisposedException.ThrowIf(_disposed, this);
+
         var channel = Channel.CreateUnbounded<T>(new UnboundedChannelOptions
         {
             SingleReader = true,
@@ -63,4 +70,19 @@ public void Complete()
             channel.Writer.TryComplete();
         }
     }
+
+    /// <summary>
+    ///     Disposes the stream, completing all client channels.
+    /// </summary>
+    /// <returns>A <see cref="ValueTask"/> representing the asynchronous dispose operation.</returns>
+    public ValueTask DisposeAsync()
+    {
+        if (_disposed)
+            return ValueTask.CompletedTask;
+
+        _disposed = true;
+        Complete();
+        _channels.Clear();
+        return ValueTask.CompletedTask;
+    }
 }

SWEN3.Paperless.RabbitMq.Tests/Unit/SseStreamTests.cs

@@ -1,9 +1,11 @@
 namespace SWEN3.Paperless.RabbitMq.Tests.Unit;
 
-public class SseStreamTests
+public class SseStreamTests : IAsyncDisposable
 {
     private readonly SseStream<Messages.SimpleMessage> _sseStream = new();
 
+    public ValueTask DisposeAsync() => _sseStream.DisposeAsync();
+
     [Fact]
     public void Subscribe_ShouldReturnChannelReader()
     {
@@ -85,4 +87,38 @@ public void Publish_WhenClientIsSlow_ShouldDropOldestMessages()
         receivedMessages[0].Id.Should().Be(ExpectedFirstMessageId); // Should start from 11
         receivedMessages[^1].Id.Should().Be(MessagesToSend); // Should end at 110
     }
+
+    [Fact]
+    public async Task Subscribe_AfterDispose_ShouldThrowObjectDisposedException()
+    {
+        await _sseStream.DisposeAsync();
+
+        var act = () => _sseStream.Subscribe(Guid.NewGuid());
+
+        act.Should().Throw<ObjectDisposedException>();
+    }
+
+    [Fact]
+    public async Task Publish_AfterDispose_ShouldBeNoOp()
+    {
+        var reader = _sseStream.Subscribe(Guid.NewGuid());
+        await _sseStream.DisposeAsync();
+
+        _sseStream.Publish(new Messages.SimpleMessage(1)); // Should not throw
+
+        // Channel was completed by dispose, so no message available
+        reader.TryRead(out _).Should().BeFalse();
+    }
+
+    [Fact]
+    public async Task DisposeAsync_ShouldCompleteAllClientChannels()
+    {
+        var reader1 = _sseStream.Subscribe(Guid.NewGuid());
+        var reader2 = _sseStream.Subscribe(Guid.NewGuid());
+
+        await _sseStream.DisposeAsync();
+
+        reader1.Completion.IsCompleted.Should().BeTrue();
+        reader2.Completion.IsCompleted.Should().BeTrue();
+    }
 }

SWEN3.Paperless.RabbitMq/Sse/ISseStream.cs

@@ -12,7 +12,7 @@ namespace SWEN3.Paperless.RabbitMq.Sse;
 /// <typeparam name="T">
 ///     The type of event object being streamed.
 /// </typeparam>
-public interface ISseStream<T>
+public interface ISseStream<T> : IAsyncDisposable
 {
     /// <summary>
     ///     Gets the number of currently active client subscriptions.
@@ -28,6 +28,7 @@ public interface ISseStream<T>
     /// <returns>
     ///     A <see cref="ChannelReader{T}"/> that the client can read from to receive asynchronous event updates.
     /// </returns>
+    /// <exception cref="ObjectDisposedException">Thrown if the stream has been disposed.</exception>
     ChannelReader<T> Subscribe(Guid clientId);
 
     /// <summary>
@@ -47,6 +48,7 @@ public interface ISseStream<T>
     /// <remarks>
     ///     The event is written to each client's individual channel. If a client's channel is full or closed,
     ///     the delivery strategy (e.g., dropping the message) is determined by the implementation.
+    ///     If the stream has been disposed, this method is a no-op.
     /// </remarks>
     void Publish(T item);
 }

SWEN3.Paperless.RabbitMq/Sse/SseExtensions.cs

@@ -92,6 +92,45 @@ public static RouteHandlerBuilder MapSse<T>(this IEndpointRouteBuilder endpoints
 #endif
     }
 
+    /// <summary>
+    ///     Maps a GET endpoint that streams Server-Sent Events (SSE) to connected clients,
+    ///     serializing the entire event object as the payload using camelCase naming.
+    /// </summary>
+    /// <typeparam name="T">
+    ///     The type of the event object being streamed.
+    /// </typeparam>
+    /// <param name="endpoints">
+    ///     The <see cref="IEndpointRouteBuilder"/> to add the route to.
+    /// </param>
+    /// <param name="pattern">
+    ///     The URL pattern for the endpoint (e.g., <c>"/api/events"</c>).
+    /// </param>
+    /// <param name="eventTypeSelector">
+    ///     A function that determines the SSE "event" type string (e.g., "message", "update") for the client to listen for.
+    /// </param>
+    /// <returns>
+    ///     A <see cref="RouteHandlerBuilder"/> that can be used to further configure the endpoint (e.g., adding authorization).
+    /// </returns>
+    /// <remarks>
+    ///     <para>
+    ///         This is a simplified overload that serializes the entire event object as JSON.
+    ///         Use this when you don't need to transform the event before sending.
+    ///     </para>
+    ///     <para>
+    ///         The event object is serialized using <see cref="JsonSerializerDefaults.Web"/> (camelCase property naming).
+    ///     </para>
+    /// </remarks>
+    /// <example>
+    ///     <code>
+    ///         app.MapSse&lt;NotificationEvent&gt;("/api/notifications", evt => evt.Type);
+    ///     </code>
+    /// </example>
+    public static RouteHandlerBuilder MapSse<T>(this IEndpointRouteBuilder endpoints, string pattern,
+        Func<T, string> eventTypeSelector) where T : class
+    {
+        return endpoints.MapSse(pattern, static item => item, eventTypeSelector);
+    }
+
 #if NET10_0_OR_GREATER
     /// <summary>
     ///     .NET 10+ implementation using native ServerSentEvents API.
@@ -122,6 +161,7 @@ async IAsyncEnumerable<SseItem<object>> StreamEventsAsync([EnumeratorCancellatio
             }
         });
     }
+
 #endif
 
 #if !NET10_0_OR_GREATER

SWEN3.Paperless.RabbitMq/Sse/SseStream.cs

@@ -10,12 +10,17 @@ namespace SWEN3.Paperless.RabbitMq.Sse;
 internal sealed class SseStream<T> : ISseStream<T>
 {
     private readonly ConcurrentDictionary<Guid, Channel<T>> _channels = new();
+    private volatile bool _disposed;
 
     /// <inheritdoc />
     public int ClientCount => _channels.Count;
 
+    /// <inheritdoc />
+    /// <exception cref="ObjectDisposedException">Thrown if the stream has been disposed.</exception>
     public ChannelReader<T> Subscribe(Guid clientId)
     {
+        ObjectDisposedException.ThrowIf(_disposed, this);
+
         var channel = Channel.CreateBounded<T>(new BoundedChannelOptions(100)
         {
             FullMode = BoundedChannelFullMode.DropOldest,
@@ -27,6 +32,7 @@ public ChannelReader<T> Subscribe(Guid clientId)
         return channel.Reader;
     }
 
+    /// <inheritdoc />
     public void Unsubscribe(Guid clientId)
     {
         if (_channels.TryRemove(clientId, out var channel))
@@ -38,9 +44,32 @@ public void Unsubscribe(Guid clientId)
     /// <inheritdoc />
     public void Publish(T item)
     {
+        if (_disposed)
+            return;
+
         foreach (var channel in _channels.Values)
         {
             channel.Writer.TryWrite(item);
         }
     }
+
+    /// <summary>
+    ///     Disposes the stream, completing all client channels for graceful shutdown.
+    /// </summary>
+    /// <returns>A <see cref="ValueTask"/> representing the asynchronous dispose operation.</returns>
+    public ValueTask DisposeAsync()
+    {
+        if (_disposed)
+            return ValueTask.CompletedTask;
+
+        _disposed = true;
+
+        foreach (var channel in _channels.Values)
+        {
+            channel.Writer.TryComplete();
+        }
+
+        _channels.Clear();
+        return ValueTask.CompletedTask;
+    }
 }