UniTask -> UniRx変換を行う際に見落としてしまったところ

概要

https://github.com/Cysharp/UniTask
https://github.com/neuecc/UniRx

UniTaskとUniRxは同じ作者(neueccさん)のものでして、今はリポジトリが分割されましたが
元々は単一プラグインでした。
非同期を扱う特徴としてはある程度似たようなところもありますが、
それぞれ場面による長所があり両方を一緒に使っているプロジェクトを世の中でよく見かけます。

UniRx -> UniTaskの変換はObservable.Awaiter.csでAwaiterが実装されており、変換せずに awaitを書ける、もしくは明示的にToUniTask()メソッドを使い変換させる方法があります。
(ToUniTaskを使うとCancellationTokenを渡せるのと、useFirstValueが用意されており
後者が望ましいと思います)

UniTask -> UniRxに関しては ToObservable()メソッドが用意されており、これを使えばRxとして扱うことが出来ます。

今回はUniTask -> UniRxを変換する際にミスしやすいところに関して軽く話してみようと思います。

本題

ToObservable()の内部実装としては、AsyncSubjectを使い結果を受け取る処理となっております。
ここでわかると思いますが、Fire().Forget()となっているため
UniTaskを非同期に走らせ、その結果をObservableに通知させる
ことになります。

Observer/Observableの役割としてはある処理の監視を行い通知することで終わるので、
走った処理をキャンセルする義務はないものに意味合い的には間違いないと思います。

Taskのキャンセルですが、ObservableのようにDisposableとして購読を破棄するわけではなく
キャンセル処理を自前で実装しないといけないため、実装が確立されているUniTask型に対して後ほど外部からキャンセルさせることは出来なくなっております。

とのことで、こういうところを見落としてしまいました。
1.ToObservableをしたとしても内部ではUniTaskが普通に走るため、
SubscribeしなかったらもちろんObservableは走らないが普通にUniTaskの処理は走る

使うつもりながらSubscribeしないまま放置するObservableは普通にコードミスだと考えて良いでしょうが、SelectMany/ContinueWith等の引数としてFuncでないIObservableを受け取るところで使われる際には明示的に遅延させないと意図しないタイミングで走る可能性があったためDeferを囲むことになりました。

private async UniTask HogeAsync()
{
    ...
}

//変更前
button.OnClickAsObservable()
    .SelectMany(HogeAsync().ToObservable()) // 多分Select().Switch()が望ましいですが
    .Subscribe()
    .AddTo(disposable);

//変更後
button.OnClickAsObservable()
    .SelectMany(Observable.Defer(() => HogeAsync().ToObservable()))
    .Subscribe()
    .AddTo(disposable);

2.ObservableがDisposeされることで、(もちろん)UniTaskもキャンセルされるわけではない

2に関しては、キャンセルさせないといけないUniTaskに関しては以下のコードを書くことで解決しました。

private async UniTask HogeAsync(CancellationToken ct)
{
    //ctを用いてキャンセル処理を行う
    ...
}

//変更前
//複数Clickを行っても前のHogeAsyncはキャンセルされず最後まで走る
button.OnClickAsObservable()
    .Select(_ => HogeAsync().ToObservable())
    .Switch()
    .Subscribe()
    .AddTo(disposable);

//変更後
private IObservable<Unit> HogeAsObservable()
{
    return Observable.Create<Unit>(observer =>
    {
        var disposable1 = new CancellationDisposable();
        var disposable2 = HogeAsync(disposable1.Token).ToObservable()
            .Subscribe(observer);              

        return new StableCompositeDisposable(disposable1, disposable2);
    });
}

button.OnClickAsObservable()
    .Select(_ => HogeAsObservable())
    .Switch()
    .Subscribe()
    .AddTo(disposable);

複数のCancellationTokenを使わないといけないなら CancellationTokenSource.CreateLinkedTokenSourceを使う感じですかね。

終わりに

UniTaskのキャンセル処理をきちんとケアするのが面倒くさい割りに、UniTaskを使うと可読性があがるのと簡単に値を返すだけの操作にはUniTaskが向いているので、どっちかを外す選択は出来なく悩ましいですね。逆にUniRxのストリームの細かい動作の操作も欲しい場合があるので、個人的にはキャンセルをそんなに考慮しなくても良いDomain層での非同期処理に関してはUniTaskを、ストリームの操作が頻繁に必要となる際にはUniRxを使う感じになってます。(複数の値と単体の値で区分けもできますが)
しばらくは両方適切に混ぜて使う形になりそうです。