Creation factories¶

You have a value, a list, a task, an event or a clock. None of them is a stream. A creation factory turns one of them into a stream, so the operators can work on it.

A stream is a source that pushes values to you over time. Its .NET type is IObservable<T>. You call Subscribe on it to start listening. It then pushes you each value, and ends in one of two ways: it completes when it has nothing more to send, or it errors when something went wrong.

Every factory on this page is a static method on the Signal class.

using ReactiveUI.Primitives;
using ReactiveUI.Primitives.Signals;

Your first factory¶

Say you have an async method that fetches a price. You want it as a stream.

1. Pick the factory that matches what you have. You have a method that returns a task, so use FromAsync.

IObservable<decimal> prices = Signal.FromAsync(() => FetchPriceAsync());

2. Nothing has run yet. Building a stream does not start it. FetchPriceAsync has not been called.

3. Subscribe. This starts the work.

IDisposable subscription = prices.Subscribe(
    price => Console.WriteLine($"price: {price}"),
    error => Console.WriteLine($"failed: {error.Message}"),
    () => Console.WriteLine("done"));

Output:

price: 42.50
done

4. Dispose when you are finished. Disposing stops you listening. Here it also cancels the fetch.

subscription.Dispose();

That is the whole shape. The rest of this page is one section per factory.

Cold and hot¶

Nearly every factory here builds a cold stream. Cold means each subscriber gets its own private run. Two subscribers to the prices stream above call FetchPriceAsync twice.

A hot stream runs once and shares that one run with everyone listening. Signal.FromTask is the one factory on this page that is hot, because the task it wraps is already running.

This matters more often than it sounds, so each section below says which kind it builds.

Sequencers¶

Many factories take an ISequencer. A sequencer decides which thread runs your callback, and when.

using ReactiveUI.Primitives.Concurrency;

Leave the argument off and the factory picks a sensible default for what it does.

One value, then done¶

Emit¶

Emit sends one value and completes. Cold, so every subscriber gets the value.

IObservable<int> answer = Signal.Emit(42);

answer.Subscribe(x => Console.WriteLine(x), () => Console.WriteLine("done"));

Output:

42
done

RxVoid is a type with exactly one value. Use it when only the fact that something happened matters, and there is no value to carry.

IObservable<RxVoid> saved = Signal.EmitRxVoid();

saved.Subscribe(_ => Console.WriteLine("the save finished"));

Output:

the save finished

Empty¶

Empty completes straight away and sends no values at all.

IObservable<int> nothing = Signal.Empty<int>();

nothing.Subscribe(
    x => Console.WriteLine($"value: {x}"),
    () => Console.WriteLine("done"));

Output:

done

Two overloads take a value they never send. They read the type off that value, so you do not have to write the type parameter yourself:

int example = 0;

IObservable<int> a = Signal.Empty<int>();      // you name the type
IObservable<int> b = Signal.None(example);     // the type is read from example

Silent¶

Silent sends nothing and never ends. Use it as a stream that should never fire, such as a "no timeout" placeholder.

IObservable<int> quiet = Signal.Silent<int>();

quiet.Subscribe(
    x => Console.WriteLine($"value: {x}"),
    () => Console.WriteLine("done"));

Output: nothing at all. Neither callback ever runs.

Signal.Silent(example) does the same, and reads the type from a value it never sends.

Empty against Silent¶

They look similar and behave very differently.

An operator waiting for a stream to finish, such as Concat, moves on immediately after Empty and waits forever after Silent.

Fail¶

Fail sends no values and errors as soon as you subscribe.

IObservable<int> broken = Signal.Fail<int>(new InvalidOperationException("no connection"));

broken.Subscribe(
    x => Console.WriteLine($"value: {x}"),
    error => Console.WriteLine($"error: {error.Message}"));

Output:

error: no connection

Repeating, counting and generating¶

Repeat¶

Repeat sends the same value again and again.

IObservable<string> beats = Signal.Repeat("tick", 3);

beats.Subscribe(x => Console.WriteLine(x), () => Console.WriteLine("done"));

Output:

tick
tick
tick
done

Leave the count off and it repeats forever, so dispose the subscription to stop it.

IObservable<string> forever = Signal.Repeat("tick");

Range¶

Range sends a run of consecutive integers, then completes.

IObservable<int> numbers = Signal.Range(10, 4);

numbers.Subscribe(x => Console.WriteLine(x), () => Console.WriteLine("done"));

Output:

10
11
12
13
done

