System.Text.Json: Add IAsyncEnumerable support

src/libraries/System.Text.Json/ref/System.Text.Json.cs

@@ -192,6 +192,7 @@ public static partial class JsonSerializer
         public static object? Deserialize(ref System.Text.Json.Utf8JsonReader reader, [System.Diagnostics.CodeAnalysis.DynamicallyAccessedMembersAttribute(System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.PublicConstructors | System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.PublicFields | System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.PublicProperties)] System.Type returnType, System.Text.Json.JsonSerializerOptions? options = null) { throw null; }
         public static System.Threading.Tasks.ValueTask<object?> DeserializeAsync(System.IO.Stream utf8Json, [System.Diagnostics.CodeAnalysis.DynamicallyAccessedMembersAttribute(System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.PublicConstructors | System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.PublicFields | System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.PublicProperties)] System.Type returnType, System.Text.Json.JsonSerializerOptions? options = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
         public static System.Threading.Tasks.ValueTask<TValue?> DeserializeAsync<[System.Diagnostics.CodeAnalysis.DynamicallyAccessedMembersAttribute(System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.PublicConstructors | System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.PublicFields | System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.PublicProperties)] TValue>(System.IO.Stream utf8Json, System.Text.Json.JsonSerializerOptions? options = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
+        public static System.Collections.Generic.IAsyncEnumerable<TValue?> DeserializeAsyncEnumerable<[System.Diagnostics.CodeAnalysis.DynamicallyAccessedMembersAttribute(System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.PublicConstructors | System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.PublicFields | System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.PublicProperties)] TValue>(System.IO.Stream utf8Json, System.Text.Json.JsonSerializerOptions? options = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
         public static TValue? Deserialize<[System.Diagnostics.CodeAnalysis.DynamicallyAccessedMembersAttribute(System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.PublicConstructors | System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.PublicFields | System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.PublicProperties)] TValue>(System.ReadOnlySpan<byte> utf8Json, System.Text.Json.JsonSerializerOptions? options = null) { throw null; }
         public static TValue? Deserialize<[System.Diagnostics.CodeAnalysis.DynamicallyAccessedMembersAttribute(System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.PublicConstructors | System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.PublicFields | System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.PublicProperties)] TValue>(string json, System.Text.Json.JsonSerializerOptions? options = null) { throw null; }
         public static TValue? Deserialize<[System.Diagnostics.CodeAnalysis.DynamicallyAccessedMembersAttribute(System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.PublicConstructors | System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.PublicFields | System.Diagnostics.CodeAnalysis.DynamicallyAccessedMemberTypes.PublicProperties)] TValue>(System.ReadOnlySpan<char> json, System.Text.Json.JsonSerializerOptions? options = null) { throw null; }

src/libraries/System.Text.Json/src/Resources/Strings.resx

@@ -357,6 +357,9 @@
   <data name="SerializationNotSupportedType" xml:space="preserve">
     <value>The type '{0}' is not supported.</value>
   </data>
+  <data name="TypeRequiresAsyncSerialization" xml:space="preserve">
+    <value>The type '{0}' can only be serialized using async serialization methods.</value>
+  </data>
   <data name="InvalidCharacterAtStartOfComment" xml:space="preserve">
     <value>'{0}' is invalid after '/' at the beginning of the comment. Expected either '/' or '*'.</value>
   </data>
@@ -557,4 +560,4 @@
   <data name="SerializerConverterFactoryReturnsJsonConverterFactory" xml:space="preserve">
     <value>The converter '{0}' cannot return an instance of JsonConverterFactory.</value>
   </data>
-</root>
+</root>

src/libraries/System.Text.Json/src/System.Text.Json.csproj

@@ -73,6 +73,8 @@
     <Compile Include="System\Text\Json\Serialization\Converters\Collection\ConcurrentStackOfTConverter.cs" />
     <Compile Include="System\Text\Json\Serialization\Converters\Collection\DictionaryDefaultConverter.cs" />
     <Compile Include="System\Text\Json\Serialization\Converters\Collection\DictionaryOfTKeyTValueConverter.cs" />
+    <Compile Include="System\Text\Json\Serialization\Converters\Collection\IAsyncEnumerableConverterFactory.cs" />
+    <Compile Include="System\Text\Json\Serialization\Converters\Collection\IAsyncEnumerableOfTConverter.cs" />
     <Compile Include="System\Text\Json\Serialization\Converters\Collection\ICollectionOfTConverter.cs" />
     <Compile Include="System\Text\Json\Serialization\Converters\Collection\IDictionaryConverter.cs" />
     <Compile Include="System\Text\Json\Serialization\Converters\Collection\IDictionaryOfTKeyTValueConverter.cs" />
@@ -173,6 +175,7 @@
     <Compile Include="System\Text\Json\Serialization\PreserveReferenceHandler.cs" />
     <Compile Include="System\Text\Json\Serialization\PreserveReferenceResolver.cs" />
     <Compile Include="System\Text\Json\Serialization\PropertyRef.cs" />
+    <Compile Include="System\Text\Json\Serialization\ReadAsyncBufferState.cs" />
     <Compile Include="System\Text\Json\Serialization\ReadStack.cs" />
     <Compile Include="System\Text\Json\Serialization\ReadStackFrame.cs" />
     <Compile Include="System\Text\Json\Serialization\ReferenceHandler.cs" />
@@ -233,11 +236,9 @@
   <ItemGroup Condition="$(TargetFramework.StartsWith('netstandard')) or $(TargetFramework.StartsWith('net4'))">
     <Compile Include="System\Collections\Generic\StackExtensions.netstandard.cs" />
     <!-- Common or Common-branched source files -->
-    <Compile Include="$(CommonPath)System\Buffers\ArrayBufferWriter.cs"
-             Link="Common\System\Buffers\ArrayBufferWriter.cs" />
+    <Compile Include="$(CommonPath)System\Buffers\ArrayBufferWriter.cs" Link="Common\System\Buffers\ArrayBufferWriter.cs" />
   </ItemGroup>
-  <ItemGroup Condition="'$(TargetFramework)' == '$(NetCoreAppCurrent)' or
-                        '$(TargetFramework)' == 'netcoreapp3.0'">
+  <ItemGroup Condition="'$(TargetFramework)' == '$(NetCoreAppCurrent)' or '$(TargetFramework)' == 'netcoreapp3.0'">
     <Reference Include="System.Buffers" />
     <Reference Include="System.Collections" />
     <Reference Include="System.Collections.Concurrent" />

src/libraries/System.Text.Json/src/System/Text/Json/Serialization/ClassType.cs

@@ -20,7 +20,7 @@ internal enum ClassType : byte
         Value = 0x2,
         // JsonValueConverter<> - simple values that need to re-enter the serializer such as KeyValuePair<TKey, TValue>.
         NewValue = 0x4,
-        // JsonIEnumerbleConverter<> - all enumerable collections except dictionaries.
+        // JsonIEnumerableConverter<> - all enumerable collections except dictionaries.
         Enumerable = 0x8,
         // JsonDictionaryConverter<,> - dictionary types.
         Dictionary = 0x10,

src/libraries/System.Text.Json/src/System/Text/Json/Serialization/Converters/Collection/IAsyncEnumerableConverterFactory.cs

@@ -0,0 +1,31 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Diagnostics.CodeAnalysis;
+using System.Text.Json.Serialization.Converters;
+
+namespace System.Text.Json.Serialization
+{
+    /// <summary>
+    /// Converter for streaming <see cref="IAsyncEnumerable{T}" /> values.
+    /// </summary>
+    internal sealed class IAsyncEnumerableConverterFactory : JsonConverterFactory
+    {
+        public override bool CanConvert(Type typeToConvert) => GetAsyncEnumerableInterface(typeToConvert) is not null;
+
+        public override JsonConverter CreateConverter(Type typeToConvert, JsonSerializerOptions options)
+        {
+            Type? asyncEnumerableInterface = GetAsyncEnumerableInterface(typeToConvert);
+            Debug.Assert(asyncEnumerableInterface is not null, $"{typeToConvert} not supported by converter.");
+
+            Type elementType = asyncEnumerableInterface.GetGenericArguments()[0];
+            Type converterType = typeof(IAsyncEnumerableOfTConverter<,>).MakeGenericType(typeToConvert, elementType);
+            return (JsonConverter)Activator.CreateInstance(converterType)!;
+        }
+
+        private static Type? GetAsyncEnumerableInterface(Type type)
+            => IEnumerableConverterFactoryHelpers.GetCompatibleGenericInterface(type, typeof(IAsyncEnumerable<>));
+    }
+}

src/libraries/System.Text.Json/src/System/Text/Json/Serialization/Converters/Collection/IAsyncEnumerableOfTConverter.cs

@@ -0,0 +1,126 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace System.Text.Json.Serialization.Converters
+{
+    internal sealed class IAsyncEnumerableOfTConverter<TAsyncEnumerable, TElement>
+        : IEnumerableDefaultConverter<TAsyncEnumerable, TElement>
+        where TAsyncEnumerable : IAsyncEnumerable<TElement>
+    {
+        internal override bool OnTryRead(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options, ref ReadStack state, out TAsyncEnumerable value)
+        {
+            if (!typeToConvert.IsAssignableFrom(typeof(IAsyncEnumerable<TElement>)))
+            {
+                ThrowHelper.ThrowNotSupportedException_CannotPopulateCollection(TypeToConvert, ref reader, ref state);
+            }
+
+            return base.OnTryRead(ref reader, typeToConvert, options, ref state, out value!);
+        }
+
+        protected override void Add(in TElement value, ref ReadStack state)
+        {
+            ((BufferedAsyncEnumerable)state.Current.ReturnValue!)._buffer.Add(value);
+        }
+
+        protected override void CreateCollection(ref Utf8JsonReader reader, ref ReadStack state, JsonSerializerOptions options)
+        {
+            state.Current.ReturnValue = new BufferedAsyncEnumerable();
+        }
+
+        internal override bool OnTryWrite(Utf8JsonWriter writer, TAsyncEnumerable value, JsonSerializerOptions options, ref WriteStack state)
+        {
+            if (!state.SupportContinuation)
+            {
+                ThrowHelper.ThrowNotSupportedException_TypeRequiresAsyncSerialization(TypeToConvert);
+            }
+
+            return base.OnTryWrite(writer, value, options, ref state);
+        }
+
+        [Diagnostics.CodeAnalysis.SuppressMessage("Reliability", "CA2012:Use ValueTasks correctly", Justification = "Converter needs to consume ValueTask's in a non-async context")]
+        protected override bool OnWriteResume(Utf8JsonWriter writer, TAsyncEnumerable value, JsonSerializerOptions options, ref WriteStack state)
+        {
+            IAsyncEnumerator<TElement> enumerator;
+            ValueTask<bool> moveNextTask;
+
+            if (state.Current.AsyncEnumerator is null)
+            {
+                enumerator = value.GetAsyncEnumerator(state.CancellationToken);
+                moveNextTask = enumerator.MoveNextAsync();
+                // we always need to attach the enumerator to the stack
+                // since it will need to be disposed asynchronously.
+                state.Current.AsyncEnumerator = enumerator;
+            }
+            else
+            {
+                Debug.Assert(state.Current.AsyncEnumerator is IAsyncEnumerator<TElement>);
+                enumerator = (IAsyncEnumerator<TElement>)state.Current.AsyncEnumerator;
+
+                if (state.Current.AsyncEnumeratorIsPendingCompletion)
+                {
+                    // converter was previously suspended due to a pending MoveNextAsync() task
+                    Debug.Assert(state.PendingTask is Task<bool> && state.PendingTask.IsCompleted);
+                    moveNextTask = new ValueTask<bool>((Task<bool>)state.PendingTask);
+                    state.Current.AsyncEnumeratorIsPendingCompletion = false;
+                    state.PendingTask = null;
+                }
+                else
+                {
+                    // converter was suspended for a different reason;
+                    // the last MoveNextAsync() call can only have completed with 'true'.
+                    moveNextTask = new ValueTask<bool>(true);
+                }
+            }
+
+            JsonConverter<TElement> converter = GetElementConverter(ref state);
+
+            // iterate through the enumerator while elements are being returned synchronously
+            for (; moveNextTask.IsCompleted; moveNextTask = enumerator.MoveNextAsync())
+            {
+                if (!moveNextTask.Result)
+                {
+                    return true;
+                }
+
+                if (ShouldFlush(writer, ref state))
+                {
+                    return false;
+                }
+
+                TElement element = enumerator.Current;
+                if (!converter.TryWrite(writer, element, options, ref state))
+                {
+                    return false;
+                }
+            }
+
+            // we have a pending MoveNextAsync() call;
+            // wrap inside a regular task so that it can be awaited multiple times;
+            // mark the current stackframe as pending completion.
+            Debug.Assert(state.PendingTask is null);
+            state.PendingTask = moveNextTask.AsTask();
+            state.Current.AsyncEnumeratorIsPendingCompletion = true;
+            return false;
+        }
+
+        private sealed class BufferedAsyncEnumerable : IAsyncEnumerable<TElement>
+        {
+            public readonly List<TElement> _buffer = new();
+
+#pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously
+            public async IAsyncEnumerator<TElement> GetAsyncEnumerator(CancellationToken _)
+            {
+                foreach (TElement element in _buffer)
+                {
+                    yield return element;
+                }
+            }
+#pragma warning restore CS1998 // Async method lacks 'await' operators and will run synchronously
+        }
+    }
+}

src/libraries/System.Text.Json/src/System/Text/Json/Serialization/Converters/Collection/IEnumerableDefaultConverter.cs

@@ -218,7 +218,7 @@ internal override bool OnTryRead(
             return true;
         }
 
-        internal sealed override bool OnTryWrite(
+        internal override bool OnTryWrite(
             Utf8JsonWriter writer,
             TCollection value,
             JsonSerializerOptions options,

src/libraries/System.Text.Json/src/System/Text/Json/Serialization/JsonResumableConverterOfT.cs

@@ -36,7 +36,15 @@ public sealed override void Write(Utf8JsonWriter writer, T value, JsonSerializer
 
             WriteStack state = default;
             state.Initialize(typeof(T), options, supportContinuation: false);
-            TryWrite(writer, value, options, ref state);
+            try
+            {
+                TryWrite(writer, value, options, ref state);
+            }
+            catch
+            {
+                state.DisposePendingDisposablesOnException();
+                throw;
+            }
         }
 
         public sealed override bool HandleNull => false;