【Unity】R3で Observable.Create() を使って Observable を自作する方法

はじめに

この記事では UniRx の後継である R3 で Observable.Create() を使用して Observable を自作する方法を解説します。

今回はこのコードのように引数に渡した Key が押されたらメッセージを発行してその押された Key を渡すような Observable を作成します。
このコードではスペースキーが押されたら、そのキーの表示名をログに表示しています。

OnPressedKeyObservable(Key.Space) //←これを作る
    .Subscribe(pressedKey => Debug.Log(Keyboard.current[pressedKey].displayName))
    .AddTo(this);

つまり、上のコードでやろうとしていることは以下のコードとほぼ同じです。

this.UpdateAsObservable()
    .Select(_ => Key.Space)
    .Where(key => Keyboard.current[key].wasPressedThisFrame)
    .Subscribe(pressedKey => Debug.Log(Keyboard.current[pressedKey].displayName))
    .AddTo(this);

この記事では Observable.Create() を使って、1つ目のコードの OnPressedKeyObservable() を作ってみます。

この記事での検証環境は以下の通りです。

• Unity 6000.2.6f2
• ObservableCollections.R3 3.3.4
• R3 1.3.0
• UniTask 2.5.10

InputSystem.Key について

ここで InputSystem.Key についても少し解説しておきます。

KeyCode と InputSystem.Key の違い

Unity でキーの一覧が列挙子となっている列挙型で最も有名なのは KeyCode かと思いますが、この KeyCode は主に「旧」の入力システムである Input Manager で使用します。
なので先ほどの UpdateAsObservable() を使用したコードで KeyCode を使うと以下のコードのようになります。

this.UpdateAsObservable()
    .Select(_ => KeyCode.Space) // KeyCode を流す
    .Where(keyCode => Input.GetKeyDown(keyCode)) // Input.GetKeyDown() を使用する
    .Subscribe(pressedKeyCode => Debug.Log(pressedKeyCode.ToString())) // KeyCode の名前を取得する方法は恐らく ToString() しかないはず
    .AddTo(this);

「Edit ＞ Project Settings… ＞ Player ＞ Active Input Handling」が「Input System Package (New)」になっている状態でそのコードの処理を実行すると「InvalidOperationException: You are trying to read Input using the UnityEngine.Input class, but you have switched active Input handling to Input System package in Player Settings.」というエラーが発生します。

このエラーを日本語に訳すと以下のようになります。

InvalidOperationException： UnityEngine.Inputクラスを使用してInputを読み込もうとしていますが、Player SettingsでアクティブなInput処理をInput Systemパッケージに切り替えています。

DeepL

この場合は「Edit ＞ Project Settings… ＞ Player ＞ Active Input Handling」を「Input Manager (Old)」か「Both」に変えるとエラーが発生しなくなりますが、「Both」の状態だと「The Input Manager is a legacy feature and not recommended for new projects. For new projects you should use the Input System Package.」という警告が表示されます。

これも日本語に訳すと以下のようになります。

インプット・マネージャーはレガシー機能であり、新しいプロジェクトには推奨されません。新しいプロジェクトにはInput System Packageを使用してください。

DeepL

プロジェクトで使用しているパッケージが Input Manager に依存してしまっているなどの、どうしても Input Manager を使わざるを得ない場合を除いて基本的には Input System を使用した方が良さそうです。

KeyCode の公式ドキュメントにも「Consider using the new Input System package instead of KeyCode-based input.」と書かれています。

KeyCodeベースの入力の代わりに、新しいInput Systemパッケージの使用を検討する。

DeepL

「新」入力システムである Input System では KeyCode ではなく InputSystem.Key を使用します。
InputSystem.Key の公式ドキュメントにはいろいろ書かれていますが、とりあえず今回の場合は KeyCode ではなく InputSystem.Key を使用した方が良さそうです。

isPressed と wasPressedThisFrame の違い

Input Manager でキーの押下判定を取得するときは Input.GetKeyDown() を使用していましたが、Input System では ButtonControl.isPressed や ButtonControl.wasPressedThisFrame を使用するようです。

これらの違いは簡単で、isPressed を使用して以下のようなコードを書くとスペースキーが押されている間、毎フレーム「Space」とログに表示されます。

this.UpdateAsObservable()
    .Select(_ => Key.Space)
    .Where(key => Keyboard.current[key].isPressed) // isPressed を使用
    .Subscribe(pressedKey => Debug.Log(Keyboard.current[pressedKey].displayName))
    .AddTo(this);

wasPressedThisFrame を使用して以下のようなコードを書くとスペースキーが押された時に一度だけ true になり、ログに「Space」と表示されます。
スペースキーを押し続けても isPressed のように毎フレーム、ログに表示され続けるわけではありません。

this.UpdateAsObservable()
    .Select(_ => Key.Space)
    .Where(key => Keyboard.current[key].wasPressedThisFrame) // wasPressedThisFrame を使用
    .Subscribe(pressedKey => Debug.Log(Keyboard.current[pressedKey].displayName))
    .AddTo(this);