The first argument is where to start. The second is how many to send, not where to stop. It also takes an optional sequencer.

Unfold¶

Unfold walks a value forward step by step. You give it four things: the value to start at, a test that says whether to keep going, a step that produces the next value, and a projection that turns each value into what you send.

IObservable<int> squares = Signal.Unfold(
    initialState: 1,
    condition: state => state <= 5,
    iterate: state => state + 1,
    resultSelector: state => state * state);

squares.Subscribe(x => Console.WriteLine(x), () => Console.WriteLine("done"));

Output:

1
4
9
16
25
done

Building a stream by hand¶

When no factory fits, write the subscribe logic yourself.

Create¶

Create takes a lambda that receives a subscriber and returns an IDisposable. Call OnNext for each value, then OnCompleted or OnError. Return a disposable that cleans up.

using ReactiveUI.Primitives.Disposables;

IObservable<int> counted = Signal.Create<int>(witness =>
{
    witness.OnNext(1);
    witness.OnNext(2);
    witness.OnCompleted();

    return new ActionDisposable(() => Console.WriteLine("cleaned up"));
});

counted.Subscribe(x => Console.WriteLine(x));

Output:

1
2
cleaned up

Return EmptyDisposable.Instance when there is nothing to clean up.

Your lambda runs once per subscriber, so Create builds a cold stream.

CreateSafe¶

CreateSafe is Create with one difference. It matters when your subscriber's own callback throws.

Use CreateSafe when your stream is cold and each subscriber owns its own resources, so a throwing subscriber releases the resources it was using.

var source = new Signal<int>();

IObservable<int> safe = Signal.CreateSafe<int>(witness =>
{
    IDisposable inner = source.Subscribe(witness.OnNext);
    return new ActionDisposable(() =>
    {
        inner.Dispose();
        Console.WriteLine("released");
    });
});

safe.Subscribe(x => throw new InvalidOperationException("boom"));

try
{
    source.OnNext(1);
}
catch (InvalidOperationException error)
{
    Console.WriteLine($"subscriber threw: {error.Message}");
}

source.OnNext(2);   // nothing: the subscription is released

Output:

released
subscriber threw: boom

The exception still reaches the code that pushed the value, after the subscription is released. With Create, the same subscriber stays subscribed and gets 2 too.

CreateWithState¶

CreateWithState passes a state value through to your lambda. That lets you mark the lambda static, so it captures nothing and allocates no closure. See mark lambdas static.

IObservable<int> fromState = Signal.CreateWithState<int, int>(
    state: 5,
    subscribe: static (count, witness) =>
    {
        for (var i = 0; i < count; i++)
        {
            witness.OnNext(i);
        }

        witness.OnCompleted();
        return EmptyDisposable.Instance;
    });

fromState.Subscribe(x => Console.Write($"{x} "));

Output:

0 1 2 3 4

The result is the same as Create. Only the allocation differs.

Deciding when someone subscribes¶

Lazy¶

Lazy calls your factory once per subscriber, at the moment they subscribe. Use it when the stream must read something current.

var clicks = 0;

IObservable<int> latest = Signal.Lazy(() => Signal.Emit(clicks));

clicks = 1;
latest.Subscribe(x => Console.WriteLine($"first: {x}"));

clicks = 2;
latest.Subscribe(x => Console.WriteLine($"second: {x}"));

Output:

first: 1
second: 2

Without Lazy, Signal.Emit(clicks) would read clicks once, when you built the stream, and both subscribers would see 0.

When your factory needs to await, call Signal.Defer instead:

If¶

If checks a condition each time someone subscribes, then uses one stream or the other.

var useCache = true;

IObservable<string> rows = Signal.If(
    () => useCache,
    thenSource: Signal.Emit("from cache"),
    elseSource: Signal.Emit("from server"));

rows.Subscribe(x => Console.WriteLine(x));

useCache = false;
rows.Subscribe(x => Console.WriteLine(x));

Output:

from cache
from server

With no elseSource, a false condition gives you an empty stream that completes at once.

Case¶

Case picks a stream out of a dictionary, using a key it reads at subscribe time.

var mode = "fast";

var sources = new Dictionary<string, IObservable<string>>
{
    ["fast"] = Signal.Emit("cached answer"),
    ["slow"] = Signal.Emit("computed answer"),
};

IObservable<string> answer = Signal.Case(() => mode, sources);

answer.Subscribe(x => Console.WriteLine(x));

mode = "slow";
answer.Subscribe(x => Console.WriteLine(x));

