Instruction file imported from reactivemarbles/DynamicData (.github/instructions/rx.instructions.md). Copyright stays with the author.

Reactive Extensions (Rx) — Comprehensive Guide

Reference: ReactiveX Observable Contract | Rx.NET GitHub | IntroToRx.com

Core Concepts

Observables are Composable

Rx's power comes from composition. Every operator returns a new IObservable<T>, enabling fluent chaining:

source
    .Where(x => x.IsValid)           // filter
    .Select(x => x.Transform())      // project
    .DistinctUntilChanged()           // deduplicate
    .ObserveOn(RxApp.MainThreadScheduler) // marshal to UI thread
    .Subscribe(x => UpdateUI(x));     // consume

Each operator in the chain is a separate subscription. Disposing the final subscription cascades disposal upstream through the entire chain. This composability is what makes Rx powerful — and what makes contract violations devastating, since a bug in any operator corrupts the entire downstream chain.

Hot vs Cold Observables

Cold: starts producing items when subscribed to. Each subscriber gets its own sequence. Created with Observable.Create, Observable.Defer, Observable.Return, etc.

// Cold: each subscriber triggers a new HTTP call
var cold = Observable.FromAsync(() => httpClient.GetAsync(url));

Hot: produces items regardless of subscribers. All subscribers share the same sequence. Examples: Subject<T>, Observable.FromEventPattern, UI events.

// Hot: events fire whether or not anyone is listening
var hot = Observable.FromEventPattern<EventArgs>(button, nameof(button.Click));

Converting: Publish() + Connect() or Publish().RefCount() converts cold to hot (shared).

var shared = coldSource.Publish().RefCount(); // auto-connect on first sub, auto-disconnect on last unsub

The Observable Contract

1. Serialized Notifications (THE critical rule)

OnNext, OnError, and OnCompleted calls MUST be serialized — they must never execute concurrently. This is the most commonly violated rule and causes the most insidious bugs.

// WRONG: two sources can call OnNext concurrently
source1.Subscribe(x => observer.OnNext(Process(x)));  // thread A
source2.Subscribe(x => observer.OnNext(Process(x)));  // thread B — RACE!

// RIGHT: use Synchronize to serialize
source1.Synchronize(gate).Subscribe(observer);
source2.Synchronize(gate).Subscribe(observer);

// RIGHT: use Merge (serializes internally)
source1.Merge(source2).Subscribe(observer);

// RIGHT: use Subject (serializes OnNext calls via Synchronize)
var subject = new Subject<T>();
source1.Subscribe(subject);  // Subject.OnNext is NOT thread-safe by default!
// Use Subject with Synchronize if multiple threads call OnNext

Why it matters: operators maintain mutable internal state (caches, dictionaries, counters). Concurrent OnNext calls corrupt this state silently — no exception, just wrong data.

2. Terminal Notifications

Grammar: OnNext* (OnError | OnCompleted)?

• Zero or more OnNext, followed by at most one terminal notification
• OnError and OnCompleted are mutually exclusive — emit one or neither, never both
• After a terminal notification, no further notifications of any kind
• Operators receiving a terminal notification should release resources

3. Subscription Lifecycle

• Subscribe returns IDisposable — disposing it unsubscribes
• After disposal, no further notifications should be delivered
• Disposal must be idempotent (safe to call multiple times) and thread-safe
• Operators should stop producing when their subscription is disposed

4. Error Handling

• Exceptions thrown inside OnNext handlers propagate synchronously to the producing operator
• Use SubscribeSafe instead of Subscribe to route subscriber exceptions to OnError:

// Subscribe: exception in handler crashes the source
source.Subscribe(x => MayThrow(x));  // if MayThrow throws, exception propagates up

// SubscribeSafe: exception in handler routes to OnError
source.SubscribeSafe(Observer.Create<T>(
    onNext: x => MayThrow(x),
    onError: ex => HandleError(ex)));  // MayThrow exception goes here

Schedulers

Schedulers control when and where work executes. They are Rx's abstraction over threading.

Common Schedulers

Using Schedulers

