R3 Reactive Extensions

R3: Modern Reactive Extensions for .NET

R3 is Cysharp's ground-up reimplementation of Reactive Extensions — "the new future of dotnet/reactive and UniRx." It keeps the LINQ-over-events programming model but rebuilds the core types, error contract, and scheduler to fix long-standing problems in System.Reactive (Rx.NET). Use this skill when composing event streams, UI input, timers, or push-based pipelines in C#.

Canonical sources (link to these from code and docs):

• Repository: https://github.com/Cysharp/R3
• README (full operator reference): https://github.com/Cysharp/R3/blob/main/README.md
• Author's design rationale: https://neuecc.medium.com/r3-a-new-modern-reimplementation-of-reactive-extensions-for-c-cf29abcc5826

When to Use This Skill

Use this skill when:

• Composing events over time — UI input, sensor/feed updates, websocket messages, domain events
• You need operators like debounce, throttle, merge, combine-latest, distinct-until-changed
• Building MVVM state with ReactiveProperty / BindableReactiveProperty
• Bridging push-based streams with Task / async and IAsyncEnumerable
• Migrating from System.Reactive, UniRx, or IObservable<T> code
• You hit Rx pain points: subscriptions dying on exceptions, scheduler overhead, or leak hunting

Not the right tool for: request/response I/O (use async/await), bounded producer/consumer with backpressure (use System.Threading.Channels), or server-side stream processing with batching/backpressure (use Akka.NET Streams). R3, like all Rx, is push-based with no backpressure. See the csharp-concurrency-patterns skill for choosing between these.

Reference Files

• rx-net-differences.md: Every meaningful difference vs System.Reactive (Rx.NET) — the new core types, the error model, operator renames, dropped APIs, the scheduler swap, and a migration checklist.
• async-and-integration-patterns.md: Common patterns — async dispatch with AwaitOperation, Task integration, IAsyncEnumerable round-tripping, ReactiveProperty/MVVM, subjects, and subscription lifecycle.
• scheduling-and-concurrency.md: How R3 handles concurrent updates (the threading contract, Synchronize, ObserveOn), TimeProvider vs FrameProvider, when each is necessary, and deterministic testing with fake providers.

Everything in this skill was validated empirically against R3 1.3.1. Captured output appears in the reference files as evidence.

Why R3 Exists (the "why use it")

The author (neuecc) built R3 to fix concrete defects in System.Reactive:

1. Exceptions silently kill subscriptions. In Rx, one exception in the pipeline calls OnError and unsubscribes forever — "a billion-dollar mistake" for long-lived event streams (a single bad UI event tears down the whole subscription). R3 routes errors to OnErrorResume and keeps the subscription alive by default.
2. IScheduler is heavy and confusing. ImmediateScheduler/Merge were measured causing real server memory/CPU bloat. R3 deletes IScheduler and uses .NET 8's TimeProvider (wall-clock) plus a new FrameProvider (frame-clock).
3. Subscription leaks are hard to find. R3 makes every Observable<T> an abstract class so all subscriptions funnel through one place, enabling ObservableTracker to list every live subscription with stack traces.
4. Rx and async were awkwardly fused. R3 treats Rx as event-first and adds explicit bridges (AwaitOperation, FromAsync, ToAsyncEnumerable) instead of pretending events are pull-based sequences.
5. One library, every UI. A platform-neutral core plus thin provider packages for Unity, Godot, WPF, WinForms, Avalonia, WinUI3, MAUI, Stride, MonoGame, and Blazor.

Install

dotnet add package R3
# Platform glue (pick what applies): R3.WPF, R3.Avalonia, R3.WinForms, R3.Unity (UPM),
# R3.Godot, ObservableCollections.R3, etc. See the repo README for the full list.

using R3;

The Mental Model

R3 replaces Rx's interfaces with abstract classes, and replaces Rx's two-method error contract with a single completion that carries a result.

public abstract class Observable<T>
{
    public IDisposable Subscribe(Observer<T> observer);     // tracked centrally
    protected abstract IDisposable SubscribeCore(Observer<T> observer);
}

public abstract class Observer<T> : IDisposable               // the observer IS the subscription
{
    public void OnNext(T value);
    public void OnErrorResume(Exception error);               // error WITHOUT unsubscribing
    public void OnCompleted(Result result);                   // success OR failure terminates
}

The grammar is (OnNext | OnErrorResume)* OnCompleted(Result)?. Note the difference from Rx's OnNext* (OnError | OnCompleted)?: errors and termination are decoupled. An error is just a notification; only OnCompleted ends the stream, and it carries a Result that is either Result.Success or Result.Failure(exception).

Quick start

using R3;

var subscription = Observable
    .EveryValueChanged(model, m => m.SearchText)   // emits when the property changes
    .Debounce(TimeSpan.FromMilliseconds(300))      // Rx called this "Throttle" (see differences)
    .DistinctUntilChanged()
    .SubscribeAwait(async (text, ct) =>
    {
        var results = await _api.SearchAsync(text, ct);
        Render(results);
    }, AwaitOperation.Switch);                      // cancel the in-flight search on a new keystroke

// Dispose to unsubscribe; or route into a DisposableBag / AddTo(token).
subscription.Dispose();

Core Behavior, Verified

Errors do not terminate by default

var subject = new Subject<int>();
subject.Select(x => 100 / x).Subscribe(
    onNext:        x => Console.WriteLine($"next {x}"),
    onErrorResume: e => Console.WriteLine($"errorResume {e.GetType().Name}"),
    onCompleted:   (Result r) => Console.WriteLine($"completed IsSuccess={r.IsSuccess}"));

subject.OnNext(2);   // next 50
subject.OnNext(0);   // errorResume DivideByZeroException   <-- NOT terminated
subject.OnNext(5);   // next 20                             <-- subscription is still alive!
subject.OnCompleted(); // completed IsSuccess=True

This is the single biggest behavioral change from Rx. To opt back into classic "an error terminates the sequence" behavior, insert .OnErrorResumeAsFailure() — the error then flows to OnCompleted(Result.Failure(e)) and downstream OnNexts stop. Recover with Catch. Full captured runs and the (deliberately absent) Retry story are in rx-net-differences.md.

Async dispatch is explicit

R3's async operators (SubscribeAwait, SelectAwait, WhereAwait, …) take an AwaitOperation that decides what happens when values arrive faster than the async work completes:

These were verified to behave exactly as described (including Switch cancelling the superseded operation's CancellationToken). See async-and-integration-patterns.md.

Task and IAsyncEnumerable bridges

// Task -> Observable
await Observable.FromAsync(async ct => await LoadAsync(ct)).FirstAsync();

// Observable -> Task (terminal operators return Task<T>)
List<int> all = await source.ToListAsync();
int last      = await source.LastAsync();

// IAsyncEnumerable -> Observable, and back
await asyncEnumerable.ToObservable().ForEachAsync(Handle);
await foreach (var x in source.ToAsyncEnumerable()) { /* ... */ }

All verified working. Details and the full terminal-operator list are in async-and-integration-patterns.md.