Agent skill

iasyncenumerable

USE FOR: Streaming data asynchronously using IAsyncEnumerable<T> with yield return, including database result streaming, API pagination, file processing, and gRPC server streaming. DO NOT USE FOR: Event-based reactive streams (use IObservable<T>/Rx), producer/consumer queues (use System.Threading.Channels), or fire-and-forget background work.

View SKILL.md on GitHub Repository

Stars 163

Forks 31

Install this agent skill to your Project

npx add-skill https://github.com/majiayu000/claude-skill-registry/tree/main/skills/other/other/iasyncenumerable

Metadata

Additional technical details for this skill

author: Tyler-R-Kendrick
version: 1.0.0
displayName: IAsyncEnumerable

SKILL.md

IAsyncEnumerable<T>

Overview

IAsyncEnumerable<T> is the async counterpart of IEnumerable<T>, enabling pull-based streaming of data where each element is produced asynchronously. It uses yield return inside async iterator methods and is consumed with await foreach. Unlike Task<List<T>> which waits for all items before returning, IAsyncEnumerable<T> yields items one at a time as they become available, reducing memory pressure and time-to-first-result for large datasets.

IAsyncEnumerable<T> is supported natively in ASP.NET Core (controller actions, minimal APIs), Entity Framework Core (streaming query results), gRPC (server streaming), and System.Text.Json (streaming serialization). It was introduced in C# 8 / .NET Core 3.0 and requires no additional packages.

Basic Producer and Consumer

csharp

using System.Collections.Generic;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;

namespace MyApp.Services;

public class SensorDataService
{
    // Producer: yields sensor readings one at a time
    public async IAsyncEnumerable<SensorReading> StreamReadingsAsync(
        string sensorId,
        [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        while (!cancellationToken.IsCancellationRequested)
        {
            var reading = await ReadSensorAsync(sensorId, cancellationToken);
            yield return reading;
            await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken);
        }
    }

    private async Task<SensorReading> ReadSensorAsync(
        string sensorId, CancellationToken ct)
    {
        // Simulate async I/O
        await Task.Delay(10, ct);
        return new SensorReading(
            SensorId: sensorId,
            Value: Random.Shared.NextDouble() * 100,
            Timestamp: DateTime.UtcNow);
    }
}

public record SensorReading(string SensorId, double Value, DateTime Timestamp);

csharp

using System;
using System.Threading;
using System.Threading.Tasks;
using MyApp.Services;

// Consumer: processes items as they arrive
public class SensorMonitor
{
    private readonly SensorDataService _service;

    public SensorMonitor(SensorDataService service)
    {
        _service = service;
    }

    public async Task MonitorAsync(string sensorId, CancellationToken ct)
    {
        await foreach (var reading in _service.StreamReadingsAsync(sensorId, ct))
        {
            if (reading.Value > 90.0)
            {
                Console.WriteLine(
                    $"ALERT: Sensor {reading.SensorId} reading {reading.Value:F1} " +
                    $"at {reading.Timestamp:HH:mm:ss}");
            }
        }
    }
}

Entity Framework Core Streaming

EF Core's AsAsyncEnumerable() streams query results without loading all rows into memory.

csharp

using System.Collections.Generic;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Threading;
using Microsoft.EntityFrameworkCore;

namespace MyApp.Data;

public class OrderRepository
{
    private readonly AppDbContext _context;

    public OrderRepository(AppDbContext context)
    {
        _context = context;
    }

    // Streams orders from the database without loading all into memory
    public async IAsyncEnumerable<Order> GetLargeOrdersAsync(
        decimal minTotal,
        [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        await foreach (var order in _context.Orders
            .Where(o => o.Total >= minTotal)
            .OrderByDescending(o => o.CreatedAt)
            .AsAsyncEnumerable()
            .WithCancellation(cancellationToken))
        {
            yield return order;
        }
    }

    // Batch processing with streaming
    public async IAsyncEnumerable<Order[]> GetOrderBatchesAsync(
        int batchSize,
        [EnumeratorCancellation] CancellationToken cancellationToken = default)
    {
        var batch = new List<Order>(batchSize);

        await foreach (var order in _context.Orders
            .AsAsyncEnumerable()
            .WithCancellation(cancellationToken))
        {
            batch.Add(order);
            if (batch.Count >= batchSize)
            {
                yield return batch.ToArray();
                batch.Clear();
            }
        }

        if (batch.Count > 0)
        {
            yield return batch.ToArray();
        }
    }
}

public class Order
{
    public int Id { get; set; }
    public decimal Total { get; set; }
    public DateTime CreatedAt { get; set; }
}

public class AppDbContext : DbContext
{
    public DbSet<Order> Orders => Set<Order>();

    public AppDbContext(DbContextOptions<AppDbContext> options) : base(options) { }
}

ASP.NET Core Integration

ASP.NET Core natively supports IAsyncEnumerable<T> in controller actions and minimal APIs, streaming JSON array elements as they are produced.

csharp

using System.Collections.Generic;
using System.Runtime.CompilerServices;
using System.Threading;
using Microsoft.AspNetCore.Mvc;
using MyApp.Data;

namespace MyApp.Controllers;

[ApiController]
[Route("api/[controller]")]
public class OrdersController : ControllerBase
{
    private readonly OrderRepository _repository;

    public OrdersController(OrderRepository repository)
    {
        _repository = repository;
    }

    // ASP.NET Core streams the JSON array elements one at a time
    [HttpGet("large")]
    public IAsyncEnumerable<Order> GetLargeOrders(
        [FromQuery] decimal minTotal = 1000,
        CancellationToken cancellationToken = default)
    {
        return _repository.GetLargeOrdersAsync(minTotal, cancellationToken);
    }
}

csharp

// Minimal API equivalent
app.MapGet("/api/orders/stream", (
    OrderRepository repo,
    CancellationToken ct) =>
{
    return repo.GetLargeOrdersAsync(1000, ct);
});

LINQ Operators for IAsyncEnumerable

System.Linq.Async (NuGet package) provides LINQ operators for IAsyncEnumerable<T>.

csharp

using System.Collections.Generic;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;

namespace MyApp.Services;

public class AnalyticsService
{
    private readonly SensorDataService _sensors;

    public AnalyticsService(SensorDataService sensors)
    {
        _sensors = sensors;
    }

    // Filter, transform, and take from an async stream
    public async IAsyncEnumerable<Alert> GetAlertsAsync(
        string sensorId,
        [EnumeratorCancellation] CancellationToken ct = default)
    {
        await foreach (var alert in _sensors.StreamReadingsAsync(sensorId, ct)
            .Where(r => r.Value > 85.0)
            .Select(r => new Alert(
                r.SensorId,
                $"High reading: {r.Value:F1}",
                r.Timestamp))
            .Take(50)
            .WithCancellation(ct))
        {
            yield return alert;
        }
    }

    // Aggregate over a stream
    public async Task<double> GetAverageAsync(
        string sensorId, int sampleCount, CancellationToken ct)
    {
        return await _sensors.StreamReadingsAsync(sensorId, ct)
            .Take(sampleCount)
            .AverageAsync(r => r.Value, ct);
    }

    // Buffer into chunks
    public async IAsyncEnumerable<SensorReading[]> GetBufferedAsync(
        string sensorId,
        int bufferSize,
        [EnumeratorCancellation] CancellationToken ct = default)
    {
        await foreach (var chunk in _sensors.StreamReadingsAsync(sensorId, ct)
            .Buffer(bufferSize)
            .WithCancellation(ct))
        {
            yield return chunk.ToArray();
        }
    }
}

public record Alert(string SensorId, string Message, DateTime Timestamp);