// Time-based operators accept an optional scheduler
Observable.Timer(TimeSpan.FromSeconds(1), scheduler)
Observable.Interval(TimeSpan.FromMilliseconds(100), scheduler)
source.Delay(TimeSpan.FromMilliseconds(500), scheduler)
source.Throttle(TimeSpan.FromMilliseconds(300), scheduler)
source.Buffer(TimeSpan.FromSeconds(1), scheduler)
source.Timeout(TimeSpan.FromSeconds(5), scheduler)
source.Sample(TimeSpan.FromMilliseconds(100), scheduler)

// ObserveOn: deliver notifications on a specific scheduler
source.ObserveOn(RxApp.MainThreadScheduler)  // marshal to UI thread

// SubscribeOn: subscribe (and produce) on a specific scheduler
source.SubscribeOn(TaskPoolScheduler.Default)  // subscribe on background thread

Scheduler Injection for Testability

Always inject schedulers instead of using defaults. This enables deterministic testing:

// WRONG: hardcoded scheduler — untestable time-dependent behavior
public IObservable<T> GetData() =>
    _source.Throttle(TimeSpan.FromMilliseconds(300));

// RIGHT: injectable scheduler — testable
public IObservable<T> GetData(IScheduler? scheduler = null) =>
    _source.Throttle(TimeSpan.FromMilliseconds(300), scheduler ?? Scheduler.Default);

// TEST: use TestScheduler for deterministic time control
var testScheduler = new TestScheduler();
var results = new List<T>();
GetData(testScheduler).Subscribe(results.Add);
testScheduler.AdvanceBy(TimeSpan.FromMilliseconds(300).Ticks);
results.Should().HaveCount(1);

Disposable Helpers

Rx provides several IDisposable implementations for managing subscription lifecycles:

Disposable.Create

Creates a disposable from an action. The action runs exactly once on first disposal.

var cleanup = Disposable.Create(() =>
{
    connection.Close();
    Log("Cleaned up");
});
// Later: cleanup.Dispose() runs the action once

Disposable.Empty

A no-op disposable. Useful as a default or placeholder.

public IDisposable Subscribe(IObservable<T> source) =>
    isEnabled ? source.Subscribe(handler) : Disposable.Empty;

CompositeDisposable

Collects multiple disposables and disposes them all at once. The workhorse of Rx resource management.

var cleanup = new CompositeDisposable();

cleanup.Add(source1.Subscribe(handler1));
cleanup.Add(source2.Subscribe(handler2));
cleanup.Add(Disposable.Create(() => Log("All done")));

// Later: disposes ALL contained disposables
cleanup.Dispose();

Use in Observable.Create to manage multiple subscriptions:

Observable.Create<T>(observer =>
{
    var cleanup = new CompositeDisposable();
    cleanup.Add(source1.Subscribe(observer));
    cleanup.Add(source2.Subscribe(x => observer.OnNext(Transform(x))));
    cleanup.Add(Disposable.Create(() => cache.Clear()));
    return cleanup;
});

SerialDisposable

Holds a single disposable that can be replaced. Disposing the previous value when a new one is set. Useful for "switch" patterns.

var serial = new SerialDisposable();

// Each assignment disposes the previous
serial.Disposable = source1.Subscribe(handler);  // subscribes to source1
serial.Disposable = source2.Subscribe(handler);  // disposes source1 sub, subscribes to source2
serial.Disposable = Disposable.Empty;             // disposes source2 sub

// Disposing the SerialDisposable disposes the current inner
serial.Dispose();

SingleAssignmentDisposable

Like SerialDisposable but can only be assigned once. Throws on second assignment. Useful when a subscription is created asynchronously but disposal might happen before it's ready.

var holder = new SingleAssignmentDisposable();

// Start async subscription
Task.Run(() =>
{
    var sub = source.Subscribe(handler);
    holder.Disposable = sub;  // safe even if Dispose was already called
});

// Can dispose before assignment — the subscription will be disposed when assigned
holder.Dispose();

RefCountDisposable

Tracks multiple "dependent" disposables. The underlying resource is only disposed when all dependents (plus the primary) are disposed.

var primary = new RefCountDisposable(expensiveResource);

