ReactiveUI.Primitives and System.Reactive¶

System.Reactive is the original Reactive Extensions library for .NET, and ReactiveUI supports it. This page sets out how ReactiveUI.Primitives differs from it, why, how to get the most from Primitives, and how to use the two together.

The short version¶

• Both are built on the same interfaces, IObservable<T> and IObserver<T>, which ship in .NET itself. A stream from one library works with operators from the other.
• Most operators answer to their System.Reactive names, such as Throttle and DistinctUntilChanged. Most also have a Primitives name, such as Calm and Unique. Both names build the same thing.
• A few types sit outside .NET, so Primitives has its own: RxVoid for Unit, ISequencer for IScheduler, and signals for subjects.
• Primitives builds each operator as its own purpose-built sink. That is a design choice: more types in the library, in exchange for speed and fewer allocations.
• Primitives has no reflection-based APIs, so every part of it works with trimming and Native AOT.
• The .Reactive packages compile the same Primitives source against System.Reactive's types, for apps and libraries that must keep them.

Words that differ¶

Type mapping¶

Signals and subjects¶

See signals you push values into.

Disposables¶

See disposables.

Schedulers and sequencers¶

Operator names¶

The operators Primitives shares with System.Reactive answer to the System.Reactive name, so most System.Reactive code reads and compiles the same. Most also have a Primitives name. Both names build the same thing, and you can use whichever suits you. The System.Reactive names also match other reactive libraries, such as RxJS.

The rest of this site uses the LINQ name when LINQ has the operator, and the Primitives name otherwise.

Where LINQ has no such operator, this site uses the Primitives name:

Some operators exist only in Primitives, such as Choose, SwitchMap, MapWith and BlendUnique. The operator pages cover each one.

Reattempt and Retry build the same operator, but count differently. Reattempt(n) allows n extra tries, while Retry(n) counts the first run too, so Retry(3) is Reattempt(2).

Behaviour that matches¶

These were checked side by side against System.Reactive 7.0.0, and behave the same:

• A subscriber that leaves out the error callback has the error thrown at the code that called OnError.
• A subscriber that throws from OnNext has the exception thrown at the code that called OnNext.
• OnNext on a disposed signal throws ObjectDisposedException.
• Values sent after OnCompleted are dropped, and a subscriber that arrives later gets the completion at once.
• BehaviorSignal<T>, ReplaySignal<T> and AsyncSignal<T> replay and await as their subject counterparts do.
• Calm and Probe send their pending value when the source completes, as Throttle and Sample do.
• Expire restarts its timer on each value, as Timeout(TimeSpan) does.
• Retry(count) and Repeat(count) run the source up to count times in total.
• Fold, Aggregate with a seed, Buffer(count, skip), Unique, SyncLatest, Blend(maxConcurrent), OnErrorResumeNext, and FirstAsync on an empty stream.

Behaviour that differs¶

System.Reactive's FirstAsync, FirstOrDefaultAsync, LastAsync and LastOrDefaultAsync return an IObservable<T> with one value. The Primitives methods of the same name return a Task<T>, so you await them directly:

int first = await signal.FirstAsync();

GetAwaiter and RunAsync return an IAwaitSignal<T> rather than System.Reactive's AsyncSubject<T>. You still await a stream the same way.

Signal.FromEventPattern(addHandler, removeHandler) for a plain EventHandler sends EventPattern<EventArgs>, where System.Reactive sends EventPattern<object>. EventPattern<TEventArgs> is Primitives' own type, in ReactiveUI.Primitives.Core.

ToEnumerable blocks the calling thread until the stream completes, then hands back every value. System.Reactive's ToEnumerable hands back each value as it arrives. Do not call it on a stream that never completes.

Operators that look alike but behave differently¶

Some Primitives operators and extension helpers look close to a System.Reactive operator, but behave differently. Each difference below was checked by running both libraries side by side. Check the row before you swap one for the other.

What Primitives does not have¶

Operators Primitives does not have yet¶

These System.Reactive operators have no Primitives version in this release:

