【Godot C#】Godot C# における Rx、または ProcessAsObservable を自前で用意する方法

【追記】

UniRx の Cysharp から次世代 Rx ライブラリがリリースされ（2024/1～）、明示的に Godot がサポートも含まれています。オレオレ Rx をする理由がなくなりました。

Godot における Rx (Reactive Extensions) について、GDScript には GodotRx がある。

GitHub - Neroware/GodotRx: Reactive Extensions for Godot Engine version 4

一方、C# においては次のプロジェクトがあるが、最終コミットは記事現在で3年前。メンテナンスされていないようだ。

GitHub - semickolon/GodotRx: Reactive extensions for Godot C#

Godot 4.x で使いたい場合には、自前でのキャッチアップが必要になる。（なお、フォークを覗くと 4.x 対応させていそうなものがあるのも触れておく（宣伝しているわけではなさそうなので触れる程度に））

改めて考えてみると、Rx を使いたいモチベーションは次の2点。

IObservableで通知 (push) 型の処理を書きたい
UI を楽に取り扱いたい

UIに応じたAPIが必要な後者はともかく、前者については ReactiveExtensions や ReactiveProperty を導入すれば実現自体は可能。

問題は、そのままだと Godot に寄り添っていない点。例えば、処理ループをIObservable化するProcessAsObservabe的なものはほしい。

というわけで、Godot 理解の推進も兼ねてそれを用意してみる。

`ProcessAsObservable`の用意

参考：前掲の C# 版 GodotRx 、UniRx
ReactiveExtensions 導入前提

処理ループをIObservable化するには、Subject<T>を用意してIObservable<T>として公開し、処理ループのなかでOnNext(T)を発行すればよい。以下のようなクラスを用意する。

internal partial class ObservableProcessTrigger : Node
{
    public const string SearchTerm = "__ObservableProcessTrigger__";

    private Subject<double> process;
    private Subject<double> physicsProcess;
    private Subject<InputEvent> input;

    public IObservable<double> ProcessAsObservable() => process ??= new Subject<double>();
    public IObservable<double> PhysicsProcessAsObservable() => physicsProcess ??= new Subject<double>();
    public IObservable<InputEvent> InputAsObservable() => input ??= new Subject<InputEvent>();

    public override void _Process(double delta) => process?.OnNext(delta);
    public override void _PhysicsProcess(double delta) => physicsProcess?.OnNext(delta);
    public override void _Input(InputEvent @event) => input?.OnNext(@event);

    protected override void Dispose(bool disposing)
    {
        static void disposeCore<T>(Subject<T> subject)
        {
            if(subject is not null)
            {
                subject.OnCompleted();
                subject.Dispose();
            }
        }

        if(disposing)
        {
            disposeCore(process);
            disposeCore(physicsProcess);
            disposeCore(input);
        }
        base.Dispose(disposing);
    }
}

IObservableは用意できたが、このままではこのノードのものでしかない。

そこで、これをProcessAsObservableしたい対象ノードに潜り込ませる。

internal static class Observable
{
    public static IObservable<double> ProcessAsObservable(this Node node)
        => GetOrAdd(node).ProcessAsObservable();
    public static IObservable<double> PhysicsProcessAsObservable(this Node node)
        => GetOrAdd(node).PhysicsProcessAsObservable();
    public static IObservable<InputEvent> InputAsObservable(this Node node)
        => GetOrAdd(node).InputAsObservable();

    private static ObservableProcessTrigger GetOrAdd(Node node)
    {
        var trigger = node.GetNodeOrNull<ObservableProcessTrigger>(ObservableProcessTrigger.SearchTerm);
        if(trigger is null)
        {
            trigger = new ObservableProcessTrigger() { Name = ObservableProcessTrigger.SearchTerm };
            node.AddChild(trigger);
        }

        return trigger;
    }
}

これでNode.ProcessAsObservable()が書けるようになった。

試用

単純にジャンプをするような例で試してみる。

ジャンプには接地が必要
ジャンプにはクールダウンあり
クールダウン状況をUI表示する
ReactiveExtensions, ReactiveProperty 利用

public partial class Player : CharacterBody2D
{
    CompositeDisposable disposables = new();

    private bool isOnGround = false;
    private Vector2 motion = Vector2.Zero;
    private ReactivePropertySlim<int> Cooldown { get; set; } = new();
    private ReadOnlyReactivePropertySlim<bool> NeedsCooldown { get; set; }

    public override void _Ready()
    {
        const float jumpPower = -1000f;
        const float g = 50f;
        const int jumpCooldownFrame = 90;

        UpDirection = Vector2.Up;

        NeedsCooldown = Cooldown
            .Select(v => 0 < v)
            .ToReadOnlyReactivePropertySlim().AddTo(disposables);

        // ProcessAsObservable でクールダウン
        this.ProcessAsObservable()
            .Where(_ => NeedsCooldown.Value)
            .Subscribe(_ => --Cooldown.Value).AddTo(disposables);

        // InputAsObservable でジャンプ設定
        this.InputAsObservable()
            .Where(ev => ev.IsActionPressed("ui_accept"))
            .Where(_ => isOnGround && NeedsCooldown.Value is false)
            .Subscribe(_ =>
            {
                motion.Y = jumpPower;
                Cooldown.Value = jumpCooldownFrame;

            }).AddTo(disposables);

        // PhysicsProcessAsObservable で物理
        this.PhysicsProcessAsObservable()
            .Subscribe(_ =>
            {
                isOnGround = IsOnFloor();
                if (isOnGround is false)
                {
                    motion.Y += g;
                }

                Velocity = motion;
                MoveAndSlide();

            }).AddTo(disposables);

        // クールダウンのプログレスバー反映
        var progress = GetParent().GetNode<ProgressBar>("ProgressBar");
        progress.MaxValue = jumpCooldownFrame;
        progress.Value = jumpCooldownFrame;
        Cooldown
            .Subscribe(v => progress.Value = jumpCooldownFrame - v).AddTo(disposables);
    }

    protected override void Dispose(bool disposing)
    {
        if (disposing) disposables.Dispose();
        base.Dispose(disposing);
    }
}