Table of Contents

Streaming sequences

While typical SerializeAsync and DeserializeAsync methods exist, these are methods that complete only when the entire job is done. In particular for DeserializeAsync, this means that no subset of the deserialized data is available until it is fully deserialized.

There are times however where progressively obtaining the deserialized elements can be useful. For example, perhaps the stream contains a very long sequence of elements, and processing them incrementally instead of all at once can save memory or improve performance. Or perhaps the stream is intentionally a long-lived generator stream that emits values over long periods of time, and it is important to the receiver that values are produced and available right away, before the stream ends.

The DeserializeEnumerableAsync and DeserializePathEnumerableAsync methods address such use cases.

One must first classify the presentation of msgpack streaming values to be deserialized. Two forms are supported:

  1. A stream that contains multiple msgpack structures without any envelope (e.g. a msgpack array).
  2. A stream that contains a msgpack structure, within which is a sequence to be streamed (e.g. a msgpack array of elements).

Sequence with no envelope

A sequence of msgpack structures without an array or any other data is said to have no envelope. To asynchronously enumerate each of these structures, we use the DeserializeEnumerableAsync methods.

private static readonly MessagePackSerializer Serializer = new();

async Task ReadListAsync(PipeReader reader)
{
    await foreach (Person? item in Serializer.DeserializeEnumerableAsync<Person>(reader))
    {
        // Process item here.
    }
}

[GenerateShape]
internal partial record Person(int Age);

Sequence within an envelope

A sequence of msgpack structures that are found within a larger structure (e.g. a msgpack array) is said to have an envelope. To asynchronously enumerate each of these structures requires first parsing through the envelope preamble to navigate to the sequence. After enumerating the sequence, the remainder of the envelope is parsed in order to leave the reader positioned at valid position, at the end of the overall msgpack structure.

Navigating through the envelope is done by an expression provided to the MessagePackSerializer.StreamingEnumerationOptions<T, TElement> argument passed to any of the DeserializePathEnumerableAsync methods that accept that as a parameter.

private static readonly MessagePackSerializer Serializer = new();

async Task ReadFamilyMembersAsync(PipeReader reader)
{
    MessagePackSerializer.StreamingEnumerationOptions<Family, Person> options = new(f => f.Members);
    await foreach (Person? item in Serializer.DeserializePathEnumerableAsync(reader, options))
    {
        // Process item here.
    }
}

[GenerateShape]
internal partial record Family(Person[] Members);

internal record Person(int Age);

The paths from envelope to sequence may include stepping through properties, indexing into arrays or even dictionaries. However, not every valid C# expression will be accepted as a path.

Serializing a sequence

SerializeEnumerableAsync writes an IAsyncEnumerable<T> as concatenated top-level MessagePack values, with no array around them. That is the same framing DeserializeEnumerableAsync reads. An empty sequence writes no bytes. Each value is written as it is produced, and the destination is flushed when the unflushed buffer exceeds UnflushedBytesThreshold, so a long or unbounded source stays bounded and applies backpressure. That byte count is kept by the serializer, so a PipeWriter that cannot report how many bytes are unflushed still flushes on the same threshold. The caller keeps ownership of the pipe or stream: pipes are not completed, and streams are not closed. Cancellation is checked before each value is written and before a pending chunk is copied to the caller's pipe. If serialization fails before that copy, the pending bytes are dropped. If the copy or the destination flush then fails, the pipe may already hold that chunk and must be abandoned, because a pipe writer cannot roll those bytes back. Bytes that were flushed successfully remain.

private static readonly MessagePackSerializer Serializer = new();

async Task WritePeopleAsync(PipeWriter writer, IAsyncEnumerable<Person?> people)
{
    // Each person is one top-level MessagePack value. There is no array around them.
    await Serializer.SerializeEnumerableAsync(writer, people);
    await writer.FlushAsync();
}

[GenerateShape]
internal partial record Person(int Age);

Reference preservation is not supported for this framing. A reference id would refer to another top-level value, and DeserializeEnumerableAsync rejects reference preservation for the same reason.

Writing one array

A MessagePack array stores its length before any element, so SerializeEnumerableAsync does not write one array. Callers that need one array should buffer the values and use SerializeAsync of a T[] or IEnumerable<T>. That call still buffers every element and then writes one ordinary array.

Edit this page