The async library, ReactiveUI.Primitives.Async, does have GroupBy, ForEachAsync, SingleAsync and SingleOrDefaultAsync.

Some System.Reactive overloads also have no Primitives version. Among them are:

• Aggregate and Scan without a seed.
• Where, SelectMany, SkipWhile and TakeWhile with an index.
• Skip(TimeSpan).
• Buffer with a closing selector, a time shift, or a time and a count.
• Throttle, Delay and Timeout that take a stream per value.
• Zip over more than two streams.
• RefCount with a disconnect delay.
• Subscribe with a CancellationToken.

If you rely on one of these, keep that part of your code on System.Reactive. The two work together, as shown below.

Covered under another name¶

Some System.Reactive calls have no overload of the same name, but Primitives does the same job another way:

Left out on purpose: APIs built on reflection¶

Reflection is .NET code that looks up types, members and events by name while the app runs, instead of the compiler binding them at build time. A trimmer cannot see a member that is only named in a string, so it can remove it. The failure then happens on the user's device, not at build time. Supporting trimming and Native AOT everywhere was a large part of why Primitives exists, so it leaves out every System.Reactive API that relies on reflection.

Only a few parts of System.Reactive use reflection: 18 overloads of FromEvent and FromEventPattern, plus the IQbservable<T> query operators. Every other System.Reactive operator is compiled normally.

Primitives binds every event handler at compile time instead. Signal.FromEventPattern works with EventHandler<TEventArgs>, so it needs TEventArgs to derive from EventArgs. For an event whose argument is any other type, or whose delegate is not an EventHandler, use Signal.FromEvent with a conversion. It places no limit on the argument type:

IObservable<int> ticks = Signal.FromEvent<EventHandler<int>, int>(
    conversion: action => (_, value) => action(value),
    addHandler: handler => ticker.Ticked += handler,
    removeHandler: handler => ticker.Ticked -= handler);

Native AOT¶

Native AOT compiles an app to machine code ahead of time, with no JIT. It trims away code the app does not use, so anything the library finds by reflection can break. Every Primitives package is marked IsAotCompatible.

The same app was published with Native AOT against each library, on .NET 10, for linux-x64. The app uses a signal or subject, Select, Where, Scan, CombineLatest, Throttle, DistinctUntilChanged, SelectMany, Buffer, Replay, RefCount, FromEventPattern, Timeout, Catch, ObserveOn, Merge, Switch and Materialize. Both native binaries ran and printed the same output as under the JIT.

Keeping the whole library shows what a library author sees when every public API must be AOT-safe. System.Reactive's 26 warnings all come from its IQbservable query providers and join patterns, which build expression trees at run time.

Calling a reflection-based API from an app is where it matters most. The same Native AOT publish with Observable.FromEventPattern<TickArgs>(ticker, "Ticked") gives 2 trim warnings, and the app then fails when it runs:

Unhandled exception. System.InvalidOperationException: Could not find event 'Ticked' on object of type 'Ticker'.

Plain strongly typed System.Reactive code is AOT-clean. The difference is that nothing in Primitives can take you down the reflection path.

Performance¶

The Primitives repository benchmarks each operator against the closest System.Reactive version of the same work, timing each call and counting the bytes it allocates. In its published run, Primitives was faster than System.Reactive in 151 of 157 comparisons. A few rows from that run, as mean time and bytes allocated per call:

The last two rows are cases where System.Reactive is as fast or faster. The full table, the method, and how to run it yourself are in the Primitives benchmarks.

Why the operators are built differently¶

System.Reactive builds operators from a small set of general parts. Synchronize, for example, is one operator you add to any chain when you need values one at a time. That keeps the library small, and it reads well.

Primitives takes the other path. Each operator has its own sink, the object that receives values from the source and passes results on. A sink that needs to deliver one value at a time builds that in, with the threading that suits its job. The result is more types in the library, and no general layer between the sink and your code.

Neither path is wrong. ReactiveUI chose the dedicated path because streams sit under every ReactiveUI library, so their speed and allocations reach every app. The same choice shows in the public types:

