feat: Simplify SSE formatting (#18)

Author: stephentoub

Directory.Packages.props

@@ -3,13 +3,14 @@
     <!-- Enable central package management, https://learn.microsoft.com/en-us/nuget/consume-packages/Central-Package-Management -->
     <ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
   </PropertyGroup>
+  
   <ItemGroup>
-
     <!-- Product dependencies netstandard -->
     <PackageVersion Include="System.Text.Json" Version="8.0.5" />
 
     <!-- Product dependencies shared -->
-    <PackageVersion Include="System.Net.ServerSentEvents" Version="9.0.6" />
+    <PackageVersion Include="System.Net.ServerSentEvents" Version="10.0.0-preview.5.25277.114" />
+    <PackageVersion Include="System.Linq.AsyncEnumerable" Version="10.0.0-preview.5.25277.114" />
 
     <!-- Testing / samples dependencies -->
     <PackageVersion Include="JsonSchema.Net" Version="7.3.4" />

src/A2A.AspNetCore/A2AJsonRpcProcessor.cs

@@ -1,7 +1,7 @@
 using Microsoft.AspNetCore.Http;
 using System.Diagnostics;
-using System.IO.Pipelines;
-using System.Text;
+using System.Net.ServerSentEvents;
+using System.Text.Encodings.Web;
 using System.Text.Json;
 
 namespace A2A.AspNetCore;
@@ -192,28 +192,15 @@ public async Task ExecuteAsync(HttpContext httpContext)
         httpContext.Response.ContentType = "text/event-stream";
         httpContext.Response.Headers.Append("Cache-Control", "no-cache");
 
-        await foreach (var taskEvent in _events)
-        {
-            var sseItem = new A2ASseItem()
-            {
-                Data = JsonRpcResponse.CreateJsonRpcResponse(requestId, taskEvent),
-            };
-            await sseItem.WriteAsync(httpContext.Response.BodyWriter);
-        }
-    }
-}
-
-public class A2ASseItem
-{
-    public JsonRpcResponse? Data { get; set; }
-
-    public async Task WriteAsync(PipeWriter writer)
-    {
-        if (Data != null)
-        {
-            string json = JsonSerializer.Serialize(Data, A2AJsonUtilities.DefaultOptions.GetTypeInfo(typeof(JsonRpcResponse)));
-            await writer.WriteAsync(Encoding.UTF8.GetBytes($"data: {json}\n\n"));
-            await writer.FlushAsync();
-        }
+        var responseTypeInfo = A2AJsonUtilities.DefaultOptions.GetTypeInfo(typeof(JsonRpcResponse));
+        await SseFormatter.WriteAsync(
+            _events.Select(e => new SseItem<JsonRpcResponse>(JsonRpcResponse.CreateJsonRpcResponse(requestId, e))),
+            httpContext.Response.Body,
+            (item, writer) =>
+            {
+                using Utf8JsonWriter json = new(writer, new() { Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping });
+                JsonSerializer.Serialize(json, item.Data, responseTypeInfo);
+            },
+            httpContext.RequestAborted);
     }
 }

src/A2A/A2A.csproj

@@ -23,6 +23,7 @@
   <!-- Dependencies needed by all -->
   <ItemGroup>
     <PackageReference Include="Microsoft.Extensions.Logging" />
+    <PackageReference Include="System.Linq.AsyncEnumerable" />
     <PackageReference Include="System.Net.ServerSentEvents" />
   </ItemGroup>