Table of Contents

Frame records from a pipe

Advanced · C#. A PipeReader hands over whatever bytes have arrived; parse the whole records, return the partial tail as examined but not consumed, and read the count of a runtime-sized message before waiting for the rest.

Run this example

Prerequisites: the repository's .NET 10 SDK and a checkout of this source. Run from the repository root:

dotnet run --project docs/examples/CStructSharp.Docs.Examples.csproj -c Release -- pipe-reader

The runner checks five fixed-size frames delivered in three chunks that cross frame boundaries, parsed with ParseMany over the whole frames and AdvanceTo for the rest, then three count-prefixed messages parsed only once every byte of each has arrived. Success includes PASS pipe-reader.

This example uses the C# API. Browser capabilities and result shapes are described in the browser guide.

Complete program

The layout, options, input bytes, helper methods, and required types are all included. To adapt it outside the repository, create a .NET 10 console project, add CStructSharp, and replace Program.cs with this complete file. These examples follow the source version; use a matching package when testing a release.

Download the complete C# source.

// Generated from executable documentation examples. Edit the source region, then regenerate.
using System;
using System.IO;
using System.IO.Pipelines;
using System.Linq;
using System.Buffers;
using System.Collections.Generic;
using System.Dynamic;
using System.Globalization;
using System.Numerics;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;
using CStructSharp;
using CStructSharp.Codecs;
using CStructSharp.Diagnostics;
using CStructSharp.Introspection;
using CStructSharp.Values;

internal static partial class Program
{
    public static async Task Main()
    {
        await PipeReaderFraming();
        Console.WriteLine("PASS pipe-reader");
    }

    /// <summary>Frames fixed-size and count-prefixed messages, retaining incomplete input across pipe reads.</summary>
    private static async Task PipeReaderFraming()
    {
        // Four-byte frames arrive on a pipe in chunks that do not line up with the frame boundaries - exactly what a
        // socket delivers. A PipeReader over a NetworkStream (PipeReader.Create(stream)) is used the same way.
        var layout = new CStruct("struct frame { uint16 id; uint16 value; };");
        int size = layout.GetStructSizeInBytes("frame");
        var pipe = new Pipe();

        // Supply chunks independently of the consumer so record boundaries need not match read boundaries.
        Task producer = Task.Run(async () =>
        {
            byte[] frames = [1, 0, 10, 0, 2, 0, 20, 0, 3, 0, 30, 0, 4, 0, 40, 0, 5, 0, 50, 0];
            foreach (int[] chunk in new[] { new[] { 0, 6 }, new[] { 6, 7 }, new[] { 13, 7 } })
            {
                await pipe.Writer.WriteAsync(frames.AsMemory(chunk[0], chunk[1]));
            }

            await pipe.Writer.CompleteAsync();
        });

        var ids = new List<ushort>();
        while (true)
        {
            var result = await pipe.Reader.ReadAsync();
            ReadOnlySequence<byte> buffer = result.Buffer;

            // Parse the whole frames the buffer holds - ParseMany reads a ReadOnlySequence<byte> directly, one frame per
            // step - and hand the partial frame at the end back to the pipe: consumed up to the last whole frame,
            // examined to the end, so the next ReadAsync waits for more bytes instead of returning the same ones.
            long whole = buffer.Length / size * size;
            foreach (StructValue frame in layout.ParseMany(buffer.Slice(0, whole), "frame"))
            {
                ids.Add(frame.Get<ushort>("id"));
            }

            pipe.Reader.AdvanceTo(buffer.GetPosition(whole), buffer.End);
            if (result.IsCompleted)
            {
                True(buffer.Length == whole, "the producer ended on a frame boundary");
                break;
            }
        }

        await producer;
        await pipe.Reader.CompleteAsync();
        Equal("1,2,3,4,5", string.Join(",", ids));

        // A count-prefixed message has no fixed size: read the count first (a whole header, or wait), work out the
        // message's length from it, and parse the message only when every byte of it is there.
        var messages = new CStruct("struct message { uint8 count; uint8 payload[count]; };");
        var chunks = new Pipe();
        await chunks.Writer.WriteAsync(new byte[] { 2, 0xAA, 0xBB, 3, 0xCC });
        await chunks.Writer.WriteAsync(new byte[] { 0xDD, 0xEE, 0 });
        await chunks.Writer.CompleteAsync();
        var lengths = new List<int>();
        while (true)
        {
            var result = await chunks.Reader.ReadAsync();
            ReadOnlySequence<byte> buffer = result.Buffer;
            while (messages.TryReadValue(buffer, "message.count", out byte count) && buffer.Length >= 1 + count)
            {
                StructValue message = messages.Parse(buffer.Slice(0, 1 + count), "message");
                lengths.Add(message.Get<byte[]>("payload").Length);
                buffer = buffer.Slice(1 + count);
            }

            chunks.Reader.AdvanceTo(buffer.Start, buffer.End);
            if (result.IsCompleted)
            {
                break;
            }
        }

        await chunks.Reader.CompleteAsync();
        Equal("2,3,0", string.Join(",", lengths));
    }

    private static void Equal<T>(T expected, T actual)
    {
        if (!EqualityComparer<T>.Default.Equals(expected, actual))
        {
            throw new InvalidOperationException($"Expected '{expected}', received '{actual}'.");
        }
    }

    private static void True(bool condition, string message)
    {
        if (!condition)
        {
            throw new InvalidOperationException(message);
        }
    }
}

Try it and diagnose mistakes

Advance to buffer.End as consumed instead of the last whole frame.

Answer: The partial frame is discarded with the chunk and the next chunk starts mid-frame; the ids after the first chunk boundary are wrong or the parse fails. The program contains assertions for its original inputs. When changing an input intentionally, update the expected assertion too; an unchanged assertion is not evidence that the new value is wrong.

Continue with the related guide or choose another recipe.

Edit this page