【C#】C#8.0から追加されたIAsyncEnumerableを使って非同期ストリームを実装する

はじめに

恥ずかしながら最近になってやっとIAsyncEnumerableというものを知りました。
learn.microsoft.com
ufcpp.net

これは非同期ストリームと呼ばれるもので、ざっくり言うとasync/awaitとyieldが共存できるようになったというイメージでしょうか。

C#8.0より追加されたそうです。

またよくReactive Extensions(Rx)とAsync Streams(非同期ストリーム)の使い所の違いについて議論されているようですね。
www.infoq.com

簡単にではありますが、使い方について記述しておきます。

使い方

// EnumeratorCancellation属性をつけることでCancellationTokenを受け取れるようになる
private static async IAsyncEnumerable<int> SampleAsync([EnumeratorCancellation] CancellationToken token = default)
{
    foreach (var item in Enumerable.Range(1, 100))
    {
        await Task.Delay(1000, token);
        yield return item;
    }
}

// データの消費側(利用者側)
var cts = new CancellationTokenSource();

// 非同期foreach
await foreach (var item in SampleAsync(cts.Token))
{
    Console.WriteLine(item);
}

内部構造

namespace System.Collections.Generic
{
  public interface IAsyncEnumerable<out T>
  {
    IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default (CancellationToken));
  }
}

namespace System.Collections.Generic
{
  public interface IAsyncEnumerator<out T> : IAsyncDisposable
  {
    T Current { get; }

    ValueTask<bool> MoveNextAsync();
  }
}

namespace System
{
  public interface IAsyncDisposable
  {
    ValueTask DisposeAsync();
  }
}

未確認飛行さんの記事にも書いてありますが、非同期foreachはパターンベースであり、IAsyncEnumerableを実装していなくても以下さえあれば動作するようですね。

• GetAsyncEnumerator
• Current
• MoveNextAsync

興味がある方は該当箇所を読んで、自分で実装してみると面白いかもしれません。