var dep1 = primary.GetDisposable();  // increment ref count
var dep2 = primary.GetDisposable();  // increment ref count

dep1.Dispose();    // decrement — resource still alive
primary.Dispose(); // decrement — resource still alive (dep2 still holds)
dep2.Dispose();    // decrement to 0 — resource disposed!

BooleanDisposable / CancellationDisposable

// BooleanDisposable: check if disposed
var bd = new BooleanDisposable();
bd.IsDisposed; // false
bd.Dispose();
bd.IsDisposed; // true — useful for cancellation checks

// CancellationDisposable: bridges IDisposable and CancellationToken
var cd = new CancellationDisposable();
cd.Token; // CancellationToken that cancels on Dispose
cd.Dispose(); // triggers cancellation

Standard Rx Operators Reference

Creation

Transformation

Filtering

Combining

Aggregation

Error Handling

Scheduling & Threading

Utility

Writing Custom Operators

Composition First — Observable.Create is a Last Resort

The Rx contracts are axioms, not guidelines. Merge subscribes sequentially. Defer evaluates at subscription time. Do fires synchronously during delivery. Concat subscribes to the second source only after the first completes. These guarantees are unconditional — they hold in every case, on every scheduler, under every threading model. If they didn't, nothing in Rx would work.

Trust the contracts completely. When you compose operators, you can reason about ordering, state, and lifecycle because the contracts are absolute. The moment you doubt them and add "safety" wrappers, you've abandoned the very thing that makes Rx code correct by construction.

Before reaching for Observable.Create, ask: can this be expressed as a composition of existing operators? Rx operators already handle subscription lifecycle, error propagation, disposal, and serialization. Manual observer forwarding inside Observable.Create reimplements all of that — and introduces bugs that the operators would have prevented.

The smell: If you're writing observer.OnNext(x) / observer.OnError(ex) / observer.OnCompleted() inside Observable.Create, you're manually reimplementing what Subscribe already does. Stop and look for the composition.

// ❌ WRONG: imperative Observable.Create with manual forwarding
// This reimplements Subscribe, adds boilerplate, and is harder to reason about.
return Observable.Create<Optional<TObject>>(observer =>
{
    var seenValue = false;
    var sub = source.ToObservableOptional(key)
        .Subscribe(
            value =>
            {
                seenValue = true;
                observer.OnNext(value);
            },
            observer.OnError,
            observer.OnCompleted);

    if (!seenValue)
        observer.OnNext(Optional.None<TObject>());

    return sub;
});

// ✅ RIGHT: declarative composition using existing operators
// Each operator does one thing. The intent is immediately clear.
return Observable.Defer(() =>
{
    var seenValue = false;
    return source.ToObservableOptional(key)
        .Do(_ => seenValue = true)
        .Merge(Observable.Defer(() => seenValue
            ? Observable.Empty<Optional<TObject>>()
            : Observable.Return(Optional.None<TObject>())));
});

Why the composition wins:

• Defer creates per-subscription state (the seenValue bool) — no shared mutable state
• Do captures a side effect without altering the stream — no manual forwarding
• Merge with inner Defer evaluates the condition after the synchronous subscription phase — the Defer factory runs when Merge subscribes to its second source, which happens after the first source's synchronous emissions
• Error propagation, completion, and disposal are all handled by the operators — zero manual wiring

When Observable.Create IS appropriate:

• You need to manage non-Rx resources (event handlers, timers, native resources) tied to subscription lifetime
• You're building a genuinely novel source (not transforming existing observables)
• You need fine-grained control over when/how disposal cascades
• The operator maintains complex mutable state that doesn't map to any existing operator's semantics

Even then, prefer Observable.Using for resource lifecycle and Observable.Defer for per-subscription state before reaching for Observable.Create.

The Defer Pattern — Per-Subscription State Without Observable.Create

Observable.Defer is the key to per-subscription mutable state in a purely compositional style:

// Per-subscription counter without Observable.Create
public static IObservable<(T Item, int Index)> WithIndex<T>(this IObservable<T> source) =>
    Observable.Defer(() =>
    {
        var index = 0;
        return source.Select(item => (item, index++));
    });

