Unity2020.2 で 非同期ストリーム（IAsyncEnumerable）が使いたい！

# Unity 2020.2

`Unity2020.2`でついに`C# 8.0`がサポートされるようになりました。

`C# 8.0`にはいろいろ新機能が追加されているのですが、今回取り上げるのは「非同期ストリーム」です。

# 非同期ストリーム

`C# 8.0`の目玉機能の1つとして、「非同期ストリーム」があります。
非同期ストリームとは`IEnumerable`の`async/await`対応版である、[`IAsyncEnumerable`](https://docs.microsoft.com/ja-jp/dotnet/api/system.collections.generic.iasyncenumerable-1?WT.mc_id=DT-MVP-5003135&view=dotnet-plat-ext-5.0)とその周辺機能を指します。

非同期ストリームが使えるとたとえば次のようなことが簡単に実装できるようになります。

- `foreach`の値の列挙に`async/await`が利用できる（`await foreach`）
- `IEnumerator`時に`yield return`と`await`が併用できる（`非同期イテレータ`）

（`Rx`の`Observable`に似てますけど、用途は結構異なります。）


# 残念だったな！Unity2020.2は`.NET Standard 2.0`だ！

ではこの非同期ストリームが`Unity2020.2`でそのまま使えるのかというと、**残念ながら使えません。**

というのも`IAsyncEnumerable`など、`C# 8.0`で利用する想定のインタフェース群は`.NET Standard 2.1`準拠のAPIセットとして提供されているからです。
ですが`Unity 2020.2`では`.NET Standard 2.0`準拠のAPIにしか対応していません。

そのため事実上、`IAsyncEnumerable`を含む非同期ストリームを素の`Unity2020.2`で使うことはできません。

# 代替としてUniTaskを使おう

非同期ストリームはとても便利なため可能であれば今すぐ使いたいです。
そこで代替として利用可能なライブラリが`UniTask`です。

- [UniTask](https://github.com/Cysharp/UniTask)

# UniTask

## UniTaskAsyncEnumerable

`UniTask`では`IAsyncEnumerable<T>`の代替として`IUniTaskAsyncEnumerable<T>`を用意してくれています。
（同様に`IUniTaskAsyncEnumerator<T>`もあります）

これらを用いることで`C# 8.0`の非同期ストリームと同等のことを実現することができます。
（さらにいうと`C# 7.3`でもこの`UniTaskAsyncEnumerable`を利用することができます）

## IUniTaskAsyncEnumerable<T>とawait foreach

`UniTask`で非同期ストリームを利用する場合は、`IUniTaskAsyncEnumerable<T>`を用いればOKです。
`Unity 2020.2`でも`C# 8.0`の`await foreach`は問題なく利用できるので、これと組み合わせて使いましょう。

```cs
// 非同期ストリームをIUniTaskAsyncEnumerableで実現
private async UniTask ForEachSampleAsync(IUniTaskAsyncEnumerable<string> messages, CancellationToken token)
{
    // await foreach は Unity 2020.2 でも利用可能
    await foreach (var value in messages.WithCancellation(token))
    {
        Debug.Log(value);
        await UniTask.Delay(TimeSpan.FromSeconds(1), cancellationToken: token);
    }
}
```

## UniTaskで非同期イテレータ

`C# 8.0` + `.NET Standard 2.1`の環境では次のように書くことができます。

```cs
// 10秒間カウントアップする
// not Unity 2020.2
async IAsyncEnumerable<int> CreateStreamAsync()
{
    for (int i = 0; i < 10; i++)
    {
        yield return i;
        await Task.Delay(TimeSpan.FromSeconds(1));
    }
}
```

これを`Unity 2020.2` + `UniTask`で代替するとこうなります。

```cs
IUniTaskAsyncEnumerable<int> CreateStreamAsync()
{
    return UniTaskAsyncEnumerable.Create<int>(async (writer, token) =>
    {
        for (int i = 0; i < 10; i++)
        {
            await writer.YieldAsync(i);
            await UniTask.Delay(TimeSpan.FromSeconds(1), cancellationToken: token);
        }
    });
}
```

イテレータが使えないので代わりに`UniTaskAsyncEnumerable.Create<T>`を利用してあげれば同等のことができます。

# まとめ

`C# 8.0`の非同期イテレータを待ち望んでいた人は代替として`UniTask`を使うとよいでしょう。

詳しい資料はこちら。

- [UniTask Ver2 、UniTaskAsyncEnumerableまとめ](https://qiita.com/toRisouP/items/8f66fd952eaffeaf3107)
- [UniRx/UniTask 完全理解](https://asciidwango.jp/post/631460628113014785/unirxunitask%E5%AE%8C%E5%85%A8%E7%90%86%E8%A7%A3)

# おまけ

## ForEachAsync, ForEachAwaitAsync, Subscribe, SubscribeAwait → await foreach

`UniTaskAsyncEnumerable`は`C# 8.0`未満の`C#`でも利用できるようにいくつか購読用APIが用意されていました。
これらを`await foreach`に書き換える場合はどうしたらいいのかを列挙します。


### ForEachAsync

値を同期で消費するパターンのもの。そのまま書くだけです。
むしろ`ForEachAsync`で書いたほうが記述量少ない。

```cs
private async UniTask ConvertedAsync(IUniTaskAsyncEnumerable<string> messages, CancellationToken token)
{
    // ForEachAsync
    await messages.ForEachAsync(x => Debug.Log(x), token);

    // await foreach
    await foreach (var x in messages.WithCancellation(token))
    {
        Debug.Log(x);
    }
}
```

### ForEachAwaitAsync

値を非同期で`await`しながら消費するパターンのもの。

`CancellationToken`を扱う場合、パフォーマンスを考えると`ForEachAwaitWithCancellationAsync`を使ったほうがよいためちょっと長ったらしくなります。
そのため`CancellationToken`を使うんであれば`await foreach`の方が素直に書けます。

```cs
private async UniTask ConvertedAsync(IUniTaskAsyncEnumerable<string> messages, CancellationToken token)
{
    // ForEachAwaitAsync + CancellationToken
    await messages.ForEachAwaitWithCancellationAsync(async (x, ct) =>
    {
        Debug.Log(x);
        await UniTask.Delay(TimeSpan.FromSeconds(1), cancellationToken: ct);
    }, token);

    
    // await foreach
    await foreach (var x in messages.WithCancellation(token))
    {
        Debug.Log(x);
        await UniTask.Delay(TimeSpan.FromSeconds(1), cancellationToken: token);
    }
}
```

### Subscribe

`Subscribe`は`IUniTaskAsyncEnumerable<T>`を同期メソッドの文脈上で消費することができます（`async/await`の文脈上でなくても消費できる）。

用途がいくつかあるのでそれぞれ紹介します。

#### Forget目的で使うとき

値を非同期で`Forget`しながら消費するパターンのもの。
こちらは`CancellationToken`を含んでいても`Subscribe`の方が記述量は少ないです。

```cs
private async UniTask ConvertedAsync(IUniTaskAsyncEnumerable<string> messages, CancellationToken token)
{
    // Subscribe
    messages.Subscribe(async (x, ct) =>
    {
        Debug.Log(x);
        await UniTask.Delay(TimeSpan.FromSeconds(1), cancellationToken: ct);
    }, token);


    // await foreach
    await foreach (var x in messages.WithCancellation(token))
    {
        // ここが UniTask.Void になる
        UniTask.Void(async ct =>
        {
            Debug.Log(x);
            await UniTask.Delay(TimeSpan.FromSeconds(1), cancellationToken: ct);
        }, token);
    }
}
```

#### IObservable<T>.Subscribeライクに使うとき

`IObservable<T>.Subscribe()`のように、`async/await`とは関係がない文脈上で`IUniTaskAsyncEnumerable<T>`を消費したいときに使い方です。
（`OnError`や`OnCompleted`を使いたい場合）

こっちは明らかに`Subscribe`で書いたほうがキレイになります。

```cs
// ↓ asyncメソッドではない！
private void ConvertedAsync(IUniTaskAsyncEnumerable<string> messages, CancellationToken token)
{
    // Subscribe
    messages.Subscribe(
        onNext: async x =>
        {
            Debug.Log(x);
            await UniTask.Delay(TimeSpan.FromSeconds(1), cancellationToken: token)
        },
        onCompleted: () => Debug.Log("Done"), cancellationToken: token);


    // await foreach
    {
        UniTask.Void(async token2 =>
        {
            try
            {
                await foreach (var x in messages.WithCancellation(token2))
                {
                    // ここが UniTask.Void になる
                    UniTask.Void(async ct =>
                    {
                        Debug.Log(x);
                        await UniTask.Delay(TimeSpan.FromSeconds(1), cancellationToken: ct);
                    }, token2);
                }

                // OnCompleted 相当
                Debug.Log("Done");
            }
            catch (Exception e) when (!(e is OperationCanceledException))
            {
                // OnError 相当
                Debug.LogError(e);
            }
        }, token);
    }
}
```

### SubscribeAwait

`SubscribeAwait`は`Subscribe`の`Forget`ではない版です。
`ForEachAwaitAsync`に挙動が近いものです（ややこしい）。

こちらも`SubscribeAwait`の方が記述量は減ります。

```cs
// ↓ asyncメソッドではない！
private void ConvertedAsync(IUniTaskAsyncEnumerable<string> messages, CancellationToken token)
{
    // SubscribeAwait
    messages.SubscribeAwait(
        onNext: async x =>
        {
            Debug.Log(x);
            await UniTask.Delay(TimeSpan.FromSeconds(1), cancellationToken: token);
        },
        onCompleted: () => Debug.Log("Done"), cancellationToken: token);


    // await foreach
    {
        UniTask.Void(async token2 =>
        {
            try
            {
                await foreach (var x in messages.WithCancellation(token2))
                {
                    Debug.Log(x);

                    // ここをちゃんとawaitし終わってからイテレートするのがSubscribeとの違い
                    await UniTask.Delay(TimeSpan.FromSeconds(1), cancellationToken: token2);
                }

                // OnCompleted 相当
                Debug.Log("Done");
            }
            catch (Exception e) when (!(e is OperationCanceledException))
            {
                // OnError 相当
                Debug.LogError(e);
            }
        }, token);
    }
}
```

## いろいろあるけどどれを使えばいいのか

`ForEachAsync`、`ForEachAwaitAsync`、`Subscribe`、`SubscribeAwait`、これらは上手く使えば記述量を減らせるメリットがあります。
ですがどれもシグネチャが似ており、挙動をしっかり把握してから使わないと事故を起こすリスクがあります。
（あとクロージャを生成するので書き方によってはアロケートが発生したりする可能性あり）

一方の`await foreach`は手続き的に書けるため挙動は把握しやすいですが、その分記述量が増えてしまします。
ただ挙動のわかりやすさとしては`await foreach`なので、困ったときは`await foreach`で書いておけばいいでしょう。