From R3 and UniRx

Use these tables to move R3 or UniRx code onto Onity.Reactive name by name. The vocabulary is familiar: Subject<T>, ReactiveProperty<T>, Where, Select, Merge, CombineLatest, Scan. The one public stream contract is IOnityObservable<T> (R3’s Observable<T>), implemented by Subject<T>, ReactiveProperty<T>, every operator and the message bridge Observe<T>(), so messages and state share one operator surface. Every mapping below is verified against the shipped API (Onity.Reactive, Onity.Unity.Reactive, Onity.Unity.Async, Onity.Composition).

The differences that change code:

• Trailing-edge throttle is ThrottleLast; Throttle is the leading-edge cool-down (R3 ThrottleFirst); UniRx’s Throttle is Debounce.
• No Publish, Share or RefCount: a Subject<T> is already multicast, so share one instance through DI.
• Lifetime is explicit: every Subscribe returns an IDisposable that you scope with AddTo(this) (a Component), TakeUntilDisable(this), AddTo(compositeDisposable) or AddTo(scope). There is no AddTo(GameObject).
• IReadOnlyReactiveProperty<T> is not an IOnityObservable<T>; operators run over the ReactiveProperty<T> itself.
• Errors do not travel down the chain; see Errors.

The measured comparison with R3 1.3.0 and UniRx 7.1.0 is in Reactive vs R3 and UniRx; the Comparisons hub collects every measured comparison.

Primitives

R3 / UniRx	Onity	Namespace	Notes
new Subject<T>()	new Subject<T>()	Onity.Reactive	Multicast source. OnNext allocates nothing in steady state; OnNext and Subscribe after Dispose() throw ObjectDisposedException.
subject.OnNext(v)	subject.OnNext(v)	Onity.Reactive	Same. There is no OnCompleted / OnError on the subject; dispose it instead.
new ReactiveProperty<T>(initial)	new ReactiveProperty<T>(initial)	Onity.Reactive	Equal values are skipped (DistinctUntilChanged built in, optional IEqualityComparer<T>). Subscribing emits the current value first.
A property registered by hand in DI	container.BindReactiveProperty(initialValue)	Onity.Composition	Binds one instance as ReactiveProperty<T> and IReadOnlyReactiveProperty<T>; the scope disposes it.
rp.Value = x	rp.Value = x	Onity.Reactive	Notifies only when the value changed.
(no bool-returning set)	rp.SetValue(x) -> bool	Onity.Reactive	Sets and returns whether the value changed.
rp.Subscribe(...)	rp.Subscribe(...)	Onity.Reactive	Current value first, then each change. rp.Subscribe(onNext, emitCurrentValue: false) skips the initial emission.
ReadOnlyReactiveProperty<T>	IReadOnlyReactiveProperty<T>	Onity.Reactive	Value plus Subscribe(observer, emitCurrentValue = true). Not an IOnityObservable<T>: expose the ReactiveProperty<T> when consumers need operators, or wrap with new OnityObservable<T>(observer => readOnly.Subscribe(observer)).
new CompositeDisposable()	new CompositeDisposable()	Onity.Reactive	Add, Remove, Clear, Count, Dispose. Not an ICollection<IDisposable>.
Observable.FromEvent<T>(add, remove)	OnityObservable.FromEvent<T>(add, remove)	Onity.Reactive	Factory.
Observable.Return(v)	OnityObservable.Return<T>(v)	Onity.Reactive	Factory.
Observable.Empty<T>()	OnityObservable.Empty<T>()	Onity.Reactive	Factory.
Observable.Create<T>(subscribe)	new OnityObservable<T>(subscribe)	Onity.Reactive	The constructor takes Func<Observer<T>, IDisposable>.
Observable.Never<T>(), Observable.Defer(...)	no factory		A custom source through the constructor above.
Observable<T>	IOnityObservable<T>	Onity.Reactive	Observer<T> is delegate void Observer<T>(T value); the Subscribe(Action<T>) you write is an extension that wraps it.

Synchronous operators

All return IOnityObservable<T> and allocate only when subscribed (OnityObservableExtensions, Onity.Reactive).