// Per-subscription flag for conditional behavior
public static IObservable<T> EmitDefaultIfEmpty<T>(this IObservable<T> source, T defaultValue) =>
    Observable.Defer(() =>
    {
        var hasEmitted = false;
        return source
            .Do(_ => hasEmitted = true)
            .Concat(Observable.Defer(() => hasEmitted
                ? Observable.Empty<T>()
                : Observable.Return(defaultValue)));
    });

Each subscription gets its own index / hasEmitted — cold observable semantics, no shared state, no locking.

The Observable.Create Pattern

When you genuinely need Observable.Create, follow this structure:

public static IObservable<TResult> MyOperator<TSource, TResult>(
    this IObservable<TSource> source,
    Func<TSource, TResult> selector)
{
    return Observable.Create<TResult>(observer =>
    {
        return source.SubscribeSafe(Observer.Create<TSource>(
            onNext: item =>
            {
                try
                {
                    var result = selector(item);
                    observer.OnNext(result);
                }
                catch (Exception ex)
                {
                    observer.OnError(ex);
                }
            },
            onError: observer.OnError,
            onCompleted: observer.OnCompleted));
    });
}

Multi-Source Operator Pattern

When combining multiple sources, serialize their notifications:

public static IObservable<T> MyMerge<T>(
    this IObservable<T> source1,
    IObservable<T> source2)
{
    return Observable.Create<T>(observer =>
    {
        var gate = new object();

        var sub1 = source1.Synchronize(gate).SubscribeSafe(observer);
        var sub2 = source2.Synchronize(gate).SubscribeSafe(observer);

        return new CompositeDisposable(sub1, sub2);
    });
}

Note: Synchronize(gate) holds the lock during downstream OnNext delivery. This ensures serialization but means the lock is held for the duration of all downstream processing. Keep downstream chains lightweight when using shared gates.

Operator Checklist

When writing or reviewing an Rx operator:

• Serialized delivery: can OnNext be called concurrently? If multiple sources, are they serialized?
• Terminal semantics: does OnError/OnCompleted propagate correctly? No notifications after terminal?
• Disposal: does disposing the subscription clean up all resources? Is it idempotent?
• Error handling: does SubscribeSafe catch subscriber exceptions? Are errors propagated, not swallowed?
• Back-pressure: does the operator buffer unboundedly? Could it cause memory issues?
• Scheduler: are time-dependent operations using an injectable scheduler?
• Cold/Hot: is the observable cold (deferred via Observable.Create)? If hot, is sharing handled correctly?
• Thread safety: is mutable state protected? Are there race conditions between subscribe/dispose/OnNext?

Common Pitfalls

1. Subscribing Multiple Times to a Cold Observable

// WRONG: two HTTP calls!
var data = Observable.FromAsync(() => httpClient.GetAsync(url));
data.Subscribe(handler1);  // call 1
data.Subscribe(handler2);  // call 2 — probably not intended

// RIGHT: share the result
var shared = data.Publish().RefCount();
shared.Subscribe(handler1);  // shares
shared.Subscribe(handler2);  // same result

2. Forgetting to Dispose Subscriptions

// WRONG: subscription leaks — handler keeps running forever
source.Subscribe(x => UpdateUI(x));

// RIGHT: track and dispose
_cleanup.Add(source.Subscribe(x => UpdateUI(x)));
// In Dispose: _cleanup.Dispose();

3. Blocking on Rx (sync-over-async)

// WRONG: blocks the thread, can hang on UI thread
var result = source.FirstAsync().Wait();

// RIGHT: use async/await
var result = await source.FirstAsync();

4. Using Subject as a Public API

// WRONG: exposes mutation to consumers
public Subject<int> Values { get; } = new();

// RIGHT: expose as IObservable, keep Subject private
private readonly Subject<int> _values = new();
public IObservable<int> Values => _values.AsObservable();

5. Not Handling OnError

// WRONG: unhandled OnError crashes the app (routes to DefaultExceptionHandler)
source.Subscribe(x => Process(x));

// RIGHT: always handle errors
source.Subscribe(
    x => Process(x),
    ex => LogError(ex),
    () => LogComplete());