IAsyncEnumerable<T> en C# — Streaming de Datos de la Forma Correcta

// Cargar todo antes de hacer algo — el uso de memoria aumenta con el número de filas
public async Task<IActionResult> ExportProducts()
{
    List<Product> products = await _db.Products
        .AsNoTracking()
        .ToListAsync();                 // TODAS las filas en memoria de una vez
 
    return Ok(products);               // luego serializar todo
}

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 trabajo async por elemento
        yield return i;
    }
}

public async IAsyncEnumerable<int> FetchFirstAsync()
{
    yield return await Task.FromResult(42); // una sentencia, ambas palabras clave — 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))
    {
        // transformación ligera por elemento
        product.Price = Math.Round(product.Price, 2);
        yield return product;
 
        // verificación explícita si el cuerpo del bucle es 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 descompone aproximadamente en:
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 el reader y devuelve filas de una en una
    await foreach (var p in _db.Products
                               .AsNoTracking()           // sin seguimiento de cambios para lecturas
                               .Where(p => p.IsActive)
                               .Select(p => new ProductDto(p.Id, p.Name, p.Price))
                               .AsAsyncEnumerable()
                               .WithCancellation(ct))
    {
        yield return p;
    }
}

// ToListAsync — almacena todo en buffer, luego procesa
var products = await _db.Products.ToListAsync(ct);
foreach (var p in products)
    await SendToClientAsync(p, ct);
 
// AsAsyncEnumerable — procesa cada fila a medida que llega
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;
 
    // La respuesta es un array JSON transmitido a medida que llegan las filas — sin 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 acotado provee contrapresión — el productor bloquea cuando está lleno
    var channel = Channel.CreateBounded<OrderEvent>(new BoundedChannelOptions(100)
    {
        FullMode = BoundedChannelFullMode.Wait
    });
 
    // Productor: se ejecuta independientemente, escribe en el 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();  // señal de fin de stream
        }
    }, ct);
 
    // Consumidor: expone el canal como IAsyncEnumerable
    await foreach (var evt in channel.Reader.ReadAllAsync(ct))
        yield return evt;
 
    await producer;   // propagar excepciones del productor
}

// Enfoque A: almacenar todo en 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;
}
 
// Enfoque B: transmitir fila por fila
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 a RequestAborted automáticamente
    {
        var fromDate = from.ToDateTime(TimeOnly.MinValue);
        var toDate   = to.ToDateTime(TimeOnly.MaxValue);
 
        return StreamOrdersAsync(fromDate, toDate, ct);
    }
 
    // Método privado separado mantiene la lógica del iterador limpia
    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 progreso cada 1000 filas sin bloquear el stream
            if (count % 1000 == 0)
                _logger.LogInformation("Exportadas {Count} órdenes hasta ahora", count);
        }
 
        _logger.LogInformation("Exportación completa. Total de filas: {Count}", count);
    }
}
 
public record OrderExportRow(
    Guid Id,
    string CustomerName,
    decimal Total,
    DateTime CreatedAt,
    int ItemCount);

dotnet add package System.Linq.Async   # solo necesario en .NET 9 y anteriores

using System.Linq;
 
// Filtrar y 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 en memoria después del 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 "primero";
    await Task.Delay(10, ct);
    throw new InvalidOperationException("algo salió mal");
    // nada después del throw se alcanza jamás
}
 
try
{
    await foreach (var item in RiskyStreamAsync())
        Console.WriteLine(item);
}
catch (InvalidOperationException ex)
{
    Console.WriteLine($"Stream falló: {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
    {
        // se ejecuta incluso si el llamador lanza excepción o cancela
        _logger.LogInformation("Reader cerrado");
    }
}