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 AsyncResultCollection<T> : System.ClientModel.ClientResult, System.Collections.Generic.IAsyncEnumerable<T>
+    {
+        protected internal AsyncResultCollection() { }
+        protected internal AsyncResultCollection(System.ClientModel.Primitives.PipelineResponse response) { }
+        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,13 @@ protected BinaryContent() { }
     }
     public partial class ClientResult
     {
+        protected ClientResult() { }
         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
     {
@@ -36,10 +44,17 @@ public ClientResultException(string message, System.ClientModel.Primitives.Pipel
     }
     public partial class ClientResult<T> : System.ClientModel.ClientResult
     {
-        protected internal ClientResult(T value, System.ClientModel.Primitives.PipelineResponse response) : base (default(System.ClientModel.Primitives.PipelineResponse)) { }
+        protected internal ClientResult(T value, System.ClientModel.Primitives.PipelineResponse response) { }
         public virtual T Value { get { throw null; } }
         public static implicit operator T (System.ClientModel.ClientResult<T> result) { throw null; }
     }
+    public abstract partial class ResultCollection<T> : System.ClientModel.ClientResult, System.Collections.Generic.IEnumerable<T>, System.Collections.IEnumerable
+    {
+        protected internal ResultCollection() { }
+        protected internal ResultCollection(System.ClientModel.Primitives.PipelineResponse response) { }
+        public abstract System.Collections.Generic.IEnumerator<T> GetEnumerator();
+        System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator() { throw null; }
+    }
 }
 namespace System.ClientModel.Primitives
 {

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 AsyncResultCollection<T> : System.ClientModel.ClientResult, System.Collections.Generic.IAsyncEnumerable<T>
+    {
+        protected internal AsyncResultCollection() { }
+        protected internal AsyncResultCollection(System.ClientModel.Primitives.PipelineResponse response) { }
+        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,13 @@ protected BinaryContent() { }
     }
     public partial class ClientResult
     {
+        protected ClientResult() { }
         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
     {
@@ -36,10 +44,17 @@ public ClientResultException(string message, System.ClientModel.Primitives.Pipel
     }
     public partial class ClientResult<T> : System.ClientModel.ClientResult
     {
-        protected internal ClientResult(T value, System.ClientModel.Primitives.PipelineResponse response) : base (default(System.ClientModel.Primitives.PipelineResponse)) { }
+        protected internal ClientResult(T value, System.ClientModel.Primitives.PipelineResponse response) { }
         public virtual T Value { get { throw null; } }
         public static implicit operator T (System.ClientModel.ClientResult<T> result) { throw null; }
     }
+    public abstract partial class ResultCollection<T> : System.ClientModel.ClientResult, System.Collections.Generic.IEnumerable<T>, System.Collections.IEnumerable
+    {
+        protected internal ResultCollection() { }
+        protected internal ResultCollection(System.ClientModel.Primitives.PipelineResponse response) { }
+        public abstract System.Collections.Generic.IEnumerator<T> GetEnumerator();
+        System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator() { throw null; }
+    }
 }
 namespace System.ClientModel.Primitives
 {

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

@@ -0,0 +1,43 @@
+// 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;
+
+/// <summary>
+/// Represents a collection of results returned from a cloud service operation.
+/// </summary>
+public abstract class AsyncResultCollection<T> : ClientResult, IAsyncEnumerable<T>
+{
+    /// <summary>
+    /// Create a new instance of <see cref="AsyncResultCollection{T}"/>.
+    /// </summary>
+    /// <remarks>If no <see cref="PipelineResponse"/> is provided when the
+    /// <see cref="ClientResult"/> instance is created, it is expected that
+    /// a derived type will call <see cref="ClientResult.SetRawResponse(PipelineResponse)"/>
+    /// prior to a user calling <see cref="ClientResult.GetRawResponse"/>.
+    /// This constructor is indended for use by collection implementations that
+    /// postpone sending a request until <see cref="GetAsyncEnumerator(CancellationToken)"/>
+    /// is called. Such implementations will typically be returned from client
+    /// convenience methods so that callers of the methods don't need to
+    /// dispose the return value. </remarks>
+    protected internal AsyncResultCollection() : base()
+    {
+    }
+
+    /// <summary>
+    /// Create a new instance of <see cref="AsyncResultCollection{T}"/>.
+    /// </summary>
+    /// <param name="response">The <see cref="PipelineResponse"/> holding the
+    /// items in the collection, or the first set of the items in the collection.
+    /// </param>
+    protected internal AsyncResultCollection(PipelineResponse response) : base(response)
+    {
+    }
+
+    /// <inheritdoc/>
+    public abstract IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default);
+}

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

@@ -11,7 +11,18 @@ namespace System.ClientModel;
 /// </summary>
 public class ClientResult
 {
-    private readonly PipelineResponse _response;
+    private PipelineResponse? _response;
+
+    /// <summary>
+    /// Create a new instance of <see cref="ClientResult"/>.
+    /// </summary>
+    /// <remarks>If no <see cref="PipelineResponse"/> is provided when the
+    /// <see cref="ClientResult"/> instance is created, it is expected that
+    /// a derived type will call <see cref="SetRawResponse(PipelineResponse)"/>
+    /// prior to a user calling <see cref="GetRawResponse"/>.</remarks>
+    protected ClientResult()
+    {
+    }
 
     /// <summary>
     /// Create a new instance of <see cref="ClientResult"/> from a service
@@ -31,7 +42,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="AsyncResultCollection{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 +87,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 +107,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 +139,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/ResultCollectionOfT.cs

@@ -0,0 +1,45 @@
+// 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;
+
+/// <summary>
+/// Represents a collection of results returned from a cloud service operation.
+/// </summary>
+public abstract class ResultCollection<T> : ClientResult, IEnumerable<T>
+{
+    /// <summary>
+    /// Create a new instance of <see cref="ResultCollection{T}"/>.
+    /// </summary>
+    /// <remarks>If no <see cref="PipelineResponse"/> is provided when the
+    /// <see cref="ClientResult"/> instance is created, it is expected that
+    /// a derived type will call <see cref="ClientResult.SetRawResponse(PipelineResponse)"/>
+    /// prior to a user calling <see cref="ClientResult.GetRawResponse"/>.
+    /// This constructor is indended for use by collection implementations that
+    /// postpone sending a request until <see cref="GetEnumerator()"/>
+    /// is called. Such implementations will typically be returned from client
+    /// convenience methods so that callers of the methods don't need to
+    /// dispose the return value. </remarks>
+    protected internal ResultCollection() : base()
+    {
+    }
+
+    /// <summary>
+    /// Create a new instance of <see cref="ResultCollection{T}"/>.
+    /// </summary>
+    /// <param name="response">The <see cref="PipelineResponse"/> holding the
+    /// items in the collection, or the first set of the items in the collection.
+    /// </param>
+    protected internal ResultCollection(PipelineResponse response) : base(response)
+    {
+    }
+
+    /// <inheritdoc/>
+    public abstract IEnumerator<T> GetEnumerator();
+
+    IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
+}

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

@@ -0,0 +1,67 @@
+// 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;
+
+/// <summary>
+/// Represents a collection of SSE events that can be enumerated as a C# async stream.
+/// </summary>
+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,32 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+namespace System.ClientModel.Internal;
+
+/// <summary>
+/// Represents an SSE event.
+/// See SSE specification: https://html.spec.whatwg.org/multipage/server-sent-events.html
+/// </summary>
+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 ? null :
+            int.TryParse(retry, out int time) ? TimeSpan.FromMilliseconds(time) : null;
+    }
+}