R3 / UniRx	Onity	Namespace	Notes
Where(predicate)	Where(Predicate<T>)	Onity.Reactive
Select(selector)	Select(Func<TSource, TResult>)	Onity.Reactive
DistinctUntilChanged()	DistinctUntilChanged(IEqualityComparer<T> = null)	Onity.Reactive
Skip(n) / SkipWhile(p)	Skip(int) / SkipWhile(Predicate<T>)	Onity.Reactive	A negative count throws ArgumentOutOfRangeException.
Take(n) / TakeWhile(p)	Take(int) / TakeWhile(Predicate<T>)	Onity.Reactive	Take disposes the upstream subscription after n values.
R3 Prepend(v), UniRx StartWith(v)	StartWith(T)	Onity.Reactive
Scan(seed, accumulator)	Scan<TState>(TState seed, Func<TState, T, TState>)	Onity.Reactive
Pairwise()	Pairwise() -> IOnityObservable<OnityPair<T>>	Onity.Reactive	Previous / Current; the first value only primes the pair.
Merge(a, b, ...)	Merge(params IOnityObservable<T>[])	Onity.Reactive
CombineLatest(other, selector)	CombineLatest<T1, T2, TResult>(IOnityObservable<T2>, Func<T1, T2, TResult>)	Onity.Reactive	Two sources only.
UniRx Sample(sampler)	Sample<TSignal>(IOnityObservable<TSignal> sampler)	Onity.Reactive	Emits the latest value on each sampler signal.
R3 Chunk(count), UniRx Buffer(count)	Buffer(int count) -> IOnityObservable<IReadOnlyList<T>>	Onity.Reactive	A new list per batch.
Subscribe(onNext)	Subscribe(Action<T>)	Onity.Reactive	Returns IDisposable.
Subscribe(onNext, onError, onCompleted)	Subscribe(Action<T>, Action<Exception>, Action<OnityResult>)	Onity.Reactive	onCompleted runs when the subscription is disposed; see Errors.
TakeUntil(cancellationToken)	TakeUntilCancellation(CancellationToken) or TakeUntil(CancellationToken)	Onity.Reactive	The first is synchronous; the second is the async form.
FirstAsync(ct)	FirstAsync(CancellationToken = default) -> Task<T>	Onity.Reactive	The first value, or OperationCanceledException. FirstOnityTask(ct) -> OnityTask<T> is the OnityTask form (Onity.Unity.Async).
R3 LastAsync(ct)	ToOnityTask(ct) -> OnityTask<T>	Onity.Unity.Async	The last value when the source completes; a hot source needs the token. ToTask(ct) (Onity.Reactive) awaits one Unit.

Fragment, in a MonoBehaviour that holds ReactiveProperty<int> m_health:

m_health.Where(value => value <= 0)
        .Select(_ => "dead")
        .Subscribe(message => Debug.Log(message))
        .AddTo(this);

Time and async operators

Each time operator takes an optional OnityTimeProvider; without one it waits on wall-clock Task.Delay. In gameplay pass an OnityTimeProviders member so the wait follows a Unity loop and time mode.

R3 / UniRx	Onity	Namespace	Notes
R3 Debounce(dueTime), UniRx Throttle(dueTime)	Debounce(TimeSpan dueTime, OnityTimeProvider = null)	Onity.Reactive	The last value after a quiet window.
R3 ThrottleLast(interval), UniRx Sample(interval)	ThrottleLast(TimeSpan interval, OnityTimeProvider = null)	Onity.Reactive	The latest value once per interval.
ThrottleFirst(interval)	Throttle(TimeSpan interval, OnityTimeProvider = null)	Onity.Reactive	The first value, then a cool-down.
R3 Chunk(timeSpan), UniRx Buffer(timeSpan)	Buffer(TimeSpan timeSpan, OnityTimeProvider = null) -> IOnityObservable<IReadOnlyList<T>>	Onity.Reactive	One list per window; empty windows emit nothing.
TakeUntil(otherObservable)	TakeUntil(CancellationToken) / TakeUntil(Task)	Onity.Reactive	The signal is a token or a task, not another observable.
SelectAwait(async selector)	SelectAwait(Func<T, CancellationToken, ValueTask<TResult>>)	Onity.Reactive	Sequential; resumes on a thread-pool thread, so follow it with ObserveOnMainThread() before any Unity call.
WhereAwait(async predicate)	WhereAwait(Func<T, CancellationToken, ValueTask<bool>>)	Onity.Reactive	Same rule.
ObserveOn(frameProvider)	ObserveOn(OnityFrameProvider)	Onity.Reactive	Queues values and delivers them on the provider’s next tick: OnityFrameProviders.Update / FixedUpdate / LateUpdate.
ObserveOn(ThreadPoolScheduler)	ObserveOnThreadPool()	Onity.Reactive	One worker, source order preserved.
CPU work on the thread pool	SelectOnThreadPool(selector, maxConcurrency)	Onity.Reactive	Results emit as workers finish when maxConcurrency is greater than one.
ObserveOnMainThread()	ObserveOnMainThread() / ObserveOnMainThread(OnityUnityFrameProvider)	Onity.Unity.Reactive	The hop back onto the Update loop (or a chosen loop).
Awaiting the next value of a property	property.WaitAsync(ct) -> OnityTask<T>	Onity.Unity.Async	The next accepted change of an IReadOnlyReactiveProperty<T>; pooled, no subscription to manage.
Awaiting a property condition	property.WaitUntilAsync(predicate, ct) -> OnityTask<T>	Onity.Unity.Async	Completes synchronously when the current value already matches.

