Summary - Async streams
Combine IAsyncEnumerable, await foreach and Channel to write a log stream.
Summary - Async streams
In this group you learned the basics of async streams: IAsyncEnumerable<T> produces an async value stream; await foreach consumes it; yield return produces elements. CancellationToken cancels the stream; System.Threading.Channels builds a producer-consumer queue.
The key lesson: stream large data in pieces, not all at once. This lowers memory use and reduces latency. Now we will combine all of this in a small app.
Exercise Combine IAsyncEnumerable with cancellation
Define static async IAsyncEnumerable<int> Squares(CancellationToken token). In its body open a for (int i = 1; i <= 3; i++) loop; on each step call token.ThrowIfCancellationRequested();, await Task.CompletedTask; and yield return i * 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 Squares(cts.Token)) { sum += n; } and print with Console.WriteLine(sum);.
Show example solution
In Squares body write for (int i = 1; i <= 3; i++) { token.ThrowIfCancellationRequested(); await Task.CompletedTask; yield return i * i; }. Make Main static async Task Main() and write var cts = new CancellationTokenSource(); int sum = 0; await foreach (int n in Squares(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> Squares(CancellationToken token)
{
for (int i = 1; i <= 3; i++)
{
token.ThrowIfCancellationRequested();
await Task.CompletedTask;
yield return i * i;
}
}
static async Task Main()
{
var cts = new CancellationTokenSource();
int sum = 0;
await foreach (int n in Squares(cts.Token))
{
sum += n;
}
Console.WriteLine(sum);
}
}Tips
A few tips: stream large data with IAsyncEnumerable; avoid awaiting the whole list. Add a CancellationToken parameter to every async stream and call ThrowIfCancellationRequested before each element.
Use System.Threading.Channels for producer-consumer; it is ideal when production and consumption speeds differ. await foreach is the consumer's standard tool; yield return is the producer's standard tool.
Exercise Mini app: Stream multiples
Define static async IAsyncEnumerable<int> MultiplesOf(int k, CancellationToken token). In its body open a for (int i = 1; i <= 4; i++) loop; on each step call token.ThrowIfCancellationRequested();, await Task.CompletedTask; and yield return k * i;. Make Main a static async Task Main; create the source with var cts = new CancellationTokenSource();. Sum with int sum = 0; await foreach (int n in MultiplesOf(5, cts.Token)) { sum += n; } and print with Console.WriteLine(sum); (5+10+15+20 = 50).
Show example solution
In MultiplesOf body write for (int i = 1; i <= 4; i++) { token.ThrowIfCancellationRequested(); await Task.CompletedTask; yield return k * i; }. Make Main static async Task Main() and write var cts = new CancellationTokenSource(); int sum = 0; await foreach (int n in MultiplesOf(5, 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> MultiplesOf(int k, CancellationToken token)
{
for (int i = 1; i <= 4; i++)
{
token.ThrowIfCancellationRequested();
await Task.CompletedTask;
yield return k * i;
}
}
static async Task Main()
{
var cts = new CancellationTokenSource();
int sum = 0;
await foreach (int n in MultiplesOf(5, cts.Token))
{
sum += n;
}
Console.WriteLine(sum);
}
}Test yourself
-
Which construct consumes IAsyncEnumerable?
- await foreach
- for
- while
- lock
Show answer
await foreach — Correct. await foreach consumes IAsyncEnumerable.