Channels and Dataflow

Two companion packages read query results straight into a pipeline instead of a collection.

Package	Entry points	Pipeline type
Open.Database.Extensions.Channel	ToChannel<T>(), ToChannelAsync<T>()	System.Threading.Channels
Open.Database.Extensions.Dataflow	ToTargetBlock<T>(), ToTargetBlockAsync<T>(), AsSourceBlock<T>()	System.Threading.Tasks.Dataflow

They work on expressive commands, on IDbCommand and on readers, and map rows the same way as Map<T>().

Channels

ToChannel<T> returns a ChannelReader<T> immediately. The channel is completed when every row has been written, or completed with the exception if reading fails:

// Returns right away; rows are produced in the background and the channel completes when they are all written.
ChannelReader<Post> reader = connectionFactory
    .Command("select * from posts order by id")
    .ToChannel<Post>(singleReader: true);

await foreach (Post post in reader.ReadAllAsync())
    received.Add(post);

Dataflow

ToTargetBlock<T> posts each row to a block. The complete argument controls the target: with true, the block is completed after every row has been delivered and the reader has been released, or faulted with the exception if reading or mapping failed. With false, the block is left open and is never completed or faulted by the call. The returned number counts the rows read and handed to the internal buffer; delivery to the target can still be in progress.

var block = new BufferBlock<Post>();

// complete: true completes the block after every row has been delivered (or faults it if reading fails).
long count = await connectionFactory
    .Command("select * from posts order by id")
    .ToTargetBlockAsync<Post>(block, complete: true);

// The block completes once every row has been delivered and received.
var received = new List<Post>();
while (await block.OutputAvailableAsync())
    received.Add(await block.ReceiveAsync());

Honest limits

• These are conveniences for pipelines, not speed-ups. In the published benchmarks, draining a channel or a BufferBlock is slower than reading rows in a plain loop; see Performance.
• The mapped overloads read raw rows into a bounded buffer of about 1,024 rows and map them in a second stage. The channel study in the benchmark results explains the trade-off.
• The unmapped ToChannel overloads can still lose the row that was in flight when a read fails while the target is full. This is a known limitation of the preview.