ReactiveCommand / AsyncReactiveCommand

ReactiveCommand は IObservable を利用して，処理の実行許可/不許可を制御できる機構です．
AsyncReactiveCommand は ReactiveCommand の非同期処理に対応しており，
AsyncReactiveCommand が活躍する場面として，コマンド実行時に非同期処理を含む場合，
その処理が完了するまでは再度コマンドを実行出来なくすることが簡単に出来ると理解しています．

利用方法

まずはサンプルスクリプトです．

using System;
using System.Threading;
using Cysharp.Threading.Tasks;
using TMPro;
using UniRx;
using UnityEngine;
using UnityEngine.UI;

namespace Denity.Experimental
{
    public class AsyncReactiveCommandExample : MonoBehaviour
    {
        [SerializeField] TMP_Text _text;
        [SerializeField] Button _buttonDownload;

        /// <summary>
        /// 実行許可・不許可を制御するBoolのIObservable (SharedCanExecuteSource)
        /// </summary>
        readonly BoolReactiveProperty _downloadTrigger = new BoolReactiveProperty(true);

        void Awake()
        {
            // IObservable<bool> から ToReactiveCommand で生成
            var asyncReactiveCommand = _downloadTrigger.ToAsyncReactiveCommand();
            // // AsyncReactiveCommand のインスタンス生成でも可能
            // var asyncReactiveCommand = new AsyncReactiveCommand();
            // 画面遷移ボタンにバインド
            asyncReactiveCommand
                .BindTo(_buttonDownload)
                .AddTo(this);
            // 購読処理
            asyncReactiveCommand
                .Subscribe(_ =>
                {
                    // ボタンクリックした直後に実行したい処理
                    var ct = this.GetCancellationTokenOnDestroy();
                    DisplayText("Now Loading...");

                    // IObservable<Unit>を返す
                    return
                        // 非同期なロード処理
                        HeavyLoad(ct)
                            // UniTask -> Observable 変換
                            .ToObservable()
                            // OnCompleted になった時の処理
                            .ForEachAsync(_ => DisplayText("Load Complete."));
                })
                .AddTo(this);
        }

        /// <summary>
        /// 重いロード処理
        /// </summary>
        /// <param name="token">CancellationToken</param>
        private async UniTask HeavyLoad(CancellationToken token)
        {
            // 今回は簡易のためDelayを利用して疑似的にHeavyなロード処理を実現
            await UniTask.Delay(TimeSpan.FromSeconds(1.0f), cancellationToken: token);
        }

        private void DisplayText(string message)
        {
            _text.text = message;
            Debug.Log(message);
        }
    }
}

流れとして，まずは AsyncReactiveCommand インスタンスを生成します．
IObservable<bool>(IReactiveProperty<bool>) から ToAsyncReactiveCommand で生成していますが，
new AsyncReactiveCommand() で生成することも可能です．

            // IObservable<bool> から ToAsyncReactiveCommand で生成
            var asyncReactiveCommand = _downloadTrigger.ToAsyncReactiveCommand();
            // // AsyncReactiveCommand のインスタンス生成でも可能
            // var asyncReactiveCommand = new AsyncReactiveCommand();

次に，AsyncReactiveCommand として登録するコマンドをバインドします．
今回は Download ボタンの Click をコマンドとしてバインドしています．

            // 画面遷移ボタンにバインド
            asyncReactiveCommand
                .BindTo(_buttonDownload)
                .AddTo(this);

IAsyncReactiveCommand の BindTo 拡張メソッドは以下のような処理になってます．

        /// <summary>
        /// Bind AsyncRaectiveCommand to button's interactable and onClick.
        /// </summary>
        public static IDisposable BindTo(this IAsyncReactiveCommand<Unit> command, UnityEngine.UI.Button button)
        {
            var d1 = command.CanExecute.SubscribeToInteractable(button);
            var d2 = button.OnClickAsObservable().SubscribeWithState(command, (x, c) => c.Execute(x));

            return StableCompositeDisposable.Create(d1, d2);
        }

