close

Articles Home » Programming Languages » C# » Building Custom Async Enumerables with IAsyncEnumerable in C#

Building Custom Async Enumerables with IAsyncEnumerable in C#

Asynchronous programming has become a cornerstone of modern software development, and with the advent of IAsyncEnumerable in C# 8.0, we gained an elegant way to work with asynchronous data streams. This tutorial will take you through building custom async enumerables step by step. By the end, you will have a solid understanding of how to implement and use IAsyncEnumerable effectively.

Introduction to IAsyncEnumerable

Before diving into building custom implementations, let’s briefly understand what IAsyncEnumerable is and why it’s useful. IAsyncEnumerable is an interface that allows iteration over data asynchronously using the await keyword. It’s particularly useful when working with streaming data, such as database queries, file reading, or web responses, where data may not be immediately available.

Here’s an example of using IAsyncEnumerable:

public async IAsyncEnumerable<int> GetNumbersAsync()
{
    for (int i = 1; i <= 5; i++)
    {
        await Task.Delay(1000); // Simulate asynchronous work
        yield return i;
    }
}

public async Task ProcessNumbersAsync()
{
    await foreach (var number in GetNumbersAsync())
    {
        Console.WriteLine(number);
    }
}Code language: C# (cs)

In this example, GetNumbersAsync yields numbers asynchronously, and ProcessNumbersAsync consumes them using await foreach.

Core Concepts of IAsyncEnumerable

To effectively implement IAsyncEnumerable, you need to understand the following key concepts:

  1. IAsyncEnumerable<T>: The main interface representing an asynchronous stream.
  2. IAsyncEnumerator<T>: Represents the enumerator used to iterate asynchronously over the elements.
  3. MoveNextAsync: A method of IAsyncEnumerator<T> that advances to the next element asynchronously.
  4. Current: A property of IAsyncEnumerator<T> that gets the current element in the collection.

With these basics in mind, let’s start building custom async enumerables.

Step 1: Creating a Custom Async Enumerable Class

We’ll begin by implementing a simple custom async enumerable that produces a sequence of numbers asynchronously. For this, we’ll create a class that implements IAsyncEnumerable<T> and IAsyncEnumerator<T>.

Implementing IAsyncEnumerable<T>

The first step is to implement the IAsyncEnumerable<T> interface. This requires defining a method GetAsyncEnumerator, which returns an instance of IAsyncEnumerator<T>.

public class AsyncNumberSequence : IAsyncEnumerable<int>
{
    private readonly int _start;
    private readonly int _count;

    public AsyncNumberSequence(int start, int count)
    {
        _start = start;
        _count = count;
    }

    public IAsyncEnumerator<int> GetAsyncEnumerator(CancellationToken cancellationToken = default)
    {
        return new AsyncNumberEnumerator(_start, _count);
    }
}Code language: C# (cs)

In this class, the constructor takes a starting number and a count. The GetAsyncEnumerator method returns an instance of a custom enumerator class that we’ll define next.

Implementing IAsyncEnumerator<T>

Next, we need to implement the IAsyncEnumerator<T> interface. This involves defining the MoveNextAsync method and the Current property.

public class AsyncNumberEnumerator : IAsyncEnumerator<int>
{
    private readonly int _start;
    private readonly int _count;
    private int _currentIndex;

    public AsyncNumberEnumerator(int start, int count)
    {
        _start = start;
        _count = count;
        _currentIndex = -1;
    }

    public int Current { get; private set; }

    public async ValueTask<bool> MoveNextAsync()
    {
        await Task.Delay(500); // Simulate asynchronous work

        if (_currentIndex + 1 < _count)
        {
            _currentIndex++;
            Current = _start + _currentIndex;
            return true;
        }

        return false;
    }

    public ValueTask DisposeAsync()
    {
        // Clean up resources if necessary
        return ValueTask.CompletedTask;
    }
}Code language: C# (cs)

The MoveNextAsync method advances the enumerator to the next item in the sequence, simulating asynchronous work with Task.Delay. The DisposeAsync method is included for cleanup, although it’s not required in this simple example.

Step 2: Consuming the Custom Async Enumerable

With the custom async enumerable and enumerator in place, let’s write code to consume it using await foreach.

