Initial sketch of an SSE JSON serializer helper.

eng/Versions.props

@@ -61,6 +61,7 @@
     <SystemSecurityCryptographyPkcsVersion>9.0.0-rc.2.24473.5</SystemSecurityCryptographyPkcsVersion>
     <SystemSecurityCryptographyXmlVersion>9.0.0-rc.2.24473.5</SystemSecurityCryptographyXmlVersion>
     <SystemTextEncodingsWebVersion>9.0.0-rc.2.24473.5</SystemTextEncodingsWebVersion>
+    <SystemNetServerSentEventsVersion>9.0.0-rc.2.24473.5</SystemNetServerSentEventsVersion>
     <SystemNumericsTensorsVersion>9.0.0-rc.2.24473.5</SystemNumericsTensorsVersion>
     <SystemTextJsonVersion>9.0.0-rc.2.24473.5</SystemTextJsonVersion>
     <!-- Dependencies from https://github.com/aspnet/AspNetCore -->

eng/packages/TestOnly.props

@@ -16,6 +16,7 @@
     <PackageVersion Include="Polly.Testing" Version="8.4.2" />
     <PackageVersion Include="StrongNamer" Version="0.2.5" />
     <PackageVersion Include="System.Configuration.ConfigurationManager" Version="$(SystemConfigurationConfigurationManagerVersion)" />
+    <PackageVersion Include="System.Net.ServerSentEvents" Version="$(SystemNetServerSentEventsVersion)" />
     <PackageVersion Include="System.Numerics.Tensors" Version="$(SystemNumericsTensorsVersion)" />
     <PackageVersion Include="Verify.Xunit" Version="20.4.0" />
     <PackageVersion Include="Xunit.Combinatorial" Version="1.6.24" />

src/Libraries/Microsoft.Extensions.AI/Microsoft.Extensions.AI.csproj

@@ -23,6 +23,7 @@
   <PropertyGroup>
     <InjectSharedEmptyCollections>true</InjectSharedEmptyCollections>
     <InjectStringSyntaxAttributeOnLegacy>true</InjectStringSyntaxAttributeOnLegacy>
+    <InjectSharedBufferWriterPool>true</InjectSharedBufferWriterPool>
     <DisableMicrosoftExtensionsLoggingSourceGenerator>false</DisableMicrosoftExtensionsLoggingSourceGenerator>
   </PropertyGroup>
 

src/Libraries/Microsoft.Extensions.AI/Utilities/AIJsonUtilities.Sse.cs