Fragment, in a MonoBehaviour’s OnEnable:

OnityEvent.Observe<PlayerDied>(this)
          .SelectAwait((message, ct) => UploadRunAsync(ct))   // ValueTask<bool>, runs on the thread pool
          .ObserveOnMainThread()                              // back onto the Update loop
          .Subscribe(ShowUploadResult)
          .TakeUntilDisable(this);

Unity bridges

R3.Unity / UniRx	Onity	Namespace	Notes
Observable.EveryUpdate()	OnityUnityObservable.EveryUpdate() -> IOnityObservable<Unit>	Onity.Unity.Reactive	One shared stream, pumped by a hidden component.
EveryFixedUpdate() / EveryLateUpdate()	OnityUnityObservable.EveryFixedUpdate() / EveryLateUpdate()	Onity.Unity.Reactive	Also with a CancellationToken, and with an OnityUnityThreadMode job boundary.
Observable.Timer(t)	OnityUnityObservable.Timer(float seconds, bool useUnscaledTime = false)	Onity.Unity.Reactive	One Unit after the delay.
Observable.Interval(t)	OnityUnityObservable.Interval(float seconds, bool useUnscaledTime = false) -> IOnityObservable<int>	Onity.Unity.Reactive	The cumulative tick count, starting at 1.
.Delay(t)	.Delay(float seconds, bool useUnscaledTime = false)	Onity.Unity.Reactive	Per value, on a Unity countdown timer.
.AddTo(this)	subscription.AddTo(this)	Onity.Unity.Reactive	this is a Component; disposes on destroy (the same as TakeUntilDestroy).
.AddTo(gameObject)	no equivalent		Pass the component.
.TakeUntilDestroy(this)	subscription.TakeUntilDestroy(this)	Onity.Unity.Reactive	Component.
.TakeUntilDisable(this)	subscription.TakeUntilDisable(this)	Onity.Unity.Reactive	Behaviour; also disposes on destroy.
.AddTo(compositeDisposable)	subscription.AddTo(compositeDisposable)	Onity.Reactive	Plain-C# owners.
(none)	subscription.AddTo(scope)	Onity.DI	Disposes when the injected IOnityScopeLifetime ends.
ObserveEveryValueChanged(x => x.Value)	OnityAsyncEnumerable.EveryValueChanged(target, x => x.Value, timing)	Onity.Unity.Async	A pull-based IOnityAsyncEnumerable<TProperty> polled at a PlayerLoop timing, not an IOnityObservable<T>; hold the value in a ReactiveProperty<T> when a push stream is needed.
Input: OnPointerClickAsObservable() and friends	action.PerformedAsObservable(), StartedAsObservable(), CanceledAsObservable()	Onity.Unity.Input	Input System only; there are no uGUI trigger observables, OnityTask covers UI events.

The lifetime helpers extend IDisposable, so they go after Subscribe: stream.Subscribe(...).AddTo(this);.

Time providers

Fragment, in a MonoBehaviour’s OnEnable:

OnityEvent.Observe<PlayerDamaged>(this)
          .Debounce(TimeSpan.FromSeconds(2), OnityTimeProviders.UpdateScaled)   // two quiet seconds after the last hit
          .Subscribe(_ => HideHitMarkers())
          .TakeUntilDisable(this);

OnityTimeProviders offers UpdateScaled, UpdateUnscaled, UpdateRealtime and the Fixed* and Late* variants. In EditMode tests, subclass the abstract OnityTimeProvider and override DelayAsync to drive time by hand; no manual or fake provider ships.

