using System;

using System.Collections;

using System.Collections.Generic;

using System.Linq;

using System.Text;

using System.Runtime.Serialization.Formatters;

using System.IO;

using System.Data;

using System.ComponentModel.DataAnnotations;

using System.Globalization;

using System.Reflection;

using System.Collections.ObjectModel;

using System.Threading.Tasks;

using System.Threading;

using System.Runtime.CompilerServices;

​

using Newtonsoft.Json;

using Newtonsoft.Json.Linq;

using Newtonsoft.Json.Converters;

using Newtonsoft.Json.Serialization;    

​

using NUnit.Framework; // Throws NUnit.Framework.AssertionException

​

//https://stackoverflow.com/questions/72451669/deserializing-to-asyncenumerable-using-newtonsoft-json

​

#nullable enable

​

    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;

        }

    }

    // Test code

​

    public class SurrogateStream : Stream

    {

        Stream? baseStream;

        Stream BaseStream => baseStream ?? throw new ObjectDisposedException(nameof(BaseStream));

        public SurrogateStream(Stream baseStream) => this.baseStream = baseStream ?? throw new ArgumentNullException(nameof(baseStream));

        public override long Length => BaseStream.Length;

        public override long Position { get => BaseStream.Position; set => BaseStream.Position = value; }

        public override bool CanRead => BaseStream.CanRead;

        public override bool CanWrite => BaseStream.CanWrite;

        public override bool CanSeek => BaseStream.CanSeek;

        public override bool CanTimeout => BaseStream.CanTimeout;

        public override IAsyncResult BeginRead (byte[] buffer, int offset, int count, AsyncCallback? callback, object? state) => BaseStream.BeginRead(buffer, offset, count, callback, state);

        public override IAsyncResult BeginWrite (byte[] buffer, int offset, int count, AsyncCallback? callback, object? state) => BaseStream.BeginWrite(buffer, offset, count, callback, state);

        //public override void Close () => BaseStream.Close();

        public override void CopyTo (System.IO.Stream destination, int bufferSize) => BaseStream.CopyTo(destination, bufferSize);

        public override Task CopyToAsync (System.IO.Stream destination, int bufferSize, System.Threading.CancellationToken cancellationToken) => BaseStream.CopyToAsync(destination, bufferSize, cancellationToken);

        //public override System.Runtime.Remoting.ObjRef CreateObjRef (Type requestedType) => BaseStream.CreateObjRef(requestedType);

        protected override void Dispose (bool disposing)

        {

            try

            {

                if (disposing)

                    Interlocked.Exchange(ref baseStream, null)?.Dispose();

            }

            finally

            {

                base.Dispose(disposing);

            }

        }

        public override ValueTask DisposeAsync() => Interlocked.Exchange(ref baseStream, null)?.DisposeAsync() ?? ValueTask.CompletedTask;

        public override int EndRead (IAsyncResult asyncResult) => BaseStream.EndRead(asyncResult);

        public override void EndWrite (IAsyncResult asyncResult) => BaseStream.EndWrite(asyncResult);

        public override void Flush() => BaseStream.Flush();

        public override Task FlushAsync (CancellationToken cancellationToken) => BaseStream.FlushAsync(cancellationToken);

        public override int Read (byte[] buffer, int offset, int count) => BaseStream.Read(buffer, offset, count);

        public override int Read (Span<byte> buffer) => BaseStream.Read(buffer);

        public override Task<int> ReadAsync (byte[] buffer, int offset, int count, CancellationToken cancellationToken) => BaseStream.ReadAsync(buffer, offset, count, cancellationToken);

        public override ValueTask<int> ReadAsync (Memory<byte> buffer, System.Threading.CancellationToken cancellationToken = default) => BaseStream.ReadAsync(buffer, cancellationToken);

        public override int ReadByte () => BaseStream.ReadByte();

        public override long Seek (long offset, System.IO.SeekOrigin origin) => BaseStream.Seek(offset, origin);

        public override void SetLength (long value) => BaseStream.SetLength(value);

        public override void Write (byte[] buffer, int offset, int count) => BaseStream.Write(buffer, offset, count);

        public override void Write (ReadOnlySpan<byte> buffer) => BaseStream.Write(buffer);

        public override System.Threading.Tasks.Task WriteAsync (byte[] buffer, int offset, int count, System.Threading.CancellationToken cancellationToken) => BaseStream.WriteAsync(buffer, offset, count, cancellationToken);

        public override System.Threading.Tasks.ValueTask WriteAsync (ReadOnlyMemory<byte> buffer, System.Threading.CancellationToken cancellationToken = default) => BaseStream.WriteAsync(buffer, cancellationToken);

        public override void WriteByte (byte value) => BaseStream.WriteByte(value);

    }

