ClientModel: Add SSE result collection

sdk/core/System.ClientModel/api/System.ClientModel.net6.0.cs

@@ -7,6 +7,12 @@ public ApiKeyCredential(string key) { }
         public static implicit operator System.ClientModel.ApiKeyCredential (string key) { throw null; }
         public void Update(string key) { }
     }
+    public abstract partial class AsyncClientResultCollection<T> : System.ClientModel.ClientResult, System.Collections.Generic.IAsyncEnumerable<T>
+    {
+        protected internal AsyncClientResultCollection(System.ClientModel.Primitives.PipelineResponse response) : base (default(System.ClientModel.Primitives.PipelineResponse)) { }
+        public static System.ClientModel.AsyncClientResultCollection<TValue> Create<TValue>(System.ClientModel.Primitives.PipelineResponse response) where TValue : System.ClientModel.Primitives.IJsonModel<TValue> { throw null; }
+        public abstract System.Collections.Generic.IAsyncEnumerator<T> GetAsyncEnumerator(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken));
+    }
     public abstract partial class BinaryContent : System.IDisposable
     {
         protected BinaryContent() { }
@@ -20,11 +26,12 @@ protected BinaryContent() { }
     }
     public partial class ClientResult
     {
-        protected ClientResult(System.ClientModel.Primitives.PipelineResponse response) { }
+        protected ClientResult(System.ClientModel.Primitives.PipelineResponse? response) { }
         public static System.ClientModel.ClientResult<T?> FromOptionalValue<T>(T? value, System.ClientModel.Primitives.PipelineResponse response) { throw null; }
         public static System.ClientModel.ClientResult FromResponse(System.ClientModel.Primitives.PipelineResponse response) { throw null; }
         public static System.ClientModel.ClientResult<T> FromValue<T>(T value, System.ClientModel.Primitives.PipelineResponse response) { throw null; }
         public System.ClientModel.Primitives.PipelineResponse GetRawResponse() { throw null; }
+        protected void SetRawResponse(System.ClientModel.Primitives.PipelineResponse response) { }
     }
     public partial class ClientResultException : System.Exception
     {

sdk/core/System.ClientModel/api/System.ClientModel.netstandard2.0.cs

@@ -7,6 +7,12 @@ public ApiKeyCredential(string key) { }
         public static implicit operator System.ClientModel.ApiKeyCredential (string key) { throw null; }
         public void Update(string key) { }
     }
+    public abstract partial class AsyncClientResultCollection<T> : System.ClientModel.ClientResult, System.Collections.Generic.IAsyncEnumerable<T>
+    {
+        protected internal AsyncClientResultCollection(System.ClientModel.Primitives.PipelineResponse response) : base (default(System.ClientModel.Primitives.PipelineResponse)) { }
+        public static System.ClientModel.AsyncClientResultCollection<TValue> Create<TValue>(System.ClientModel.Primitives.PipelineResponse response) where TValue : System.ClientModel.Primitives.IJsonModel<TValue> { throw null; }
+        public abstract System.Collections.Generic.IAsyncEnumerator<T> GetAsyncEnumerator(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken));
+    }
     public abstract partial class BinaryContent : System.IDisposable
     {
         protected BinaryContent() { }
@@ -20,11 +26,12 @@ protected BinaryContent() { }
     }
     public partial class ClientResult
     {
-        protected ClientResult(System.ClientModel.Primitives.PipelineResponse response) { }
+        protected ClientResult(System.ClientModel.Primitives.PipelineResponse? response) { }
         public static System.ClientModel.ClientResult<T?> FromOptionalValue<T>(T? value, System.ClientModel.Primitives.PipelineResponse response) { throw null; }
         public static System.ClientModel.ClientResult FromResponse(System.ClientModel.Primitives.PipelineResponse response) { throw null; }
         public static System.ClientModel.ClientResult<T> FromValue<T>(T value, System.ClientModel.Primitives.PipelineResponse response) { throw null; }
         public System.ClientModel.Primitives.PipelineResponse GetRawResponse() { throw null; }
+        protected void SetRawResponse(System.ClientModel.Primitives.PipelineResponse response) { }
     }
     public partial class ClientResultException : System.Exception
     {

sdk/core/System.ClientModel/src/Convenience/AsyncClientResultCollectionOfT.cs

@@ -0,0 +1,27 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using System.ClientModel.Internal;
+using System.ClientModel.Primitives;
+using System.Collections.Generic;
+using System.Threading;
+
+namespace System.ClientModel;
+
+#pragma warning disable CS1591 // public XML comments
+public abstract class AsyncClientResultCollection<T> : ClientResult, IAsyncEnumerable<T>
+{
+    protected internal AsyncClientResultCollection(PipelineResponse response) : base(response)
+    {
+    }
+
+    public abstract IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default);
+
+    // TODO: take CancellationToken?
+    //public static ClientResultCollection<T> Create<TValue>(PipelineResponse response) where TValue : IJsonModel<T>
+    public static AsyncClientResultCollection<TValue> Create<TValue>(PipelineResponse response) where TValue : IJsonModel<TValue>
+    {
+        return StreamingClientResult<TValue>.Create<TValue>(response);
+    }
+}
+#pragma warning restore CS1591 // public XML comments

