Coroutine Streams
Coroutine stream methods move one bounded synchronous operation onto the coroutine worker service. This page covers eager tasks, asynchronous block and line generators, timeout preservation, shared ownership, cancellation, and continuation execution.
Await Bounded Input
coRead... methods return an eager CoTask.
Work can begin before the caller awaits it.
Awaiting consumes the single result and resumes when the matching bounded read completes.
The result is unchanged from synchronous input: callers still handle Data, Finished, and Timeout.
Exceptions raised on the worker are rethrown when the task result is observed.
/// Stream `CoTask` operations start eagerly and run the matching bounded synchronous call on the worker service.
/// Awaiting preserves the normal result status and rethrows stream errors at the observation point.
void awaitByteRead() {
const auto input = std::make_shared<ScriptedByteInputStream>(std::vector<uint8_t>{37U});
auto task = readRiverGauge(input);
waitForTask(task);
el::io::printLine("Vandstand: "_el, task.result(), " cm"_el);
}
Vandstand: 37 cm
Await Owned Atomic Output
Asynchronous output takes an owned ByteBlock or String so no borrowed view outlives the caller’s frame.
It preserves atomic acceptance: success means the complete request was queued, while timeout means none was accepted.
Use it when waiting for output back pressure must not stall a parser or event-loop coroutine.
/// Asynchronous text output owns its string until the complete atomic request is accepted.
/// This is useful when a producer coroutine must not wait for output back pressure on its current thread.
void awaitTextWrite() {
const auto output = el::AnyStringBuilderStream::create();
auto task = output->coWriteLine("Flodprofil: rolig strøm ved østbredden"_el);
waitForTask(task);
el::io::printLine("Skrivning accepteret: "_el, task.result().isSuccess());
el::io::print(output->takeString());
}
Skrivning accepteret: true
Flodprofil: rolig strøm ved østbredden
Generate Byte or Text Blocks
coReadBlocks() returns a lazy CoAsyncGenerator.
Each next() operation is awaited and yields an owned read result.
Timeout results are visible to the consumer; Finished completes the generator without yielding another value.
Only one next() operation may be outstanding.
/// `coReadBlocks()` is a lazy, single-pass sequence of owned blocks.
/// Advancing it is asynchronous; data and timeout results are yielded, while end-of-stream completes the generator.
void processBlockGenerator() {
const auto input = std::make_shared<ScriptedByteInputStream>(std::vector<uint8_t>{1U, 1U, 2U, 3U, 5U}, 2U);
auto task = countRiverBlocks(input);
waitForTask(task);
el::io::printLine("Data blocks from the stream: "_el, task.result());
}
Data blocks from the stream: 3
Generate Complete Lines
coReadLines() applies the same lazy sequence to decoded text.
Line endings, maximum-length fragments, final unterminated lines, and code-point boundaries match synchronous
readLine() behavior.
/// `coReadLines()` combines decoded, code-point-safe line input with asynchronous iteration.
/// It yields complete lines or bounded fragments and completes normally after the final unterminated line.
void processLineGenerator() {
const auto directory = createStreamDemoDirectory("flod"_el);
const auto path = directory->path() / "målinger.txt"_el;
path.content().writeTextOrThrow("Nord: 12 cm\nMidte: 37 cm\nSyd: 19 cm"_el);
const auto input = path.content().openTextInputStream();
auto task = countRiverLines(input);
waitForTask(task);
el::io::printLine("Measurement lines: "_el, task.result());
}
Measurement lines: 3
Keep Timeout as Flow Control
Moving work to another thread does not create an unbounded operation. The configured stream timeout remains active, and the task or generator yields the same timeout result. The coroutine can then observe cancellation, update progress, or retry.
/// Coroutine operations preserve timeout as a normal stream result.
/// Awaiting moves bounded work away from the caller thread; it does not convert flow control into an exception.
void handleCoroutineTimeout() {
const auto input = std::make_shared<ScriptedByteInputStream>(std::vector<uint8_t>{9U}, 1U, 1U);
auto first = input->coRead(el::ByteLength{1U});
waitForTask(first);
el::io::printLine("First coroutine timeout: "_el, first.result().isTimeout());
auto second = input->coRead(el::ByteLength{1U});
waitForTask(second);
el::io::printLine("Anden coroutine fik data: "_el, second.result().hasData());
}
First coroutine timeout: true
Anden coroutine fik data: true
Cancel by Releasing the Task
Destroying or cancelling an incomplete task requests cancellation. Already-running bounded native work may finish, but the cancelled user continuation is not resumed afterward. Destroying a generator ends iteration and destroys its producer frame.
/// Destroying or cancelling an incomplete `CoTask` suppresses its continuation.
/// Already-running bounded stream work may finish, and its retained stream ownership is released afterward.
void cancelPendingTask() {
auto input = std::make_shared<ScriptedByteInputStream>(std::vector<uint8_t>{8U}, 1U, 0U, true, true);
auto weak = std::weak_ptr<ScriptedByteInputStream>{input};
auto task = input->coRead(el::ByteLength{1U});
while (input->readCount() == 0U) {
std::this_thread::yield();
}
task.cancel();
input->release();
input.reset();
const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds{2};
while (!weak.expired() && std::chrono::steady_clock::now() < deadline) {
std::this_thread::yield();
}
el::io::printLine("Annulleret opgave frigav streamen: "_el, weak.expired());
}
Annulleret opgave frigav streamen: true
Continuations run on the thread that completes the awaited operation. There is no automatic return to the initiating thread, so marshal UI or thread-affine work explicitly.