​

    public class AsyncOnlySurogateStream : SurrogateStream

    {

        public AsyncOnlySurogateStream(Stream baseStream) : base(baseStream) { }

        public override void CopyTo (System.IO.Stream destination, int bufferSize) => throw new InvalidOperationException(string.Format("Non async method {0} called", System.Reflection.MethodBase.GetCurrentMethod()?.Name));

        protected override void Dispose (bool disposing)

        {

            base.Dispose(disposing);

            if (disposing)

            {

                throw new InvalidOperationException(string.Format("Non async method {0} called", System.Reflection.MethodBase.GetCurrentMethod()?.Name));

            }

        }

        public override void Flush() => throw new InvalidOperationException(string.Format("Non async method {0} called", System.Reflection.MethodBase.GetCurrentMethod()?.Name));

        public override int Read (byte[] buffer, int offset, int count) => throw new InvalidOperationException(string.Format("Non async method {0} called", System.Reflection.MethodBase.GetCurrentMethod()?.Name));

        public override int Read (Span<byte> buffer) => throw new InvalidOperationException(string.Format("Non async method {0} called", System.Reflection.MethodBase.GetCurrentMethod()?.Name));

        public override int ReadByte () => throw new InvalidOperationException(string.Format("Non async method {0} called", System.Reflection.MethodBase.GetCurrentMethod()?.Name));

        public override void Write (byte[] buffer, int offset, int count) => throw new InvalidOperationException(string.Format("Non async method {0} called", System.Reflection.MethodBase.GetCurrentMethod()?.Name));

        public override void Write (ReadOnlySpan<byte> buffer) => throw new InvalidOperationException(string.Format("Non async method {0} called", System.Reflection.MethodBase.GetCurrentMethod()?.Name));

        public override void WriteByte (byte value) => throw new InvalidOperationException(string.Format("Non async method {0} called", System.Reflection.MethodBase.GetCurrentMethod()?.Name));

    }

​

    public class Model

    {

        public int Id { get; set; }

        public string? Name { get; set; }

    }

