Response streaming with IAsyncEnumerable

Response streaming lets a grain method return IAsyncEnumerable<T> so one caller can consume a logically single grain call’s response incrementally. The caller addresses a grain as for any other grain call, and then pulls results as they’re produced.

Use response streaming for a query or command whose results are naturally incremental. A response stream doesn’t create a durable subscription, retain results, or multicast them to other consumers. For those capabilities, consider Orleans Streams.

Define and implement a response-streaming method

Declare IAsyncEnumerable<T> directly on the grain interface. A cancellation token is optional, as with other grain methods:

public interface IExportGrain : IGrainWithStringKey
{
    IAsyncEnumerable<ExportRow> ExportRows(
        CancellationToken cancellationToken = default);
}

[GenerateSerializer]
public sealed record ExportRow(
    [property: Id(0)] int RowNumber,
    [property: Id(1)] string Payload);

An async iterator can produce each result with yield return. Apply EnumeratorCancellationAttribute so the iterator observes cancellation requested while the caller enumerates it:

public sealed class ExportGrain : Grain, IExportGrain
{
    public async IAsyncEnumerable<ExportRow> ExportRows(
        [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        for (var rowNumber = 0; rowNumber < 1_000; rowNumber++)
        {
            await Task.Delay(
                TimeSpan.FromMilliseconds(10),
                cancellationToken);
            yield return new ExportRow(rowNumber, $"row-{rowNumber}");
        }
    }
}

Consume a streamed response

Use await foreach to process each result. The response stream starts when the caller requests the first element, not when the grain method returns the enumerable:

public static async Task ConsumeStream(IExportGrain grain)
{
    await foreach (var row in grain.ExportRows())
    {
        await ProcessRow(row);
    }
}

Leaving an await foreach loop disposes its enumerator, including when the loop exits with break or an exception.

Control response batching

Orleans batches synchronously available elements to reduce network round trips, up to 100 elements by default. Use WithBatchSize to change that limit:

public static async Task ConsumeInBatches(IExportGrain grain)
{
    await foreach (var row in grain.ExportRows().WithBatchSize(25))
    {
        await ProcessRow(row);
    }
}

Call WithBatchSize directly on the value returned by the grain method and before wrappers such as WithCancellation. After another operator wraps the enumerable, WithBatchSize has no Orleans request to configure and has no effect. A batch size of 1 sends one element per request.

Batching doesn’t cause Orleans to read an unbounded number of elements ahead. The caller’s next MoveNextAsync request drives production, and a batch contains only elements that become synchronously available, up to the configured limit.

Cancel response streaming

Supply a token as a grain method argument, through WithCancellation, or both. Orleans links distinct tokens so cancellation of either stops the enumeration. Call WithBatchSize first when using both extensions:

public static async Task ConsumeWithCancellation(
    IExportGrain grain,
    CancellationToken cancellationToken)
{
    try
    {
        await foreach (var row in grain
            .ExportRows()
            .WithBatchSize(25)
            .WithCancellation(cancellationToken))
        {
            await ProcessRow(row);
        }
    }
    catch (OperationCanceledException)
        when (cancellationToken.IsCancellationRequested)
    {
        // The caller requested cancellation.
    }
}

Cancellation is cooperative and surfaces to the caller as OperationCanceledException. The response-streaming method must observe its token and pass it to cancellation-aware operations. See Cancel Orleans grain calls for delivery and failure semantics.

Handle an interrupted response stream

An exception thrown while producing the response stream propagates to the caller with its original exception type. The caller instead receives EnumerationAbortedException if the grain deactivates during enumeration or the silo removes an enumerator which the caller left idle:

public static async Task ConsumeWithInterruptionHandling(
    IExportGrain grain)
{
    try
    {
        await foreach (var row in grain.ExportRows())
        {
            await ProcessRow(row);
        }
    }
    catch (EnumerationAbortedException)
    {
        // The grain deactivated or Orleans removed an idle enumerator.
    }
}

Idle-enumerator cleanup runs periodically using ResponseTimeout as its interval. Don’t hold an enumerator open while doing unrelated long-running work. If processing an element can take a long time, decouple that work from pulling the next element or use a messaging abstraction with a lifetime independent of one grain call.

Choose between response streaming and Orleans Streams

Concern	Response streaming with IAsyncEnumerable<T>	Orleans Streams
Communication shape	Logically one grain call, one producer, and one caller	Multicast pub/sub with independent producers and subscribers
Lifetime	One live enumeration, ending on completion, disposal, cancellation, deactivation, or idle cleanup	Independent of any one grain call; subscriptions can survive activation changes
Flow control	Pull-based; MoveNextAsync drives production, with bounded batching	Provider-dependent delivery and buffering
Persistence and replay	None	Optional and provider-dependent
Best fit	Progressively return one command or query result	Publish events to multiple or long-lived subscriptions

See Choose an Orleans messaging abstraction for observers, broadcast channels, and other alternatives.