📝 Lesson 3.2: Async Streams and Channels

Data That Arrives Over Time

You know two related tools already: iterators (yield return, Lesson 1.3) produce a sequence lazily but synchronously; async (Task<T>, Intermediate) awaits a single result. What about a sequence where producing each item requires awaiting?

Perfect fits: reading pages from a web API one at a time, tailing a growing file, or receiving messages as they arrive. You don't want to wait for all the data before processing any of it — you want to handle each item the moment it's ready.

Producing an Async Stream

An async iterator method returns IAsyncEnumerable<T>, is marked async, and combines await with yield return — the two features you already know, together:

public static async IAsyncEnumerable<int> GenerateAsync(int count)
{
    for (int i = 1; i <= count; i++)
    {
        await Task.Delay(500);   // simulate awaiting each item (e.g. an API page)
        yield return i;          // produce it as soon as it's ready
    }
}

A realistic example — paging through a web API, yielding each page's items as they load:

public static async IAsyncEnumerable<Item> GetAllItemsAsync(HttpClient client)
{
    int page = 1;
    while (true)
    {
        var items = await client.GetFromJsonAsync<List<Item>>($"/items?page={page}");
        if (items is null || items.Count == 0) yield break;   // no more pages

        foreach (var item in items)
        {
            yield return item;    // stream each item without loading every page first
        }
        page++;
    }
}

await foreach

You consume an async stream with await foreach — like a normal foreach, but it awaits each item as it's produced:

await foreach (int n in GenerateAsync(5))
{
    Console.WriteLine($"Received {n}");   // prints one every ~500ms, as they arrive
}

Received 1
Received 2
Received 3
Received 4
Received 5

LINQ-style operators are available too, via the System.Linq.Async package (or built-in helpers), so async streams compose much like regular sequences.

Cancelling a Stream

Async streams integrate with the cancellation model from Lesson 3.1. The producer accepts a token marked [EnumeratorCancellation]; the consumer supplies one via WithCancellation:

using System.Runtime.CompilerServices;

public static async IAsyncEnumerable<int> GenerateAsync(
    int count,
    [EnumeratorCancellation] CancellationToken token = default)
{
    for (int i = 1; i <= count; i++)
    {
        await Task.Delay(500, token);   // token flows into the wait
        yield return i;
    }
}

using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(2));

try
{
    await foreach (int n in GenerateAsync(100).WithCancellation(cts.Token))
    {
        Console.WriteLine(n);
    }
}
catch (OperationCanceledException)
{
    Console.WriteLine("Stream cancelled.");
}

Channels

Async streams are great when one consumer pulls from one producer. But sometimes producers and consumers are separate, independent pieces running concurrently — one (or many) generating work, another (or many) processing it. That's the producer/consumer pattern, and System.Threading.Channels is the modern, async-friendly tool for it.

using System.Threading.Channels;

Channel<int> channel = Channel.CreateUnbounded<int>();

// Producer: runs concurrently, writes items, then signals completion
Task producer = Task.Run(async () =>
{
    for (int i = 1; i <= 5; i++)
    {
        await channel.Writer.WriteAsync(i);
        await Task.Delay(200);
    }
    channel.Writer.Complete();          // "no more items"
});

// Consumer: reads until the channel is completed and drained
await foreach (int item in channel.Reader.ReadAllAsync())
{
    Console.WriteLine($"Consumed {item}");
}

await producer;

Backpressure

An unbounded channel grows without limit — if the producer is faster than the consumer, memory balloons. A bounded channel caps its capacity and applies backpressure: when full, WriteAsync awaits until the consumer makes room, naturally throttling the producer to the consumer's pace.

var channel = Channel.CreateBounded<int>(capacity: 10);

// If the buffer holds 10 unconsumed items, this WriteAsync will
// asynchronously wait until the consumer reads one — no runaway memory.
await channel.Writer.WriteAsync(item);

Exercise & Quiz

using System.Threading.Channels;

// Part A
await foreach (int sq in SquaresAsync(5))
    Console.WriteLine($"square: {sq}");

// Part B
var channel = Channel.CreateBounded<int>(5);
// TODO: producer Task.Run writing 1..20 then Complete()
// TODO: consumer await foreach over channel.Reader.ReadAllAsync()

static async IAsyncEnumerable<int> SquaresAsync(int n)
{
    // TODO: for 1..n, await Task.Delay(100), yield return i*i
}

using System.Threading.Channels;

// Part A
await foreach (int sq in SquaresAsync(5))
    Console.WriteLine($"square: {sq}");

// Part B
var channel = Channel.CreateBounded<int>(5);

Task producer = Task.Run(async () =>
{
    for (int i = 1; i <= 20; i++)
    {
        await channel.Writer.WriteAsync(i);   // waits when the buffer is full
        Console.WriteLine($"  produced {i}");
    }
    channel.Writer.Complete();
});

await foreach (int x in channel.Reader.ReadAllAsync())
{
    Console.WriteLine($"consumed {x}");
    await Task.Delay(150);                     // slow consumer
}

await producer;

static async IAsyncEnumerable<int> SquaresAsync(int n)
{
    for (int i = 1; i <= n; i++)
    {
        await Task.Delay(100);
        yield return i * i;
    }
}

Summary

📚 Additional Resources

• Generate and consume async streams — Microsoft Docs
• IAsyncEnumerable<T> — API reference
• System.Threading.Channels — Microsoft Docs

🚀 What's Next?

You can now stream and coordinate async data. To close the module, we sweat the details that make async fast and correct. In Lesson 3.3: Async Performance and Pitfalls, you'll meet ValueTask, ConfigureAwait, and the classic deadlock traps.