LineDelimitedProgram.cs

CSharp · Console

/Program.cs

using System;
using System.Buffers;
using System.Collections.Generic;
using System.IO;
using System.IO.Pipelines;
using System.Text;
using System.Text.Json;
using System.Threading.Tasks;

class Program
{
	static readonly byte[] NewLineChars = {(byte)'\r', (byte)'\n'};
	static readonly byte[] WhiteSpaceChars = {(byte)'\r', (byte)'\n', (byte)' ', (byte)'\t'};
	private static async Task Main()
	{
		JsonSerializerOptions jsonOptions = new(JsonSerializerDefaults.Web);
		var json = "{\"some\":\"thing1\"}\r\n{\"some\":\"thing2\"}\r\n{\"some\":\"thing3\"}";
		var contentStream = new MemoryStream(Encoding.UTF8.GetBytes(json));
		var pipeReader = PipeReader.Create(contentStream);
		await foreach (var foo in ReadItemsAsync<Foo>(pipeReader, jsonOptions))
		{
			Console.WriteLine($"foo: {foo.Some}");
		}
	}

	static async IAsyncEnumerable<TValue> ReadItemsAsync<TValue>(PipeReader pipeReader, JsonSerializerOptions jsonOptions = null)
	{
		while (true)
		{
			var result = await pipeReader.ReadAsync();
			var buffer = result.Buffer;
			bool isCompleted = result.IsCompleted;
			SequencePosition bufferPosition = buffer.Start;
			while (true)
			{
				var(value, advanceSequence) = TryReadNextItem<TValue>(buffer, ref bufferPosition, isCompleted, jsonOptions);
				if (value != null)
				{
					yield return value;
				}

				if (advanceSequence)
				{
					pipeReader.AdvanceTo(bufferPosition, buffer.End); //advance our position in the pipe
					break;
				}
			}

			if (isCompleted)
				yield break;
		}
	}

	static (TValue, bool) TryReadNextItem<TValue>(ReadOnlySequence<byte> sequence, ref SequencePosition sequencePosition, bool isCompleted, JsonSerializerOptions jsonOptions)
	{
		var reader = new SequenceReader<byte>(sequence.Slice(sequencePosition));
		while (!reader.End) // loop until we've come to the end or read an item
		{
			if (reader.TryReadToAny(out ReadOnlySpan<byte> itemBytes, NewLineChars, advancePastDelimiter: true))
			{
				sequencePosition = reader.Position;
				if (itemBytes.TrimStart(WhiteSpaceChars).IsEmpty)
				{
					continue;
				}

				return (JsonSerializer.Deserialize<TValue>(itemBytes, jsonOptions), false);
			}
			else if (isCompleted)
			{
				// read last item
				var remainingReader = sequence.Slice(reader.Position);
				using var memoryOwner = MemoryPool<byte>.Shared.Rent((int)reader.Remaining);
				remainingReader.CopyTo(memoryOwner.Memory.Span);
				reader.Advance(remainingReader.Length); // advance reader to the end
				sequencePosition = reader.Position;
				if (!itemBytes.TrimStart(WhiteSpaceChars).IsEmpty)
				{
					return (JsonSerializer.Deserialize<TValue>(memoryOwner.Memory.Span, jsonOptions), true);
				}
				else
				{
					return (default, true);
				}
			}
			else
			{
				// no more items in sequence
				break;
			}
		}

		// PipeReader needs to read more
		return (default, true);
	}
}

public class Foo
{
	public string Some
	{
		get;
		set;
	}
}