Output:

cached answer
computed answer

A key that is not in the dictionary gives an empty stream. Pass a third argument to use a fallback stream instead.

Use¶

Use ties a resource to the subscription. It builds the resource when someone subscribes, builds the stream from it, and disposes the resource when the stream ends or the subscriber leaves.

IObservable<string> lines = Signal.Use(
    resourceFactory: () => new StreamReader("names.txt"),
    signalFactory: reader => Signal.FromEnumerable(ReadAll(reader)));

lines.Subscribe(x => Console.WriteLine(x), () => Console.WriteLine("reader disposed"));

Output:

Ada
Grace
reader disposed

Each subscriber gets its own reader, and its own reader is disposed.

From a collection¶

FromEnumerable¶

FromEnumerable walks a collection and sends each item, then completes.

IObservable<string> names = Signal.FromEnumerable(["Ada", "Grace", "Katherine"]);

names.Subscribe(x => Console.WriteLine(x), () => Console.WriteLine("done"));

Output:

Ada
Grace
Katherine
done

It is cold, so each subscriber walks the collection again. Pass a CancellationToken as a second argument to stop walking when the token fires.

FromAsyncEnumerable¶

FromAsyncEnumerable does the same for an IAsyncEnumerable<T>.

IObservable<string> rows = Signal.FromAsyncEnumerable(QueryRowsAsync());

rows.Subscribe(x => Console.WriteLine(x));

Output:

row 1
row 2

It is not available on .NET Framework targets. Every other factory on this page is available everywhere.

ToSignal on a collection¶

ToSignal is the same thing written as an extension method, so it reads left to right.

IObservable<int> numbers = new[] { 1, 2, 3 }.ToSignal();

numbers.Subscribe(x => Console.Write($"{x} "));

Output:

1 2 3

From a task¶

FromTask against FromAsync¶

These two look alike and differ in the way that matters most: when the work runs.

// The task is already running. Everyone shares its one result.
Task<int> running = FetchAsync();
IObservable<int> hot = Signal.FromTask(running);

// The factory runs once per subscriber. Two subscribers means two fetches.
IObservable<int> cold = Signal.FromAsync(() => FetchAsync());

Subscribe twice to each:

FromTask also takes a short cut when the task has already finished: it sends the result without scheduling anything.

IObservable<int> done = Signal.FromTask(Task.FromResult(7));

done.Subscribe(x => Console.WriteLine(x), () => Console.WriteLine("done"));

Output:

7
done

FromAsync¶

FromAsync starts the work when someone subscribes and sends the result.

IObservable<string> body = Signal.FromAsync(token => httpClient.GetStringAsync(url, token));

IDisposable subscription = body.Subscribe(x => Console.WriteLine(x.Length));

subscription.Dispose();   // cancels the request

ToSignal on a task¶

ToSignal on a Task<T> is the extension form of Signal.FromTask. It is hot, because the task is already running.

IObservable<int> result = FetchAsync().ToSignal();

result.Subscribe(x => Console.WriteLine(x));

FromTask with a cancellation source¶

There is a second FromTask family that hands your work the CancellationTokenSource the subscription owns. It returns an ITaskSignal<T>, which is a stream you can also cancel directly.

ITaskSignal<int> job = Signal.FromTask(async cts =>
{
    await Task.Delay(1000, cts.Token);
    return 7;
});

IDisposable subscription = job.Subscribe(x => Console.WriteLine(x));

subscription.Dispose();   // cancels the token source, so the delay stops

Each form takes the work first, then optional extras:

Each of the three has a generic form that returns ITaskSignal<TResult> instead of ITaskSignal<RxVoid>.

From an event¶

An event is already a source of values over time. These factories turn one into a stream so you can filter and combine it.

FromEvent¶

FromEvent takes a pair of callbacks: one that attaches a handler, one that removes it.

IObservable<string> messages = Signal.FromEvent<string>(
    addHandler: handler => chat.MessageReceived += handler,
    removeHandler: handler => chat.MessageReceived -= handler);

IDisposable subscription = messages.Subscribe(x => Console.WriteLine(x));

chat.Receive("hello");    // the chat client raises MessageReceived

subscription.Dispose();   // removes the handler
chat.Receive("goodbye");  // nobody is listening now

Output:

hello

The handler is attached when someone subscribes and removed when they dispose, so you cannot leak it.

When the event uses its own delegate type, pass a conversion that wraps an Action<TEventArgs> in that type:

