Exporting reader rows through a channel

For several active report workers with an HTTP status/cancellation API, follow queued customer reports. This recipe focuses on copying a forward-only database reader into a channel safely.

A background export has two different kinds of forward-only source: a database reader advances one cursor, while a channel transfers independent work items between stages. Keeping that difference visible prevents a subtle bug: retaining several references to the same reader row does not produce several saved rows.

This example reads document identifiers and text from one query, snapshots each row, and writes a bounded stream of export jobs to a file worker. It uses ordinary DbCommand, ChannelWriter, ChannelReader and File APIs. No custom adapter factory is required for these recognized reader/channel sources.

Projecting rows before handing them off

The query returns two non-null columns in this order: a Guid identifier and a String body. The caller owns an open connection and its command. The producer owns only the reader it creates; it borrows the command and channel writer.

import Global.System
import Global.System.Data.Common
import Global.System.IO
import Global.System.Threading.Channels

public class ExportJob
    public Id as Guid = Guid.Empty
    public Contents as String = ""
end class

public class DocumentExportProducer
    public async action LoadAsync(command as DbCommand, output as ChannelWriter <of ExportJob>)
        require command is not nothing
        require output is not nothing
        Cancellation.ThrowIfCanceled()

        using reader = await command.ExecuteReaderAsync(Cancellation.Token)
            consume row from reader
                Cancellation.ThrowIfCanceled()
                if row.IsDBNull(0) or row.IsDBNull(1) then
                    throw new InvalidDataException("Document identifier and body must be present")
                end if

                let job = new ExportJob()
                job.Id = row.GetGuid(0)
                job.Contents = row.GetString(1)
                if job.Id = Guid.Empty then
                    throw new InvalidDataException("Document identifier must not be empty")
                end if
                await output.WriteAsync(job, Cancellation.Token)
            end consume
        end using
    end action
end class

The job contains independent values before WriteAsync can wait for channel capacity. The next ReadAsync can change the row view without changing an earlier job. Do not enqueue row itself. The producer publishes each job once and does not mutate it afterwards.

Reader consumption ends at the current result set. It does not call NextResult, close the writer or dispose the caller's command/connection. The surrounding using owns the created reader and performs its resolved type-directed resource cleanup on success, exception or cancellation. Asynchronous acquisition and row reads do not transfer that cleanup responsibility to the channel.

Consuming jobs sequentially

The worker receives a trusted, configured directory. Filenames come from Guid identifiers, never from text stored in the database. This avoids treating arbitrary row content as a relative filesystem path.

import Global.System.IO
import Global.System.Threading.Channels

public class DocumentExportWorker
    public async action RunAsync(jobs as ChannelReader <of ExportJob>, outputDirectory as String)
        require jobs is not nothing
        require outputDirectory is not whitespace
        Cancellation.ThrowIfCanceled()
        Directory.CreateDirectory(outputDirectory)

        consume job from jobs
            Cancellation.ThrowIfCanceled()
            require job is not nothing
            let filename = job.Id.ToString("N") + ".txt"
            let path = Path.Combine(outputDirectory, filename)
            await File.WriteAllTextAsync(path, job.Contents, Cancellation.Token)
        end consume
    end action
end class

Plain consume waits for the next item while an open channel is empty. Once the writer completes and its buffer is drained, the loop ends. Each file write finishes before the next job is processed; this is asynchronous I/O without overlapping worker bodies.

The explicit cancellation checks cover CPU-only work before the native cancellable operations. Cancellation is cooperative. A cancelled write can leave a partial file, and a completed file is not rolled back if a later job fails. The sample replaces an existing file with the same identifier. A production export may need temporary files and a publication step, a manifest, checksums or an external checkpoint; none is supplied implicitly by consume.

Connect the producer and consumer

The producer and file worker above contain the application logic in DirectSharp. A bounded channel connects them: the producer projects each database row into a retained job, and the worker writes one job at a time. This keeps the reader's borrowed current row out of later work.

The submitting operation owns the connection, command and linked cancellation source. It starts both stages, completes the writer when production finishes, and observes both outcomes before releasing their resources. A producer failure must release a waiting consumer; a worker failure must release a producer blocked by a full channel. Cancellation callbacks can fail too, so the hosting policy records those failures separately from the stage outcome.

A channel transfers process-local work; it is not a durable queue or an automatic acknowledgement/retry protocol. The buffer limits job count rather than bytes. The table supplies unique non-empty Id values and non-null Body text. Ordering makes successful output reproducible for a fixed query result; database isolation still determines which rows belong to that result.

The language example stays focused on the two DirectSharp stages. Their caller chooses and owns the surrounding cancellation and settlement policy.

Failure cases to exercise

See consumable contracts and the data access example for the surrounding resource conventions.