IAsyncEnumerable<T> em C# — Streaming de Dados do Jeito Certo

// Carregar tudo antes de fazer qualquer coisa — o uso de memória cresce com o número de linhas
public async Task<IActionResult> ExportProducts()
{
    List<Product> products = await _db.Products
        .AsNoTracking()
        .ToListAsync();                 // TODAS as linhas em memória de uma vez
 
    return Ok(products);               // depois serializar tudo
}

public interface IAsyncEnumerable<out T>
{
    IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default);
}
 
public interface IAsyncEnumerator<out T> : IAsyncDisposable
{
    T Current { get; }
    ValueTask<bool> MoveNextAsync();
}

public async IAsyncEnumerable<int> GenerateSequenceAsync(int count)
{
    for (int i = 0; i < count; i++)
    {
        await Task.Delay(10);   // simular trabalho async por item
        yield return i;
    }
}

public async IAsyncEnumerable<int> FetchFirstAsync()
{
    yield return await Task.FromResult(42); // uma instrução, ambas as palavras-chave — legal
}

using System.Runtime.CompilerServices;
 
public async IAsyncEnumerable<Product> StreamProductsAsync(
    int categoryId,
    [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    await using var connection = await _dbFactory.OpenConnectionAsync(cancellationToken);
 
    await foreach (var product in FetchFromDbAsync(connection, categoryId, cancellationToken))
    {
        // transformação leve por item
        product.Price = Math.Round(product.Price, 2);
        yield return product;
 
        // verificação explícita se o corpo do loop for lento
        cancellationToken.ThrowIfCancellationRequested();
    }
}

var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
 
await foreach (var product in StreamProductsAsync(categoryId: 5)
                   .WithCancellation(cts.Token))
{
    Console.WriteLine(product.Name);
}

await foreach (var item in source)
{
    Process(item);
}
 
// se decompõe aproximadamente em:
await using var enumerator = source.GetAsyncEnumerator(cancellationToken);
while (await enumerator.MoveNextAsync())
{
    Process(enumerator.Current);
}

await foreach (var item in source.ConfigureAwait(false))
{
    Process(item);
}

public async IAsyncEnumerable<ProductDto> StreamProductsAsync(
    [EnumeratorCancellation] CancellationToken ct = default)
{
    // AsAsyncEnumerable() abre o reader e retorna linhas uma por vez
    await foreach (var p in _db.Products
                               .AsNoTracking()           // sem rastreamento de mudanças para leituras
                               .Where(p => p.IsActive)
                               .Select(p => new ProductDto(p.Id, p.Name, p.Price))
                               .AsAsyncEnumerable()
                               .WithCancellation(ct))
    {
        yield return p;
    }
}

// ToListAsync — armazena tudo em buffer, depois processa
var products = await _db.Products.ToListAsync(ct);
foreach (var p in products)
    await SendToClientAsync(p, ct);
 
// AsAsyncEnumerable — processa cada linha conforme chega
await foreach (var p in _db.Products.AsAsyncEnumerable().WithCancellation(ct))
    await SendToClientAsync(p, ct);

[ApiController]
[Route("api/products")]
public class ProductsController : ControllerBase
{
    private readonly AppDbContext _db;
 
    public ProductsController(AppDbContext db) => _db = db;
 
    // A resposta é um array JSON transmitido conforme as linhas chegam — sem buffer completo
    [HttpGet("export")]
    public IAsyncEnumerable<ProductDto> ExportAsync(CancellationToken ct)
    {
        return _db.Products
                  .AsNoTracking()
                  .Select(p => new ProductDto(p.Id, p.Name, p.Price))
                  .AsAsyncEnumerable();
    }
}

app.MapGet("/api/products/export", (AppDbContext db, CancellationToken ct) =>
    db.Products
      .AsNoTracking()
      .Select(p => new ProductDto(p.Id, p.Name, p.Price))
      .AsAsyncEnumerable());

public async IAsyncEnumerable<OrderEvent> WatchOrdersAsync(
    string customerId,
    [EnumeratorCancellation] CancellationToken ct = default)
{
    // Canal limitado fornece backpressure — o produtor bloqueia quando cheio
    var channel = Channel.CreateBounded<OrderEvent>(new BoundedChannelOptions(100)
    {
        FullMode = BoundedChannelFullMode.Wait
    });
 
    // Produtor: executa independentemente, escreve no canal
    var producer = Task.Run(async () =>
    {
        try
        {
            await foreach (var evt in _messageBus.SubscribeAsync(customerId, ct))
                await channel.Writer.WriteAsync(evt, ct);
        }
        finally
        {
            channel.Writer.Complete();  // sinaliza fim do stream
        }
    }, ct);
 
    // Consumidor: expõe o canal como IAsyncEnumerable
    await foreach (var evt in channel.Reader.ReadAllAsync(ct))
        yield return evt;
 
    await producer;   // propagar exceções do produtor
}

// Abordagem A: armazenar tudo em buffer
public async Task<long> BufferedApproachAsync()
{
    List<Product> products = await _db.Products
        .AsNoTracking()
        .ToListAsync();
 
    long total = 0;
    foreach (var p in products)
        total += (long)p.Price;
 
    return total;
}
 
// Abordagem B: transmitir linha por linha
public async Task<long> StreamingApproachAsync(CancellationToken ct = default)
{
    long total = 0;
 
    await foreach (var p in _db.Products
                               .AsNoTracking()
                               .AsAsyncEnumerable()
                               .WithCancellation(ct))
    {
        total += (long)p.Price;
    }
 
    return total;
}

[ApiController]
[Route("api/reports")]
public class ReportController : ControllerBase
{
    private readonly AppDbContext _db;
    private readonly ILogger<ReportController> _logger;
 
    public ReportController(AppDbContext db, ILogger<ReportController> logger)
    {
        _db = db;
        _logger = logger;
    }
 
    [HttpGet("orders/export")]
    [Produces("application/json")]
    public IAsyncEnumerable<OrderExportRow> ExportOrdersAsync(
        [FromQuery] DateOnly from,
        [FromQuery] DateOnly to,
        CancellationToken ct)                   // vinculado ao RequestAborted automaticamente
    {
        var fromDate = from.ToDateTime(TimeOnly.MinValue);
        var toDate   = to.ToDateTime(TimeOnly.MaxValue);
 
        return StreamOrdersAsync(fromDate, toDate, ct);
    }
 
    // Método privado separado mantém a lógica do iterador limpa
    private async IAsyncEnumerable<OrderExportRow> StreamOrdersAsync(
        DateTime fromDate,
        DateTime toDate,
        [EnumeratorCancellation] CancellationToken ct)
    {
        int count = 0;
 
        await foreach (var order in _db.Orders
                                       .AsNoTracking()
                                       .Where(o => o.CreatedAt >= fromDate && o.CreatedAt <= toDate)
                                       .OrderBy(o => o.CreatedAt)
                                       .Select(o => new
                                       {
                                           o.Id,
                                           o.CustomerName,
                                           o.Total,
                                           o.CreatedAt,
                                           ItemCount = o.Items.Count
                                       })
                                       .AsAsyncEnumerable()
                                       .WithCancellation(ct))
        {
            yield return new OrderExportRow(
                order.Id,
                order.CustomerName,
                order.Total,
                order.CreatedAt,
                order.ItemCount);
 
            count++;
 
            // registrar progresso a cada 1000 linhas sem bloquear o stream
            if (count % 1000 == 0)
                _logger.LogInformation("Exportadas {Count} ordens até agora", count);
        }
 
        _logger.LogInformation("Exportação concluída. Total de linhas: {Count}", count);
    }
}
 
public record OrderExportRow(
    Guid Id,
    string CustomerName,
    decimal Total,
    DateTime CreatedAt,
    int ItemCount);

dotnet add package System.Linq.Async   # necessário apenas no .NET 9 e anteriores

using System.Linq;
 
// Filtrar e transformar antes de consumir
var expensiveProducts = _db.Products
    .AsAsyncEnumerable()
    .Where(p => p.Price > 100)
    .Select(p => new { p.Name, p.Price })
    .OrderBy(p => p.Price);       // nota: ordena em memória após o streaming
 
await foreach (var p in expensiveProducts)
    Console.WriteLine($"{p.Name}: {p.Price:C}");

public async IAsyncEnumerable<string> RiskyStreamAsync(
    [EnumeratorCancellation] CancellationToken ct = default)
{
    yield return "primeiro";
    await Task.Delay(10, ct);
    throw new InvalidOperationException("algo deu errado");
    // nada depois do throw é alcançado
}
 
try
{
    await foreach (var item in RiskyStreamAsync())
        Console.WriteLine(item);
}
catch (InvalidOperationException ex)
{
    Console.WriteLine($"Stream falhou: {ex.Message}");
}

public async IAsyncEnumerable<Row> StreamWithCleanupAsync(
    [EnumeratorCancellation] CancellationToken ct = default)
{
    await using var reader = await OpenReaderAsync(ct);
    try
    {
        while (await reader.ReadAsync(ct))
            yield return reader.GetRow();
    }
    finally
    {
        // executado mesmo se o chamador lançar exceção ou cancelar
        _logger.LogInformation("Reader fechado");
    }
}