initial shift to proposed BCL api - WIP

sdk/core/System.ClientModel/src/Internal/SSE/AsyncSseItemCollection.cs

@@ -0,0 +1,105 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using System.Collections.Generic;
+using System.Formats.Sse;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace System.ClientModel.Internal;
+
+internal class AsyncSseItemCollection<T> : IAsyncEnumerable<SseItem<T>>
+{
+    private readonly Func<Task<Stream>> _getSseStreamAsync;
+    private readonly Func<string, T> _parseItem;
+    private readonly Func<SseItem<T>, bool>? _isTerminalEvent;
+
+    public AsyncSseItemCollection(Func<Task<Stream>> getSseStreamAsync,
+        Func<string, T> parseItem,
+        Func<SseItem<T>, bool>? isTerminalEvent = default)
+    {
+        _getSseStreamAsync = getSseStreamAsync;
+        _parseItem = parseItem;
+        _isTerminalEvent = isTerminalEvent;
+    }
+
+    public IAsyncEnumerator<SseItem<T>> GetAsyncEnumerator(CancellationToken cancellationToken = default)
+    {
+        return new AsyncSseItemEnumerator(_getSseStreamAsync, _parseItem, _isTerminalEvent, cancellationToken);
+    }
+
+    private sealed class AsyncSseItemEnumerator : IAsyncEnumerator<SseItem<T>>
+    {
+        private readonly Func<Task<Stream>> _getSseStreamAsync;
+        private readonly Func<string, T> _parseItem;
+        private readonly Func<SseItem<T>, bool>? _isTerminalEvent;
+        private readonly CancellationToken _cancellationToken;
+
+        private ServerSentEventReader<T>? _reader;
+        private bool _started;
+        private SseItem<T> _current;
+
+        public SseItem<T> Current => _current;
+
+        public AsyncSseItemEnumerator(Func<Task<Stream>> getSseStreamAsync,
+        Func<string, T> parseItem,
+        Func<SseItem<T>, bool>? isTerminalEvent = default,
+        CancellationToken cancellationToken = default)
+        {
+            _getSseStreamAsync = getSseStreamAsync;
+            _parseItem = parseItem;
+            _isTerminalEvent = isTerminalEvent;
+            _cancellationToken = cancellationToken;
+        }
+
+        public async ValueTask<bool> MoveNextAsync()
+        {
+            if (_reader is null && _started)
+            {
+                throw new ObjectDisposedException(nameof(AsyncSseItemCollection<T>));
+            }
+
+            // TODO: consolidate with above
+            if (_reader is null)
+            {
+                Stream stream = await _getSseStreamAsync().ConfigureAwait(false);
+                _reader = new(stream, _parseItem);
+                _started = true;
+            }
+
+            SseItem<T>? nextEvent = await _reader.TryGetNextEventAsync(_cancellationToken).ConfigureAwait(false);
+
+            if (nextEvent.HasValue)
+            {
+                if (_isTerminalEvent is not null &&
+                    _isTerminalEvent(nextEvent.Value))
+                {
+                    _current = default;
+                    return false;
+                }
+
+                _current = nextEvent.Value;
+                return true;
+            }
+
+            return false;
+        }
+
+        public async ValueTask DisposeAsync()
+        {
+            await DisposeAsyncCore().ConfigureAwait(false);
+
+            GC.SuppressFinalize(this);
+        }
+
+        private async ValueTask DisposeAsyncCore()
+        {
+            if (_reader is not null)
+            {
+                await _reader.DisposeAsync().ConfigureAwait(false);
+                _reader = null;
+            }
+        }
+    }
+}

sdk/core/System.ClientModel/src/Internal/SSE/PendingSseItem.cs

@@ -0,0 +1,66 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using System.Collections.Generic;
+using System.Diagnostics;
+
+namespace System.Formats.Sse;
+
+internal struct PendingSseItem
+{
+    private const char LF = '\n';
+
+    private List<ServerSentEventField>? _dataFields;
+
+    // TODO: I think making the fields nullable makes them take up more space?
+
+    public int DataLength { get; set; }
+    public List<ServerSentEventField> DataFields => _dataFields ??= new();
+    public ServerSentEventField? EventTypeField { get; set; }
+    public ServerSentEventField? IdField { get; set; }
+    public ServerSentEventField? RetryField { get; set; }
+
+    public SseItem<T> ToSseItem<T>(Func<string, T> itemParser)
+    {
+        SseItem<T> item = default;
+
+        // Per spec, if event type buffer is empty, set event.type to "message".
+        item.EventType = EventTypeField.HasValue ?
+            EventTypeField.Value.Value.ToString() :
+            "message";
+
+        if (IdField.HasValue)
+        {
+            item.Id = IdField.Value.Value.ToString();
+        }
+
+        if (RetryField.HasValue)
+        {
+#if NETSTANDARD2_0
+            item.ReconnectionTime = int.TryParse(RetryField.Value.Value.ToString(), out int retry) ? TimeSpan.FromMilliseconds(retry) : null;
+#else
+            item.ReconnectionTime = int.TryParse(RetryField.Value.Value.Span, out int retry) ? TimeSpan.FromMilliseconds(retry) : null;
+#endif
+        }
+
+        Debug.Assert(DataLength > 0);
+
+        Memory<char> buffer = new(new char[DataLength]);
+
+        int curr = 0;
+
+        foreach (ServerSentEventField field in DataFields)
+        {
+            Debug.Assert(field.FieldType == ServerSentEventFieldKind.Data);
+
+            field.Value.Span.CopyTo(buffer.Span.Slice(curr));
+            buffer.Span[curr + field.Value.Length] = LF;
+            curr += field.Value.Length + 1;
+        }
+
+        // remove trailing LF and parse as T.
+        item.Data = itemParser(buffer.Slice(0, buffer.Length - 1).ToString());
+
+        return item;
+    }
+}

sdk/core/System.ClientModel/src/Internal/SSE/ServerSentEventField.cs

@@ -1,7 +1,7 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License.
 
-namespace System.ClientModel.Internal;
+namespace System.Formats.Sse;
 
 // SSE specification: https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream
 internal readonly struct ServerSentEventField

sdk/core/System.ClientModel/src/Internal/SSE/ServerSentEventFieldKind.cs

@@ -1,7 +1,7 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License.
 
-namespace System.ClientModel.Internal;
+namespace System.Formats.Sse;
 
 // SSE specification: https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream
 internal enum ServerSentEventFieldKind