r3-reactive-extensions
GitHub用于在C#中构建响应式或事件驱动应用，基于Cysharp的R3库。涵盖Observable模型、错误处理、异步集成、调度及与System.Reactive的差异，适用于UI输入、数据流组合和MVVM状态管理。
Trigger Scenarios
Install
npx skills add NeverSight/learn-skills.dev --skill r3-reactive-extensions -g -y
SKILL.md
Frontmatter
{
"name": "r3-reactive-extensions",
"invocable": false,
"description": "Build reactive\/event-driven C# with R3 (Cysharp's modern reimplementation of Reactive Extensions). Covers the Observable<T>\/Observer<T> model, the OnErrorResume error contract, async dispatch with AwaitOperation, Task\/IAsyncEnumerable integration, TimeProvider\/FrameProvider scheduling, the concurrency contract, and how R3 differs from System.Reactive (Rx.NET)."
}
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/asyncandIAsyncEnumerable - Migrating from
System.Reactive, UniRx, orIObservable<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,Taskintegration,IAsyncEnumerableround-tripping,ReactiveProperty/MVVM, subjects, and subscription lifecycle. - scheduling-and-concurrency.md: How R3 handles concurrent updates (the threading contract,
Synchronize,ObserveOn),TimeProvidervsFrameProvider, 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:
- Exceptions silently kill subscriptions. In Rx, one exception in the pipeline calls
OnErrorand unsubscribes forever — "a billion-dollar mistake" for long-lived event streams (a single bad UI event tears down the whole subscription). R3 routes errors toOnErrorResumeand keeps the subscription alive by default. IScheduleris heavy and confusing.ImmediateScheduler/Mergewere measured causing real server memory/CPU bloat. R3 deletesISchedulerand uses .NET 8'sTimeProvider(wall-clock) plus a newFrameProvider(frame-clock).- Subscription leaks are hard to find. R3 makes every
Observable<T>an abstract class so all subscriptions funnel through one place, enablingObservableTrackerto list every live subscription with stack traces. - 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. - 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:
AwaitOperation |
Overlap behavior | Typical use |
|---|---|---|
Sequential (default) |
Queue values, run one at a time | Ordered processing |
Drop |
Ignore new values while one is running | Debounced submit / cooldown |
Switch |
Cancel the running one, start the new | Search-as-you-type, latest-wins |
Parallel |
Run all concurrently | Independent fan-out |
SequentialParallel |
Run concurrently, emit results in order | Parallel map, ordered output |
ThrottleFirstLast |
Run first + last of a burst | Leading/trailing sampling |
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.
How R3 Handles Concurrent Updates
R3 does not serialize concurrent producers. Like Rx, it assumes the Rx grammar: OnNext
must not be called concurrently or re-entrantly from multiple threads. Operators (Where,
Select, Subject, …) are not internally locked. Pushing OnNext from many threads at once
into a stateful downstream corrupts state — in testing, 20,000 concurrent OnNext calls into
a List<T> subscriber lost ~half the items and threw inside the operator chain.
The fix is to make the boundary explicit:
// Multiple producer threads -> one serialized consumer
subject.Synchronize() // lock-based gate; delivery becomes single-threaded
.Where(x => x.IsValid)
.Subscribe(Handle); // verified: 10000/10000 items, no corruption
// Or marshal onto a context/threadpool, which also serializes delivery:
source.ObserveOnThreadPool().Subscribe(Handle);
// For shared MVVM state written from many threads:
var counter = new SynchronizedReactiveProperty<int>(0); // thread-safe writes
Practical rule: if more than one thread can publish into a stream, put Synchronize() (or an
ObserveOn*) immediately after the source, or use SynchronizedReactiveProperty. Full race
reproductions and outputs are in scheduling-and-concurrency.md.
Time vs Frames: TimeProvider and FrameProvider
R3 has two notions of "when," and both are abstractions you can fake in tests:
TimeProvider(the .NET 8 BCL type) = wall-clock time. Used byDelay,Debounce,Interval,Timer,Timeout. This is what server/business code uses.FrameProvider(R3-specific) = a frame clock. Used byEveryUpdate,DelayFrame(n),IntervalFrame(n), etc.
When is a FrameProvider necessary? Whenever "progress" is measured in render/update ticks instead of elapsed time:
- Game engines (Unity, Godot, Stride, MonoGame) — logic ticks with the engine's update loop, so it respects pause and time-scale and stays in lockstep with rendering.
- UI render loops (WPF/Avalonia/WinUI composition frames) — react per frame.
- Deterministic tests —
FakeFrameProvider.Advance(n)drives frames with zero real time, exactly asFakeTimeProvider.Advance(timeSpan)drives the clock.
Plain server/business code virtually never needs FrameProvider — that's TimeProvider
territory. Both fakes make time-dependent pipelines fully deterministic; examples in
scheduling-and-concurrency.md.
Best Practices Summary
DO
- Treat
OnErrorResumeas the default: design streams that survive individual bad events. - Add
.OnErrorResumeAsFailure()when you genuinely want an error to terminate the stream. - Choose an
AwaitOperationdeliberately for every async operator (Switchfor latest-wins,Sequentialfor ordering,Dropfor cooldowns). - Put
Synchronize()/ObserveOn*after any source that multiple threads publish into. - Pass a
TimeProviderto time operators and aFrameProviderto frame operators so tests can useFakeTimeProvider/FakeFrameProvider. - Manage lifetime: route subscriptions into a
DisposableBag,CompositeDisposable, or.AddTo(cancellationToken); turn onObservableTrackerin dev to catch leaks. - Use
ReactivePropertyfor de-duplicated observable state;BindableReactivePropertyfor XAML-bound state.
DON'T
- Don't assume an exception ends the stream (that's Rx, not R3).
- Don't reach for Rx names that R3 renamed: it's
Debounce(notThrottle),ThrottleLast(notSample),Chunk(notBuffer).Retry,GroupBy,Finally, and plainBufferare absent in 1.3.1 — see the differences file for replacements. - Don't call
OnNextconcurrently/re-entrantly from multiple threads withoutSynchronize(). - Don't use R3 for backpressured throughput pipelines — use Channels or Akka.NET Streams.
- Don't block on terminal operators (
.Result/.Wait()); they returnTask<T>—awaitthem.
Additional Resources
- R3 repository: https://github.com/Cysharp/R3
- Full README / operator reference: https://github.com/Cysharp/R3/blob/main/README.md
- Design rationale (neuecc): https://neuecc.medium.com/r3-a-new-modern-reimplementation-of-reactive-extensions-for-c-cf29abcc5826
- NuGet: https://www.nuget.org/packages/R3
TimeProvider(BCL): https://learn.microsoft.com/en-us/dotnet/api/system.timeproviderFakeTimeProvider: https://www.nuget.org/packages/Microsoft.Extensions.TimeProvider.Testing- Related skill:
csharp-concurrency-patterns(choosing R3 vs async/await vs Channels vs Akka.NET)
Version History
- e0220ca Current 2026-07-05 22:12