sdk/core/System.ClientModel/src/Convenience/ClientResult.cs

@@ -11,18 +11,16 @@ namespace System.ClientModel;
 /// </summary>
 public class ClientResult
 {
-    private readonly PipelineResponse _response;
+    private PipelineResponse? _response;
 
     /// <summary>
     /// Create a new instance of <see cref="ClientResult"/> from a service
     /// response.
     /// </summary>
     /// <param name="response">The <see cref="PipelineResponse"/> received
     /// from the service.</param>
-    protected ClientResult(PipelineResponse response)
+    protected ClientResult(PipelineResponse? response)
     {
-        Argument.AssertNotNull(response, nameof(response));
-
         _response = response;
     }
 
@@ -31,7 +29,37 @@ protected ClientResult(PipelineResponse response)
     /// </summary>
     /// <returns>the <see cref="PipelineResponse"/> received from the service.
     /// </returns>
-    public PipelineResponse GetRawResponse() => _response;
+    /// <exception cref="InvalidOperationException">No
+    /// <see cref="PipelineResponse"/> value is currently available for this
+    /// <see cref="ClientResult"/> instance.  This can happen when the instance
+    /// is a collection type like <see cref="AsyncClientResultCollection{T}"/>
+    /// that has not yet been enumerated.</exception>
+    public PipelineResponse GetRawResponse()
+    {
+        if (_response is null)
+        {
+            throw new InvalidOperationException("No response is associated " +
+                "with this result.  If the result is a collection result " +
+                "type, this may be because no request has been sent to the " +
+                "server yet.");
+        }
+
+        return _response;
+    }
+
+    /// <summary>
+    /// Update the value returned from <see cref="GetRawResponse"/>.
+    /// </summary>
+    /// <remarks>This method may be called from types derived from
+    /// <see cref="ClientResult"/> that poll the service for status updates
+    /// or to retrieve additional collection values to update the raw response
+    /// to the response most recently returned from the service.</remarks>
+    /// <param name="response">The <see cref="PipelineResponse"/> to return
+    /// from <see cref="GetRawResponse"/>.</param>
+    protected void SetRawResponse(PipelineResponse response)
+    {
+        _response = response;
+    }
 
     #region Factory methods for ClientResult and subtypes
 
@@ -44,7 +72,11 @@ protected ClientResult(PipelineResponse response)
     /// provided <paramref name="response"/>.
     /// </returns>
     public static ClientResult FromResponse(PipelineResponse response)
-        => new ClientResult(response);
+    {
+        Argument.AssertNotNull(response, nameof(response));
+
+        return new ClientResult(response);
+    }
 
     /// <summary>
     /// Creates a new instance of <see cref="ClientResult{T}"/> that holds the
@@ -60,6 +92,8 @@ public static ClientResult FromResponse(PipelineResponse response)
     /// </returns>
     public static ClientResult<T> FromValue<T>(T value, PipelineResponse response)
     {
+        Argument.AssertNotNull(response, nameof(response));
+
         if (value is null)
         {
             string message = "ClientResult<T> contract guarantees that ClientResult<T>.Value is non-null. " +
@@ -90,7 +124,11 @@ public static ClientResult<T> FromValue<T>(T value, PipelineResponse response)
     /// provided <paramref name="value"/> and <paramref name="response"/>.
     /// </returns>
     public static ClientResult<T?> FromOptionalValue<T>(T? value, PipelineResponse response)
-        => new ClientResult<T?>(value, response);
+    {
+        Argument.AssertNotNull(response, nameof(response));
+
+        return new ClientResult<T?>(value, response);
+    }
 
     #endregion
 }

sdk/core/System.ClientModel/src/Convenience/ClientResultCollectionOfT.cs

@@ -0,0 +1,31 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using System.ClientModel.Internal;
+using System.ClientModel.Primitives;
+using System.Collections;
+using System.Collections.Generic;
+
+namespace System.ClientModel;
+
+// TODO: Re-enable sync version
+
+//#pragma warning disable CS1591 // public XML comments
+//public abstract class ClientResultCollection<T> : ClientResult, IEnumerable<T>
+//{
+//    protected internal ClientResultCollection(PipelineResponse response) : base(response)
+//    {
+//    }
+
+//    public abstract IEnumerator<T> GetEnumerator();
+
+//    IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
+
+//    // TODO: take CancellationToken?
+//    //public static ClientResultCollection<T> Create<TValue>(PipelineResponse response) where TValue : IJsonModel<T>
+//    public static ClientResultCollection<TValue> Create<TValue>(PipelineResponse response) where TValue : IJsonModel<TValue>
+//    {
+//        return StreamingClientResult<TValue>.Create<TValue>(response);
+//    }
+//}
+//#pragma warning restore CS1591 // public XML comments

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

@@ -0,0 +1,67 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace System.ClientModel.Internal;
+
+internal sealed class AsyncServerSentEventEnumerator : IAsyncEnumerator<ServerSentEvent>, IDisposable, IAsyncDisposable
+{
+    // TODO: make this configurable per coming from TypeSpec
+    private static readonly ReadOnlyMemory<char> _doneToken = "[DONE]".AsMemory();
+
+    private readonly ServerSentEventReader _reader;
+    private readonly CancellationToken _cancellationToken;
+    private bool _disposedValue;
+
+    public ServerSentEvent Current { get; private set; }
+
+    public AsyncServerSentEventEnumerator(ServerSentEventReader reader, CancellationToken cancellationToken = default)
+    {
+        _reader = reader;
+        _cancellationToken = cancellationToken;
+    }
+
+    public async ValueTask<bool> MoveNextAsync()
+    {
+        ServerSentEvent? nextEvent = await _reader.TryGetNextEventAsync(_cancellationToken).ConfigureAwait(false);
+        if (nextEvent.HasValue)
+        {
+            if (nextEvent.Value.Data.Span.SequenceEqual(_doneToken.Span))
+            {
+                return false;
+            }
+            Current = nextEvent.Value;
+            return true;
+        }
+        return false;
+    }
+
+    private void Dispose(bool disposing)
+    {
+        if (!_disposedValue)
+        {
+            if (disposing)
+            {
+                _reader.Dispose();
+            }
+
+            _disposedValue = true;
+        }
+    }
+
+    public void Dispose()
+    {
+        // Do not change this code. Put cleanup code in 'Dispose(bool disposing)' method
+        Dispose(disposing: true);
+        GC.SuppressFinalize(this);
+    }
+
+    public ValueTask DisposeAsync()
+    {
+        Dispose();
+        return new ValueTask();
+    }
+}

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

@@ -0,0 +1,61 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using System.ClientModel.Primitives;
+using System.Collections.Generic;
+using System.Text.Json;
+using System.Threading.Tasks;
+
+namespace System.ClientModel.Internal;
+
+internal class AsyncServerSentEventJsonDataEnumerator<T> : IAsyncEnumerator<T>, IDisposable, IAsyncDisposable
+    where T : IJsonModel<T>
+{
+    private readonly AsyncServerSentEventEnumerator _eventEnumerator;
+
+    private T? _current;
+
+    // TODO: is null supression the correct pattern here?
+    public T Current { get => _current!; }
+
+    public AsyncServerSentEventJsonDataEnumerator(AsyncServerSentEventEnumerator eventEnumerator)
+    {
+        Argument.AssertNotNull(eventEnumerator, nameof(eventEnumerator));
+
+        _eventEnumerator = eventEnumerator;
+    }
+
+    public async ValueTask<bool> MoveNextAsync()
+    {
+        if (await _eventEnumerator.MoveNextAsync().ConfigureAwait(false))
+        {
+            using JsonDocument eventDocument = JsonDocument.Parse(_eventEnumerator.Current.Data);
+            BinaryData eventData = BinaryData.FromObjectAsJson(eventDocument.RootElement);
+            T? jsonData = ModelReaderWriter.Read<T>(eventData);
+
+            // TODO: should we stop iterating if we can't deserialize?
+            if (jsonData is null)
+            {
+                _current = default;
+                return false;
+            }
+
+            if (jsonData is T singleInstanceData)
+            {
+                _current = singleInstanceData;
+                return true;
+            }
+        }
+        return false;
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        await _eventEnumerator.DisposeAsync().ConfigureAwait(false);
+    }
+
+    public void Dispose()
+    {
+        _eventEnumerator.Dispose();
+    }
+}