public async Task RunAsync()
{
    var numbers = new AsyncNumberSequence(1, 10);

    await foreach (var number in numbers)
    {
        Console.WriteLine(number);
    }
}Code language: C# (cs)

This will output numbers from 1 to 10 with a delay of 500 milliseconds between each.

Step 3: Adding Cancellation Support

Asynchronous operations should respect cancellation to improve responsiveness and resource management. To support cancellation, we use the CancellationToken passed to GetAsyncEnumerator.

Updating the Enumerator

Modify the AsyncNumberEnumerator to check the cancellation token in the MoveNextAsync method:

public class AsyncNumberEnumerator : IAsyncEnumerator<int>
{
    private readonly int _start;
    private readonly int _count;
    private readonly CancellationToken _cancellationToken;
    private int _currentIndex;

    public AsyncNumberEnumerator(int start, int count, CancellationToken cancellationToken)
    {
        _start = start;
        _count = count;
        _cancellationToken = cancellationToken;
        _currentIndex = -1;
    }

    public int Current { get; private set; }

    public async ValueTask<bool> MoveNextAsync()
    {
        _cancellationToken.ThrowIfCancellationRequested();
        await Task.Delay(500, _cancellationToken); // Support cancellation

        if (_currentIndex + 1 < _count)
        {
            _currentIndex++;
            Current = _start + _currentIndex;
            return true;
        }

        return false;
    }

    public ValueTask DisposeAsync()
    {
        return ValueTask.CompletedTask;
    }
}Code language: C# (cs)

Passing the Cancellation Token

Update GetAsyncEnumerator to pass the cancellation token:

public IAsyncEnumerator<int> GetAsyncEnumerator(CancellationToken cancellationToken = default)
{
    return new AsyncNumberEnumerator(_start, _count, cancellationToken);
}Code language: C# (cs)

Using Cancellation in Consumer Code

You can now use a cancellation token in the consuming code:

public async Task RunWithCancellationAsync(CancellationToken cancellationToken)
{
    var numbers = new AsyncNumberSequence(1, 10);

    try
    {
        await foreach (var number in numbers.WithCancellation(cancellationToken))
        {
            Console.WriteLine(number);
        }
    }
    catch (OperationCanceledException)
    {
        Console.WriteLine("Operation was canceled.");
    }
}Code language: C# (cs)

Step 4: Handling Exceptions Gracefully

Errors can occur during asynchronous iteration. To handle exceptions, wrap the logic in MoveNextAsync with try-catch blocks.

public async ValueTask<bool> MoveNextAsync()
{
    try
    {
        _cancellationToken.ThrowIfCancellationRequested();
        await Task.Delay(500, _cancellationToken);

        if (_currentIndex + 1 < _count)
        {
            _currentIndex++;
            Current = _start + _currentIndex;
            return true;
        }

        return false;
    }
    catch (Exception ex)
    {
        Console.WriteLine($"Error: {ex.Message}");
        throw;
    }
}Code language: C# (cs)

Step 5: Advanced Scenarios

Streaming Data from APIs

You can use IAsyncEnumerable to stream data from APIs or other sources. For example, fetching data from a paginated API:

public async IAsyncEnumerable<string> FetchDataAsync()
{
    int page = 1;

    while (true)
    {
        var data = await FetchPageAsync(page++);

        if (data == null || !data.Any())
            yield break;

        foreach (var item in data)
        {
            yield return item;
        }
    }
}

private async Task<List<string>> FetchPageAsync(int page)
{
    // Simulate fetching data
    await Task.Delay(1000);
    return page <= 3 ? new List<string> { "Item1", "Item2" } : null;
}Code language: C# (cs)

Processing Large Files

IAsyncEnumerable is useful for processing large files line by line:

public async IAsyncEnumerable<string> ReadLinesAsync(string filePath)
{
    using var reader = new StreamReader(filePath);
    while (!reader.EndOfStream)
    {
        yield return await reader.ReadLineAsync();
    }
}Code language: C# (cs)

In this tutorial, we explored the step-by-step process of building custom async enumerables using IAsyncEnumerable in C#. We started with the basics, added cancellation support, and handled exceptions. Finally, we looked at advanced scenarios like streaming data from APIs and processing large files.

Related posts: