Streamable HTTP resumability + redelivery + SSE polling via server-side disconnect

src/Common/ServerSentEvents/ArrayBuffer.cs

@@ -0,0 +1,198 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+// Copied from https://github.com/dotnet/runtime/blob/main/src/libraries/Common/src/System/Net/ArrayBuffer.cs
+
+using System.Buffers;
+using System.Diagnostics;
+using System.Runtime.CompilerServices;
+using System.Runtime.InteropServices;
+
+namespace System.Net.ServerSentEvents;
+
+// Warning: Mutable struct!
+// The purpose of this struct is to simplify buffer management.
+// It manages a sliding buffer where bytes can be added at the end and removed at the beginning.
+// [ActiveSpan/Memory] contains the current buffer contents; these bytes will be preserved
+// (copied, if necessary) on any call to EnsureAvailableBytes.
+// [AvailableSpan/Memory] contains the available bytes past the end of the current content,
+// and can be written to in order to add data to the end of the buffer.
+// Commit(byteCount) will extend the ActiveSpan by [byteCount] bytes into the AvailableSpan.
+// Discard(byteCount) will discard [byteCount] bytes as the beginning of the ActiveSpan.
+
+[StructLayout(LayoutKind.Auto)]
+internal struct ArrayBuffer : IDisposable
+{
+#if NET
+    private static int ArrayMaxLength => Array.MaxLength;
+#else
+        private const int ArrayMaxLength = 0X7FFFFFC7;
+#endif
+
+    private readonly bool _usePool;
+    private byte[] _bytes;
+    private int _activeStart;
+    private int _availableStart;
+
+    // Invariants:
+    // 0 <= _activeStart <= _availableStart <= bytes.Length
+
+    public ArrayBuffer(int initialSize, bool usePool = false)
+    {
+        Debug.Assert(initialSize > 0 || usePool);
+
+        _usePool = usePool;
+        _bytes = initialSize == 0
+            ? Array.Empty<byte>()
+            : usePool ? ArrayPool<byte>.Shared.Rent(initialSize) : new byte[initialSize];
+        _activeStart = 0;
+        _availableStart = 0;
+    }
+
+    public ArrayBuffer(byte[] buffer)
+    {
+        Debug.Assert(buffer.Length > 0);
+
+        _usePool = false;
+        _bytes = buffer;
+        _activeStart = 0;
+        _availableStart = 0;
+    }
+
+    public void Dispose()
+    {
+        _activeStart = 0;
+        _availableStart = 0;
+
+        byte[] array = _bytes;
+        _bytes = null!;
+
+        if (array is not null)
+        {
+            ReturnBufferIfPooled(array);
+        }
+    }
+
+    // This is different from Dispose as the instance remains usable afterwards (_bytes will not be null).
+    public void ClearAndReturnBuffer()
+    {
+        Debug.Assert(_usePool);
+        Debug.Assert(_bytes is not null);
+
+        _activeStart = 0;
+        _availableStart = 0;
+
+        byte[] bufferToReturn = _bytes!;
+        _bytes = Array.Empty<byte>();
+        ReturnBufferIfPooled(bufferToReturn);
+    }
+
+    public int ActiveLength => _availableStart - _activeStart;
+    public Span<byte> ActiveSpan => new Span<byte>(_bytes, _activeStart, _availableStart - _activeStart);
+    public ReadOnlySpan<byte> ActiveReadOnlySpan => new ReadOnlySpan<byte>(_bytes, _activeStart, _availableStart - _activeStart);
+    public Memory<byte> ActiveMemory => new Memory<byte>(_bytes, _activeStart, _availableStart - _activeStart);
+
+    public int AvailableLength => _bytes.Length - _availableStart;
+    public Span<byte> AvailableSpan => _bytes.AsSpan(_availableStart);
+    public Memory<byte> AvailableMemory => _bytes.AsMemory(_availableStart);
+    public Memory<byte> AvailableMemorySliced(int length) => new Memory<byte>(_bytes, _availableStart, length);
+
+    public int Capacity => _bytes.Length;
+    public int ActiveStartOffset => _activeStart;
+
+    public byte[] DangerousGetUnderlyingBuffer() => _bytes;
+
+    public void Discard(int byteCount)
+    {
+        Debug.Assert(byteCount <= ActiveLength, $"Expected {byteCount} <= {ActiveLength}");
+        _activeStart += byteCount;
+
+        if (_activeStart == _availableStart)
+        {
+            _activeStart = 0;
+            _availableStart = 0;
+        }
+    }
+
+    public void Commit(int byteCount)
+    {
+        Debug.Assert(byteCount <= AvailableLength);
+        _availableStart += byteCount;
+    }
+
+    // Ensure at least [byteCount] bytes to write to.
+    [MethodImpl(MethodImplOptions.AggressiveInlining)]
+    public void EnsureAvailableSpace(int byteCount)
+    {
+        if (byteCount > AvailableLength)
+        {
+            EnsureAvailableSpaceCore(byteCount);
+        }
+    }
+
+    private void EnsureAvailableSpaceCore(int byteCount)
+    {
+        Debug.Assert(AvailableLength < byteCount);
+
+        if (_bytes.Length == 0)
+        {
+            Debug.Assert(_usePool && _activeStart == 0 && _availableStart == 0);
+            _bytes = ArrayPool<byte>.Shared.Rent(byteCount);
+            return;
+        }
+
+        int totalFree = _activeStart + AvailableLength;
+        if (byteCount <= totalFree)
+        {
+            // We can free up enough space by just shifting the bytes down, so do so.
+            Buffer.BlockCopy(_bytes, _activeStart, _bytes, 0, ActiveLength);
+            _availableStart = ActiveLength;
+            _activeStart = 0;
+            Debug.Assert(byteCount <= AvailableLength);
+            return;
+        }
+
+        int desiredSize = ActiveLength + byteCount;
+
+        if ((uint)desiredSize > ArrayMaxLength)
+        {
+            throw new OutOfMemoryException();
+        }
+
+        // Double the existing buffer size (capped at Array.MaxLength).
+        int newSize = Math.Max(desiredSize, (int)Math.Min(ArrayMaxLength, 2 * (uint)_bytes.Length));
+
+        byte[] newBytes = _usePool ?
+            ArrayPool<byte>.Shared.Rent(newSize) :
+            new byte[newSize];
+        byte[] oldBytes = _bytes;
+
+        if (ActiveLength != 0)
+        {
+            Buffer.BlockCopy(oldBytes, _activeStart, newBytes, 0, ActiveLength);
+        }
+
+        _availableStart = ActiveLength;
+        _activeStart = 0;
+
+        _bytes = newBytes;
+        ReturnBufferIfPooled(oldBytes);
+
+        Debug.Assert(byteCount <= AvailableLength);
+    }
+
+    public void Grow()
+    {
+        EnsureAvailableSpaceCore(AvailableLength + 1);
+    }
+
+    [MethodImpl(MethodImplOptions.AggressiveInlining)]
+    private void ReturnBufferIfPooled(byte[] buffer)
+    {
+        // The buffer may be Array.Empty<byte>()
+        if (_usePool && buffer.Length > 0)
+        {
+            ArrayPool<byte>.Shared.Return(buffer);
+        }
+    }
+}

