Add line and Server-Sent Events readers for streaming responses

README.md

@@ -1,7 +1,7 @@
 # SimpleHttpClient
 [![CI](https://github.com/Mako88/SimpleHttpClient/actions/workflows/ci.yml/badge.svg)](https://github.com/Mako88/SimpleHttpClient/actions/workflows/ci.yml)
 
-An easy-to-use .NET wrapper for `HttpClient`. No extension methods, included interfaces allow for easy unit test mocking, and straightforward properties allow for easier debugging (the response body is available as a string, byte array, and/or a typed object).
+An easy-to-use .NET wrapper for `HttpClient`. No extension methods, included interfaces for easy unit test mocking, and straightforward properties for easier debugging (the response body is available as a string, byte array, and/or a typed object). It also supports streaming responses, with built-in helpers for reading line-delimited streams and Server-Sent Events.
 
 ## Contents
 - [Installation](#installation)
@@ -15,6 +15,7 @@ An easy-to-use .NET wrapper for `HttpClient`. No extension methods, included int
   - [Headers](#headers)
   - [Request Bodies](#request-bodies)
 - [Streaming Responses](#streaming-responses)
+  - [Reading lines and Server-Sent Events](#reading-lines-and-server-sent-events)
 - [Configuration](#configuration)
   - [Timeouts](#timeouts)
   - [Additional Successful Status Codes](#additional-successful-status-codes)
@@ -200,7 +201,46 @@ while ((line = await reader.ReadLineAsync()) != null)
 A few things to keep in mind:
 - **Dispose the response.** The underlying `HttpResponseMessage` and connection are held open until you dispose the returned `ISimpleStreamResponse`. A `using` block is the simplest way to guarantee this.
 - **`MakeStreamRequest` accepts a `CancellationToken`.** Pass one to cancel sending the request, waiting for the headers, and reading the stream (e.g. when a user aborts mid-stream). The token is observed by reads too, even through a `StreamReader` that gives you no place to pass it — so the `await reader.ReadLineAsync()` loop above stops promptly when the token fires. Async reads honor it even mid-read; synchronous reads observe it between reads, so to abort a synchronous read already blocked on the socket, dispose the response. A caller-requested cancellation surfaces as an `OperationCanceledException`; a timeout still surfaces as a `TimeoutException`.
-- **The body is yours to frame.** `SimpleStreamResponse.Body` is a plain `Stream`, leaving any protocol-specific framing (such as SSE `event:`/`data:` parsing) to you.
+- **The body is yours to frame.** `SimpleStreamResponse.Body` is a plain `Stream`. For common cases there are helpers (below) that read it as lines or Server-Sent Events; otherwise you can frame it however you like.
+
+### Reading lines and Server-Sent Events
+`ISimpleStreamResponse` has two methods that cover the most common streaming formats. Both are `IAsyncEnumerable<T>`, so you consume them with `await foreach`, and both honor a `CancellationToken`:
+
+```csharp
+// Line-delimited streams (NDJSON, plain text, etc.)
+await foreach (var line in response.ReadLinesAsync(cancellationToken))
+{
+    Console.WriteLine(line);
+}
+```
+
+```csharp
+// Server-Sent Events (text/event-stream)
+await foreach (var sse in response.ReadServerSentEventsAsync(cancellationToken))
+{
+    // sse.Data, sse.EventType, sse.Id, sse.Retry
+    Console.WriteLine(sse.Data);
+}
+```
+
+`ReadServerSentEventsAsync` parses the SSE wire format per the [WHATWG specification](https://html.spec.whatwg.org/multipage/server-sent-events.html): events are separated by blank lines, multiple `data:` lines are joined with newlines, and comment/keep-alive lines (starting with `:`) are skipped. It deliberately does **not** handle application-specific conventions — most notably it does not treat any sentinel value specially and does not deserialize the payload — so those stay in your hands:
+
+```csharp
+using var response = await client.MakeStreamRequest(request, cancellationToken);
+
+await foreach (var sse in response.ReadServerSentEventsAsync(cancellationToken))
+{
+    if (sse.Data == "end") // a sentinel some APIs send to mark the end - your convention, not the library's
+    {
+        break;
+    }
+
+    var chunk = client.Serializer.Deserialize<MyChunk>(sse.Data);
+    // ...handle chunk
+}
+```
+
+This keeps SimpleHttpClient general-purpose: the SSE framing is a web standard (the same format `EventSource` consumes in browsers), while which sentinel terminates the stream and how each `data` payload is shaped are specific to the API you're calling.
 
 ## Configuration
 
@@ -236,11 +276,11 @@ You can supply your own serializer by implementing `ISimpleHttpSerializer`.
 #### JSON serialization
 The default JSON serializer (`SimpleHttpDefaultJsonSerializer`) is backed by `System.Text.Json`. It serializes with camelCase names, omits null values, writes indented output, and deserializes case-insensitively. For smoother interop it also reads numbers from JSON strings (e.g. `"123"`) and tolerates trailing commas and comments while reading. The equivalent `SimpleHttpSystemTextJsonSerializer` is also available for callers who reference it explicitly.
 
-> **Upgrading from v4?** As of **v5.0.0** the default serializer moved from `Newtonsoft.Json` to `System.Text.Json` and the `Newtonsoft.Json` dependency was removed. The defaults above cover the most common differences, but `System.Text.Json` is stricter in two ways it won't soften:
+> **Note:** the defaults above cover the most common cases, but deserialization is strict in two ways they don't soften:
 > - **Non-public parameterless constructors** aren't used — add a public constructor or a `[JsonConstructor]`.
-> - **Wrong-shape values aren't coerced** — a field that's sometimes a string and sometimes an object (and similar) threw nothing under Newtonsoft but throws here. For such fields, attach a custom `JsonConverter` to the property.
+> - **Wrong-shape values aren't coerced** — a field that's sometimes a string and sometimes an object (and similar) will throw. For such fields, attach a custom `JsonConverter` to the property.
 >
-> If you'd rather keep the old behavior wholesale, implement `ISimpleHttpSerializer` with your own `Newtonsoft.Json` serializer and set it on the client.
+> If you need different behavior wholesale, implement `ISimpleHttpSerializer` with your own serializer and set it on the client.
 
 ### Logging
 You can log requests and responses by setting the `LogRequest` and `LogResponse` delegates (called immediately before a request is sent and immediately after a response is received), or by providing an `ISimpleHttpLogger`:

src/SimpleHttpClient.Tests.Integration/ServerSentEventsTests.cs

@@ -0,0 +1,251 @@
+using Moq;
+using SimpleHttpClient.Models;
+using System.Collections.Generic;
+using System.IO;
+using System.Net;
+using System.Net.Http;
+using System.Text;
+using System.Threading;
+using WireMock.RequestBuilders;
+using WireMock.ResponseBuilders;
+using WireMock.Server;
+
+namespace SimpleHttpClient.Tests
+{
+    public class ServerSentEventsTests
+    {
+        // ---- ReadLinesAsync ----
+
+        [Fact]
+        public async Task ReadLines_SplitsOnNewlines()
+        {
+            using var response = StreamFrom("a\nb\nc\n");
+
+            var lines = await Collect(response.ReadLinesAsync());
+
+            Assert.Equal(new[] { "a", "b", "c" }, lines);
+        }
+
+        [Fact]
+        public async Task ReadLines_HandlesCrlf()
+        {
+            using var response = StreamFrom("a\r\nb\r\n");
+
+            var lines = await Collect(response.ReadLinesAsync());
+
+            Assert.Equal(new[] { "a", "b" }, lines);
+        }
+
+        // ---- ReadServerSentEventsAsync ----
+
+        [Fact]
+        public async Task Sse_SingleEvent_ParsesData()
+        {
+            using var response = StreamFrom("data: hello\n\n");
+
+            var events = await Collect(response.ReadServerSentEventsAsync());
+
+            var sse = Assert.Single(events);
+            Assert.Equal("hello", sse.Data);
+            Assert.Null(sse.EventType);
+            Assert.Null(sse.Id);
+            Assert.Null(sse.Retry);
+        }
+
+        [Fact]
+        public async Task Sse_MultipleDataLines_JoinedWithNewline()
+        {
+            using var response = StreamFrom("data: a\ndata: b\ndata: c\n\n");
+
+            var events = await Collect(response.ReadServerSentEventsAsync());
+
+            Assert.Equal("a\nb\nc", Assert.Single(events).Data);
+        }
+
+        [Fact]
+        public async Task Sse_NoSpaceAfterColon_IsAccepted()
+        {
+            using var response = StreamFrom("data:hello\n\n");
+
+            Assert.Equal("hello", Assert.Single(await Collect(response.ReadServerSentEventsAsync())).Data);
+        }
+
+        [Fact]
+        public async Task Sse_CommentLines_AreIgnored()
+        {
+            using var response = StreamFrom(": this is a keep-alive\ndata: real\n\n");
+
+            Assert.Equal("real", Assert.Single(await Collect(response.ReadServerSentEventsAsync())).Data);
+        }
+
+        [Fact]
+        public async Task Sse_ParsesEventTypeIdAndRetry()
+        {
+            using var response = StreamFrom("event: update\nid: 42\nretry: 1500\ndata: payload\n\n");
+
+            var sse = Assert.Single(await Collect(response.ReadServerSentEventsAsync()));
+            Assert.Equal("update", sse.EventType);
+            Assert.Equal("42", sse.Id);
+            Assert.Equal(1500, sse.Retry);
+            Assert.Equal("payload", sse.Data);
+        }
+
+        [Fact]
+        public async Task Sse_Id_CarriesForwardToLaterEvents()
+        {
+            using var response = StreamFrom("id: 1\ndata: a\n\ndata: b\n\n");
+
+            var events = await Collect(response.ReadServerSentEventsAsync());
+
+            Assert.Equal(2, events.Count);
+            Assert.Equal("1", events[0].Id);
+            Assert.Equal("1", events[1].Id); // id carries forward per the SSE spec
+            Assert.Equal("b", events[1].Data);
+        }
+
+        [Fact]
+        public async Task Sse_EventTypeAndRetry_ResetBetweenEvents()
+        {
+            using var response = StreamFrom("event: first\nretry: 100\ndata: a\n\ndata: b\n\n");
+
+            var events = await Collect(response.ReadServerSentEventsAsync());
+
+            Assert.Equal("first", events[0].EventType);
+            Assert.Equal(100, events[0].Retry);
+            Assert.Null(events[1].EventType); // event type does not carry forward
+            Assert.Null(events[1].Retry);
+        }
+
+        [Fact]
+        public async Task Sse_DoneSentinel_PassesThroughAsData()
+        {
+            // The library must NOT special-case [DONE]; the caller handles it.
+            using var response = StreamFrom("data: {\"x\":1}\n\ndata: [DONE]\n\n");
+
+            var events = await Collect(response.ReadServerSentEventsAsync());
+
+            Assert.Equal(2, events.Count);
+            Assert.Equal("{\"x\":1}", events[0].Data);
+            Assert.Equal("[DONE]", events[1].Data);
+        }
+
+        [Fact]
+        public async Task Sse_EventWithoutData_IsNotDispatched()
+        {
+            using var response = StreamFrom("event: ping\n\ndata: real\n\n");
+
+            var events = await Collect(response.ReadServerSentEventsAsync());
+
+            // Per spec, an event with no data is not dispatched.
+            var sse = Assert.Single(events);
+            Assert.Equal("real", sse.Data);
+        }
+
+        [Fact]
+        public async Task Sse_IncompleteTrailingEvent_IsNotDispatched()
+        {
+            using var response = StreamFrom("data: complete\n\ndata: incomplete");
+
+            var events = await Collect(response.ReadServerSentEventsAsync());
+
+            Assert.Equal("complete", Assert.Single(events).Data);
+        }
+
+        [Fact]
+        public async Task Sse_PreCancelledToken_Throws()
+        {
+            using var response = StreamFrom("data: a\n\n");
+            using var cts = new CancellationTokenSource();
+            cts.Cancel();
+
+            await Assert.ThrowsAnyAsync<System.OperationCanceledException>(
+                async () => await Collect(response.ReadServerSentEventsAsync(cts.Token)));
+        }
+
+        [Fact]
+        public async Task Sse_EndToEnd_OverMakeStreamRequest()
+        {
+            var server = WireMockServer.Start();
+            server.Given(Request.Create().WithPath("/sse").UsingGet())
+                .RespondWith(Response.Create()
+                    .WithStatusCode(HttpStatusCode.OK)
+                    .WithHeader("Content-Type", "text/event-stream")
+                    .WithBody("data: one\n\ndata: two\n\ndata: [DONE]\n\n"));
+
+            var client = new SimpleClient(server.Url);
+
+            var datas = new List<string>();
+            using (var response = await client.MakeStreamRequest(new SimpleRequest("/sse")))
+            {
+                await foreach (var sse in response.ReadServerSentEventsAsync())
+                {
+                    if (sse.Data == "[DONE]")
+                    {
+                        break;
+                    }
+
+                    datas.Add(sse.Data);
+                }
+            }
+
+            Assert.Equal(new[] { "one", "two" }, datas);
+
+            server.Stop();
+        }
+
+        [Fact]
+        public async Task ReadServerSentEventsAsync_IsMockable()
+        {
+            // The readers are interface members (not extension methods) specifically so callers
+            // can mock them. This test fails to compile if they ever become extension methods.
+            var events = new[] { new ServerSentEvent("mocked") };
+            var mock = new Mock<ISimpleStreamResponse>();
+            mock.Setup(x => x.ReadServerSentEventsAsync(It.IsAny<CancellationToken>()))
+                .Returns(ToAsync(events));
+
+            var result = await Collect(mock.Object.ReadServerSentEventsAsync());
+
+            Assert.Equal("mocked", Assert.Single(result).Data);
+        }
+
+        // ---- helpers ----
+
+        private static async IAsyncEnumerable<T> ToAsync<T>(IEnumerable<T> items)
+        {
+            foreach (var item in items)
+            {
+                yield return item;
+            }
+
+            await Task.CompletedTask;
+        }
+
+        private static ISimpleStreamResponse StreamFrom(string body, string contentType = "text/event-stream")
+        {
+            var stream = new MemoryStream(Encoding.UTF8.GetBytes(body));
+            var response = new SimpleStreamResponse(new HttpResponseMessage(HttpStatusCode.OK), stream)
+            {
+                StatusCode = HttpStatusCode.OK,
+                IsSuccessful = true,
+            };
+
+            if (contentType != null)
+            {
+                response.Headers["Content-Type"] = contentType;
+            }
+
+            return response;
+        }
+
+        private static async Task<List<T>> Collect<T>(IAsyncEnumerable<T> sequence)
+        {
+            var list = new List<T>();
+            await foreach (var item in sequence)
+            {
+                list.Add(item);
+            }
+
+            return list;
+        }
+    }
+}

src/SimpleHttpClient/Models/ISimpleStreamResponse.cs

@@ -2,6 +2,7 @@
 using System.Collections.Generic;
 using System.IO;
 using System.Net;
+using System.Threading;
 
 namespace SimpleHttpClient.Models
 {
@@ -32,5 +33,29 @@ public interface ISimpleStreamResponse : IDisposable
         /// Read from this directly to consume the response as it arrives.
         /// </summary>
         Stream Body { get; }
+
+        /// <summary>
+        /// Read the response body as a sequence of text lines as they arrive. Lines are split on
+        /// CR, LF, or CRLF. The encoding is taken from the response's Content-Type charset, falling
+        /// back to UTF-8.
+        /// </summary>
+        /// <param name="cancellationToken">A token to cancel reading (observed mid-read, not just between lines).</param>
+        /// <returns>An async sequence of lines (without their line terminators).</returns>
+        IAsyncEnumerable<string> ReadLinesAsync(CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Read the response body as a sequence of Server-Sent Events (<c>text/event-stream</c>),
+        /// parsing the SSE wire format per the WHATWG/W3C specification: events are separated by
+        /// blank lines, multiple <c>data:</c> lines are joined with newlines, and lines beginning
+        /// with a colon (comments / keep-alives) are ignored.
+        /// <para>
+        /// Application-specific conventions are intentionally NOT handled here - for example a
+        /// sentinel <c>data</c> value that marks the end of the stream, or deserializing
+        /// <see cref="ServerSentEvent.Data"/>. Inspect each event and handle those in your own loop.
+        /// </para>
+        /// </summary>
+        /// <param name="cancellationToken">A token to cancel reading.</param>
+        /// <returns>An async sequence of <see cref="ServerSentEvent"/>.</returns>
+        IAsyncEnumerable<ServerSentEvent> ReadServerSentEventsAsync(CancellationToken cancellationToken = default);
     }
 }