@@ -0,0 +1,105 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.IO;
+using System.Text;
+using System.Text.Json;
+using System.Text.Json.Serialization.Metadata;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Shared.Diagnostics;
+using Microsoft.Shared.Pools;
+
+#pragma warning disable SA1114 // Parameter list should follow declaration
+
+namespace Microsoft.Extensions.AI;
+
+public static partial class AIJsonUtilities
+{
+    private static readonly byte[] _sseEventFieldPrefix = "event: "u8.ToArray();
+    private static readonly byte[] _sseDataFieldPrefix = "data: "u8.ToArray();
+    private static readonly byte[] _sseIdFieldPrefix = "id: "u8.ToArray();
+    private static readonly byte[] _sseLineBreak = Encoding.UTF8.GetBytes(Environment.NewLine);
+
+    /// <summary>
+    /// Serializes the specified server-sent events to the provided stream as JSON data.
+    /// </summary>
+    /// <typeparam name="T">Specifies the type of data payload in the event.</typeparam>
+    /// <param name="stream">The UTF-8 stream to write the server-sent events to.</param>
+    /// <param name="sseEvents">The events to serialize to the stream.</param>
+    /// <param name="options">The options configuring serialization.</param>
+    /// <param name="cancellationToken">The token taht can be used to cancel the write operation.</param>
+    /// <returns>A task representing the asynchronous write operation.</returns>
+    public static async ValueTask SerializeAsSseAsync<T>(
+        Stream stream,
+        IAsyncEnumerable<SseEvent<T>> sseEvents,
+        JsonSerializerOptions? options = null,
+        CancellationToken cancellationToken = default)
+    {
+        _ = Throw.IfNull(stream);
+        _ = Throw.IfNull(sseEvents);
+
+        options ??= DefaultOptions;
+        options.MakeReadOnly();
+        var typeInfo = (JsonTypeInfo<T>)options.GetTypeInfo(typeof(T));
+
+        BufferWriter<byte> bufferWriter = BufferWriterPool.SharedBufferWriterPool.Get();
+
+        try
+        {
+            // Build a custom Utf8JsonWriter that ignores indentation configuration from JsonSerializerOptions.
+            using Utf8JsonWriter writer = new(bufferWriter);
+
+            await foreach (var sseEvent in sseEvents.WithCancellation(cancellationToken).ConfigureAwait(false))
+            {
+                JsonSerializer.Serialize(writer, sseEvent.Data, typeInfo);
+#pragma warning disable CA1849 // Call async methods when in an async method
+                writer.Flush();
+#pragma warning restore CA1849 // Call async methods when in an async method
+                Debug.Assert(bufferWriter.WrittenSpan.IndexOf((byte)'\n') == -1, "The buffer writer should not contain any newline characters.");
+
+                if (sseEvent.EventType is { } eventType)
+                {
+                    await stream.WriteAsync(_sseEventFieldPrefix, cancellationToken).ConfigureAwait(false);
+                    await stream.WriteAsync(Encoding.UTF8.GetBytes(eventType), cancellationToken).ConfigureAwait(false);
+                    await stream.WriteAsync(_sseLineBreak, cancellationToken).ConfigureAwait(false);
+                }
+
+                await stream.WriteAsync(_sseDataFieldPrefix, cancellationToken).ConfigureAwait(false);
+                await stream.WriteAsync(
+#if NET
+                    bufferWriter.WrittenMemory,
+#else
+                    bufferWriter.WrittenMemory.ToArray(),
+#endif
+                    cancellationToken).ConfigureAwait(false);
+
+                await stream.WriteAsync(_sseLineBreak, cancellationToken).ConfigureAwait(false);
+
+                if (sseEvent.Id is { } id)
+                {
+                    await stream.WriteAsync(_sseIdFieldPrefix, cancellationToken).ConfigureAwait(false);
+                    await stream.WriteAsync(Encoding.UTF8.GetBytes(id), cancellationToken).ConfigureAwait(false);
+                    await stream.WriteAsync(_sseLineBreak, cancellationToken).ConfigureAwait(false);
+                }
+
+                await stream.WriteAsync(_sseLineBreak, cancellationToken).ConfigureAwait(false);
+
+                bufferWriter.Reset();
+                writer.Reset();
+            }
+        }
+        finally
+        {
+            BufferWriterPool.SharedBufferWriterPool.Return(bufferWriter);
+        }
+    }
+
+#if !NET
+    private static Task WriteAsync(this Stream stream, byte[] buffer, CancellationToken cancellationToken = default)
+        => stream.WriteAsync(buffer, 0, buffer.Length, cancellationToken);
+#endif
+}

src/Libraries/Microsoft.Extensions.AI/Utilities/SseEvent.cs

@@ -0,0 +1,28 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+#pragma warning disable SA1114 // Parameter list should follow declaration
+#pragma warning disable CA1815 // Override equals and operator equals on value types
+
+namespace Microsoft.Extensions.AI;
+
+/// <summary>Represents a server-sent event.</summary>
+/// <typeparam name="T">Specifies the type of data payload in the event.</typeparam>
+public readonly struct SseEvent<T>
+{
+    /// <summary>Initializes a new instance of the <see cref="SseEvent{T}"/> struct.</summary>
+    /// <param name="data">The event's payload.</param>
+    public SseEvent(T data)
+    {
+        Data = data;
+    }
+
+    /// <summary>Gets the event's payload.</summary>
+    public T Data { get; }
+
+    /// <summary>Gets the event's type.</summary>
+    public string? EventType { get; init; }
+
+    /// <summary>Gets the event's identifier.</summary>
+    public string? Id { get; init; }
+}

