What are async streams and IAsyncEnumerable of T? How do you produce, consume and cancel them?

1. What the compiler generates

An async iterator compiles to a class that implements IAsyncEnumerable<T>, IAsyncEnumerator<T>, IAsyncDisposable and IValueTaskSource<bool>:

MoveNextAsync():
    resume state machine until:
        yield return x  -> Current = x; complete the ValueTask<bool> with true
        await (incomplete) -> return pending ValueTask<bool> (backed by ManualResetValueTaskSourceCore)
        end of method   -> complete with false
DisposeAsync():
    run pending finally blocks / await using disposals

Because MoveNextAsync returns a reusable ValueTask<bool>, streaming items doesn't allocate a task per item.

2. Cancellation plumbing

// Consumer
await foreach (var x in source.WithCancellation(ct)) { ... }

// What happens
source.GetAsyncEnumerator(ct)  ->  compiler routes ct into the [EnumeratorCancellation] parameter

3. ASP.NET Core streaming

app.MapGet("/orders", (OrdersDb db, CancellationToken ct) =>
    db.Orders.AsNoTracking().OrderBy(o => o.Id).AsAsyncEnumerable());

System.Text.Json writes the array incrementally, flushing as it goes. Things to know:

• Response headers are sent before all data is ready, so a mid-stream exception can't become a clean 500; the client sees a truncated response.
• Keep per-item work light: the DB reader (and connection) stays open for the whole stream.
• Use AsNoTracking() to avoid the change tracker growing with each row.

4. Writing CSV manually

app.MapGet("/export", async (HttpContext http, OrdersDb db, CancellationToken ct) =>
{
    http.Response.ContentType = "text/csv";
    await using var writer = new StreamWriter(http.Response.Body);
    int n = 0;
    await foreach (var o in db.Orders.AsNoTracking().AsAsyncEnumerable().WithCancellation(ct))
    {
        await writer.WriteLineAsync($"{o.Id},{o.Total}");
        if (++n % 1000 == 0) await writer.FlushAsync(ct);
    }
});

5. Pitfalls

6. Version notes

7. Try it yourself

1. Run the code example: break after a high reading and observe the iterator's finally message.
2. Remove [EnumeratorCancellation] and read warning CS8425; confirm the timeout no longer stops the feed.
3. Return IAsyncEnumerable<T> from a minimal API and watch the response arrive progressively with curl -N.
4. Compare memory for ToListAsync() vs AsAsyncEnumerable() over 1 million rows using dotnet-counters.
5. Enumerate the same async iterator twice and count how many times the source runs.
6. On .NET 10, use Where/Select/ToListAsync on an async stream without extra packages.

using System.Runtime.CompilerServices;

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

await foreach (var reading in SensorFeed.ReadAsync("pump-7").WithCancellation(cts.Token))
{
    Console.WriteLine($"{reading.At:HH:mm:ss.fff} {reading.Value}");
    if (reading.Value > 95) break; // disposes the enumerator, runs the iterator's finally
}

// .NET 10: LINQ over IAsyncEnumerable<T> is built in (System.Linq.AsyncEnumerable).
var highs = await SensorFeed.ReadAsync("pump-8")
    .Where(r => r.Value > 50)
    .Take(3)
    .ToListAsync();
Console.WriteLine($"High readings: {highs.Count}");

public sealed record Reading(DateTime At, int Value);

public static class SensorFeed
{
    public static async IAsyncEnumerable<Reading> ReadAsync(
        string sensorId,
        [EnumeratorCancellation] CancellationToken ct = default)
    {
        var random = new Random(sensorId.GetHashCode());
        try
        {
            while (true)
            {
                await Task.Delay(100, ct);          // simulate waiting for the next reading
                yield return new Reading(DateTime.UtcNow, random.Next(0, 100));
            }
        }
        finally
        {
            Console.WriteLine($"Feed {sensorId} closed"); // runs on break, cancellation or error
        }
    }
}