Async Streams with Yield Return in C

This article provides a practical overview of asynchronous streams in C#, explaining how the yield return statement functions alongside IAsyncEnumerable<T>. You will learn the core concepts behind async streams, see how they solve the limitations of traditional synchronous iterators and bulk asynchronous tasks, and discover how to produce and consume streaming data efficiently using await foreach and cancellation tokens.

What Are Async Streams?

Prior to C# 8.0, developers faced a trade-off when handling collections of data asynchronously:

• Synchronous Iterators (IEnumerable<T> with yield return): Elements are generated on demand (lazily), saving memory, but generation blocks the executing thread.
• Asynchronous Operations (Task<IEnumerable<T>>): The thread is not blocked during execution, but all items must be retrieved, buffered, and returned at once, leading to latency and higher memory usage for large datasets.

Async streams bridge this gap by combining the asynchronous behavior of Task<T> with the pull-based iteration of IEnumerable<T>. Represented by the IAsyncEnumerable<T> interface, async streams allow a method to yield items individually as soon as they become available without blocking the calling thread.

Creating an Async Stream Generator

An async stream generator is a method that meets three criteria:

1. It is declared with the async modifier.
2. It returns IAsyncEnumerable<T>.
3. It uses the yield return statement to emit values.

Here is a practical implementation simulating data retrieval over a network:

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

public class DataService
{
    public async IAsyncEnumerable<int> FetchDataStreamAsync([System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        for (int i = 1; i <= 5; i++)
        {
            // Simulate an asynchronous I/O operation
            await Task.Delay(1000, cancellationToken);

            // Yield the item to the caller without completing the method
            yield return i;
        }
    }
}

In this example, every time yield return is executed, the item is pushed to the consumer. The execution state of the method is preserved, and the next iteration does not run until the consumer requests the next item.

Consuming an Async Stream

To consume an IAsyncEnumerable<T>, C# provides the await foreach construct. This pulls items from the stream asynchronously, awaiting each item as it is produced:

public class Program
{
    public static async Task Main()
    {
        var service = new DataService();
        var cts = new CancellationTokenSource();

        try
        {
            await foreach (var item in service.FetchDataStreamAsync(cts.Token))
            {
                Console.WriteLine($"Received item: {item} at {DateTime.Now:T}");
            }
        }
        catch (OperationCanceledException)
        {
            Console.WriteLine("Stream processing was canceled.");
        }
    }
}

The loop executes asynchronously; if an item is not yet ready, the loop yields control back to the runtime rather than blocking the thread.

Handling Cancellation and Configuration

Async streams support cancellation via CancellationToken. To properly bind a consumer's token to the generator method, apply the [EnumeratorCancellation] attribute to the parameter in the generator:

public async IAsyncEnumerable<string> ReadLogStreamAsync(
    [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    while (!cancellationToken.IsCancellationRequested)
    {
        string? line = await ReadNextLineAsync(cancellationToken);
        if (line == null) break;
        yield return line;
    }
}

If you are developing a library or need to configure synchronization context switching, you can use WithCancellation and ConfigureAwait:

await foreach (var item in service.FetchDataStreamAsync()
                                  .WithCancellation(cancellationToken)
                                  .ConfigureAwait(false))
{
    // Process item without capturing the original synchronization context
}

When to Use Async Streams

Async streams are ideal for scenarios involving high-latency or continuous data sources, such as:

• Reading records line-by-line from a large file or network socket.
• Querying paginated HTTP APIs without loading all pages into memory at once.
• Processing real-time event feeds, such as message queues or IoT sensor data.