Streaming Async Enumerable - lobodava/artisan-orm GitHub Wiki

For queries that return more rows than you want to buffer in memory, ReadToAsyncEnumerable<T> and ReadAsAsyncEnumerable<T> stream rows one at a time as IAsyncEnumerable<T>, with first-class CancellationToken support.

Added in v4 — see What's New in v4.

Use streaming over ReadToList<T> when any of these is true:

• The result set is large enough that holding it all in RAM is uncomfortable (CSV exports, batch processing pipelines, archival reads).
• You want to start working on each row immediately without waiting for the whole list (latency-sensitive feeds, server-sent events, gRPC streaming responses).
• You want the consumer to control how many rows to read (early break after finding what you need, take-N patterns).

Use ReadToList<T> when:

• The result is small and bounded (a page of 50 users, a lookup table).
• You will iterate the rows multiple times.
• You need to materialise the count up front.

Both exist on SqlCommand, RepositoryBase, and SqlDataReader. The command/repository variants own the reader they create. The reader variants use a reader already owned by the caller and do not advance to another result set.

Command and reader overloads can also accept an explicit factory. In the source being prepared for 4.1.0, that factory handles each row even when its output is scalar/nullable and column zero is SQL NULL. It chooses the source columns and null substitution. Without a factory, the normal registered-mapper/scalar path applies. See Inline Mappers; this correction does not change the lifetime and cancellation boundaries below.

public class RecordRepository : RepositoryBase
{
    public RecordRepository(string connStr) : base(connStr) { }

    public IAsyncEnumerable<Record> GetAllRecords(CancellationToken ct = default) =>
        ReadToAsyncEnumerable<Record>("dbo.GetRecords", ct);
}

// Caller
await foreach (var record in repo.GetAllRecords(stoppingToken))
{
    await ProcessAsync(record, stoppingToken);
}

Once enumeration starts, the underlying reader stays open until enumeration finishes or the enumerator is disposed. await foreach disposes the enumerator on normal completion, break, or an exception, including an observed cancellation.

Command/repository streams dispose the reader they create. Reader disposal closes the connection only if the stream opened it; a connection already open remains the caller's responsibility. The connection object itself belongs to the repository/caller. Keep the repository and any caller-owned command alive throughout enumeration. Reader-level streams do not dispose the supplied reader.

Requesting cancellation does not by itself dispose an idle enumerator. Always finish or dispose enumeration; an unenumerated sequence has not opened a reader/connection at all.

Two ways to pass the token, both equivalent:

// Pass at call time
await foreach (var r in repo.ReadToAsyncEnumerable<Record>("dbo.GetRecords", ct))
    await ProcessAsync(r, ct);

// Pass via WithCancellation when the producer didn't take a CT
await foreach (var r in cmd.ReadToAsyncEnumerable<Record>().WithCancellation(ct))
    await ProcessAsync(r, ct);

The token is forwarded to command execution and row reads. Cancellation may surface as OperationCanceledException (including derived types), but an in-flight SQL Server cancellation can also surface as a provider SqlException. Do not accept an arbitrary exception as proof of cancellation in tests.

In 4.1.0, opening a closed connection also uses OpenAsync(ct). The library checks cancellation between rows, including when the provider has buffered results. Synchronous mapper/application work must check the token itself when appropriate. These corrections do not change resource ownership: a connection opened by the caller remains caller-owned.

Iteration is lazy — rows are pulled from SQL Server only as you consume them. Breaking out of the loop is cheap and clean:

await foreach (var record in repo.ReadToAsyncEnumerable<Record>("dbo.GetAllRecords"))
{
    if (record.Name == "STOP") break;   // stops consuming rows; the provider may already have buffered data
    await ProcessAsync(record);
}

When you need to configure parameters or share the command with other reads:

public async IAsyncEnumerable<Record> StreamRecordsAsync(
    int minId,
    [EnumeratorCancellation] CancellationToken ct = default)
{
    using var cmd = CreateCommand();
    cmd.UseProcedure("dbo.GetRecordsByMinId");
    cmd.AddIntParam("@MinId", minId);

    await foreach (var record in cmd.ReadToAsyncEnumerable<Record>(ct))
        yield return record;
}

The [EnumeratorCancellation] attribute lets the caller's WithCancellation(...) token flow into your inner streaming call.

ReadAsAsyncEnumerable<T> works the same way but builds the mapper via reflection on the first call and caches it (see Auto-Mapping):

public IAsyncEnumerable<UserSummary> StreamUsers(CancellationToken ct = default) =>
    ReadAsAsyncEnumerable<UserSummary>(
        "select Id, Login, Name, Email from dbo.Users",
        ct);

The underlying SqlConnection is held open for the entire iteration. Two implications:

1. Don't issue other repository calls inside the loop on the same repo. The connection is busy with the open reader. If you need a second query mid-iteration, take that query's data first (into a list / dictionary) and use it during the iteration, or open a second repository instance.

2. Dispose a started enumerator promptly. await foreach does this automatically, including after break. If manually calling GetAsyncEnumerator, use await using or DisposeAsync; abandoning a started enumerator may retain its reader/connection. Merely storing a sequence without starting enumeration does not open a connection.

reader.ReadToAsyncEnumerable<T>(ct) and reader.ReadAsAsyncEnumerable<T>(ct) consume only the current result set. They do not call NextResult() and do not dispose the supplied reader. After finishing enumeration, use await reader.NextResultAsync(ct) explicitly if another result set is part of the contract. This differs from materializing reader methods such as ReadToList<T>(), which advance by default.

Do not return an async stream from GetByReaderAsync's projection callback and consume it later. That helper accepts a synchronous callback and disposes the reader as soon as it returns. Own the reader for the entire enumeration instead.

ReadToAsyncEnumerable<int>, <string>, <DateTime> etc. work too, useful for "ids only" passes:

await foreach (var id in repo.ReadToAsyncEnumerable<int>(
    "select Id from dbo.Records order by Id"))
{
    // ...
}

Sources: command streaming, reader streaming, and repository lifetime.

See also:

• Read Methods Understanding
• Auto-Mapping
• What's New in v4