[Json] Avoid writing to PipeWriter if IAsyncEnumerable throws before first item (#113503)

* [Json] Avoid writing to PipeWriter if IAsyncEnumerable throws before first item

* update

Author: BrennanConroy

src/libraries/System.Text.Json/src/System/Text/Json/Serialization/Metadata/JsonTypeInfoOfT.WriteHelpers.cs

@@ -169,7 +169,6 @@ rootValue is not null &&
                         try
                         {
                             isFinalBlock = EffectiveConverter.WriteCore(writer, rootValue, Options, ref state);
-                            writer.Flush();
 
                             if (state.SuppressFlush)
                             {
@@ -179,6 +178,7 @@ rootValue is not null &&
                             }
                             else
                             {
+                                writer.Flush();
                                 FlushResult result = await pipeWriter.FlushAsync(cancellationToken).ConfigureAwait(false);
                                 if (result.IsCanceled || result.IsCompleted)
                                 {
@@ -230,6 +230,9 @@ rootValue is not null &&
                 }
                 catch
                 {
+                    // Reset the writer in exception cases as we don't want the writer.Dispose() call to flush any pending bytes.
+                    writer.Reset();
+                    writer.Dispose();
                     // On exception, walk the WriteStack for any orphaned disposables and try to dispose them.
                     await state.DisposePendingDisposablesOnExceptionAsync().ConfigureAwait(false);
                     throw;

src/libraries/System.Text.Json/tests/System.Text.Json.Tests/Serialization/JsonSerializerWrapper.Reflection.cs

@@ -886,11 +886,13 @@ private int ReadExactlyFromSource(byte[] buffer, int offset, int count)
         }
 
         // TODO: Deserialize to use PipeReader overloads once implemented
-        private class AsyncPipelinesSerializerWrapper : JsonSerializerWrapper
+        private class AsyncPipelinesSerializerWrapper : StreamingJsonSerializerWrapper
         {
             public override JsonSerializerOptions DefaultOptions => JsonSerializerOptions.Default;
             public override bool SupportsNullValueOnDeserialize => true;
 
+            public override bool IsAsyncSerializer => true;
+
             public override async Task<T> DeserializeWrapper<T>(string json, JsonSerializerOptions options = null)
             {
                 return await JsonSerializer.DeserializeAsync<T>(new MemoryStream(Encoding.UTF8.GetBytes(json)), options);
@@ -915,6 +917,31 @@ public override async Task<object> DeserializeWrapper(string json, Type type, Js
                 return await JsonSerializer.DeserializeAsync(new MemoryStream(Encoding.UTF8.GetBytes(json)), type, context);
             }
 
+            public override Task<object> DeserializeWrapper(Stream utf8Json, Type returnType, JsonSerializerOptions? options = null)
+            {
+                return JsonSerializer.DeserializeAsync(utf8Json, returnType, options).AsTask();
+            }
+
+            public override Task<T> DeserializeWrapper<T>(Stream utf8Json, JsonSerializerOptions? options = null)
+            {
+                return JsonSerializer.DeserializeAsync<T>(utf8Json, options).AsTask();
+            }
+
+            public override Task<object> DeserializeWrapper(Stream utf8Json, Type returnType, JsonSerializerContext context)
+            {
+                return JsonSerializer.DeserializeAsync(utf8Json, returnType, context).AsTask();
+            }
+
+            public override Task<T> DeserializeWrapper<T>(Stream utf8Json, JsonTypeInfo<T> jsonTypeInfo)
+            {
+                return JsonSerializer.DeserializeAsync(utf8Json, jsonTypeInfo).AsTask();
+            }
+
+            public override Task<object> DeserializeWrapper(Stream utf8Json, JsonTypeInfo jsonTypeInfo)
+            {
+                return JsonSerializer.DeserializeAsync(utf8Json, jsonTypeInfo).AsTask();
+            }
+
             public override async Task<string> SerializeWrapper(object value, Type inputType, JsonSerializerOptions options = null)
             {
                 Pipe pipe = new Pipe();
@@ -969,6 +996,71 @@ public override async Task<string> SerializeWrapper(object value, JsonTypeInfo j
                 pipe.Reader.AdvanceTo(result.Buffer.End);
                 return stringResult;
             }
+
+            public override async Task SerializeWrapper(Stream stream, object value, Type inputType, JsonSerializerOptions? options = null)
+            {
+                var writer = PipeWriter.Create(stream);
+                try
+                {
+                    await JsonSerializer.SerializeAsync(writer, value, inputType, options);
+                }
+                finally
+                {
+                    await writer.FlushAsync();
+                }
+            }
+
+            public override async Task SerializeWrapper<T>(Stream stream, T value, JsonSerializerOptions? options = null)
+            {
+                var writer = PipeWriter.Create(stream);
+                try
+                {
+                    await JsonSerializer.SerializeAsync(writer, value, options);
+                }
+                finally
+                {
+                    await writer.FlushAsync();
+                }
+            }
+
+            public override async Task SerializeWrapper(Stream stream, object value, Type inputType, JsonSerializerContext context)
+            {
+                var writer = PipeWriter.Create(stream);
+                try
+                {
+                    await JsonSerializer.SerializeAsync(writer, value, inputType, context);
+                }
+                finally
+                {
+                    await writer.FlushAsync();
+                }
+            }
+
+            public override async Task SerializeWrapper<T>(Stream stream, T value, JsonTypeInfo<T> jsonTypeInfo)
+            {
+                var writer = PipeWriter.Create(stream);
+                try
+                {
+                    await JsonSerializer.SerializeAsync(writer, value, jsonTypeInfo);
+                }
+                finally
+                {
+                    await writer.FlushAsync();
+                }
+            }
+
+            public override async Task SerializeWrapper(Stream stream, object value, JsonTypeInfo jsonTypeInfo)
+            {
+                var writer = PipeWriter.Create(stream);
+                try
+                {
+                    await JsonSerializer.SerializeAsync(writer, value, jsonTypeInfo);
+                }
+                finally
+                {
+                    await writer.FlushAsync();
+                }
+            }
         }
     }
 }