IObservable<PriceArgs> prices = Signal.FromEvent<PriceHandler, PriceArgs>(
    conversion: action => new PriceHandler(args => action(args)),
    addHandler: handler => feed.PriceChanged += handler,
    removeHandler: handler => feed.PriceChanged -= handler);

A third argument sets the sequencer used to attach and remove the handler.

FromEventPattern¶

Most .NET events follow the (object sender, TEventArgs args) shape. FromEventPattern handles that shape and hands you both parts.

IObservable<EventPattern<PropertyChangedEventArgs>> changes =
    Signal.FromEventPattern<PropertyChangedEventHandler, PropertyChangedEventArgs>(
        handler => person.PropertyChanged += handler,
        handler => person.PropertyChanged -= handler);

changes.Subscribe(e => Console.WriteLine(e.EventArgs.PropertyName));

person.Name = "Ada";   // person raises PropertyChanged for Name

Output:

Name

The family covers the shapes you meet:

Every one of them takes an optional ISequencer as a final argument.

> [!NOTE]

> TEventArgs must derive from EventArgs. For an event whose argument is some other type, such as

> EventHandler<int>, use FromEvent instead.

Running work¶

Start¶

Start runs a piece of work once and sends the result.

IObservable<int> total = Signal.Start(() => CountLines("names.txt"));

total.Subscribe(x => Console.WriteLine($"lines: {x}"));

Output:

lines: 2

The work runs on Sequencer.Default, which is a background thread, so it does not block the caller. Pass a sequencer as a second argument to choose another one.

An Action overload runs work with no result and sends RxVoid when it finishes.

IObservable<RxVoid> saved = Signal.Start(() => File.WriteAllText("out.txt", "hi"));

saved.Subscribe(_ => Console.WriteLine("written"));

Output:

written

Start runs the work once per subscriber.

From a clock¶

After¶

After sends the long value 0 once, after a delay, then completes.

IObservable<long> ready = Signal.After(TimeSpan.FromSeconds(2));

ready.Subscribe(x => Console.WriteLine($"fired: {x}"), () => Console.WriteLine("done"));

Output, two seconds later:

fired: 0
done

Give it a second TimeSpan and it keeps going, counting up.

IObservable<long> ticks = Signal.After(TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(2));

ticks.Subscribe(x => Console.WriteLine(x));

Output: 0 after one second, then 1, 2, 3 every two seconds after that. It never completes, so dispose the subscription to stop it.

Each of the three also takes a sequencer.

Every¶

Every counts up on a fixed period, forever. There is no initial delay to set; the first value arrives after one full period.

IObservable<long> seconds = Signal.Every(TimeSpan.FromSeconds(1));

IDisposable subscription = seconds.Subscribe(x => Console.WriteLine(x));

Output, one per second:

0
1
2

Dispose the subscription to stop it. It also takes an optional sequencer.

After against Every¶

IObservable<long> once = Signal.After(TimeSpan.FromSeconds(5));                            // 0 at 5s, then completes
IObservable<long> delayedTicks = Signal.After(TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(1)); // 0 at 5s, 1 at 6s, 2 at 7s, …
IObservable<long> ticks = Signal.Every(TimeSpan.FromSeconds(1));                           // 0 at 1s, 1 at 2s, 2 at 3s, …

Use After for a deadline. Use Every for a heartbeat.

Emitting inside Subscribe¶

Observables.Return sends one value and completes, like Signal.Emit. The difference is that it does the work inside your call to Subscribe, before Subscribe returns.

using ReactiveUI.Primitives.Extensions;

Console.WriteLine("before");
Observables.Return(1).Subscribe(x => Console.WriteLine(x));
Console.WriteLine("after");

Output:

before
1
after

Reach for it when you need the value to have arrived by the time the next line runs.

Factories that join streams¶

Signal also carries static factories that build one stream out of several: Blend, Concat, Race, Switch, Pair, SyncLatest, PairLatest, ForkJoin, OnErrorResumeNext and Recover. They are covered with the operators that do the same job, on the combination, transformation and error handling pages.

Factories that build a stream you push values into yourself, such as Signal.Serialized, Signal.Scheduled and Signal.Delayable, are covered on the signals page.

Every factory at a glance¶

The two package flavours¶

Every type here ships twice. ReactiveUI.Primitives puts them under ReactiveUI.Primitives.*. ReactiveUI.Primitives.Reactive is the same source compiled against System.Reactive, and puts them under ReactiveUI.Primitives.Reactive.*. The factories behave the same in both.