Add SseFormatter

src/libraries/System.Net.ServerSentEvents/ref/System.Net.ServerSentEvents.cs

@@ -6,15 +6,22 @@
 
 namespace System.Net.ServerSentEvents
 {
+    public static partial class SseFormatter
+    {
+        public static System.Threading.Tasks.Task WriteAsync(System.Collections.Generic.IAsyncEnumerable<System.Net.ServerSentEvents.SseItem<string>> source, System.IO.Stream destination, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
+        public static System.Threading.Tasks.Task WriteAsync<T>(System.Collections.Generic.IAsyncEnumerable<System.Net.ServerSentEvents.SseItem<T>> source, System.IO.Stream destination, System.Action<System.Net.ServerSentEvents.SseItem<T>, System.Buffers.IBufferWriter<byte>> itemFormatter, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
+    }
     public delegate T SseItemParser<out T>(string eventType, System.ReadOnlySpan<byte> data);
     public readonly partial struct SseItem<T>
     {
         private readonly T _Data_k__BackingField;
         private readonly object _dummy;
         private readonly int _dummyPrimitive;
-        public SseItem(T data, string? eventType) { throw null; }
+        public SseItem(T data, string? eventType = null) { throw null; }
         public T Data { get { throw null; } }
+        public string? EventId { get { throw null; } init { } }
         public string EventType { get { throw null; } }
+        public System.TimeSpan? ReconnectionInterval { get { throw null; } init { } }
     }
     public static partial class SseParser
     {

src/libraries/System.Net.ServerSentEvents/ref/System.Net.ServerSentEvents.csproj

@@ -8,6 +8,10 @@
     <Compile Include="System.Net.ServerSentEvents.cs" />
   </ItemGroup>
 
+  <ItemGroup Condition="'$(TargetFrameworkIdentifier)' != '.NETCoreApp'">
+    <Compile Include="$(CoreLibSharedDir)System\Runtime\CompilerServices\IsExternalInit.cs" />
+  </ItemGroup>
+
   <ItemGroup Condition="'$(TargetFrameworkIdentifier)' != '.NETCoreApp'">
     <PackageReference Include="System.Memory" Version="$(SystemMemoryVersion)" />
     <ProjectReference Include="$(LibrariesProjectRoot)Microsoft.Bcl.AsyncInterfaces\src\Microsoft.Bcl.AsyncInterfaces.csproj" />

src/libraries/System.Net.ServerSentEvents/src/Resources/Strings.resx

@@ -120,4 +120,10 @@
   <data name="InvalidOperation_EnumerateOnlyOnce" xml:space="preserve">
     <value>The enumerable may be enumerated only once.</value>
   </data>
+  <data name="ArgumentException_CannotContainLineBreaks" xml:space="preserve">
+    <value>The argument cannot contain line breaks.</value>
+  </data>
+  <data name="ArgumentException_CannotBeNegative" xml:space="preserve">
+    <value>The argument cannot be a negative value.</value>
+  </data>
 </root>

src/libraries/System.Net.ServerSentEvents/src/System.Net.ServerSentEvents.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
     <TargetFrameworks>$(NetCoreAppCurrent);$(NetCoreAppPrevious);$(NetCoreAppMinimum);netstandard2.0;$(NetFrameworkMinimum)</TargetFrameworks>
@@ -11,10 +11,19 @@ System.Net.ServerSentEvents.SseParser</PackageDescription>
   </PropertyGroup>
 
   <ItemGroup>
+    <Compile Include="$(CommonPath)System\Net\ArrayBuffer.cs" Link="ProductionCode\Common\System\Net\ArrayBuffer.cs" />
+    <Compile Include="System\Net\ServerSentEvents\Helpers.cs" />
+    <Compile Include="System\Net\ServerSentEvents\PooledByteBufferWriter.cs" />
+    <Compile Include="System\Net\ServerSentEvents\SseFormatter.cs" />
     <Compile Include="System\Net\ServerSentEvents\SseParser_1.cs" />
     <Compile Include="System\Net\ServerSentEvents\SseItem.cs" />
     <Compile Include="System\Net\ServerSentEvents\SseItemParser.cs" />
     <Compile Include="System\Net\ServerSentEvents\SseParser.cs" />
+    <Compile Include="System\Net\ServerSentEvents\ThrowHelper.cs" />
+  </ItemGroup>
+
+  <ItemGroup Condition="'$(TargetFrameworkIdentifier)' != '.NETCoreApp'">
+    <Compile Include="$(CoreLibSharedDir)System\Runtime\CompilerServices\IsExternalInit.cs" />
   </ItemGroup>
 
   <ItemGroup Condition="'$(TargetFrameworkIdentifier)' != '.NETCoreApp'">

src/libraries/System.Net.ServerSentEvents/src/System/Net/ServerSentEvents/Helpers.cs

@@ -0,0 +1,100 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+using System.Buffers;
+using System.Diagnostics;
+using System.Globalization;
+using System.IO;
+using System.Runtime.InteropServices;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace System.Net.ServerSentEvents
+{
+    internal static class Helpers
+    {
+        public static void WriteUtf8Number(this IBufferWriter<byte> writer, long value)
+        {
+#if NET
+            const int MaxDecimalDigits = 20;
+            Span<byte> buffer = writer.GetSpan(MaxDecimalDigits);
+            Debug.Assert(MaxDecimalDigits <= buffer.Length);
+
+            bool success = value.TryFormat(buffer, out int bytesWritten, provider: CultureInfo.InvariantCulture);
+            Debug.Assert(success);
+            writer.Advance(bytesWritten);
+#else
+            writer.WriteUtf8String(value.ToString(CultureInfo.InvariantCulture));
+#endif
+        }
+
+        public static void WriteUtf8String(this IBufferWriter<byte> writer, ReadOnlySpan<byte> value)
+        {
+            if (value.IsEmpty)
+            {
+                return;
+            }
+
+            Span<byte> buffer = writer.GetSpan(value.Length);
+            Debug.Assert(value.Length <= buffer.Length);
+            value.CopyTo(buffer);
+            writer.Advance(value.Length);
+        }
+
+        public static unsafe void WriteUtf8String(this IBufferWriter<byte> writer, ReadOnlySpan<char> value)
+        {
+            if (value.IsEmpty)
+            {
+                return;
+            }
+
+            int maxByteCount = Encoding.UTF8.GetMaxByteCount(value.Length);
+            Span<byte> buffer = writer.GetSpan(maxByteCount);
+            Debug.Assert(maxByteCount <= buffer.Length);
+            int bytesWritten;
+#if NET
+            bytesWritten = Encoding.UTF8.GetBytes(value, buffer);
+#else
+            fixed (char* chars = value)
+            fixed (byte* bytes = buffer)
+            {
+                bytesWritten = Encoding.UTF8.GetBytes(chars, value.Length, bytes, maxByteCount);
+            }
+#endif
+            writer.Advance(bytesWritten);
+        }
+
+        public static bool ContainsLineBreaks(this ReadOnlySpan<char> text) =>
+            text.IndexOfAny('\r', '\n') >= 0;
+
+#if !NET
+
+        public static ValueTask WriteAsync(this Stream stream, ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
+        {
+            if (MemoryMarshal.TryGetArray(buffer, out ArraySegment<byte> segment))
+            {
+                return new ValueTask(stream.WriteAsync(segment.Array, segment.Offset, segment.Count, cancellationToken));
+            }
+            else
+            {
+                return WriteAsyncUsingPooledBuffer(stream, buffer, cancellationToken);
+
+                static async ValueTask WriteAsyncUsingPooledBuffer(Stream stream, ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken)
+                {
+                    byte[] sharedBuffer = ArrayPool<byte>.Shared.Rent(buffer.Length);
+                    buffer.Span.CopyTo(sharedBuffer);
+                    try
+                    {
+                        await stream.WriteAsync(sharedBuffer, 0, buffer.Length, cancellationToken).ConfigureAwait(false);
+                    }
+                    finally
+                    {
+                        ArrayPool<byte>.Shared.Return(sharedBuffer);
+                    }
+                }
+            }
+        }
+#endif
+    }
+}

src/libraries/System.Net.ServerSentEvents/src/System/Net/ServerSentEvents/PooledByteBufferWriter.cs

@@ -0,0 +1,33 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+using System.Buffers;
+using System.Diagnostics;
+
+namespace System.Net.ServerSentEvents
+{
+    internal sealed class PooledByteBufferWriter : IBufferWriter<byte>, IDisposable
+    {
+        private ArrayBuffer _buffer = new(initialSize: 256, usePool: true);
+
+        public void Advance(int count) => _buffer.Commit(count);
+
+        public Memory<byte> GetMemory(int sizeHint = 0)
+        {
+            _buffer.EnsureAvailableSpace(sizeHint);
+            return _buffer.AvailableMemory;
+        }
+
+        public Span<byte> GetSpan(int sizeHint = 0)
+        {
+            _buffer.EnsureAvailableSpace(sizeHint);
+            return _buffer.AvailableSpan;
+        }
+
+        public ReadOnlyMemory<byte> WrittenMemory => _buffer.ActiveMemory;
+        public int Capacity => _buffer.Capacity;
+        public int WrittenCount => _buffer.ActiveLength;
+        public void Reset() => _buffer.Discard(_buffer.ActiveLength);
+        public void Dispose() => _buffer.Dispose();
+    }
+}

src/libraries/System.Net.ServerSentEvents/src/System/Net/ServerSentEvents/SseFormatter.cs

@@ -0,0 +1,166 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+using System.Buffers;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace System.Net.ServerSentEvents
+{
+    /// <summary>
+    /// Provides methods for formatting server-sent events.
+    /// </summary>
+    public static class SseFormatter
+    {
+        private static readonly byte[] s_newLine = "\n"u8.ToArray();
+
+        /// <summary>
+        /// Writes the <paramref name="source"/> of server-sent events to the <paramref name="destination"/> stream.
+        /// </summary>
+        /// <param name="source">The events to write to the stream.</param>
+        /// <param name="destination">The destination stream to write the events.</param>
+        /// <param name="cancellationToken">The <see cref="CancellationToken"/> that can be used to cancel the write operation.</param>
+        /// <returns>A task that represents the asynchronous write operation.</returns>
+        public static Task WriteAsync(IAsyncEnumerable<SseItem<string>> source, Stream destination, CancellationToken cancellationToken = default)
+        {
+            if (source is null)
+            {
+                ThrowHelper.ThrowArgumentNullException(nameof(source));
+            }
+
+            if (destination is null)
+            {
+                ThrowHelper.ThrowArgumentNullException(nameof(destination));
+            }
+
+            return WriteAsyncCore(source, destination, static (item, writer) => writer.WriteUtf8String(item.Data), cancellationToken);
+        }
+
+        /// <summary>
+        /// Writes the <paramref name="source"/> of server-sent events to the <paramref name="destination"/> stream.
+        /// </summary>
+        /// <typeparam name="T">The data type of the event.</typeparam>
+        /// <param name="source">The events to write to the stream.</param>
+        /// <param name="destination">The destination stream to write the events.</param>
+        /// <param name="itemFormatter">The formatter for the data field of given event.</param>
+        /// <param name="cancellationToken">The <see cref="CancellationToken"/> that can be used to cancel the write operation.</param>
+        /// <returns>A task that represents the asynchronous write operation.</returns>
+        public static Task WriteAsync<T>(IAsyncEnumerable<SseItem<T>> source, Stream destination, Action<SseItem<T>, IBufferWriter<byte>> itemFormatter, CancellationToken cancellationToken = default)
+        {
+            if (source is null)
+            {
+                ThrowHelper.ThrowArgumentNullException(nameof(source));
+            }
+
+            if (destination is null)
+            {
+                ThrowHelper.ThrowArgumentNullException(nameof(destination));
+            }
+
+            if (itemFormatter is null)
+            {
+                ThrowHelper.ThrowArgumentNullException(nameof(itemFormatter));
+            }
+
+            return WriteAsyncCore(source, destination, itemFormatter, cancellationToken);
+        }
+
+        private static async Task WriteAsyncCore<T>(IAsyncEnumerable<SseItem<T>> source, Stream destination, Action<SseItem<T>, IBufferWriter<byte>> itemFormatter, CancellationToken cancellationToken)
+        {
+            using PooledByteBufferWriter bufferWriter = new();
+            using PooledByteBufferWriter userDataBufferWriter = new();
+
+            await foreach (SseItem<T> item in source.WithCancellation(cancellationToken).ConfigureAwait(false))
+            {
+                itemFormatter(item, userDataBufferWriter);
+
+                FormatSseEvent(
+                    bufferWriter,
+                    eventType: item._eventType, // Do not use the public property since it normalizes to "message" if null
+                    data: userDataBufferWriter.WrittenMemory.Span,
+                    eventId: item.EventId,
+                    reconnectionInterval: item.ReconnectionInterval);
+
+                await destination.WriteAsync(bufferWriter.WrittenMemory, cancellationToken).ConfigureAwait(false);
+
+                userDataBufferWriter.Reset();
+                bufferWriter.Reset();
+            }
+        }
+
+        private static void FormatSseEvent(
+            PooledByteBufferWriter bufferWriter,
+            string? eventType,
+            ReadOnlySpan<byte> data,
+            string? eventId,
+            TimeSpan? reconnectionInterval)
+        {
+            Debug.Assert(bufferWriter.WrittenCount is 0);
+
+            if (eventType is not null)
+            {
+                Debug.Assert(!eventType.ContainsLineBreaks());
+
+                bufferWriter.WriteUtf8String("event: "u8);
+                bufferWriter.WriteUtf8String(eventType);
+                bufferWriter.WriteUtf8String(s_newLine);
+            }
+
+            WriteLinesWithPrefix(bufferWriter, prefix: "data: "u8, data);
+            bufferWriter.Write(s_newLine);
+
+            if (eventId is not null)
+            {
+                Debug.Assert(!eventId.ContainsLineBreaks());
+
+                bufferWriter.WriteUtf8String("id: "u8);
+                bufferWriter.WriteUtf8String(eventId);
+                bufferWriter.WriteUtf8String(s_newLine);
+            }
+
+            if (reconnectionInterval is { } retry)
+            {
+                Debug.Assert(retry >= TimeSpan.Zero);
+
+                bufferWriter.WriteUtf8String("retry: "u8);
+                bufferWriter.WriteUtf8Number((long)retry.TotalMilliseconds);
+                bufferWriter.WriteUtf8String(s_newLine);
+            }
+
+            bufferWriter.WriteUtf8String(s_newLine);
+        }
+
+        private static void WriteLinesWithPrefix(PooledByteBufferWriter writer, ReadOnlySpan<byte> prefix, ReadOnlySpan<byte> data)
+        {
+            // Writes a potentially multi-line string, prefixing each line with the given prefix.
+            // Both \n and \r\n sequences are normalized to \n.
+
+            while (true)
+            {
+                writer.WriteUtf8String(prefix);
+
+                int i = data.IndexOfAny((byte)'\r', (byte)'\n');
+                if (i < 0)
+                {
+                    writer.WriteUtf8String(data);
+                    return;
+                }
+
+                int lineLength = i;
+                if (data[i++] == '\r' && i < data.Length && data[i] == '\n')
+                {
+                    i++;
+                }
+
+                ReadOnlySpan<byte> nextLine = data.Slice(0, lineLength);
+                data = data.Slice(i);
+
+                writer.WriteUtf8String(nextLine);
+                writer.WriteUtf8String(s_newLine);
+            }
+        }
+    }
+}