最後に，購読処理を記述します．

            // 購読処理
            asyncReactiveCommand
                .Subscribe(_ =>
                {
                    // ボタンクリックした直後に実行したい処理
                    var ct = this.GetCancellationTokenOnDestroy();
                    DisplayText("Now Loading...");

                    // IObservable<Unit>を返す
                    return
                        // 非同期なロード処理
                        HeavyLoad(ct)
                            // UniTask -> Observable 変換
                            .ToObservable()
                            // OnCompleted になった時の処理
                            .ForEachAsync(_ => DisplayText("Load Complete."));
                })
                .AddTo(this);

AsyncReactiveCommand の Subscribe の引数・返り値は以下のようになっています．

        /// <summary>Subscribe execute.</summary>
        public IDisposable Subscribe(Func<T, IObservable<Unit>> asyncAction)
        {
            lock (gate)
            {
                asyncActions = asyncActions.Add(asyncAction);
            }

            return new Subscription(this, asyncAction);
        }

IObservable<Unit> の return が必要になります．
まず非同期処理（ここでは疑似的なロード処理）を行う前に実行する処理を記述しています．

                    var ct = this.GetCancellationTokenOnDestroy();
                    DisplayText("Now Loading...");

次に非同期処理を記述しています．
返り値を IObservable<Unit> にするために，ToObservable 変換を行っています．
この時点では IObservable<Unit> ではなく， IObservable<AsyncUnit> になっています．

最後に ForEachAsync で IObservable<Unit> に変換しており，完了時の処理を記載しています．

ForEachAsync の詳細はこちら．

neue.cc

ちなみに，AsyncReactiveCommand では以下のように省略記法があるので，こちらの方が楽です．

        void Awake()
        {
            // AsyncReactiveCommand 省略記法 BindToOnClick
            _buttonDownload.BindToOnClick(_downloadTrigger, _ =>
            {
                // ボタンクリックした直後に実行したい処理
                var ct = this.GetCancellationTokenOnDestroy();
                DisplayText("Now Loading...");

                // IObservable<Unit>を返す
                return
                    // 非同期なロード処理
                    HeavyLoad(ct)
                    // UniTask -> Observable 変換
                    .ToObservable()
                    // OnCompleted になった時の処理
                    .ForEachAsync(_ => DisplayText("Load Complete."));
            });
        }

BindToOnClick 拡張メソッドの中身は以下の感じ．

        /// <summary>
        /// Bind AsyncRaectiveCommand to button's interactable and onClick and register async action to command.
        /// </summary>
        public static IDisposable BindToOnClick(this IAsyncReactiveCommand<Unit> command, UnityEngine.UI.Button button, Func<Unit, IObservable<Unit>> asyncOnClick)
        {
            var d1 = command.CanExecute.SubscribeToInteractable(button);
            var d2 = button.OnClickAsObservable().SubscribeWithState(command, (x, c) => c.Execute(x));
            var d3 = command.Subscribe(asyncOnClick);

            return StableCompositeDisposable.Create(d1, d2, d3);
        }

        /// <summary>
        /// Create AsyncReactiveCommand and bind to button's interactable and onClick and register async action to command.
        /// </summary>
        public static IDisposable BindToOnClick(this UnityEngine.UI.Button button, Func<Unit, IObservable<Unit>> asyncOnClick)
        {
            return new AsyncReactiveCommand().BindToOnClick(button, asyncOnClick);
        }

        /// <summary>
        /// Create AsyncReactiveCommand and bind sharedCanExecuteSource source to button's interactable and onClick and register async action to command.
        /// </summary>
        public static IDisposable BindToOnClick(this UnityEngine.UI.Button button, IReactiveProperty<bool> sharedCanExecuteSource, Func<Unit, IObservable<Unit>> asyncOnClick)
        {
            return sharedCanExecuteSource.ToAsyncReactiveCommand().BindToOnClick(button, asyncOnClick);
        }