test/Libraries/Microsoft.Extensions.AI.Abstractions.Tests/AIJsonUtilitiesTests.cs

@@ -1,11 +1,18 @@
 // Licensed to the .NET Foundation under one or more agreements.
 // The .NET Foundation licenses this file to you under the MIT license.
 
+using System.Collections.Generic;
 using System.ComponentModel;
+using System.IO;
+using System.Net.ServerSentEvents;
+using System.Text;
 using System.Text.Json;
 using System.Text.Json.Serialization;
+using System.Threading.Tasks;
 using Xunit;
 
+#pragma warning disable S2197 // Modulus results should not be checked for direct equality
+
 namespace Microsoft.Extensions.AI;
 
 public static class AIJsonUtilitiesTests
@@ -142,4 +149,81 @@ public enum MyEnumValue
         A = 1,
         B = 2
     }
+
+    [Fact]
+    public static async Task SerializeAsSseAsync_HasExpectedOutput()
+    {
+        using MemoryStream stream = new();
+        await AIJsonUtilities.SerializeAsSseAsync(stream, CreateEvents());
+        string output = Encoding.UTF8.GetString(stream.ToArray());
+
+        Assert.Equal("""
+            data: {"value":1}
+
+            event: eventType1
+            data: {"value":2}
+
+            data: {"value":3}
+            id: 3
+
+            event: eventType2
+            data: {"value":4}
+            id: 4
+
+
+            """,
+            output);
+
+        static async IAsyncEnumerable<SseEvent<SseValue>> CreateEvents()
+        {
+            yield return new SseEvent<SseValue>(new SseValue(1));
+            yield return new SseEvent<SseValue>(new SseValue(2)) { EventType = "eventType1" };
+            await Task.CompletedTask;
+            yield return new SseEvent<SseValue>(new SseValue(3)) { Id = "3" };
+            yield return new SseEvent<SseValue>(new SseValue(4)) { Id = "4", EventType = "eventType2" };
+        }
+    }
+
+    [Fact]
+    public static async Task SerializeAsSseAsync_CanRoundtripValues()
+    {
+        using MemoryStream stream = new();
+        await AIJsonUtilities.SerializeAsSseAsync(stream, CreateEvents(100), cancellationToken: default);
+        stream.Position = 0;
+
+        var parser = SseParser.Create(stream, (_, data) => JsonSerializer.Deserialize<SseValue>(data, AIJsonUtilities.DefaultOptions)!);
+        string expectedLastEventId = "";
+        int i = 0;
+
+        foreach (var parsedEvent in parser.Enumerate())
+        {
+            Assert.NotNull(parsedEvent.Data);
+            Assert.Equal(i, parsedEvent.Data.Value);
+            Assert.Equal((i % 7) switch { 3 => "A", 5 => "B", _ => "message" }, parsedEvent.EventType);
+
+            if (i % 10 == 9)
+            {
+                expectedLastEventId = i.ToString();
+            }
+
+            Assert.Equal(expectedLastEventId, parser.LastEventId);
+            i++;
+        }
+
+        static async IAsyncEnumerable<SseEvent<SseValue>> CreateEvents(int count)
+        {
+            for (int i = 0; i < count; i++)
+            {
+                yield return new SseEvent<SseValue>(new SseValue(i))
+                {
+                    Id = i % 10 == 9 ? i.ToString() : null,
+                    EventType = (i % 7) switch { 3 => "A", 5 => "B", _ => null },
+                };
+            }
+
+            await Task.CompletedTask;
+        }
+    }
+
+    public record SseValue(int Value);
 }

test/Libraries/Microsoft.Extensions.AI.Abstractions.Tests/Microsoft.Extensions.AI.Abstractions.Tests.csproj

@@ -15,6 +15,7 @@
 
   <ItemGroup>
     <PackageReference Include="System.Memory.Data" />
+    <PackageReference Include="System.Net.ServerSentEvents" />
   </ItemGroup>
 
   <ItemGroup>