问用AsyncEnumerable反序列化为Newtonsoft.Json

EN

public static partial class JsonExtensions
{
    /// <summary>
    /// Asynchronously load and synchronously deserialize values from a stream containing a JSON array.  The root object of the JSON stream must in fact be an array, or an exception is thrown
    /// </summary>
    public static async IAsyncEnumerable<T?> DeserializeAsyncEnumerable<T>(Stream stream, JsonSerializerSettings? settings = default, [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        var serializer = JsonSerializer.CreateDefault(settings);
        var loadSettings = new JsonLoadSettings { LineInfoHandling = LineInfoHandling.Ignore }; // For performance do not load line info.
        // StreamReader and JsonTextReader do not implement IAsyncDisposable so let the caller dispose the stream.
        using (var textReader = new StreamReader(stream, leaveOpen : true))
        using (var reader = new JsonTextReader(textReader) { CloseInput = false })
        {
            await foreach (var token in LoadAsyncEnumerable(reader, loadSettings, cancellationToken ).ConfigureAwait(false))
                yield return token.ToObject<T>(serializer);
        }
    }
    
    /// <summary>
    /// Asynchronously load and return JToken values from a stream containing a JSON array.  The root object of the JSON stream must in fact be an array, or an exception is thrown
    /// </summary>
    public static async IAsyncEnumerable<JToken> LoadJTokenAsyncEnumerable(Stream stream, JsonLoadSettings? settings = default, [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        // StreamReader and JsonTextReader do not implement IAsyncDisposable so let the caller dispose the stream.
        using (var textReader = new StreamReader(stream, leaveOpen : true))
        using (var reader = new JsonTextReader(textReader) { CloseInput = false })
        {
            await foreach (var token in LoadAsyncEnumerable(reader, settings, cancellationToken).ConfigureAwait(false))
                yield return token;
        }
    }
    
    /// <summary>
    /// Asynchronously load and return JToken values from a stream containing a JSON array.  The root object of the JSON stream must in fact be an array, or an exception is thrown
    /// </summary>
    public static async IAsyncEnumerable<JToken> LoadAsyncEnumerable(JsonTextReader reader, JsonLoadSettings? settings = default, [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        (await reader.MoveToContentAndAssertAsync().ConfigureAwait(false)).AssertTokenType(JsonToken.StartArray);
        cancellationToken.ThrowIfCancellationRequested();
        while ((await reader.ReadToContentAndAssert(cancellationToken).ConfigureAwait(false)).TokenType != JsonToken.EndArray)
        {
            cancellationToken.ThrowIfCancellationRequested();
            yield return await JToken.LoadAsync(reader, settings, cancellationToken).ConfigureAwait(false);
        }
        cancellationToken.ThrowIfCancellationRequested();
    }
    
    public static JsonReader AssertTokenType(this JsonReader reader, JsonToken tokenType) => 
        reader.TokenType == tokenType ? reader : throw new JsonSerializationException(string.Format("Unexpected token {0}, expected {1}", reader.TokenType, tokenType));
    
    public static async Task<JsonReader> ReadToContentAndAssert(this JsonReader reader, CancellationToken cancellationToken = default) =>
        await (await reader.ReadAndAssertAsync(cancellationToken).ConfigureAwait(false)).MoveToContentAndAssertAsync(cancellationToken).ConfigureAwait(false);

    public static async Task<JsonReader> MoveToContentAndAssertAsync(this JsonReader reader, CancellationToken cancellationToken = default)
    {
        if (reader == null)
            throw new ArgumentNullException();
        if (reader.TokenType == JsonToken.None)       // Skip past beginning of stream.
            await reader.ReadAndAssertAsync(cancellationToken).ConfigureAwait(false);
        while (reader.TokenType == JsonToken.Comment) // Skip past comments.
            await reader.ReadAndAssertAsync(cancellationToken).ConfigureAwait(false);
        return reader;
    }

    public static async Task<JsonReader> ReadAndAssertAsync(this JsonReader reader, CancellationToken cancellationToken = default)
    {
        if (reader == null)
            throw new ArgumentNullException();
        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
            throw new JsonReaderException("Unexpected end of JSON stream.");
        return reader;
    }
}

await foreach (var token in JsonExtensions.LoadJTokenAsyncEnumerable(stream, cancellationToken : cancellationToken))
{
    Console.WriteLine(token.ToString(Formatting.None));
}