R3 Observables and Signals

The FlowTask.R3 package connects R3’s Observable<T> and FlowTask’s Signal<T> in both directions. This page explains ToSignal, for awaiting an Observable from a flow, and ToObservable, for subscribing to a Signal as an Observable.

Installation

Reference FlowTask.R3 from NuGet. R3 comes in as a dependency. The namespace is Katout.FlowTask, the same as the core.

There is no UPM package. On Unity, you can turn an Observable’s values into a signal with the core’s FlowBridge.FromCallback.

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

In this form, the Observable’s completion and failure don’t reach the signal. The signal closes when the scope that created it ends (Signals).

Awaiting an Observable in a flow: 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() returns an EventSignal<T> owned by the current scope.

It subscribes to the Observable at the point you call it. When the scope ends (or when you call Dispose), it unsubscribes from R3 and closes the signal.
Values from OnNext are emitted to the signal. An OnNext from another thread goes through the World’s inbox and arrives at the next Tick or Flush.
A flow doesn’t resume synchronously inside OnNext. It resumes at the next flush.
If you give it a name with the name argument, the name appears in the wait description of the dump (clicks.ToSignal("Clicks")).
A value the Observable sends right when it is subscribed (the current value of a ReactiveProperty or a BehaviorSubject) isn’t received, because nobody waits yet. Read the current value first with CurrentValue (Value for a BehaviorSubject).

Next() waits only for the next value after you call it (an edge). Values that arrive while nobody is waiting are lost. If you don’t want to miss any, subscribe.

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
}

Subscriptions and buffer policies are covered in Signals.

Completion and failure

When the Observable completes, the signal closes. A completion on another thread also closes the signal after the values sent before it.

Waiting on a closed signal with Next() throws SignalClosedException. With NextOrClosed(), you get (false, default) instead.
When the Observable ends with a failure, the signal closes with that failure. A flow waiting with Next() receives a SignalClosedException whose InnerException is the original exception.
The failure is also reported to R3’s unhandled exception handler (the one you set with ObservableSystem.RegisterUnhandledExceptionHandler). When no flow is waiting, this is where you find out about the failure.
An error sent with OnErrorResume doesn’t reach the signal; it goes only to R3’s unhandled exception handler. The signal stays open, and a waiting flow keeps waiting. To receive it in a flow, add OnErrorResumeAsFailure() before ToSignal() to turn it into a completion with a failure (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() returns (false, default) both on completion and on failure. To tell a failure apart, wait with Next() and look at the InnerException of the SignalClosedException.

Subscribing to a Signal as an 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) returns an Observable that delivers the Signal’s Emits.

Each R3 subscription starts one forwarding flow in the World. This flow appears in the dump with the name R3.ToObservable(…).
Values aren’t delivered inside Emit. They are delivered in the World’s flush, in the order they were emitted.
A subscription receives the Emits from the point it subscribed.
Subscribe on the World’s thread.
When the Signal closes, it finishes delivering the buffered values and then calls OnCompleted. When the Signal is closed as a failure with Close(error), it calls it with a failure (Result.Failure) holding that error.
Disposing the R3 subscription cancels the forwarding flow. You can call Dispose from any thread. From another thread, it takes effect at the next Tick or Flush, but no values are delivered after Dispose. When the subscriber disposes, OnCompleted isn’t called.
If the forwarding flow ends before the Signal closes (because the World was disposed, or because of an exception), it calls OnCompleted with a failure holding an OperationCanceledException.

Buffer limit

Emits that arrive between two flushes are buffered in the subscription. By default, it holds up to 256 values, and going over that is an unhandled exception (values are never dropped silently). You can change the policy with an argument.

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

Details

The signal that ToSignal() creates is the same EventSignal<T> that FlowBridge.FromCallback creates. Ownership and closing are covered in Signals.
Create a Signal per World and pass it around. Don’t put it in a static (Signals).
How to send values from another thread is covered in Threads.