Events as a stream

R3 and UniRx need an adapter to turn a message bus into an observable; Onity’s messaging and reactive pillars share IOnityObservable<T>:

using Onity.Reactive;           // Where, Select, Subscribe
using Onity.Unity;              // OnityEvent
using Onity.Unity.Reactive;     // TakeUntilDisable
using UnityEngine;

OnityEvent.Observe<PlayerDamaged>(this)
          .Where(message => message.Amount >= 10)
          .Select(message => message.Amount)
          .Subscribe(amount => Debug.Log($"Took {amount}"))
          .TakeUntilDisable(this);

OnityEvent.Observe<T>(), OnityEventHub.Observe<T>(), broker.Observe<T>() and subscriber.Observe<T>() all return IOnityObservable<T>; the hub caches one stream per message type. UniRx’s MessageBroker.Default maps to OnityEvent on the active context; see Events and Messaging.

Errors

Onity has no error channel that flows through the operator chain, so R3’s OnErrorResume and UniRx’s OnError habits do not transfer:

Situation	Onity behavior	Do this
A subscriber throws (source is a Subject<T> or ReactiveProperty<T>)	Caught; routed to OnityObservableExceptionHandler.Handler; the other subscribers still run. The default handler is a no-op.	Assign a handler at startup: OnityObservableExceptionHandler.Handler = Debug.LogException;
A subscriber throws (source is a message stream from Observe<T>())	Propagates to the Publish call; later handlers for that message are skipped.	Catch inside the handler.
A SelectAwait / WhereAwait delegate throws	The subscription ends silently; onError is not called.	Catch inside the delegate and return a fallback.
A thread-pool selector throws	Routed to OnityObservableExceptionHandler; the stream continues.	Same handler.
Subscribe(onNext, onError, onCompleted)	onCompleted runs when the subscription is disposed. onError is raised only by sources that report faults (OnityTask.ToObservable(), IOnityAsyncEnumerable<T>.AsObservable()).	Do not rely on onError for core operators.
ObjectDisposedException	OnNext, Subscribe or SetValue after Dispose() (also on MessageChannel<T> and MessageBroker).	Tie subscriptions to a lifetime with AddTo.
ArgumentNullException, ArgumentOutOfRangeException	A null source or delegate; a negative count or a non-positive interval.	Validate inputs.
OperationCanceledException	FirstAsync, ToTask, the OnityTask bridges and the property waits were canceled. Normal cancellation.	Catch it where the flow starts.

OnityReactiveException exists for reactive-core failures; the shipped operators throw the standard exceptions above.

Not supported, and what to do instead

R3 / UniRx feature	Onity
Publish / Share / RefCount / Multicast	A Subject<T> is multicast; bind one instance (BindSubject<T>() in Onity.Composition, or BindInstance) and subscribe it directly.
Create / Defer / Never factories	new OnityObservable<T>(subscribe) for a custom source; Return, Empty, FromEvent for the rest.
ObserveEveryValueChanged as a push stream	OnityAsyncEnumerable.EveryValueChanged (async stream), or hold the value in a ReactiveProperty<T>, or publish a message and Observe<T>() it.
Window, Zip, Switch, Concat	Compose with Buffer, Scan, Pairwise, Merge, CombineLatest and Sample, or move to a pull-based stream: observable.AsOnityAsyncEnumerable(capacity) has Zip, Concat, Publish and Queue (Stream operators).
Catch / Retry / Timeout	Guard inside the delegate; for a timeout, cancel a token with CancelAfterSlim or OnityTimeoutController (Onity.Unity.Async) and pass it to TakeUntil or FirstAsync.
IObservable<T> adapters (System.Reactive)	Not shipped; stay on IOnityObservable<T>.
Parallel managed observers on Jobs or Burst	OnityUnityThreadMode adds a job boundary to a frame stream only. Use SelectOnThreadPool for managed CPU work, ObserveOnThreadPool for an ordered hop, and ObserveOnMainThread before Unity API access.
AddTo(gameObject)	AddTo(this) / TakeUntilDestroy(this) with the component, TakeUntilDisable(this), AddTo(compositeDisposable) or AddTo(scope).
UniTask-style awaits on a property	property.WaitAsync(ct), WaitUntilAsync(predicate, ct), AsLatestAsyncEnumerable(); see Reactive.