fix(tests): resolve SSE test hanging issue

- Refactored all SSE tests to use FakeCompletableSseStream with .Complete()
- Added .AsTask().WaitAsync() pattern for read operations with timeouts
- Fixed tests that were hanging due to never-ending SSE streams:
  - SseExtensionsTests.cs
  - SseExtensionsFallbackTests.cs
  - SseExtensionsNet10Tests.cs
  - GenAIEventStreamIntegrationTests.cs
  - OcrEventStreamIntegrationTests.cs

Test suite now completes in ~50 seconds instead of 11+ minutes.

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

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

Author: ANcpLua

README.md

@@ -61,9 +61,15 @@ services.AddPaperlessRabbitMq(config, includeOcrResultStream: true, includeGenAi
 ### Publishing
 
 ```csharp
+// Basic command
 var command = new OcrCommand(docId, fileName, storagePath);
 await publisher.PublishOcrCommandAsync(command);
 
+// With timestamp (v2.2.0+)
+var command = new OcrCommand(docId, fileName, storagePath, createdAt: DateTimeOffset.UtcNow);
+await publisher.PublishOcrCommandAsync(command);
+
+// Event
 var result = new OcrEvent(jobId, "Completed", text, DateTimeOffset.UtcNow);
 await publisher.PublishOcrEventAsync(result);
 ```
@@ -87,13 +93,28 @@ await foreach (var command in consumer.ConsumeAsync(cancellationToken))
 }
 ```
 
-### SSE Endpoint
+### SSE Endpoints
 
 ```csharp