つまり、Input.GetKey() が ButtonControl.isPressed で、Input.GetKeyDown() が ButtonControl.wasPressedThisFrame っぽいです。
ちなみにこの ButtonControl の公式ドキュメントはこちらです。

キーの表示名の取得方法

「旧」入力システムである Input Manager の KeyCode でそのキーの表示名を取得する直接的な方法は恐らく無く、enum を ToString() するしかなかったっぽいのですが、InputSystem.Key では InputControl というクラスに name や displayName、shortDisplayName などのプロパティが用意されています。

InputControl の公式ドキュメントから各プロパティの概要を引用して表にまとめてみました。

プロパティ名	概要（「Class InputControl」より引用）
name	The name of the control, i.e. the final name part in its path.
displayName	The text to display as the name of the control.
shortDisplayName	An alternate, abbreviated displayName (for example “LMB” instead of “Left Button”).

とりあえず今回は displayName でいきます。
実際の使用方法は以下の通りです。

var key = Key.Space;
Debug.Log(Keyboard.current[key].displayName); //「Space」と表示される

Observable.Create() を使用した Observable の作り方

話がだいぶ逸れてしまいましたが、本題に戻ります。
ここからは Observable.Create() を使用してオリジナルの Observable を作ってみます。

結論

早速結論から言うと以下のコードのようになりました。

private static Observable<Key> OnPressedKeyObservable(Key key)
{
    return Observable.Create<Key>(async (observer, token) =>
    {
        try
        {
            while (true)
            {
                await UniTask.WaitUntil(() => Keyboard.current[key].wasPressedThisFrame, cancellationToken: token);
                        
                observer.OnNext(key);
                
                await UniTask.Yield(token);
            }
        }
        catch (Exception e)
        {
            observer.OnCompleted(e);
        }
    });
}

素直に UpdateAsObservable() と Where() を使って実装した方がいいような気もしますが、この静的関数を使って以下のように書くと、スペースキーが押される度にログに「Space」と表示されます。

OnPressedKeyObservable(Key.Space)
    .Subscribe(pressedKey => Debug.Log(Keyboard.current[pressedKey].displayName))
    .AddTo(this);

先ほどの結論のコードを簡略化すると以下のようになります。
Observable.Create() で async/await を使って書くときは基本的にこの構文？というか書き方になるかと思います。
とてもシンプルかつ直感的に書けますね。

private static Observable<Key> OnPressedKeyObservable(Key key)
{
    return Observable.Create<Key>(async (observer, token) =>
    {
        // await で待機して適切なタイミングで Observer に通知する
    });
}

そしてR3 では Observable から Observer に通知する手段は OnNext() と OnCompleted()、OnErrorResume() の3つです。

Observer.OnNext()

先ほどの結論のコードでは while (true) の無限ループの中で「await UniTask.WaitUntil(() => Keyboard.current[key].wasPressedThisFrame , cancellationToken: token);」でキーが押されるまで待機してから Observer.OnNext() でキーが押されたことを Observer に通知しています。

 private static Observable<Key> OnPressedKeyObservable(Key key)
{
    return Observable.Create<Key>(async (observer, token) =>
    {
        try
        {
            while (true)
            {
                await UniTask.WaitUntil(() => Keyboard.current[key].wasPressedThisFrame, cancellationToken: token);
                        
                observer.OnNext(key); //←ココ
                
                await UniTask.Yield(token);
            }
        }
        catch (Exception e)
        {
            observer.OnCompleted(e);
        }
    });
}

Observer.OnCompleted()

OperationCanceledException を渡した場合

そして UniTask.WaitUntil() でも UniTask.Yield() でも CancellationToken を渡しているので Observer からキャンセルされると OperationCanceledException がスローされます。

なので試しに以下のコードのように購読を開始した3秒後に Observer（MonoBehaviour）を削除してみます。

private void Start()
{
    OnPressedKeyObservable(Key.Space)
        .Subscribe(pressedKey => Debug.Log(Keyboard.current[pressedKey].displayName))
        .AddTo(this); //自身が削除されたら Dispose する
            
    //3秒後に自身を削除する
    Destroy(this, 3f);
}

そして Observable 側に「Debug.Log(“catch: ” + e.GetType());」という1文を追加して、キャッチした例外の種類を確認してみます。

 private static Observable<Key> OnPressedKeyObservable(Key key)
{
    return Observable.Create<Key>(async (observer, token) =>
    {
        try
        {
            while (true)
            {
                await UniTask.WaitUntil(() => Keyboard.current[key].wasPressedThisFrame, cancellationToken: token);
                        
                observer.OnNext(key);
                
                await UniTask.Yield(token);
            }
        }
        catch (Exception e)
        {
            Debug.Log("catch: " + e.GetType()); //追加

            observer.OnCompleted(e);
        }
    });
}

すると「catch: System.OperationCanceledException」と表示されました。

次に Observer 側を以下のコードのように修正してみました。
3秒後にObserver を削除する処理はそのままで、Subscribe() の中で OnNext と OnErrorResume、OnCompleted の内容を確認しています。

