.NET Async Streams

Introduction

Async streams, introduced in C# 8.0 and .NET Core 3.0, provide a way to work with asynchronous sequences of data. Traditional IEnumerable<T> collections work well for synchronous data access, but what if you need to asynchronously produce or consume sequences? That's where async streams come in.

Async streams allow you to:

• Process sequences of data that arrive over time
• Handle data that's produced asynchronously
• Yield results one at a time while performing asynchronous operations

In this tutorial, you'll learn how to work with async streams using IAsyncEnumerable<T> and async iterators in C#.

Understanding Async Streams

The Problem Async Streams Solve

Before async streams, developers faced challenges when dealing with asynchronous data sequences:

1. Using Task<IEnumerable<T>> meant waiting for the entire collection before processing any items
2. Creating custom async enumeration patterns was complex and error-prone
3. Combining asynchronous operations with enumeration required workarounds

Async streams provide a natural way to handle these scenarios with language-level support.

Key Components

The async streams feature consists of:

1. IAsyncEnumerable<T>: An interface representing an asynchronous sequence
2. IAsyncEnumerator<T>: An interface for enumerating async sequences
3. async iterator methods: Methods that use yield return with async/await
4. await foreach statement: For consuming async streams

Creating Async Streams

Let's look at how to create async streams:

csharp

// Method that returns an async stream
public static async IAsyncEnumerable<int> GenerateNumbersAsync(int count, int delay)
{
    for (int i = 0; i < count; i++)
    {
        // Simulate asynchronous work
        await Task.Delay(delay);
        yield return i;
    }
}

The key elements here:

• The return type is IAsyncEnumerable<T>
• The method is marked with both async and uses yield return
• We can use await within the iterator

Consuming Async Streams

To consume an async stream, use the await foreach statement:

csharp

// Consuming an async stream
await foreach (var number in GenerateNumbersAsync(5, 1000))
{
    Console.WriteLine($"Received: {number}");
}

Output:

Received: 0
Received: 1
Received: 2
Received: 3
Received: 4

Notice that each number is printed as it becomes available, with a 1-second delay between them.

Cancellation Support

Async streams support cancellation via a special attribute:

csharp

public static async IAsyncEnumerable<string> GetDataAsync(
    [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    for (int i = 0; i < 10; i++)
    {
        // Check cancellation before doing work
        cancellationToken.ThrowIfCancellationRequested();
        
        await Task.Delay(500, cancellationToken);
        yield return $"Data {i}";
    }
}

To use cancellation with await foreach:

csharp

using var cts = new CancellationTokenSource();
cts.CancelAfter(2000); // Cancel after 2 seconds

try
{
    await foreach (var item in GetDataAsync(cts.Token))
    {
        Console.WriteLine(item);
    }
}
catch (OperationCanceledException)
{
    Console.WriteLine("Operation was cancelled");
}

ConfigureAwait Support

You can configure how the async stream is awaited:

csharp

// Use ConfigureAwait(false) with async streams
await foreach (var item in asyncStream.ConfigureAwait(false))
{
    // Process item
}

Real-World Examples

Let's look at some practical applications of async streams:

Example 1: Reading File Lines Asynchronously

csharp

public static async IAsyncEnumerable<string> ReadLinesAsync(string filePath)
{
    using var fileStream = new FileStream(filePath, FileMode.Open);
    using var reader = new StreamReader(fileStream);
    
    while (!reader.EndOfStream)
    {
        var line = await reader.ReadLineAsync();
        if (line != null)
            yield return line;
    }
}

// Usage
await foreach (var line in ReadLinesAsync("largeFile.txt"))
{
    // Process each line as it's read, without loading the entire file
    ProcessLine(line);
}

Example 2: API Pagination

csharp

public static async IAsyncEnumerable<Product> GetAllProductsAsync(HttpClient client)
{
    int page = 1;
    bool hasMore = true;
    
    while (hasMore)
    {
        var response = await client.GetAsync($"api/products?page={page}");
        response.EnsureSuccessStatusCode();
        
        var pageResult = await response.Content.ReadFromJsonAsync<PagedResponse<Product>>();
        
        if (pageResult?.Items == null || pageResult.Items.Count == 0)
        {
            hasMore = false;
        }
        else
        {
            foreach (var product in pageResult.Items)
            {
                yield return product;
            }
            
            page++;
        }
    }
}

// Usage
await foreach (var product in GetAllProductsAsync(httpClient))
{
    Console.WriteLine($"Processing product: {product.Name}");
}

Example 3: Database Streaming with Entity Framework Core

csharp

public static async IAsyncEnumerable<Customer> StreamCustomersAsync(DbContext context)
{
    // AsAsyncEnumerable is available in EF Core
    await foreach (var customer in context.Customers
                                         .AsAsyncEnumerable())
    {
        // Maybe do some enrichment or filtering
        if (customer.IsActive)
            yield return customer;
    }
}

// Usage
await foreach (var customer in StreamCustomersAsync(dbContext))
{
    await ProcessCustomerDataAsync(customer);
}

Performance Considerations

When working with async streams:

1. Lazy Evaluation: Items are produced only as needed
2. Memory Efficiency: Process large data sets without loading everything into memory
3. Responsiveness: UI or service remains responsive while processing
4. Overhead: Be aware that async operations have some overhead

When to Use Async Streams

Async streams are ideal for:

• Processing data that becomes available over time
• Working with large datasets that should be streamed rather than loaded entirely
• APIs that naturally return data in pages or chunks
• Scenarios where both asynchronous operations and enumeration are needed

They might not be necessary when:

• All data is available synchronously
• The collection is small and can be processed in memory
• You need the entire collection before processing any items

Summary

Async streams provide an elegant way to work with asynchronous sequences of data in .NET. They offer:

• A natural programming model for producing and consuming asynchronous data
• Built-in language support with async, await foreach, and IAsyncEnumerable<T>
• Efficient handling of asynchronous sequences with lazy evaluation
• Proper cancellation and configuration support

By using async streams, you can write cleaner, more efficient code when dealing with sequences that require asynchronous operations to produce or process.

Additional Resources

• Microsoft Docs on Async Streams
• .NET API documentation for IAsyncEnumerable
• GitHub sample repository on Async Streams

Exercises

1. Create an async stream that simulates reading sensor data at regular intervals
2. Implement a method that returns weather data from an API as an async stream
3. Write a program that processes a large CSV file using async streams
4. Create an async stream that monitors a directory for new files and yields their contents
5. Implement pagination with async streams for a database query using Entity Framework Core