ClientModel: AsyncResultCollection<T> and SSE event collection implementation

sdk/core/Azure.Core.TestFramework/src/MockJsonModel.cs

@@ -11,7 +11,7 @@ namespace Azure.Core.TestFramework
 {
     public class MockJsonModel : IJsonModel<MockJsonModel>
     {
-        internal MockJsonModel()
+        public MockJsonModel()
         {
         }
 

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() : base (default(System.ClientModel.Primitives.PipelineResponse)) { }
+        protected internal AsyncClientResultCollection(System.ClientModel.Primitives.PipelineResponse response) : base (default(System.ClientModel.Primitives.PipelineResponse)) { }
+        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,19 @@ 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 abstract partial class ClientResultCollection<T> : System.ClientModel.ClientResult, System.Collections.Generic.IEnumerable<T>, System.Collections.IEnumerable
+    {
+        protected internal ClientResultCollection() : base (default(System.ClientModel.Primitives.PipelineResponse)) { }
+        protected internal ClientResultCollection(System.ClientModel.Primitives.PipelineResponse response) : base (default(System.ClientModel.Primitives.PipelineResponse)) { }
+        public abstract System.Collections.Generic.IEnumerator<T> GetEnumerator();
+        System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator() { throw null; }
     }
     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() : base (default(System.ClientModel.Primitives.PipelineResponse)) { }
+        protected internal AsyncClientResultCollection(System.ClientModel.Primitives.PipelineResponse response) : base (default(System.ClientModel.Primitives.PipelineResponse)) { }
+        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,19 @@ 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 abstract partial class ClientResultCollection<T> : System.ClientModel.ClientResult, System.Collections.Generic.IEnumerable<T>, System.Collections.IEnumerable
+    {
+        protected internal ClientResultCollection() : base (default(System.ClientModel.Primitives.PipelineResponse)) { }
+        protected internal ClientResultCollection(System.ClientModel.Primitives.PipelineResponse response) : base (default(System.ClientModel.Primitives.PipelineResponse)) { }
+        public abstract System.Collections.Generic.IEnumerator<T> GetEnumerator();
+        System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator() { throw null; }
     }
     public partial class ClientResultException : System.Exception
     {

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

@@ -0,0 +1,29 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+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>
+{
+    // Constructor overload for collection implementations that postpone
+    // sending a request until GetAsyncEnumerator is called. This will typically
+    // be used by collections returned from client convenience methods.
+    protected internal AsyncClientResultCollection() : base(default)
+    {
+    }
+
+    // Constructor overload for collection implementations where the service
+    // has returned a response.  This will typically be used by collections
+    // created from the return result of a client's protocol method.
+    protected internal AsyncClientResultCollection(PipelineResponse response) : base(response)
+    {
+    }
+
+    public abstract IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default);
+}
+#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,39 @@ 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)
+    {
+        Argument.AssertNotNull(response, nameof(response));
+
+        _response = response;
+    }
 
     #region Factory methods for ClientResult and subtypes
 
@@ -44,7 +74,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 +94,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 +126,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.Primitives;
+using System.Collections;
+using System.Collections.Generic;
+
+namespace System.ClientModel;
+
+#pragma warning disable CS1591 // public XML comments
+public abstract class ClientResultCollection<T> : ClientResult, IEnumerable<T>
+{
+    // Constructor overload for collection implementations that postpone
+    // sending a request until GetAsyncEnumerator is called. This will typically
+    // be used by collections returned from client convenience methods.
+    protected internal ClientResultCollection() : base(default)
+    {
+    }
+
+    // Constructor overload for collection implementations where the service
+    // has returned a response.  This will typically be used by collections
+    // created from the return result of a client's protocol method.
+    protected internal ClientResultCollection(PipelineResponse response) : base(response)
+    {
+    }
+
+    public abstract IEnumerator<T> GetEnumerator();
+
+    IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
+}
+#pragma warning restore CS1591 // public XML comments

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

@@ -0,0 +1,64 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using System.Collections.Generic;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace System.ClientModel.Internal;
+
+internal class AsyncServerSentEventEnumerable : IAsyncEnumerable<ServerSentEvent>
+{
+    private readonly Stream _contentStream;
+
+    public AsyncServerSentEventEnumerable(Stream contentStream)
+    {
+        _contentStream = contentStream;
+    }
+
+    public IAsyncEnumerator<ServerSentEvent> GetAsyncEnumerator(CancellationToken cancellationToken = default)
+    {
+        return new AsyncServerSentEventEnumerator(_contentStream, cancellationToken);
+    }
+
+    private sealed class AsyncServerSentEventEnumerator : IAsyncEnumerator<ServerSentEvent>
+    {
+        private readonly CancellationToken _cancellationToken;
+        private readonly ServerSentEventReader _reader;
+
+        public ServerSentEvent Current { get; private set; }
+
+        public AsyncServerSentEventEnumerator(Stream contentStream, CancellationToken cancellationToken = default)
+        {
+            _reader = new(contentStream);
+            _cancellationToken = cancellationToken;
+        }
+
+        public async ValueTask<bool> MoveNextAsync()
+        {
+            ServerSentEvent? nextEvent = await _reader.TryGetNextEventAsync(_cancellationToken).ConfigureAwait(false);
+
+            if (nextEvent.HasValue)
+            {
+                Current = nextEvent.Value;
+                return true;
+            }
+
+            Current = default;
+            return false;
+        }
+
+        public ValueTask DisposeAsync()
+        {
+            // The creator of the enumerable has responsibility for disposing
+            // the content stream passed to the enumerable constructor.
+
+#if NET6_0_OR_GREATER
+            return ValueTask.CompletedTask;
+#else
+            return new ValueTask();
+#endif
+        }
+    }
+}

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

@@ -0,0 +1,30 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+namespace System.ClientModel.Internal;
+
+// SSE specification: https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream
+internal readonly struct ServerSentEvent
+{
+    // Gets the value of the SSE "event type" buffer, used to distinguish between event kinds.
+    public string EventType { get; }
+
+    // Gets the value of the SSE "data" buffer, which holds the payload of the server-sent event.
+    public string Data { get; }
+
+    // Gets the value of the "last event ID" buffer, with which a user agent can reestablish a session.
+    public string? Id { get; }
+
+    // If present, gets the defined "retry" value for the event, which represents the delay before reconnecting.
+    public TimeSpan? ReconnectionTime { get; }
+
+    public ServerSentEvent(string type, string data, string? id, string? retry)
+    {
+        EventType = type;
+        Data = data;
+        Id = id;
+        ReconnectionTime = retry is null ?
+            default :
+            int.TryParse(retry, out int time) ? TimeSpan.FromMilliseconds(time) : null;
+    }
+}

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

@@ -0,0 +1,64 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using System.Collections;
+using System.Collections.Generic;
+using System.IO;
+
+namespace System.ClientModel.Internal;
+
+internal class ServerSentEventEnumerable : IEnumerable<ServerSentEvent>
+{
+    private readonly Stream _contentStream;
+
+    public ServerSentEventEnumerable(Stream contentStream)
+    {
+        _contentStream = contentStream;
+    }
+
+    public IEnumerator<ServerSentEvent> GetEnumerator()
+    {
+        return new ServerSentEventEnumerator(_contentStream);
+    }
+
+    IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
+
+    private sealed class ServerSentEventEnumerator : IEnumerator<ServerSentEvent>
+    {
+        private readonly ServerSentEventReader _reader;
+
+        public ServerSentEventEnumerator(Stream contentStream)
+        {
+            _reader = new(contentStream);
+        }
+
+        public ServerSentEvent Current { get; private set; }
+
+        object IEnumerator.Current => Current;
+
+        public bool MoveNext()
+        {
+            ServerSentEvent? nextEvent = _reader.TryGetNextEvent();
+
+            if (nextEvent.HasValue)
+            {
+                Current = nextEvent.Value;
+                return true;
+            }
+
+            Current = default;
+            return false;
+        }
+
+        public void Reset()
+        {
+            throw new NotSupportedException("Cannot seek back in an SSE stream.");
+        }
+
+        public void Dispose()
+        {
+            // The creator of the enumerable has responsibility for disposing
+            // the content stream passed to the enumerable constructor.
+        }
+    }
+}