Stream with CancellationToken
Learn to cancel an async stream with CancellationToken.
CancellationToken in a stream
An async stream should be cancellable from outside. You add a CancellationToken parameter to enable this. Inside the stream you check with token.ThrowIfCancellationRequested() at each step.
The consumer has two ways to pass cancellation: add .WithCancellation(token) to await foreach, or pass the token directly to the method. If you add [EnumeratorCancellation] CancellationToken token to the stream method signature, foreach passes the token automatically.
ThrowIfCancellationRequested in a stream
In an async stream ThrowIfCancellationRequested is called before each element. If cancellation was requested, OperationCanceledException is thrown and the stream ends. Example: foreach (var n in Numbers()) { token.ThrowIfCancellationRequested(); yield return n; }.
This pattern lets you stop long streams safely. When the consumer cancels the token, the stream stops at the next step; the resource is released. This is especially critical in unbounded streams (e.g. real-time data).
Exercise Stream with CancellationToken
Define static async IAsyncEnumerable<int> CountUp(CancellationToken token). In its body open a for (int i = 1; i <= 5; i++) loop; on each step call token.ThrowIfCancellationRequested();, await Task.CompletedTask; and yield return i;. Make Main a static async Task Main; create the source with var cts = new CancellationTokenSource(); (do not cancel). Sum with int sum = 0; await foreach (int n in CountUp(cts.Token)) { sum += n; } and print with Console.WriteLine(sum);.
Show example solution
In CountUp body write for (int i = 1; i <= 5; i++) { token.ThrowIfCancellationRequested(); await Task.CompletedTask; yield return i; }. Make Main static async Task Main() and write var cts = new CancellationTokenSource(); int sum = 0; await foreach (int n in CountUp(cts.Token)) { sum += n; } Console.WriteLine(sum);
Example solution:
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
class Program
{
static async IAsyncEnumerable<int> CountUp(CancellationToken token)
{
for (int i = 1; i <= 5; i++)
{
token.ThrowIfCancellationRequested();
await Task.CompletedTask;
yield return i;
}
}
static async Task Main()
{
var cts = new CancellationTokenSource();
int sum = 0;
await foreach (int n in CountUp(cts.Token))
{
sum += n;
}
Console.WriteLine(sum);
}
}Test yourself
-
Match stream cancellation concepts
- WithCancellation(token)
- Passes token to foreach
- ThrowIfCancellationRequested()
- Throws if cancelled
- [EnumeratorCancellation]
- foreach passes automatically
- CancellationToken
- Cancellation control type
-
How is CancellationToken passed to a stream?
- Only through a global variable
- With Thread.Sleep
- WithCancellation(token) or as a parameter
- With Task.Run
Show answer
WithCancellation(token) or as a parameter — Correct. WithCancellation(token) or as a parameter to the method.
-
Complete the WithCancellation call
await foreach (int n in Numbers().____(token)) { Console.WriteLine(n); }Show answer
await foreach (int n in Numbers().WithCancellation(token)) { Console.WriteLine(n); }