• Signals for a job. SerializedSignal<T> orders values from many threads. StateSignal<T> holds a settable current value. CommandSignal<TResult> runs work with IsRunning and CanRun. See signals.
• Embedded state. DisposableSet and the delivery types are structs you hold as a field in your own type, with no extra object. See disposables.
• Public operator types. Each operator's sink is a public class, so you can build your own operators from the same parts. Every operator page lists them.

Getting the most from Primitives¶

• Send from many threads through SerializedSignal<T>, rather than a subject followed by Synchronize.
• Let subscribers set the pace with an async signal. Signal.Create<T>() in ReactiveUI.Primitives.Async.Signals builds a SerialSignalAsync<T>, which waits for each subscriber in turn before OnNextAsync returns. System.Reactive has nothing like it. See async signals.
• Hold changing state in StateSignal<T>. Set Value to send, and read Value at any time.
• Use the With operators with static lambdas on paths that run often. See best practices.
• Prefer Serialize to Synchronize. It holds no lock while your callback runs.
• Test with VirtualClock. The time operators take a sequencer.
• Use Signal.FromEvent with a conversion for events, so nothing depends on reflection.

Using both libraries together¶

Streams pass straight between them¶

A System.Reactive subject works with Primitives operators, and a Primitives signal works with System.Reactive operators. Both are IObservable<T>.

Importing both in one file¶

System.Reactive and Primitives both add extension methods named Subscribe, Select, Where and so on to IObservable<T>. When a project references both libraries and a file imports both namespaces, C# cannot tell which one you mean, and the call does not compile. Subscribe clashes even without the System.Reactive using, because System.Reactive declares it in the System namespace.

Two ways around it:

• Call SubscribePrimitives instead of Subscribe. It does the same job, with a name that cannot clash. See SubscribePrimitives.
• Call System.Reactive through an alias for the operators you need from it.

using ReactiveUI.Primitives;
using ReactiveUI.Primitives.Signals;
using Rx = System.Reactive.Linq.Observable;

var rxSubject = new System.Reactive.Subjects.Subject<int>();
var signal = new Signal<int>();

IObservable<int> fromRx = Rx.Where(rxSubject, x => x > 1);
IObservable<int> fromPrimitives = signal.Where(x => x > 1);

fromRx.SubscribePrimitives(x => Console.WriteLine($"rx {x}"));
fromPrimitives.SubscribePrimitives(x => Console.WriteLine($"primitives {x}"));

The .Reactive packages¶

Some code has to keep System.Reactive's types in its public API: a library other apps depend on, or a framework that takes an IScheduler. The .Reactive packages are for that. ReactiveUI.Primitives.Reactive is the same Primitives source, compiled so that ISequencer is System.Reactive's IScheduler and RxVoid is its Unit. Its namespaces start ReactiveUI.Primitives.Reactive.

using System.Reactive;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;
using System.Reactive.Subjects;
using ReactiveUI.Primitives.Reactive;
using ReactiveUI.Primitives.Reactive.Disposables;

var clicks = new Subject<Unit>();
var subscriptions = new ContainerDisposable();

clicks.Throttle(TimeSpan.FromMilliseconds(50), new EventLoopScheduler())
      .Select(_ => "clicked")
      .Subscribe(Console.WriteLine)
      .DisposeWith(subscriptions);

CompositeDisposable composite = subscriptions;

Here a System.Reactive subject, Unit and scheduler go straight into Primitives operators.

ContainerDisposable is a MultipleDisposable that converts to a CompositeDisposable, so you can hand it to code that expects one. Disposing the container also disposes anything added through the composite.

ReactiveUI.Primitives, ReactiveUI.Primitives.Async and each platform package have a .Reactive partner, such as ReactiveUI.Primitives.Wpf.Reactive.

ReactiveUI's packages¶

ReactiveUI follows the same pattern. Both sets run on Primitives. They differ only in the types their public API uses.

Choose the .Reactive packages when your app uses another library that needs System.Reactive's types, or to upgrade existing code without changing it.

Moving code across¶

The Primitives README has a step-by-step migration guide, for moving a project across in one go or for publishing a .Reactive package beside a lean one.