R3 の Observable と Signal

R3 の Observable<T> と FlowTask の Signal<T> は、FlowTask.R3 パッケージで互いにつなげます。このページでは、Observable をフローから待つ ToSignal と、Signal を Observable として購読する ToObservable を説明します。

導入

NuGet の FlowTask.R3 を参照します。R3 も依存として入ります。名前空間は本体と同じ Katout.FlowTask です。

UPM のパッケージはありません。Unity では、本体の FlowBridge.FromCallback で Observable の値をシグナルにできます。

// damaged is an Observable<int>
using var hits = FlowBridge.FromCallback<int>(emit => damaged.Subscribe(emit).Dispose);
var damage = await hits.Next();

この形では、Observable の完了と失敗はシグナルに届きません。シグナルは、作ったスコープの終わりで閉じます（シグナル）。

Observable をフローで待つ：ToSignal

async FlowTask WaitForClick(Observable<Unit> clicks)
{
    using var click = clicks.ToSignal();   // subscribed now, disposed when this scope ends
    var r = await FlowTask.Race(click.Next(), FlowTask.WaitForSeconds(1));
    if (r.Index == 0) await OnClicked();
}

ToSignal() は、今のスコープが所有する EventSignal<T> を返します。

呼んだ時点で Observable を購読します。スコープが終わると（または Dispose すると）R3 の購読を解除し、シグナルを閉じます。
OnNext の値はシグナルに Emit されます。別のスレッドからの OnNext は、World の受信箱を通って次の Tick か Flush で届きます。
フローは OnNext の中で同期には再開しません。次のフラッシュで再開します。
name 引数で名前を付けると、ダンプの待ちの表示に出ます（clicks.ToSignal("Clicks")）。
購読したときにすぐ流れる値（ReactiveProperty や BehaviorSubject の今の値）は、まだ待ち手がいないので受け取れません。今の値は、先に CurrentValue（BehaviorSubject は Value）で読みます。

Next() は呼んだ後の次の値だけを待ちます（エッジ）。待っていない間の値は失われます。取りこぼしたくないなら購読します。

using var hits = damage.ToSignal();
using var queue = hits.Subscribe(BufferPolicy.Queue(16, BufferOverflow.DropOldest));
while (true)
{
    var hit = await queue.Next();
    await HitStun(hit);   // hits during the stun are kept in the queue
}

購読とバッファの方針はシグナルにあります。

完了と失敗

Observable が完了すると、シグナルは閉じます。別のスレッドでの完了も、その前に送られた値の後で閉じます。

閉じたシグナルを Next() で待つと SignalClosedException が投げられます。NextOrClosed() なら (false, default) が返ります。
Observable が失敗で終わると、シグナルはその失敗を持って閉じます。Next() で待っていたフローには、InnerException が元の例外の SignalClosedException が届きます。
失敗は R3 の未処理例外のハンドラ（ObservableSystem.RegisterUnhandledExceptionHandler で設定するもの）にも報告されます。待っているフローがないときは、ここで失敗が分かります。
OnErrorResume で流れるエラーは、シグナルに届かず、R3 の未処理例外のハンドラだけに行きます。シグナルは閉じず、待っているフローは待ち続けます。フローで受けるなら、ToSignal() の前に OnErrorResumeAsFailure() を付けて、失敗での完了に変えます（source.OnErrorResumeAsFailure().ToSignal()）。

using var feed = source.ToSignal("Feed");
using var items = feed.Subscribe(BufferPolicy.Queue(8, BufferOverflow.Fail));
while (true)
{
    var (received, item) = await items.NextOrClosed();
    if (!received) break;                         // the observable completed or failed
    Show(item);
}

NextOrClosed() は、完了でも失敗でも (false, default) を返します。失敗を見分けるなら、Next() で待って SignalClosedException の InnerException を見ます。

Signal を Observable として購読する：ToObservable

var hits = new Signal<int>(world, "Hits");
using var log = hits.ToObservable(world)
    .Where(damage => damage >= 10)
    .Subscribe(damage => Console.WriteLine($"Big hit: {damage}"));

ToObservable(world) は、Signal の Emit を流す Observable を返します。

R3 の購読 1 つごとに、World で転送のフローを 1 つ始めます。このフローは R3.ToObservable(…) という名前でダンプに出ます。
値は Emit の中では流れず、World のフラッシュで Emit の順に流れます。
購読は、購読した時点からの Emit を受け取ります。
World のスレッドで購読してください。
Signal が閉じると、たまった値を流し終えてから OnCompleted を呼びます。Close(error) で失敗として閉じたときは、そのエラーを持った失敗（Result.Failure）で呼びます。
R3 の購読を Dispose すると、転送のフローが取り消されます。Dispose はどのスレッドからでも呼べます。別のスレッドからは次の Tick か Flush で効きますが、Dispose の後の値は流れません。購読者が Dispose したときは OnCompleted は呼ばれません。
Signal が閉じる前に転送のフローが終わったとき（World の Dispose、例外）は、OperationCanceledException を持った失敗で OnCompleted を呼びます。

バッファの上限

2 回のフラッシュの間に届いた Emit は、購読のバッファにたまります。既定では 256 個までで、それを超えると未処理の例外になります（値を黙って捨てません）。方針は引数で変えられます。

var latest = position.ToObservable(world, BufferPolicy.Latest);   // only the newest value per flush

詳しく

ToSignal() で作るシグナルは、FlowBridge.FromCallback で作るものと同じ EventSignal<T> です。所有と閉じ方はシグナルにあります。
Signal は World ごとに作って渡します。static に置かないでください（シグナル）。
別のスレッドから値を送る仕組みはスレッドにあります。