​

    class TestClass

    {

        public static async Task Test()

        {

            await TestDeserializeAsync();

            Console.WriteLine();

            await TestLoadJTokenAsync();

            Assert.ThrowsAsync<InvalidOperationException>(() => TestReadAsJsonNonAsyncFail());

            Assert.ThrowsAsync<InvalidOperationException>(() => TestReadAsTextNonAsyncFail());

            Assert.DoesNotThrowAsync(() => TestReadAsTextAsync());

            Assert.DoesNotThrowAsync(() => TestReadAsJsonAsync());

        }

        public static async Task TestDeserializeAsync(CancellationToken cancellationToken = default)

        {

            var inputs = Enumerable.Range(0,12).Select(i => new Model { Id = i, Name = $"Model {i}" }).ToList();

​

            using var innerStream = new MemoryStream();

            using (var textWriter = new StreamWriter(innerStream, leaveOpen : true))

            using (var writer = new JsonTextWriter(textWriter) { CloseOutput = false })

            {

                JsonSerializer.CreateDefault().Serialize(writer, inputs);

            }

            innerStream.Position = 0;

            await using var stream = new AsyncOnlySurogateStream(innerStream);

​

            await foreach (var token in JsonExtensions.LoadJTokenAsyncEnumerable(stream, cancellationToken : cancellationToken))

            {

                Console.WriteLine(token.ToString(Formatting.None));

            }

        }

        public static async Task TestLoadJTokenAsync(CancellationToken cancellationToken = default)

        {

            var inputs = Enumerable.Range(0,10).Select(i => new Model { Id = i, Name = $"Token  {i}" }).ToList();

​

            using var innerStream = new MemoryStream();

            using (var textWriter = new StreamWriter(innerStream, leaveOpen : true))

            using (var writer = new JsonTextWriter(textWriter) { CloseOutput = false })

            {

                JsonSerializer.CreateDefault().Serialize(writer, inputs);

            }

            innerStream.Position = 0;

            await using var stream = new AsyncOnlySurogateStream(innerStream);

​

            await foreach (var model in JsonExtensions.DeserializeAsyncEnumerable<Model>(stream, cancellationToken : cancellationToken))

            {

                // Process each item somehow

                Console.WriteLine(JsonConvert.SerializeObject(model, Formatting.None));

            }

        }

        public static async Task TestReadAsTextAsync()

        {

            using var innerStream = new MemoryStream(Encoding.UTF8.GetBytes(GetJson()));

            await using (var stream = new AsyncOnlySurogateStream(innerStream))

            {

                using (var textReader = new StreamReader(stream, leaveOpen : true))

                {

                    string? line;

                    while ((line = await textReader.ReadLineAsync()) != null)

                        ;

                }

            }

        }   

​

        public static async Task TestReadAsTextNonAsyncFail()

        {

            using var innerStream = new MemoryStream(Encoding.UTF8.GetBytes(GetJson()));

            await using (var stream = new AsyncOnlySurogateStream(innerStream))

            {

                using (var textReader = new StreamReader(stream, leaveOpen : false))

                {

                    string? line;

                    while ((line = await textReader.ReadLineAsync()) != null)

                        ;

                }

            }

        }   

​

        public static async Task TestReadAsJsonAsync()

        {

            var inputs = Enumerable.Range(0,12000).Select(i => new Model { Id = i, Name = $"Model {i}" }).ToList();

​

            using var innerStream = new MemoryStream();

            using (var textWriter = new StreamWriter(innerStream, leaveOpen : true))

            using (var writer = new JsonTextWriter(textWriter) { CloseOutput = false })

            {

                JsonSerializer.CreateDefault().Serialize(writer, inputs);

            }

            Console.WriteLine($"\ninnerStream.Length = {innerStream.Length}");

            innerStream.Position = 0;

            await using (var stream = new AsyncOnlySurogateStream(innerStream))

            {

                using (var textReader = new StreamReader(stream, leaveOpen : true))

                {

                    using (var reader = new JsonTextReader(textReader) { CloseInput = false })

                    {

                        while (await reader.ReadAsync())

                            ;

                    }

                }

            }

        }   

​

        public static async Task TestReadAsJsonNonAsyncFail()

        {

            using var innerStream = new MemoryStream(Encoding.UTF8.GetBytes(GetJson()));

            await using (var stream = new AsyncOnlySurogateStream(innerStream))

            {

                using (var textReader = new StreamReader(stream, leaveOpen : true))

                {

                    using (var reader = new JsonTextReader(textReader) { CloseInput = false })

                    {

                        while (reader.Read())

                        {

                            Console.WriteLine($"TokenType={reader.TokenType}, Value={reader.Value}");

                        }

                    }

                }

            }

        }   

        static string GetJson() => @"[{

    ""glossary"": {

        ""title"": ""example glossary"",

        ""GlossDiv"": {

            ""title"": ""S"",

            ""GlossList"": {

                ""GlossEntry"": {

                    ""ID"": ""SGML"",

                    ""SortAs"": ""SGML"",

                    ""GlossTerm"": ""Standard Generalized Markup Language"",

                    ""Acronym"": ""SGML"",

                    ""Abbrev"": ""ISO 8879:1986"",

                    ""GlossDef"": {

                        ""para"": ""A meta-markup language, used to create markup languages such as DocBook."",

                        ""GlossSeeAlso"": [""GML"", ""XML""]

                    },

                    ""GlossSee"": ""markup""

                }

            }

        }

    }

},{

    ""glossary"": {

        ""title"": ""example glossary"",

        ""GlossDiv"": {

            ""title"": ""S"",

            ""GlossList"": {

            }

        }

    }

},{

    ""glossary"": {

        ""title"": ""example glossary"",

    }

},

null,true,false,1,2,""hello"",

]

";

    }

​

    public class Program

    {

        public static async Task Main(string[] args)

        {

            Console.WriteLine("Environment version: {0} ({1})", System.Runtime.InteropServices.RuntimeInformation.FrameworkDescription , Environment.Version);

            Console.WriteLine("{0} version: {1}", typeof(JsonSerializer).Assembly.GetName().Name,  typeof(JsonSerializer).Assembly.FullName);

            Console.WriteLine();

​

            try

            {

                await TestClass.Test();

            }

            catch (Exception ex)

            {

                Console.WriteLine("Failed with unhandled exception: ");

                Console.WriteLine(ex);

                throw;

            }

        }

    }

​