first steps

.dotnet/src/Custom/Chat/ChatClient.cs

@@ -40,18 +40,18 @@ public ChatClient(string model, ApiKeyCredential credential = default, OpenAICli
     /// <param name="message"> The user message to provide as a prompt for chat completion. </param>
     /// <param name="options"> Additional options for the chat completion request. </param>
     /// <returns> A result for a single chat completion. </returns>
-     public virtual ClientResult<ChatCompletion> CompleteChat(string message, ChatCompletionOptions options = null)
-         => CompleteChat(new List<ChatRequestMessage>() { new ChatRequestUserMessage(message) }, options);
+    public virtual ClientResult<ChatCompletion> CompleteChat(string message, ChatCompletionOptions options = null)
+        => CompleteChat(new List<ChatRequestMessage>() { new ChatRequestUserMessage(message) }, options);
 
     /// <summary>
     /// Generates a single chat completion result for a single, simple user message.
     /// </summary>
     /// <param name="message"> The user message to provide as a prompt for chat completion. </param>
     /// <param name="options"> Additional options for the chat completion request. </param>
     /// <returns> A result for a single chat completion. </returns>
-     public virtual Task<ClientResult<ChatCompletion>> CompleteChatAsync(string message, ChatCompletionOptions options = null)
-        => CompleteChatAsync(
-             new List<ChatRequestMessage>() { new ChatRequestUserMessage(message) }, options);
+    public virtual Task<ClientResult<ChatCompletion>> CompleteChatAsync(string message, ChatCompletionOptions options = null)
+       => CompleteChatAsync(
+            new List<ChatRequestMessage>() { new ChatRequestUserMessage(message) }, options);
 
     /// <summary>
     /// Generates a single chat completion result for a provided set of input chat messages.
@@ -63,7 +63,7 @@ public virtual ClientResult<ChatCompletion> CompleteChat(
         IEnumerable<ChatRequestMessage> messages,
         ChatCompletionOptions options = null)
     {
-        Internal.Models.CreateChatCompletionRequest request = CreateInternalRequest(messages, options); 
+        Internal.Models.CreateChatCompletionRequest request = CreateInternalRequest(messages, options);
         ClientResult<Internal.Models.CreateChatCompletionResponse> response = Shim.CreateChatCompletion(request);
         ChatCompletion chatCompletion = new(response.Value, internalChoiceIndex: 0);
         return ClientResult.FromValue(chatCompletion, response.GetRawResponse());
@@ -93,7 +93,6 @@ public virtual async Task<ClientResult<ChatCompletion>> CompleteChatAsync(
     ///     The number of independent, alternative response choices that should be generated.
     /// </param>
     /// <param name="options"> Additional options for the chat completion request. </param>
-    /// <param name="cancellationToken"> The cancellation token for the operation. </param>
     /// <returns> A result for a single chat completion. </returns>
     public virtual ClientResult<ChatCompletionCollection> CompleteChat(
         IEnumerable<ChatRequestMessage> messages,
@@ -147,14 +146,14 @@ public virtual async Task<ClientResult<ChatCompletionCollection>> CompleteChatAs
     /// </param>
     /// <param name="options"> Additional options for the chat completion request. </param>
     /// <returns> A streaming result with incremental chat completion updates. </returns>
-   public virtual StreamingClientResult<StreamingChatUpdate> CompleteChatStreaming(
-        string message,
-        int? choiceCount = null,
-        ChatCompletionOptions options = null)
-        => CompleteChatStreaming(
-            new List<ChatRequestMessage> { new ChatRequestUserMessage(message) },
-            choiceCount,
-            options);
+    public virtual StreamingClientResult<StreamingChatUpdate> CompleteChatStreaming(
+         string message,
+         int? choiceCount = null,
+         ChatCompletionOptions options = null)
+         => CompleteChatStreaming(
+             new List<ChatRequestMessage> { new ChatRequestUserMessage(message) },
+             choiceCount,
+             options);
 
     /// <summary>
     /// Begins a streaming response for a chat completion request using a single, simple user message as input.
@@ -191,29 +190,25 @@ public virtual Task<StreamingClientResult<StreamingChatUpdate>> CompleteChatStre
     ///     The number of independent, alternative choices that the chat completion request should generate.
     /// </param>
     /// <param name="options"> Additional options for the chat completion request. </param>
-    /// <param name="cancellationToken"> The cancellation token for the operation. </param>
     /// <returns> A streaming result with incremental chat completion updates. </returns>
     public virtual StreamingClientResult<StreamingChatUpdate> CompleteChatStreaming(
         IEnumerable<ChatRequestMessage> messages,
         int? choiceCount = null,
         ChatCompletionOptions options = null)
     {
-        PipelineMessage requestMessage = CreateCustomRequestMessage(messages, choiceCount, options);
-        requestMessage.BufferResponse = false;
-        Shim.Pipeline.Send(requestMessage);
-        PipelineResponse response = requestMessage.ExtractResponse();
+        PipelineMessage message = CreateCustomRequestMessage(messages, choiceCount, options);
+        message.BufferResponse = false;
+
+        Shim.Pipeline.Send(message);
+
+        PipelineResponse response = message.Response;
 
         if (response.IsError)
         {
             throw new ClientResultException(response);
         }
 
-        ClientResult genericResult = ClientResult.FromResponse(response);
-        return StreamingClientResult<StreamingChatUpdate>.CreateFromResponse(
-            genericResult,
-            (responseForEnumeration) => SseAsyncEnumerator<StreamingChatUpdate>.EnumerateFromSseJsonStream(
-                responseForEnumeration.GetRawResponse().ContentStream,
-                StreamingChatUpdate.DeserializeSseChatUpdates));
+        return new StreamingChatResult(response);
     }
 
     /// <summary>
@@ -229,29 +224,25 @@ public virtual StreamingClientResult<StreamingChatUpdate> CompleteChatStreaming(
     ///     The number of independent, alternative choices that the chat completion request should generate.
     /// </param>
     /// <param name="options"> Additional options for the chat completion request. </param>
-    /// <param name="cancellationToken"> The cancellation token for the operation. </param>
     /// <returns> A streaming result with incremental chat completion updates. </returns>
     public virtual async Task<StreamingClientResult<StreamingChatUpdate>> CompleteChatStreamingAsync(
         IEnumerable<ChatRequestMessage> messages,
         int? choiceCount = null,
         ChatCompletionOptions options = null)
     {
-        PipelineMessage requestMessage = CreateCustomRequestMessage(messages, choiceCount, options);
-        requestMessage.BufferResponse = false;
-        await Shim.Pipeline.SendAsync(requestMessage).ConfigureAwait(false);
-        PipelineResponse response = requestMessage.ExtractResponse();
+        PipelineMessage message = CreateCustomRequestMessage(messages, choiceCount, options);
+        message.BufferResponse = false;
+
+        await Shim.Pipeline.SendAsync(message).ConfigureAwait(false);
+
+        PipelineResponse response = message.Response;
 
         if (response.IsError)
         {
             throw new ClientResultException(response);
         }
 
-        ClientResult genericResult = ClientResult.FromResponse(response);
-        return StreamingClientResult<StreamingChatUpdate>.CreateFromResponse(
-            genericResult,
-            (responseForEnumeration) => SseAsyncEnumerator<StreamingChatUpdate>.EnumerateFromSseJsonStream(
-                responseForEnumeration.GetRawResponse().ContentStream,
-                StreamingChatUpdate.DeserializeSseChatUpdates));
+        return new StreamingChatResult(response);
     }
 
     private Internal.Models.CreateChatCompletionRequest CreateInternalRequest(
@@ -326,4 +317,4 @@ private PipelineMessage CreateCustomRequestMessage(IEnumerable<ChatRequestMessag
 
     private static PipelineMessageClassifier _responseErrorClassifier200;
     private static PipelineMessageClassifier ResponseErrorClassifier200 => _responseErrorClassifier200 ??= PipelineMessageClassifier.Create(stackalloc ushort[] { 200 });
-}
+}

.dotnet/src/Custom/Chat/StreamingChatUpdate.cs

@@ -1,9 +1,9 @@
-namespace OpenAI.Chat;
-
 using System;
 using System.Collections.Generic;
 using System.Text.Json;
 
+namespace OpenAI.Chat;
+
 /// <summary>
 /// Represents an incremental item of new data in a streaming response to a chat completion request.
 /// </summary>
@@ -184,11 +184,17 @@ internal StreamingChatUpdate(
 
     internal static IEnumerable<StreamingChatUpdate> DeserializeSseChatUpdates(ReadOnlyMemory<char> _, JsonElement sseDataJson)
     {
+		// TODO: would another enumerable implementation be more performant than list?        
         List<StreamingChatUpdate> results = [];
+
+		// TODO: Do we need to validate that we didn't get null or empty?
+        // What's the contract for the JSON updates?
+
         if (sseDataJson.ValueKind == JsonValueKind.Null)
         {
             return results;
         }
+
         string id = default;
         DateTimeOffset created = default;
         string systemFingerprint = null;
@@ -333,4 +339,4 @@ Internal.Models.CreateChatCompletionResponseChoiceLogprobs internalLogprobs
         }
         return results;
     }
-}
+}

.dotnet/src/Utility/SseReader.cs

@@ -6,7 +6,7 @@
 
 namespace OpenAI;
 
-internal sealed class SseReader : IDisposable
+internal sealed class SseReader : IDisposable, IAsyncDisposable
 {
     private readonly Stream _stream;
     private readonly StreamReader _reader;
@@ -121,4 +121,10 @@ private void Dispose(bool disposing)
             _disposedValue = true;
         }
     }
+
+    ValueTask IAsyncDisposable.DisposeAsync()
+    {
+        // TODO: revisit per platforms where async dispose is available.
+        return new ValueTask();
+    }
 }

.dotnet/src/Utility/StreamingClientResultOfT.cs

@@ -0,0 +1,28 @@
+using System.ClientModel;
+using System.ClientModel.Primitives;
+using System.Collections.Generic;
+using System.Threading;
+
+namespace OpenAI;
+
+#pragma warning disable CS1591 // public XML comments
+
+/// <summary>
+/// Represents an operation response with streaming content that can be deserialized and enumerated while the response
+/// is still being received.
+/// </summary>
+/// <typeparam name="T"> The data type representative of distinct, streamable items. </typeparam>
+// TODO: Revisit the IDisposable question
+public abstract class StreamingClientResult<T> : ClientResult, IAsyncEnumerable<T>
+{
+    protected StreamingClientResult(PipelineResponse response) : base(response)
+    {
+    }
+
+    // Note that if the implementation disposes the stream, the caller can only
+    // enumerate the results once.  I think this makes sense, but we should
+    // make sure architects agree.
+    public abstract IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default);
+}
+
+#pragma warning restore CS1591 // public XML comments