src/Common/ServerSentEvents/PooledByteBufferWriter.cs

@@ -0,0 +1,35 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+// Copied from https://github.com/dotnet/runtime/blob/main/src/libraries/System.Net.ServerSentEvents/src/System/Net/ServerSentEvents/PooledByteBufferWriter.cs
+
+using System.Buffers;
+using System.Diagnostics;
+
+namespace System.Net.ServerSentEvents;
+
+internal sealed class PooledByteBufferWriter : IBufferWriter<byte>, IDisposable
+{
+    private const int MinimumBufferSize = 256;
+    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(Math.Max(sizeHint, MinimumBufferSize));
+        return _buffer.AvailableMemory;
+    }
+
+    public Span<byte> GetSpan(int sizeHint = 0)
+    {
+        _buffer.EnsureAvailableSpace(Math.Max(sizeHint, MinimumBufferSize));
+        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/Common/ServerSentEvents/SseEventWriter.cs

@@ -0,0 +1,136 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+// Based on https://github.com/dotnet/runtime/blob/main/src/libraries/System.Net.ServerSentEvents/src/System/Net/ServerSentEvents/SseFormatter.cs
+
+using System.Buffers;
+using System.Diagnostics;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace System.Net.ServerSentEvents;
+
+/// <summary>
+/// Provides methods for writing SSE events to a stream.
+/// </summary>
+internal sealed class SseEventWriter : IDisposable
+{
+    private static readonly byte[] s_newLine = "\n"u8.ToArray();
+
+    private readonly Stream _destination;
+    private readonly PooledByteBufferWriter _bufferWriter = new();
+    private readonly PooledByteBufferWriter _userDataBufferWriter = new();
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="SseEventWriter"/> class with the specified destination stream and item formatter.
+    /// </summary>
+    /// <param name="destination">The stream to write SSE events to.</param>
+    /// <exception cref="ArgumentNullException"><paramref name="destination"/> is <see langword="null"/>.</exception>
+    public SseEventWriter(Stream destination)
+    {
+        _destination = destination ?? throw new ArgumentNullException(nameof(destination));
+    }
+
+    /// <summary>
+    /// Writes an SSE item to the destination stream.
+    /// </summary>
+    /// <param name="item">The SSE item to write.</param>
+    /// <param name="itemFormatter"></param>
+    /// <param name="cancellationToken">The token to monitor for cancellation requests.</param>
+    /// <returns>A task representing the asynchronous write operation.</returns>
+    public async ValueTask WriteAsync<T>(SseItem<T> item, Action<SseItem<T>, IBufferWriter<byte>> itemFormatter, CancellationToken cancellationToken = default)
+    {
+        itemFormatter(item, _userDataBufferWriter);
+
+        FormatSseEvent(
+            _bufferWriter,
+            eventType: item.EventType,
+            data: _userDataBufferWriter.WrittenMemory.Span,
+            eventId: item.EventId,
+            reconnectionInterval: item.ReconnectionInterval);
+
+        await _destination.WriteAsync(_bufferWriter.WrittenMemory, cancellationToken).ConfigureAwait(false);
+        await _destination.FlushAsync(cancellationToken).ConfigureAwait(false);
+
+        _userDataBufferWriter.Reset();
+        _bufferWriter.Reset();
+    }
+
+    private static void FormatSseEvent(
+        IBufferWriter<byte> bufferWriter,
+        string? eventType,
+        ReadOnlySpan<byte> data,
+        string? eventId,
+        TimeSpan? reconnectionInterval)
+    {
+        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(IBufferWriter<byte> 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);
+        }
+    }
+
+    /// <inheritdoc/>
+    public void Dispose()
+    {
+        _bufferWriter.Dispose();
+        _userDataBufferWriter.Dispose();
+    }
+}