private void Start()
{
    OnPressedKeyObservable(Key.Space)
        .Subscribe(
            pressedKey => Debug.Log("OnNext: " + Keyboard.current[pressedKey].displayName),
            exception => Debug.Log("OnErrorResume: " + exception.Message),
            result => Debug.Log("OnCompleted: " + result))
        .AddTo(this);
            
    Destroy(this, 3f);
}

すると先ほどと同じように「catch: System.OperationCanceledException」とだけ表示され、「OnCompleted: ○○」は表示されません。

「Debug.Log(“catch: ” + e.GetType());」の後に Observer.OnCompleted() を呼び出して、発生した例外を渡していますが、その例外がキャンセル処理による OperationCanceledException の場合は Subscribe() の onCompleted は実行されないようです。
（たぶん）

OperationCanceledException 以外の例外を渡した場合

次はキーが押されたら Observer.OnNext() でメッセージを発行して、その後すぐに「throw new Exception();」で適当な例外を投げてみます。

 private static Observable<Key> OnPressedKeyObservable(Key key)
{
    return Observable.Create<Key>(async (observer, token) =>
    {
        try
        {
            while (true)
            {
                await UniTask.WaitUntil(() => Keyboard.current[key].wasPressedThisFrame, cancellationToken: token);
                        
                observer.OnNext(key);

                throw new Exception(); //追加
                
                await UniTask.Yield(token); //上で止まるからこれは意味ない
            }
        }
        catch (Exception e)
        {
            Debug.Log("catch: " + e.GetType());

            observer.OnCompleted(e);
        }
    });
}

この状態でプレイしてスペースキーを押すと「OnNext: Space」と「catch: System.Exception」と表示されます。
そして「OnCompleted: Failure{Exception of type ‘System.Exception’ was thrown.}」とも表示されています。
そのログの表示以降はスペースキーを押しても何も起こりません。

OperationCanceledException のときは Subscribe() の onCompleted は実行されませんでしたが、それ以外の例外のときは「失敗」として onCompleted でその例外の内容を確認することができるようです。
（たぶん）

何も渡さなかった場合

次はキーが押されたら Observer.OnNext() でメッセージを発行して、その後すぐに Observer.OnCompleted() でストリームを終了させてみます。

 private static Observable<Key> OnPressedKeyObservable(Key key)
{
    return Observable.Create<Key>(async (observer, token) =>
    {
        try
        {
            while (true)
            {
                await UniTask.WaitUntil(() => Keyboard.current[key].wasPressedThisFrame, cancellationToken: token);
                        
                observer.OnNext(key);

                observer.OnCompleted(); //追加
                
                await UniTask.Yield(token);
            }
        }
        catch (Exception e)
        {
            Debug.Log("catch: " + e.GetType());

            observer.OnCompleted(e);
        }
    });
}

すると「OnNext: Space」の後に「OnCompleted: Success」と表示されてストリームが正常に終了しているのがわかります。

その後に OperationCanceledException が発生しています。
これは Observer.OnCompleted() 後の「await UniTask.Yield(token);」の部分で発生したものです。
これ以降はストリームが終了しているのでスペースキーを押しても何も起こりません。

Observer.OnErrorResume()

次はキーが押されたら Observer.OnErrorResume() で Observer にエラーを通知しつつストリームを続行してみます。

 private static Observable<Key> OnPressedKeyObservable(Key key)
{
    return Observable.Create<Key>(async (observer, token) =>
    {
        try
        {
            while (true)
            {
                await UniTask.WaitUntil(() => Keyboard.current[key].wasPressedThisFrame, cancellationToken: token);
                        
                observer.OnNext(key);

                observer.OnErrorResume(new Exception()); //追加
                
                await UniTask.Yield(token);
            }
        }
        catch (Exception e)
        {
            Debug.Log("catch: " + e.GetType());

            observer.OnCompleted(e);
        }
    });
}

この状態でプレイしてスペースキーを押してみると「OnNext: Space」の後に「OnErrorResume: Exception of type ‘System.Exception’ was thrown.」と表示されますが、その後もスペースキーを何度も押せます。

適切なタイミングでメッセージを発行してストリームを続行させたいときは Observer.OnNext() を、ストリームを正常に終了させたいときは Observer.OnCompleted() を、エラーの発生によってストリームを終了させたいときは Observer.OnCompleted() に例外を渡し、Observer にエラーを通知しつつストリームを続行させたいときは Observer.OnErrorResume() を使うといった感じでしょうか。

最後に

参考記事

• 【Unity】【R3】非同期メソッド版のObservable.Createでストリームを直感的に書く
• 【Unity】【UniRx】独自のObservableを作る方法まとめ
• 【Unity】 R3とUniRxの比較まとめ
• 次世代Rx「R3」解説
• 【C#】async/awaitのキャンセル処理まとめ
• 【Unity】Input Systemの使い方入門
• 新InputSystemでKeyInputはどう書くの？【Unity】
• KeyCode
• Enum Key
• Class ButtonControl
• Class InputControl

お問い合わせ