Async Streams ใน C# 8.0 คืออะไร

C# 8.0 Introduce Async Streams ซึ่ง Model Streaming Source ของ Data Data Streams บ่อยครั้ง Retrieve หรือ Generate Elements แบบ Asynchronous

Code ที่ Generate Sequence สามารถใช้ yield return เพื่อ Return Elements ใน Method ที่ Declared ด้วย async Modifier

เราสามารถ Consume Async Stream โดยใช้ await foreach Loop

Syntax Comparison

// Synchronous Iterator (เก่า)
static IEnumerable Messages() {
    yield return "Hello!";
    yield return "World!";
}

// Async Iterator (C# 8.0+)
static async IAsyncEnumerable MessagesAsync() {
    await Task.Delay(2000);
    yield return "Hello!";
    await Task.Delay(2000);
    yield return "World!";
}

ตัวอย่างการใช้งาน

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

class Program {
    public static async Task Main() {
        await foreach (var item in MessagesAsync()) {
            System.Console.WriteLine(item);
        }
        Console.ReadLine();
    }
    
    static async IAsyncEnumerable MessagesAsync() {
        await Task.Delay(2000);
        yield return "Hello!";
        await Task.Delay(2000);
        yield return "World!";
    }
}

Output

Hello!
World!

(แต่ละบรรทัดห่างกัน 2 วินาที)

Use Cases จริงของ Async Streams

1. Database Streaming (Large Result Sets)

public async IAsyncEnumerable GetCustomersAsync(
    [EnumeratorCancellation] CancellationToken cancellationToken = default) {
    
    using var connection = new SqlConnection(_connectionString);
    await connection.OpenAsync(cancellationToken);
    
    using var command = new SqlCommand("SELECT * FROM Customers", connection);
    using var reader = await command.ExecuteReaderAsync(cancellationToken);
    
    while (await reader.ReadAsync(cancellationToken)) {
        yield return new Customer {
            Id = reader.GetInt32(0),
            Name = reader.GetString(1),
            Email = reader.GetString(2)
        };
    }
}

// Consumption
await foreach (var customer in GetCustomersAsync()) {
    ProcessCustomer(customer);
}

2. File Processing (Large Files)

public static async IAsyncEnumerable ReadLinesAsync(
    string filePath, 
    [EnumeratorCancellation] CancellationToken ct = default) {
    
    using var reader = new StreamReader(filePath);
    string line;
    while ((line = await reader.ReadLineAsync()) != null) {
        ct.ThrowIfCancellationRequested();
        yield return line;
    }
}

3. API Pagination (Auto-paging)

public async IAsyncEnumerable GetAllStudentsAsync(
    [EnumeratorCancellation] CancellationToken ct = default) {
    
    string nextUrl = "/api/students";
    while (!string.IsNullOrEmpty(nextUrl)) {
        var response = await _httpClient.GetAsync(nextUrl, ct);
        response.EnsureSuccessStatusCode();
        
        var page = await response.Content.ReadFromJsonAsync>(ct);
        foreach (var student in page.Items) {
            yield return student;
        }
        nextUrl = page.NextPageUrl;
    }
}

4. Real-time Data (SignalR, WebSockets)

public async IAsyncEnumerable GetStockPricesAsync(
    string symbol, 
    [EnumeratorCancellation] CancellationToken ct = default) {
    
    var connection = new HubConnectionBuilder()
        .WithUrl($"https://api.example.com/hubs/stocks/{symbol}")
        .Build();
    
    var channel = Channel.CreateUnbounded();
    
    connection.On("PriceUpdate", price => {
        channel.Writer.TryWrite(price);
    });
    
    await connection.StartAsync(ct);
    
    try {
        await foreach (var price in channel.Reader.ReadAllAsync(ct)) {
            yield return price;
        }
    } finally {
        await connection.DisposeAsync();
    }
}

IAsyncEnumerable vs IEnumerable

CancellationToken ใน Async Streams

await foreach (var item in GetDataAsync().WithCancellation(cancellationToken)) {
    // Process item
}

// หรือใน Method Definition
public async IAsyncEnumerable GetDataAsync(
    [EnumeratorCancellation] CancellationToken ct = default) {
    while (!ct.IsCancellationRequested) {
        yield return await FetchNextAsync(ct);
    }
}

ข้อจำกัด

• ต้องใช้ C# 8.0+ และ .NET Core 3.0+ / .NET Standard 2.1+
• ไม่สามารถใช้ yield return ใน try-catch block ได้ (แต่ได้ใน try-finally)
• ไม่รองรับ yield break ใน finally block
• Enumerator ต้อง Dispose อย่างถูกต้อง (await foreach ทำให้อัตโนมัติ)