Async Streams trong C# 8.0 là gì?

C# 8.0 giới thiệu async streams, mô hình hóa streaming source của data. Data streams thường retrieve hoặc generate elements asynchronously.

Code generate sequence giờ có thể dùng yield return để trả về elements trong method được khai báo với async modifier.

Ta có thể consume async stream dùng await foreach loop.

Cú pháp

// Cũ: Synchronous iterator
static IEnumerable<string> Messages() { 
    yield return "Hello!"; 
    yield return "World!"; 
}

// Mới: Async iterator (C# 8.0+)
static async IAsyncEnumerable<string> MessagesAsync() { 
    await Task.Delay(2000); 
    yield return "Hello!"; 
    await Task.Delay(2000); 
    yield return "World!"; 
}

Ví dụ

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

Output

Hello!
World!

(Mỗi line in ra sau 2 giây delay)

Key Concepts

Use Cases Thực Tế

1. Database Streaming (Entity Framework Core)

public async IAsyncEnumerable GetCustomersAsync(
    [EnumeratorCancellation] CancellationToken cancellationToken = default) {
    
    await foreach (var customer in _context.Customers
        .AsAsyncEnumerable()
        .WithCancellation(cancellationToken)) {
        yield return customer;
    }
}

2. HTTP Streaming / Large File Download

public async IAsyncEnumerable DownloadFileInChunksAsync(string url) {
    using var client = new HttpClient();
    using var response = await client.GetAsync(url, HttpCompletionOption.ResponseHeadersRead);
    response.EnsureSuccessStatusCode();
    
    using var stream = await response.Content.ReadAsStreamAsync();
    var buffer = new byte[8192];
    int bytesRead;
    
    while ((bytesRead = await stream.ReadAsync(buffer, 0, buffer.Length)) > 0) {
        var chunk = new byte[bytesRead];
        Array.Copy(buffer, chunk, bytesRead);
        yield return chunk;
    }
}

3. Real-time Data (Sensor, Log Tail, Message Queue)

public async IAsyncEnumerable TailLogFileAsync(string filePath) {
    using var fs = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.ReadWrite);
    using var reader = new StreamReader(fs);
    
    // Seek to end
    fs.Seek(0, SeekOrigin.End);
    
    while (true) {
        var line = await reader.ReadLineAsync();
        if (line != null) {
            yield return ParseLogEntry(line);
        }
        await Task.Delay(100); // Poll interval
    }
}

4. Transform Async Stream (LINQ-style)

// C# 8.0 không có built-in async LINQ, dùng System.Linq.Async (NuGet)
using System.Linq.Async;

await sourceAsync
    .Where(x => x.IsActive)
    .Select(x => x.ToDto())
    .Take(10)
    .ForEachAsync(async item => await ProcessItemAsync(item));

Cancellation Support

public async IAsyncEnumerable GetDataAsync(
    [EnumeratorCancellation] CancellationToken cancellationToken = default) {
    
    for (int i = 0; i < 100; i++) {
        cancellationToken.ThrowIfCancellationRequested();
        await Task.Delay(100, cancellationToken);
        yield return $"Item {i}";
    }
}

// Consumer
var cts = new CancellationTokenSource(TimeSpan.FromSeconds(1));
await foreach (var item in GetDataAsync(cts.Token)) {
    Console.WriteLine(item);
}

Async Streams vs Task vs IObservable

Yêu cầu

• C# 8.0+ (.NET Core 3.0+, .NET Standard 2.1+)
• Nullable reference types khuyến nghị
• using System.Collections.Generic; (IAsyncEnumerable)