【Unity】UniTaskAsyncEnumerableとは何なのか

# はじめに

`UniTask`のv2が登場し、`UniTaskAsyncEnumerable`という機能が追加されから早数年が経過しました。
当時、[UniTask Ver2 、UniTaskAsyncEnumerableまとめ](https://qiita.com/toRisouP/items/8f66fd952eaffeaf3107#unitaskasyncenumerableiunitaskasyncenumerable)という記事を執筆しました。このときも`UniTaskAsyncEnumerable`についてまとめているのですが、今回はもうちょっと掘り下げて`UniTaskAsyncEnumerable`について解説します。

とくに`Observable`（UniRx）の代替として`UniTaskAsyncEnumerable`を用いてみるパターンが増えてきたと思うので、そのあたりの比較をしていきます。

## サンプルコード

この記事中に登場するコードはこちらのGitHubにて公開しています。

* [TORISOUP/UniTaskAsyncEnumerable_Samples](https://github.com/TORISOUP/UniTaskAsyncEnumerable_Samples)

# UniTaskAsyncEnumerable概要

`UniTaskAsyncEnumerable`とは「`IAsyncEnumerable<T>`相当の機能をUnity向けの実装したもの」です。
インタフェースとしては`IUniTaskAsyncEnumerable<T>`が`IAsyncEnumerable<T>`に相当しています。

では`IAsyncEnumerable<T>`が何なのかというと、読んで字の如く「非同期で扱える`Enumerable`」です。

## 同期的なIEnumerable<T>

ではまず**いつもの**同期的な`IEnumerable<T>`がどのような挙動をするのか見てみます。
といっても`IEnumerable<T>`だけ取り出してもわかりにくいので、「`IEnumerable<T>`を`foreach`で列挙する」ときの挙動を見てみましょう。

たとえば次のコードを見て下さい。

```cs
// 同期的ないつものEnumerable
private static void EnumerableSample(IEnumerable<int> data)
{
    foreach (var i in data)
    {
        Console.WriteLine(i);
    }
}
```

このコードはシンプルに「`IEnumerable<int>`を`foreach`で列挙してログに出す」というだけの処理です。
ここで`foreach`を使っていますが、実はこのコードは実行時にコンパイラによって次のようなコードに書き換えられてします。

```cs
private static void EnumerableSample(IEnumerable<int> data)
{
    // foreachは次のコードと同義である
    var enumerator = data.GetEnumerator();
    try
    {
        while (enumerator.MoveNext())
        {
            var i = enumerator.Current;
            Console.WriteLine(i);
        }
    }
    finally
    {
        enumerator.Dispose();
    }
}
```

1. `IEnumerable<T>`から`IEnumerator<T>`を取得する
2. `IEnumerator<T>`の`MoveNext()`を実行して「次の要素」を１つ取り出す
3. `MoveNext()`から次の要素が取得できたら、その値を使って「処理」を実行する
4. `MoveNext()`から次の要素が取得できない（要素が尽きた）場合は終了

これが同期的な`Enumerable`と`foreach`の挙動です。
値を順番に同期的に列挙していき、その中身が尽きたら終了という動作をしています。

## 非同期的なIAsyncEnumerable<T>

では一方で非同期的な`IAsyncEnumerable<T>`がどうなるのか。
こちらも実際にコードを見たほうが早いです。

```cs
// 非同期的なAsyncEnumerableの列挙
private static async ValueTask EnumerableSampleAsync(IAsyncEnumerable<int> data)
{
    await foreach (var i in data)
    {
        Console.WriteLine(i);
    }
}
```

見た目はほとんど同じですが、`await foreach`に変化している部分が特徴です。
そしてこのコードはコンパイラによって次のコードに変換されます。

```cs
// 非同期的なAsyncEnumerableの列挙
private static async ValueTask EnumerableSampleAsync(IAsyncEnumerable<int> data)
{
    // await foreachは次のコードと同義である
    var enumerator = data.GetAsyncEnumerator();
    try
    {
        while (await enumerator.MoveNextAsync())
        {
            var i = enumerator.Current;
            Console.WriteLine(i);
        }
    }
    finally
    {
        await enumerator.DisposeAsync();
    }
}
```

最大の違いは「`await enumerator.MoveNextAsync()`」と、`MoveNext()`が非同期になっているところです。
つまり「値の列挙を要求したときに、その要素が返って来るのを1つずつ待てる」というのが`IAsyncEnumerable<T>`の特徴です。

## EnumerableとObservable、そしてAsyncEnumerable

さて、`Enumerable`のその対となる概念として`Observable`(UniRx)があります。
この`Observable`と`AsyncEnumerable`は何が異なるのでしょうか。

`IAsyncEnumerable`がまだ存在しなかった時代、「`IObservable`は`IEnumerable`と双対である」といわれていました。

![SyncAsync.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/47146/e4893d56-a114-e3a6-110f-f40a338ad7a0.jpeg)


というのも、`IEnumerable`はPull型駆動であるのに対して、`IObservable`はPush型駆動（本質的に非同期）でした。
この2つのインタフェースは非常によく似ており、実際にインタフェースをひっくり返して定義すれば双対であることがわかります。
参考: [Observable Everywhere - Rxの原則とUniRxにみるデータソースの見つけ方](https://www.slideshare.net/neuecc/observable-everywhere-rxuni-rx)

ということで、「`Enumerable`のPull/Pushを逆にし、非同期にコレクションを扱えるようにしたものが`Observable`だよ」という説明を**今までは**行うことができました。


ですが、`AsyncEnumerable`の登場により話がややこしくなりました。`Enumerable`の非同期版が`Observable`というのであれば、じゃあ`AsyncEnumerable`は何者なのか。さっきの図は嘘なのか。`AsyncEnumerable`をさきほどの図に追加するとどうなるのか。自分はこうなると考えます。

![SyncAsync2.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/47146/1558d898-eb9a-1400-1314-427b729e3988.jpeg)


普通に「非同期・複数」の枠に2つ収まるかなと考えます。


では根本的に`AsyncEnumerable`と`Observable`が何が違うのかというと、やはりそれは「PushであるかPullであるか」です。

## PushとPull

非同期におけるPushとPullの違いを表にすると次のとおりになります。

||型|挙動の違い|
|:--|:--|:--|
|Push|`IObservable<T>`|非同期処理の「実行側」に制御権がある。いつ値が発行されるかは待受側にはわからない。|
|Pull|`IAsyncEnumerable<T>`|非同期処理の「待受側」に制御権がある。待受側が要求しない限り値が発行されることがない。|


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/47146/0c06d241-77d9-8794-e80b-50e835348c5f.png)


`Observable`では非同期処理の実行側が制御を握っています。つまり非同期処理がいつどのようなタイミングで実行され、その結果がいつ発行されるのかはすべて`Observable`の内部が管理しています。そのため`Subscribe`側では「いつか発行される値を待つ」ということしかできません。

一方の`AsyncEnumerable`では非同期処理の待受側が制御を握っています。つまり待受側の準備が整って初めて非同期処理が実行されます。そのため待受側
でメッセージの流量の制御が行えます。

# ObservableとUniAsyncEnumerableの使い分け

`Observable`と`UniAsyncEnumerable`の使い分けをどう使い分けるか。
Unityにおいては両者を同じ雰囲気で使える場合と、明確に区別したほうがいい場面があります。

というわけでここからは「UniRxのObservable」と「UniTaskのUniTaskAsyncEnumerable」に話を限定していきます。

## だいたい似た使い方ができるパターン

* Unityのイベントメッセージの待ち受け
* `AsyncReactiveProperty`

購読が一箇所のみ、つまり「どこか一箇所だけで`Subscribe`または`await`している」みたいな状況の場合。このときは`Observable`と`UniAsyncEnumerable`も似た使い方ができます。

### Unityのイベントメッセージの待受

Unityでよくあるのは「UIイベントの待機」が挙げられます。「UIボタンが押されたら処理を行う」といったイベント処理は`Observable`で実装しても`UniAsyncEnumerable`で実装しても似た形で扱えます。


```cs
using System.Threading;
using Cysharp.Threading.Tasks;
using Cysharp.Threading.Tasks.Linq;
using UniRx;
using UnityEngine;
using UnityEngine.UI;

namespace TORISOUP.UniTaskAsyncEnumerableSamples.UIEventDiff
{
    public class UIEventDiffSample : MonoBehaviour
    {
        [SerializeField] private Button _buttonA;
        [SerializeField] private Text _outputText;

        private void Start()
        {
            SubscribeAsObservable(destroyCancellationToken);
            SubscribeAsAsyncEnumerable(destroyCancellationToken);
        }

        // ButtonをObservableとして購読
        private void SubscribeAsObservable(CancellationToken ct)
        {
            // OnClickAsObservableでObservableに変換できる
            _buttonA.OnClickAsObservable()
                .Subscribe(_ =>
                {
                    _outputText.text += "ButtonA Clicked (Observable)!\n";
                })
                .AddTo(ct);
        }

        // ButtonをUniTaskAsyncEnumerableとして購読
        private void SubscribeAsAsyncEnumerable(CancellationToken ct)
        {
            // OnClickAsAsyncEnumerableでUniTaskAsyncEnumerableに変換できる
            _buttonA.OnClickAsAsyncEnumerable(ct)
                .Subscribe(_ =>
                {
                    _outputText.text += "ButtonA Clicked (UniTaskAsyncEnumerable)!\n";
                }, ct);
        }
    }
}
```

![UIA.gif](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/47146/740aa00b-0b26-6989-b152-300d6ceb01fb.gif)


このようなUniTaskが提供するUnityイベントを`UniTaskAsyncEnumerable`に変換してくれる機構は`Observable`（UniRx）版とだいたい同じ感覚で使えます。

ただし`UniRx`の`Operator`と、`UniTask`の`LINQ`は完全に互換ではないので、まったく同じように使えるわけではありません。`Operator`の組み合わせでやっていたことを`UniTaskAsyncEnumerable`で再現しようとした場合は工夫が必要になります。

### AsyncReactiveProperty

UniRxには`ReactiveProperty`と呼ばれる`Observable`ベースの機構がありましたが、これの`UniTaskAsyncEnumerable`として`AsyncReactiveProperty`があります。

こちらについては以前にまとめているのでまずはこちらを参照してください。

* [UniTask Ver2 、UniTaskAsyncEnumerableまとめ](https://qiita.com/toRisouP/items/8f66fd952eaffeaf3107#asyncreactiveproperty)

基本的な挙動は`UniRx.ReactiveProperty`とだいたい同じです。ただ`AsyncReactiveProperty`は`async/await`と組み合わせたときに扱いやすくなっているため、その点はUniRx版より優れています。


## 挙動が異なるので気をつけたほうがよいパターン

一方で`Observable`と`UniTaskAsyncEnumerable`とで挙動が大きく異なる場合があります。
根本から大きく挙動が変わる場合があるので注意してください。


### メッセージを受け取った「後」の非同期処理の実施


#### Observableの場合

まずそもそも、`Observable`ではメッセージ購読と非同期処理（`async/await`）を組み合わせることができません。`Subscribe()`メソッドの中で`async/await`を実行したとしても、それは`Observable`の挙動に一切干渉しません（できません）。

たとえば次のような実装があったとします。「ボタンを押したときに通信をするが、連打防止のために1回実行したら数秒awaitを挟んで次のイベントを受け取らないようにする」というもの。しかしこれは意図した動作をしません。

```cs
using System;
using System.Threading;
using Cysharp.Threading.Tasks;
using UniRx;
using UnityEngine;
using UnityEngine.Networking;
using UnityEngine.UI;

namespace TORISOUP.UniTaskAsyncEnumerableSamples.ObservableAsync
{
    public class ObservableAsyncSample : MonoBehaviour
    {
        [SerializeField] private Button _buttonA;
        [SerializeField] private InputField _urlInputField;
        [SerializeField] private Text _outputText;

        private void Start()
        {
            SubscribeAsObservable(destroyCancellationToken);
        }

        // ButtonをObservableとして購読
        private void SubscribeAsObservable(CancellationToken ct)
        {
            // ボタンが押されたら通信を実行
            // 通信結果をTextに表示する
            _buttonA.OnClickAsObservable()
                 // ここでasync/awatiを使おうとしている
                .Subscribe(async _ =>
                {
                    Debug.Log("<color=red>実行開始！</color>");
                    _outputText.text = "";

                    var url = _urlInputField.text;

                    // サーバに問い合わせる
                    var result = await FetchAsync(url, ct);
                    _outputText.text = result;

                    // ボタンを連打したときに一気に通信しないように3秒待たせる
                    await UniTask.Delay(TimeSpan.FromSeconds(3), cancellationToken: ct);

                    Debug.Log("<color=green>実行終了！</color>");
                })
                .AddTo(ct);
        }

        private async UniTask<string> FetchAsync(string url, CancellationToken ct)
        {
            using var request = UnityWebRequest.Get(url);
            await request.SendWebRequest().ToUniTask(cancellationToken: ct);

            return request.downloadHandler.text;
        }
    }
}
```

`ObservableAsyncSample.cs`では**ObservableのSubscribe上でasync/awaitの実行**を行っています。
意図としては「ボタンが押されたときに通信を行うが、このときに通信が頻発しないように3秒間の待機時間を作る」です。

ですがこれは意図どおりに動きません。ボタンを連打するとその回数だけ処理が実行されます。

![UI_Renda.gif](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/47146/116d6cae-54c0-72b6-98d9-9f4fab8b37cc.gif)


これはなぜかというと、`ObservableのSubscribe()`が`async/await`をケアしてくれないためです。`Subscribe()`の中で`async/await`を使ったとしても、それは`async void`で呼び出した扱いになってしまいます。

```cs
// ButtonをObservableとして購読
private void SubscribeAsObservable(CancellationToken ct)
{
    // ボタンが押されたら通信を実行
    // 通信結果をTextに表示する
    _buttonA.OnClickAsObservable()
        .Subscribe(_ =>
        {
            // async voidのメソッドを呼び出してるのと何も変わらない
            InnerFunction(ct);
        })
        .AddTo(ct);
}

private async void InnerFunction(CancellationToken ct)
{
    Debug.Log("<color=red>実行開始！</color>");
    _outputText.text = "";

    var url = _urlInputField.text;

    // サーバに問い合わせる
    var result = await FetchAsync(url, ct);
    _outputText.text = result;

    // ボタンを連打したときに一気に通信しないように3秒待たせる
    await UniTask.Delay(TimeSpan.FromSeconds(3), cancellationToken: ct);

    Debug.Log("<color=green>実行終了！</color>");
}
```
![ObservableAsync.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/47146/aa0ece1b-7372-f3e0-56a4-bbe7000b7a26.jpeg)


つまり**Observableの場合はメッセージのハンドリングに非同期処理を絡めることができません。**
（`Operator`使えばメッセージの「流量」を制限することはできるにはできます。が、`Operator`も`async/await`には未対応です）


#### UniTaskAsyncEnumerableの場合

では`UniTaskAsyncEnumerable`の場合どうなるかですが、`Observable`と違ってメッセージ購読に`async/await`を併用することができます。

```cs
using System;
using System.Threading;
using Cysharp.Threading.Tasks;
using Cysharp.Threading.Tasks.Linq;
using UnityEngine;
using UnityEngine.Networking;
using UnityEngine.UI;

namespace TORISOUP.UniTaskAsyncEnumerableSamples.UniTaskAsyncEnumerableAsync
{
    public class UniTaskAsyncEnumerableAsyncSample : MonoBehaviour
    {
        [SerializeField] private Button _buttonA;
        [SerializeField] private InputField _urlInputField;
        [SerializeField] private Text _outputText;

        private void Start()
        {
            SubscribeAsUniTaskAsyncEnumerable(destroyCancellationToken);
        }

        // ButtonをObservableとして購読
        private void SubscribeAsUniTaskAsyncEnumerable(CancellationToken ct)
        {
            // ボタンが押されたら通信を実行
            // 通信結果をTextに表示する
            _buttonA.OnClickAsAsyncEnumerable(ct)
                // ForEachAwaitAsyncになっている点に注意
                .ForEachAwaitAsync(async _ =>
                {
                    Debug.Log("<color=red>実行開始！</color>");
                    _outputText.text = "";

                    var url = _urlInputField.text;

                    // サーバに問い合わせる
                    var result = await FetchAsync(url, ct);
                    _outputText.text = result;

                    // ボタンを連打したときに一気に通信しないように3秒待たせる
                    await UniTask.Delay(TimeSpan.FromSeconds(3), cancellationToken: ct);

                    Debug.Log("<color=green>実行終了！</color>");
                }, ct);
        }


        private async UniTask<string> FetchAsync(string url, CancellationToken ct)
        {
            using var request = UnityWebRequest.Get(url);
            await request.SendWebRequest().ToUniTask(cancellationToken: ct);
            return request.downloadHandler.text;
        }
    }
}
```

![UI_Renda2.gif](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/47146/c08ae3fa-2ecf-ae5f-9e08-0f28da2091fd.gif)


`UniTaskAsyncEnumerable`として購読を行った場合、メッセージのハンドリングに`async/await`を正しくケアしてくれます。このため意図したとおりに連打防止が働いています。

![AsyncEnumerableAsync.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/47146/c20497d5-f564-87ff-8c79-8f0e61883ba1.jpeg)


**そしてここで、UniTaskAsyncEnumerableのややこしい面がでてきます。**
`UniTaskAsyncEnumerable`ではメッセージ購読側で`async/await`を実行することができました。そのためメッセージのハンドリングタイミングを柔軟に制御できていて`Observable`の上位互換のようにも見えます。

ですがこの購読側でメッセージのハンドリング中に発行されたメッセージ、つまり「`MoveNextAsync()`を呼び出す前に発行されたメッセージ」がどのようにハンドリングされるかは`IUniTaskAsyncEnumerable<T>`の実装側に委ねられています。**つまり利用者側にとってメッセージがタイミングによってはキューに積まれるのか、それとも無視されるのか、どっちの挙動になるかがわからない**のです。

たとえば先程の「連打防止」ですが、これは「`Button.OnClickAsAsyncEnumerable`は購読側で`await`している間に発行されたメッセージはすべて無視される」という前提だからこそ正しく動作しています。
![UniTaskYaya.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/47146/34e5c8e1-ad9e-6b42-921c-6c68bc16d701.jpeg)

もし同じような連打防止機構を「キュー機能を持った`UniTaskAsyncEnumerable`」に適用した場合、短期間で処理が連発することは防げますでしょうが、連打された回数だけ時間をかけて処理を実行することになってしまいます（こっちの挙動のほうが好ましい、という場合もあるので一概に間違いとは言えない）。


このように、**UniTaskAsyncEnumerableはその実装によって挙動が全く変わる場合がある**というのを覚えておく必要があります。

### 複数箇所から同時に購読する場合

（**そもそも、複数回同時に同じUniTaskAsyncEnumerableを購読すること自体があまり推奨されません。UniTaskAsyncEnumerableに対する購読は一箇所までと制限を課したほうが扱いは簡単になります。**）

まず同期版の`IEnumerable`を考えてみます。`IEnumerable`を2回`foreach`したらどうなるでしょうか。正解は2個のイテレータが生成され、それぞれ独立した`foreach`文が実行されます（それぞれごとに値の列挙が独立して実行される）。

`UniTaskAsyncEnumerable`もこれと同じであり、購読するたびに新しい非同期イテレータのインスタンスが生成されます。

たとえば次のようなコードがあったとします。`IUniTaskAsyncEnumerable<int>`を1つ作成し、これを2回購読しています。

```cs
using Cysharp.Threading.Tasks;
using Cysharp.Threading.Tasks.Linq;
using UnityEngine;
using UnityEngine.UI;

namespace TORISOUP.UniTaskAsyncEnumerableSamples.UniTaskAsyncEnumerableInstance
{
    public class UniTaskAsyncEnumerableAsyncSample : MonoBehaviour
    {
        [SerializeField] private Button _buttonA;
        [SerializeField] private Button _buttonB;
        [SerializeField] private Text _outputText;

        // カウントアップを続けるUniTaskAsyncEnumerable
        private IUniTaskAsyncEnumerable<int> CountUpAsyncEnumerable()
        {
            return UniTaskAsyncEnumerable.Create<int>(async (writer, _) =>
            {
                for (var i = 0; i < 10; i++)
                {
                    // 発行したタイムスタンプをログに出す
                    Debug.Log($"{Time.time}: {i}");
                    await writer.YieldAsync(i);
                }
            });
        }

        private void Start()
        {
            // カウントアップを続けるUniTaskAsyncEnumerableをインスタンス化した"つもり"
            var uniTaskAsyncEnumerable = CountUpAsyncEnumerable();

            // ボタンAが押されたら1個取得する
            uniTaskAsyncEnumerable
                .ForEachAwaitAsync(async i =>
                {
                    _outputText.text += $"A: {i}\n";
                    await _buttonA.OnClickAsync(destroyCancellationToken);
                }, destroyCancellationToken);
            
            // ボタンBが押されたら1個取得する
            uniTaskAsyncEnumerable
                .ForEachAwaitAsync(async i =>
                {
                    _outputText.text += $"B: {i}\n";
                    await _buttonB.OnClickAsync(destroyCancellationToken);
                }, destroyCancellationToken);

        }
    }
}
```

![twoButton.gif](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/47146/1ad65bb9-d912-3d79-777f-115ab5e8ef84.gif)


実行結果として、**2個の独立したUniTaskAsyncEnumerableが生成されて個別に購読された**ということがわかります。このように`UniTaskAsyncEnumerable`は複数回購読するとそれぞれ独立した非同期ストリームが生成されるというのが基本挙動となります。
（Pull型挙動なので当たり前の話ではある）


では「`UniTaskAsyncEnumerable`は購読されたら**必ず**独立した非同期ストリームが生成されるのか」というと、それは**NOです**（ややこしい）。

たとえば`AsyncReactiveProperty<T>`ですが、こちら経由で`IUniTaskAsyncEnumerable<T>`を生成した場合は非同期ストリームのソースは`AsyncReactiveProperty`のインスタンスそれ1本にまとまります。

```cs
using Cysharp.Threading.Tasks;
using Cysharp.Threading.Tasks.Linq;
using UnityEngine;
using UnityEngine.UI;

namespace TORISOUP.UniTaskAsyncEnumerableSamples.AsyncReactivePropertySample
{
    public class AsyncReactivePropertySample : MonoBehaviour
    {
        [SerializeField] private Button _buttonA;
        [SerializeField] private Text _outputTextA;
        [SerializeField] private Text _outputTextB;

        // カウントアップを続けるUniTaskAsyncEnumerable
        private IUniTaskAsyncEnumerable<int> CreateAsyncEnumerable()
        {
            // AsyncReactivePropertyを使って生成する
            var asyncReactiveProperty = new AsyncReactiveProperty<int>(0);

            UniTask.Void(async () =>
            {
                while (!destroyCancellationToken.IsCancellationRequested)
                {
                    // ボタンが押されたらカウントアップ
                    await _buttonA.OnClickAsync(destroyCancellationToken);
                    asyncReactiveProperty.Value++;
                    
                    Debug.Log($"{Time.time}: {asyncReactiveProperty.Value}");
                }
            });

            Debug.Log("AsyncReactivePropertyを生成しました");
            return asyncReactiveProperty;
        }

        private void Start()
        {
            // カウントアップを続けるUniTaskAsyncEnumerableを作成
            // 実体はAsyncReactiveProperty
            var uniTaskAsyncEnumerable = CreateAsyncEnumerable();

            // 購読してText Aに出力する
            uniTaskAsyncEnumerable
                .ForEachAsync(i =>
                {
                    _outputTextA.text += $"A: {i}\n";
                }, destroyCancellationToken);

            // 購読してText Bに出力する
            uniTaskAsyncEnumerable
                .ForEachAsync(i =>
                {
                    _outputTextB.text += $"B: {i}\n";
                }, destroyCancellationToken);
        }
    }
}
```

![AsyncReactiveProperty.gif](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/47146/51c4d356-a87e-d542-a625-cc2478736608.gif)


（「AsyncReactivePropertyを生成しました」は1回しかログに出ていない。つまり1個の`AsyncReactiveProperty`が2回購読されていることである。）

---

以下、補足として`Channel`の話。折りたたんでおきます。

<details><summary>Channel.CreateSingleConsumerUnboundedは1回しか購読できない</summary><div>

手続き的に`IUniTaskAsyncEnumerable<T>`を生成できる`Channel`という機能もあるのですが、こちらは購読回数が1回に制限されています。
2回以上、同じ`IUniTaskAsyncEnumerable<T>`を購読しようとすると例外が出るようになっています。

```cs
using Cysharp.Threading.Tasks;
using Cysharp.Threading.Tasks.Linq;
using UnityEngine;
using UnityEngine.UI;

namespace TORISOUP.UniTaskAsyncEnumerableSamples.ChannelSample
{
    public class ChannelSample : MonoBehaviour
    {
        [SerializeField] private Button _buttonA;
        [SerializeField] private Text _outputTextA;
        [SerializeField] private Text _outputTextB;

        // カウントアップを続けるUniTaskAsyncEnumerable
        private IUniTaskAsyncEnumerable<int> CreateAsyncEnumerable()
        {
            // Channelを使って生成する
            var channel = Channel.CreateSingleConsumerUnbounded<int>();
            var writer = channel.Writer;
            destroyCancellationToken.Register(() => writer.TryComplete());
            
            var value = 0;
            UniTask.Void(async () =>
            {
                writer.TryWrite(value++);

                while (!destroyCancellationToken.IsCancellationRequested)
                {
                    // ボタンが押されたらカウントアップ
                    await _buttonA.OnClickAsync(destroyCancellationToken);
                    writer.TryWrite(value++);
                    Debug.Log($"{Time.time}: {value}");
                }
            });

            Debug.Log("Channelを生成しました");
            return channel.Reader.ReadAllAsync(destroyCancellationToken);
        }

        private void Start()
        {
            // カウントアップを続けるUniTaskAsyncEnumerableを作成
            // 実体はAsyncReactiveProperty
            var uniTaskAsyncEnumerable = CreateAsyncEnumerable();

            // 購読してText Aに出力する
            uniTaskAsyncEnumerable
                .ForEachAsync(i =>
                {
                    _outputTextA.text += $"A: {i}\n";
                }, destroyCancellationToken);

            
            // 購読してText Bに出力する
            // ここで例外が出る
            uniTaskAsyncEnumerable
                .ForEachAsync(i =>
                {
                    _outputTextB.text += $"B: {i}\n";
                }, destroyCancellationToken);
        }
    }
}

```

![Channel.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/47146/f697acf9-2807-d598-4360-f95637bee512.jpeg)




</div></details>


#### 複数購読についてまとめると

* `UniTaskAsyncEnumerable`は購読されるたびに新しい非同期イテレータを生成するのが基本挙動
* ただし`AsyncReactiveProperty`を使うと1本にまとまるので、UniRxの`ReactiveProperty`とだいたい同じ使い方できる

ということになります。これを改めて結論を出すとこうなります。

* `UniTaskAsyncEnumerable`は`Observable`と同じ感覚でイベント通知などに用いるべきではない
    * `UniTaskAsyncEnumerable`自体の実装方法でまったく異なる挙動をするため、意図通りに動くかの保証ができない
* もしイベント通知用途ならば`AsyncReactiveProperty`を使って、型も`IReadOnlyAsyncReactiveProperty<T>`として公開するなどしたほうが安全
    * 少なくとも「`AsyncReactiveProperty`である」ということがわかっていれば大きな事故は起きないはず


# 全体のまとめ

* `Enumerable`の非同期版が`AsyncEnumerable`
* `AsyncEnumerable`のUniTask実装版が`UniTaskAsyncEnumerable`
* 似た概念で`Observable`があるがこちらとの明確な違いは「Push」であるか「Pull」であるか
* `Observable`とだいたい同じ気持ちで使える場面もあるが、使えない場面もある
* `UniTaskAsyncEnumerable`はその実装や連結されたLINQメソッドで大きく挙動を変える可能性がある
* `UniTaskAsyncEnumerable`に対する複数回の購読は非推奨である
* わからないなら`AsyncReactiveProperty`からまず触り始めるのがよい


正直、`UniTaskAsyncEnumerable`は**難しい**です。かなりのマニア向けの機能な感じがしており、万人が常用するような機能ではないと思います。

# （個人的な感想）

## 「`UniTaskAsyncEnumerable`があるならば、UniRxはもう要らないのでは？」という意見について

この辺も一度まとめてはいます。

* [2022年現在におけるUniRxの使いみち](https://qiita.com/toRisouP/items/af7d32846ab99f493d92)

それを踏まえて改めて書きます。

これについては「Yes」でもあり、「No」でもあると思います。

UniRxを「Unityイベントの購読」or「`ReactiveProperty`」くらいにしか使ってなかったのであれば、そのあたりはUniTaskに代替しても同じ様に動くと思います。なのでそういう場合は「UniRxは不要にできるかもしれない」です。

しかし「`UniTaskAsyncEnumerable`は`Observable`の完全互換ではない」です。`Observable`はPush挙動であるが`UniTaskAsyncEnumerable`はPull挙動であるという点で大きく挙動が異なります。


とくにこの「`UniTaskAsyncEnumerable`はPullで動作する」ということの意味をちゃんと把握して、内部がどういう挙動になっているのかを把握しないと事故る可能性が高いです（複数回購読時や、`Queue`や`Publish`を挟んだときの挙動の理解ができなくなるため）。

そのためイベント駆動で使う場合においては、全体的な挙動のわかりやすさや素直さとしては`Observable`の方が優れていると自分は思います。

なので、個人的な落とし所としては次のあたりなんじゃないかなぁと思います。

---
* `UniTask`は最優先で導入して良い
* `UniRx`は無くてもなんとかなるが、`UniTask`ですべて代替は厳しい部分もある
* Unityイベントの扱いについては`UniTask`/`Observable`/`UniTaskAsyncEnumerable`どれでも好きなやり方でよい
* 自分でイベント通知を実装する場合は、`Observable`/`ReactiveProperty`/`AsyncReactiveProperty`を使う
    * 剥き身の`IUniTaskAsyncEnumerable`をイベント通知用途に使うのは怖すぎるから避ける

---

そもそも`Observable`と`async/await`って併用できるし、何なら`Observable`と`UniTaskAsyncEnumerable`は相互変換が可能です。
なので「UniRxとUniTaskを両方導入していいとこ取りする」ができればそれがベストかなと思います。

あと身も蓋もない話をすると、自分はあまり`UniTaskAsyncEnumerable`を使ってません。挙動がややこしすぎる上、ピッタリこれがハマって役に立つ場面が限定的だからです。（UnityのUIイベント周りのハンドリングに使うこともあるが、UniRxでも実現できるので自分は併用してます）