PipelinesDemo.cs

using System;
using System.Buffers;
using System.Buffers.Binary;
using System.IO.Pipelines;
using System.Numerics;
using System.Threading.Tasks;

namespace PipelinesDemo
{
    class Program
    {
        private const int delayMS = 10;
        
        static async Task Main(string[] args)
        {
            var pipeline = new Pipe();

            await Task.WhenAll(Write(pipeline.Writer), Read(pipeline.Reader));
        }

        static async Task Write(PipeWriter writer)
        {
            var random = new Random();
            while (true)
            {
                var vector = new Vector3((float)random.NextDouble() * 100, (float)random.NextDouble() * 100, (float)random.NextDouble() * 100);
                Console.WriteLine("Writing {0}", vector);
                WriteVector(vector, writer);
                await writer.FlushAsync();
                await Task.Delay(delayMS);
            }

            Console.WriteLine("Writer Completed");
            await writer.CompleteAsync();
        }

        private static void WriteVector(Vector3 vector, PipeWriter writer)
        {
            var span = writer.GetSpan(3 * sizeof(float));
            BinaryPrimitives.WriteSingleLittleEndian(span.Slice(0 * sizeof(float), sizeof(float)), vector.X);
            BinaryPrimitives.WriteSingleLittleEndian(span.Slice(1 * sizeof(float), sizeof(float)), vector.Y);
            BinaryPrimitives.WriteSingleLittleEndian(span.Slice(2 * sizeof(float), sizeof(float)), vector.Z);
            writer.Advance(sizeof(float) * 3);
        }

        static async Task Read(PipeReader reader)
        {
            while (true)
            {
                if (!reader.TryRead(out ReadResult result))
                {
                    await Task.Delay(delayMS * 2);
                    continue;
                }

                if (result.IsCanceled)
                    break;


                var buffer = result.Buffer;

                if (buffer.Length < sizeof(float) * 3)
                {
                    reader.AdvanceTo(buffer.GetPosition(0));
                    continue;
                }
                
                if (!TryReadVector(buffer, out var vector))
                {
                    reader.AdvanceTo(buffer.GetPosition(0));
                    break;
                }
                
                Console.WriteLine("Read {0}", vector);
                reader.AdvanceTo(buffer.GetPosition(sizeof(float) * 3));

                if (result.IsCompleted)
                    break;
            }

            Console.WriteLine("Reader Completed");
            await reader.CompleteAsync();
        }

        private static bool TryReadVector(ReadOnlySequence<byte> buffer, out Vector3 vector)
        {
            vector = default;
            var reader = new SequenceReader<byte>(buffer);
            if (!reader.TryReadLittleEndian(out int xInt))
                return false;
            if (!reader.TryReadLittleEndian(out int yInt))
                return false;
            if (!reader.TryReadLittleEndian(out int zInt))
                return false;
            
            vector = new Vector3(BitConverter.Int32BitsToSingle(xInt), BitConverter.Int32BitsToSingle(yInt), BitConverter.Int32BitsToSingle(zInt));
            return true;
        }
    }
}