-// Map endpoint
-app.MapOcrEventStream();
+// Built-in endpoints
+app.MapOcrEventStream();      // /api/v1/ocr-results
+app.MapGenAIEventStream();    // /api/v1/events/genai
+
+// Custom endpoint with dynamic event types (v2.3.0+)
+app.MapSse<NotificationEvent>("/api/notifications", evt => evt.Type);
+
+// Custom endpoint with fixed event type (v2.4.0+)
+app.MapSse<HeartbeatEvent>("/api/heartbeat", "heartbeat");
+
+// Full control with payload projection
+app.MapSse<OcrEvent>("/api/ocr",
+    evt => new { evt.JobId, evt.Status },  // payload
+    evt => evt.Status == "Completed" ? "done" : "error");  // event type
+```
 
-// Client-side
+**Client-side:**
+
+```javascript
 const eventSource = new EventSource('/api/v1/ocr-results');
 eventSource.addEventListener('ocr-completed', (event) => {
     const data = JSON.parse(event.data);
@@ -139,18 +160,57 @@ await _publisher.PublishGenAICommandAsync(genAiCommand);
 ## Message Types
 
 ```csharp
-public record OcrCommand(Guid JobId, string FileName, string FilePath);
+// OCR
+public record OcrCommand(Guid JobId, string FileName, string FilePath, DateTimeOffset? CreatedAt = null);
 public record OcrEvent(Guid JobId, string Status, string? Text, DateTimeOffset ProcessedAt);
+
+// GenAI
 public record GenAICommand(Guid DocumentId, string Text);
-public record GenAIEvent(Guid DocumentId, string? Summary, DateTimeOffset ProcessedAt, string? ErrorMessage = null);
+public record GenAIEvent(Guid DocumentId, string? Summary, DateTimeOffset GeneratedAt, string? ErrorMessage = null);
 ```
 
+## SSE MapSse Overloads
+
+| Overload | Use Case |
+|----------|----------|
+| `MapSse<T>(pattern, payloadSelector, eventTypeSelector)` | Full control over payload and per-event type |
+| `MapSse<T>(pattern, eventTypeSelector)` | Serialize whole object, dynamic event type |
+| `MapSse<T>(pattern, eventType)` | Serialize whole object, fixed event type |
+
 ## Installation
 
 ```bash
 dotnet add package SWEN3.Paperless.RabbitMq
 ```
 
+## Changelog
+
+### v2.4.0
+
+- Added `MapSse<T>(pattern, eventType)` overload for fixed event type streams
+- .NET 10: Uses optimized `TypedResults.ServerSentEvents` overload (no per-event `SseItem<T>` allocation)
+
+### v2.3.0
+
+- `ISseStream<T>` now extends `IAsyncDisposable` for graceful shutdown
+- Added `MapSse<T>(pattern, eventTypeSelector)` overload for simplified endpoint registration
+- `SseStream<T>` completes all client channels on dispose
+
+### v2.2.0
+
+- Added optional `CreatedAt` parameter to `OcrCommand` for document timestamp tracking
+- Updated default Gemini model to `gemini-2.5-flash`
+
+### v2.1.0
+
+- GenAI SSE streaming support
+
+### v2.0.0
+
+- Simplified `AddPaperlessGenAI()` registration
+- Added `Microsoft.Extensions.Http.Resilience` for automatic retry/circuit breaker
+- Configuration section renamed: `GenAI:Gemini` → `Gemini`
+
 ## License
 
 This project is licensed under the [MIT License](LICENSE).

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

@@ -42,9 +42,14 @@ public static async Task WaitForClientsAsync<T>(
 
         try
         {
+            // PeriodicTimer is more efficient than Task.Delay in a loop:
+            // - Returns ValueTask (often zero-alloc) vs new Task per iteration
+            // - Clearer semantics: "wait for next tick" vs "delay for X ms"
+            using var timer = new PeriodicTimer(TimeSpan.FromMilliseconds(PollingIntervalMs));
+
             while (stream.ClientCount < expectedClients)
             {
-                await Task.Delay(PollingIntervalMs, timeoutCts.Token);
+                await timer.WaitForNextTickAsync(timeoutCts.Token);
             }
 
             if (stabilize)

SWEN3.Paperless.RabbitMq.Tests/Integration/GenAIEventStreamIntegrationTests.cs

@@ -2,17 +2,19 @@ namespace SWEN3.Paperless.RabbitMq.Tests.Integration;
 
 public class GenAIEventStreamIntegrationTests
 {
+    private static readonly TimeSpan ReadTimeout = TimeSpan.FromSeconds(5);
+
     [Theory]
     [InlineData("Completed", "genai-completed")]
     [InlineData("Failed", "genai-failed")]
     public async Task MapGenAIEventStream_ShouldEmitCorrectEventType(string status, string expectedEventType)
     {
+        var fakeStream = new FakeCompletableSseStream<GenAIEvent>();
         var (host, server) = await SseTestHelpers.CreateSseTestServerAsync<GenAIEvent>(
-            configureServices: null,
+            configureServices: s => s.AddSingleton<ISseStream<GenAIEvent>>(fakeStream),
             configureEndpoints: endpoints => endpoints.MapGenAIEventStream());
 
         using var _ = host;
-        var sseStream = host.Services.GetRequiredService<ISseStream<GenAIEvent>>();
         var client = server.CreateClient();
         var ct = TestContext.Current.CancellationToken;
 
@@ -25,20 +27,21 @@ public async Task MapGenAIEventStream_ShouldEmitCorrectEventType(string status,
 
             while (true)
             {
-                var line = await reader.ReadLineAsync(ct);
+                var line = await reader.ReadLineAsync(ct).AsTask().WaitAsync(ReadTimeout, ct);
                 if (line == null)
                     return null;
                 if (line.StartsWith("event:"))
                     return line;
             }
         }, ct);
 
-        await SseTestHelpers.WaitForClientsAsync(sseStream, stabilize: false, cancellationToken: ct);
+        await SseTestHelpers.WaitForClientsAsync(fakeStream, stabilize: false, cancellationToken: ct);
 
         var genAiEvent = status == "Completed"
             ? new GenAIEvent(Guid.NewGuid(), "Test summary", DateTimeOffset.UtcNow)
             : new GenAIEvent(Guid.NewGuid(), string.Empty, DateTimeOffset.UtcNow, "Service error");
-        sseStream.Publish(genAiEvent);
+        fakeStream.Publish(genAiEvent);
+        fakeStream.Complete();
 
         var eventLine = await readTask;
 
@@ -48,26 +51,27 @@ public async Task MapGenAIEventStream_ShouldEmitCorrectEventType(string status,
     [Fact]
     public async Task MapGenAIEventStream_WithMultipleEvents_ShouldStreamInOrder()
     {
+        var fakeStream = new FakeCompletableSseStream<GenAIEvent>();
         var (host, server) = await SseTestHelpers.CreateSseTestServerAsync<GenAIEvent>(
-            configureServices: null,
+            configureServices: s => s.AddSingleton<ISseStream<GenAIEvent>>(fakeStream),
             configureEndpoints: endpoints => endpoints.MapGenAIEventStream());
 
         using var _ = host;
-        var sseStream = host.Services.GetRequiredService<ISseStream<GenAIEvent>>();
         var client = server.CreateClient();
         var ct = TestContext.Current.CancellationToken;
 
         var readTask = Task.Run(() => ReadEventsAsync(client, 3, ct), ct);
 
-        await SseTestHelpers.WaitForClientsAsync(sseStream, cancellationToken: ct);
+        await SseTestHelpers.WaitForClientsAsync(fakeStream, cancellationToken: ct);
 
-        sseStream.Publish(new GenAIEvent(Guid.NewGuid(), "Summary 1", DateTimeOffset.UtcNow));
+        fakeStream.Publish(new GenAIEvent(Guid.NewGuid(), "Summary 1", DateTimeOffset.UtcNow));
         await Task.Delay(50, ct);
 
-        sseStream.Publish(new GenAIEvent(Guid.NewGuid(), "Summary 2", DateTimeOffset.UtcNow));
+        fakeStream.Publish(new GenAIEvent(Guid.NewGuid(), "Summary 2", DateTimeOffset.UtcNow));
         await Task.Delay(50, ct);
 
-        sseStream.Publish(new GenAIEvent(Guid.NewGuid(), string.Empty, DateTimeOffset.UtcNow, "Error occurred"));
+        fakeStream.Publish(new GenAIEvent(Guid.NewGuid(), string.Empty, DateTimeOffset.UtcNow, "Error occurred"));
+        fakeStream.Complete();
 
         var events = await readTask;
 
@@ -98,7 +102,7 @@ await client.GetAsync("/api/v1/events/genai", HttpCompletionOption.ResponseHeade
 
             while (events.Count < count)
             {
-                var line = await reader.ReadLineAsync(cancellationToken);
+                var line = await reader.ReadLineAsync(cancellationToken).AsTask().WaitAsync(ReadTimeout, cancellationToken);
                 if (line == null)
                     break;
 

SWEN3.Paperless.RabbitMq.Tests/Integration/OcrEventStreamIntegrationTests.cs

@@ -2,18 +2,20 @@ namespace SWEN3.Paperless.RabbitMq.Tests.Integration;
 
 public sealed class OcrEventStreamIntegrationTests
 {
+    private static readonly TimeSpan ReadTimeout = TimeSpan.FromSeconds(5);
+
     [Theory]
     [InlineData("Completed", "ocr-completed")]
     [InlineData("Failed", "ocr-failed")]
     [InlineData("Processing", "ocr-failed")]
     public async Task MapOcrEventStream_ShouldEmitCorrectEventType(string status, string expectedEventType)
     {
+        var fakeStream = new FakeCompletableSseStream<OcrEvent>();
         var (host, server) = await SseTestHelpers.CreateSseTestServerAsync<OcrEvent>(
-            configureServices: null,
+            configureServices: s => s.AddSingleton<ISseStream<OcrEvent>>(fakeStream),
             configureEndpoints: endpoints => endpoints.MapOcrEventStream());
 
         using var _ = host;
-        var sseStream = host.Services.GetRequiredService<ISseStream<OcrEvent>>();
         var ct = TestContext.Current.CancellationToken;
 
         var readTask = Task.Run(async () =>
@@ -26,19 +28,20 @@ public async Task MapOcrEventStream_ShouldEmitCorrectEventType(string status, st
 
             while (true)
             {
-                var line = await reader.ReadLineAsync(ct);
+                var line = await reader.ReadLineAsync(ct).AsTask().WaitAsync(ReadTimeout, ct);
                 if (line == null)
                     return null;
                 if (line.StartsWith("event:"))
                     return line;
             }
         }, ct);
 
-        await SseTestHelpers.WaitForClientsAsync(sseStream, cancellationToken: ct);
+        await SseTestHelpers.WaitForClientsAsync(fakeStream, cancellationToken: ct);
 
         var ocrEvent = new OcrEvent(Guid.NewGuid(), status, status is "Completed" ? "Text" : null,
             DateTimeOffset.UtcNow);
-        sseStream.Publish(ocrEvent);
+        fakeStream.Publish(ocrEvent);
+        fakeStream.Complete();
 
         var eventLine = await readTask;