// See https://aka.ms/new-console-template for more information

​

using System;

using System.Buffers;

using System.Collections.Generic;

using System.IO;

using System.Runtime.CompilerServices;

using System.Text.Json;

using System.IO.Pipelines;

using System.Net.Http;

using System.Threading;

using System.Threading.Channels;

using System.Threading.Tasks;

​

static class Program

{

​

    public static async Task Main(string[] args)

    {

        // tweak these values based on testing

        // increasing the readBufferCapacity will increase memory usage, but might allow better throughput

        // it depends on the speed of the writer

        int readBufferCapacity = 2048;

​

        // this batches the work, set this to an appropriate batch level

        // increasing this will increase memory usage but might allow better throughput

        int writeBatchSize = 100;

​

​

        // this solution will use a PipeReader to read Sequences of bytes from the source stream

        // the source stream is a chunking networkstream when using the HttpCompletionOption.ResponseHeadersRead option

        // the pipereader will read single buffers from the stream, a sequencereader will read contiguous sequences from the stream until a newline is found

        // when a newline is found, the read sequence will be parsed to a JsonDocument and written to a Channel

        //

        // we write to a channel to optimize throughput. This allows us to prefetch items from the network stream while processing the json

        // 

        // in this example we're processing the json items in batches of writeBatchSize, this will allow us to tweak the write portion for maximum throughput

        // for more advanced scenario's like multiple async transformations we could look into projects like https://github.com/Open-NET-Libraries/Open.ChannelExtensions 

        // or https://github.com/i3arnon/MoreAsyncLINQ 

        using var httpClient = new HttpClient();

​

        await foreach (var buffer in httpClient

                           .GetAsJsonDelimitedFromAsync("https://poc-jsond.free.beeceptor.com/json", readBufferCapacity)

                           .Buffer(writeBatchSize))

        {

            //simulate processing (like writing to sql)

            Console.WriteLine($"Processing {buffer.Count} items");

            await Task.Delay(500);

        }

    }

}

​

​

public static class AsyncEnumerableExtensions

{

    public static async IAsyncEnumerable<IReadOnlyCollection<T>> Buffer<T>(this IAsyncEnumerable<T> source,

        int batchSize, [EnumeratorCancellation] CancellationToken ct = default)

​

    {

        var buffer = new List<T>(batchSize);

​

        await foreach (var item in source.WithCancellation(ct))

        {          

            buffer.Add(item);

​

            if (buffer.Count == batchSize)

            {

                // note!! we're sharing the buffer with the consumer here

                // so the consumer must not share this as that will lead to race conditions

                yield return buffer;

​

                buffer.Clear();

            }

        }

​

        // return remainder

        if (buffer.Count > 0)

        {

            yield return buffer;

        }

    }

}

​

public static class HttpClientJsonDelimitedExtensions

{

    public static async IAsyncEnumerable<JsonDocument> GetAsJsonDelimitedFromAsync(this HttpClient httpClient,

        string url, int bufferCapacity = 1024, [EnumeratorCancellation] CancellationToken ct = default)

    {

        using var response = await httpClient.GetAsync(url, HttpCompletionOption.ResponseHeadersRead, ct);

        await using var stream = await response.Content.ReadAsStreamAsync(ct);

​

        await foreach (var json in stream.ReadAsJsonAsync(bufferCapacity).WithCancellation(ct))

        {

            yield return json;

        }

    }

}

​

public static class StreamJsonDelimitedExtensions

{

    private static readonly byte[] NewLineCharacters = "\n\r"u8.ToArray();

​

    public static async IAsyncEnumerable<JsonDocument> ReadAsJsonAsync(this Stream source, int bufferCapacity = 1024, [EnumeratorCancellation] CancellationToken ct = default)

    {

        var channel = Channel.CreateBounded<JsonDocument>(new BoundedChannelOptions(bufferCapacity)

        {

            SingleWriter = true,

            SingleReader = true,

            FullMode = BoundedChannelFullMode.Wait

        });

        var writerTask = source.WriteJsonToChannelAsync(channel.Writer, ct);

​

        await foreach (var item in channel.Reader.ReadAllAsync(ct))

        {

            yield return item;

        }

​

        // await writertask to ensure it was ended and we do not leak  any running tasks

        // note: this might throw if an exception was raised in the writer

        await writerTask;

    }

​

    public static async Task WriteJsonToChannelAsync(this Stream stream, ChannelWriter<JsonDocument> writer, CancellationToken token)

    {

        var pipeReader = PipeReader.Create(stream, new StreamPipeReaderOptions(leaveOpen: true));

        var items = new Queue<JsonDocument>();

        while (!token.IsCancellationRequested)

        {

            var result = await pipeReader.ReadAsync(token); // read from the pipe

​

            var buffer = result.Buffer;

​

            var position= ReadItems(buffer, items, result.IsCompleted, token); // read complete items from the current buffer

​

            while (items.TryDequeue(out var item))

            {

                // this will block when the reader is slow

                // applying backpressure

                await writer.WriteAsync(item, token);

            }

​

            if (result.IsCompleted)

                break; // exit if we've read everything from the pipe

​

            pipeReader.AdvanceTo(position, buffer.End); //advance our position in the pipe

        }

​

        await pipeReader.CompleteAsync();

        writer.TryComplete();

    }

​

    [MethodImpl(MethodImplOptions.AggressiveInlining)]

    private static SequencePosition ReadItems(in ReadOnlySequence<byte> sequence, Queue<JsonDocument> items,

        bool isCompleted, CancellationToken token)

    {

        var reader = new SequenceReader<byte>(sequence);     

​

        while (!reader.End && !token.IsCancellationRequested) // loop until we've read the entire sequence

        {

            if (reader.TryReadToAny(out ReadOnlySequence<byte> itemBytes, NewLineCharacters, advancePastDelimiter: true)) // we have an item to handle

            {

                var item = JsonDocument.Parse(itemBytes);

                items.Enqueue(item);

            }

            else if (isCompleted) // read last item which has no final delimiter

            {

                var item = JsonDocument.Parse(sequence.Slice(reader.Position));

                items.Enqueue(item);

                reader.Advance(sequence.Length); // advance reader to the end

            }

            else // no more items in this sequence

            {

                break;

            }

        }

​

        return reader.Position;

    }

}