Combination¶

A combination operator takes two or more streams and makes one stream out of them. The operators differ in one thing: when the new stream sends a value. Some send a value whenever any source does. Some wait for every source. Some only listen to one side and read the other.

Picking the right one is mostly about that timing. The comparison near the end of this page runs four of them on the same input, so you can see the difference side by side.

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

Many examples here use a Signal<T>. A Signal<T> is a stream you push values into by hand with OnNext. That lets you choose exactly when each source sends a value, so you can see what the operator does at each step.

Your first combination¶

You have a form with a first name box and a last name box. You want a full name that updates whenever either box changes.

1. Make a stream for each box.

var first = new Signal<string>();
var last = new Signal<string>();

2. Combine them. SyncLatest keeps the newest value from each side. Whenever either side sends a value, it calls your lambda with both newest values.

IObservable<string> fullName = first.SyncLatest(last, (f, l) => $"{f} {l}");

3. Subscribe.

fullName.Subscribe(name => Console.WriteLine(name));

4. Push values and watch.

first.OnNext("Ada");         // prints nothing
last.OnNext("Lovelace");     // prints Ada Lovelace
first.OnNext("Augusta");     // prints Augusta Lovelace

Trace it one push at a time:

The first push printed nothing. SyncLatest cannot build a full name until both sides have sent at least one value. Once they have, every push from either side produces a new full name.

Adding values at the start or the end¶

Lead¶

Lead sends one value before the source's own values.

Input: 2, 3

Signal.Range(2, 2)
      .Lead(0)
      .Subscribe(x => Console.Write($"{x} "));

Output:

0 2 3

Use it to give a stream a starting value, so a subscriber has something to show before the first real value arrives.

Prepend¶

Prepend sends values before the source's own values. It takes one value, several values, or a collection.

Input: 3, 4

Signal.Range(3, 2)
      .Prepend(1, 2)
      .Subscribe(x => Console.Write($"{x} "));

Output:

1 2 3 4

With one value, Prepend(value) does the same job as Lead.

StartWith¶

StartWith sends several values before the source's own values, like Prepend. It takes the values directly or as a collection.

Input: 3, 4

var header = new List<int> { 1, 2 };

Signal.Range(3, 2)
      .StartWith(header)
      .Subscribe(x => Console.Write($"{x} "));

Output:

1 2 3 4

Append¶

Append sends one extra value after the source completes.

Input: 1, 2, then completion

Signal.Range(1, 2)
      .Append(99)
      .Subscribe(x => Console.Write($"{x} "));

Output:

1 2 99

The 99 only arrives when the source completes. A source that never completes never sends it.

One source after another¶

Concat¶

Concat runs a second stream after the first one completes. The second stream does not start until the first has finished.

Input: 1, 2, then 10, 11

Signal.Range(1, 2)
      .Concat(Signal.Range(10, 2))
      .Subscribe(x => Console.Write($"{x} "), () => Console.WriteLine("| done"));

Output:

1 2 10 11 | done

Concat on a stream of streams¶

A stream can carry other streams as its values. Concat on such a stream runs the inner streams one at a time, in the order they arrive.

Input: a stream that carries two streams, 1, 2 and 10, 11

IObservable<IObservable<int>> pages = Signal.FromEnumerable(new[]
{
    Signal.Range(1, 2),
    Signal.Range(10, 2),
});

pages.Concat().Subscribe(x => Console.Write($"{x} "), () => Console.WriteLine("| done"));

Output:

1 2 10 11 | done

Concat on a stream of tasks¶

Concat on a stream of tasks waits for each task in turn and sends its result. The results come out in the order the tasks arrived, not the order they finished.

Input: two tasks, where the second finishes first

var firstTask = new TaskCompletionSource<int>();
var secondTask = new TaskCompletionSource<int>();

var tasks = new Signal<Task<int>>();
tasks.Concat().Subscribe(x => Console.Write($"{x} "), () => Console.WriteLine("| done"));

tasks.OnNext(firstTask.Task);
tasks.OnNext(secondTask.Task);
tasks.OnCompleted();

secondTask.SetResult(2);   // finishes first, but waits its turn
firstTask.SetResult(1);

Output:

1 2 | done

The second task finished first. Concat still held its result back until the first task's result had gone out.

Several sources at once¶

Merge on two streams¶

Merge listens to two streams at the same time. It passes each value on as soon as it arrives, from either stream.

Input: a and b push values in turn

var a = new Signal<string>();
var b = new Signal<string>();

a.Merge(b).Subscribe(x => Console.Write($"{x} "));

a.OnNext("a1");
b.OnNext("b1");
b.OnNext("b2");
a.OnNext("a2");

Output:

a1 b1 b2 a2

The values come out in the order they were pushed, whichever stream they came from.

Blend¶

Blend does the same for any number of streams. It takes a stream of streams, or a collection of streams.

Input: two streams in a collection

var a = new Signal<string>();
var b = new Signal<string>();

new[] { a, b }.Blend().Subscribe(x => Console.Write($"{x} "), () => Console.WriteLine("| done"));

b.OnNext("b1");
a.OnNext("a1");
a.OnCompleted();
b.OnNext("b2");
b.OnCompleted();

Output:

b1 a1 b2 | done

Blend completes only when every stream has completed. a finished early, and Blend kept passing on values from b until b finished too.

Signal.Blend(a, b) is the same operator called as a factory.

Blend with a limit¶

Blend(maxConcurrent) caps how many streams it listens to at once. The rest wait their turn. When one stream completes, the next one starts.

Input: the same two streams, with a limit of one

var a = new Signal<string>();
var b = new Signal<string>();

new[] { a, b }.Blend(1).Subscribe(x => Console.Write($"{x} "), () => Console.WriteLine("| done"));

b.OnNext("b1");
a.OnNext("a1");
a.OnCompleted();
b.OnNext("b2");
b.OnCompleted();

Output:

a1 b2 | done

Compare that with the output above. b1 is missing.

> [!WARNING]

> A stream that waits its turn is not listening yet. b sent b1 while Blend was still busy with a, so

> nobody received b1 and it was lost. A Signal<T> sends values whether or not anyone is subscribed.

> Use a limit on streams like that only when losing early values is acceptable. A stream that starts its work

> when you subscribe, such as Signal.Range, loses nothing, because it waits for you.

BlendUnique¶

BlendUnique merges streams, then drops any value equal to the one it passed on just before.

Input: two int streams

var a = new Signal<int>();
var b = new Signal<int>();

LinqExtensions.BlendUnique(a, b).Subscribe(x => Console.Write($"{x} "));

a.OnNext(5);
b.OnNext(5);   // same as the last value, dropped
b.OnNext(6);
a.OnNext(6);   // same as the last value, dropped
a.OnNext(7);
b.OnNext(5);   // different from 7, so it gets through

Output:

5 6 7 5

It compares each value only with the one just before. The last 5 got through because the value before it was 7. BlendUnique is a static method, so you call it on LinqExtensions. A second form takes an IEqualityComparer<T> to decide what counts as equal.

Concat against Blend¶

Both turn several streams into one. They differ in whether the streams run one after another or all at once.

Use Concat when order matters, such as pages you must show in sequence. Use Blend when you want every value as soon as it exists, such as messages from several chat rooms.

The first source to react¶

Race¶

Race listens to several streams and keeps whichever one sends something first. It ignores the rest from then on.

Input: two streams, where b sends first

var a = new Signal<string>();
var b = new Signal<string>();

Signal.Race(a, b).Subscribe(x => Console.Write($"{x} "));

b.OnNext("b1");   // b wins
a.OnNext("a1");   // ignored from now on
b.OnNext("b2");
a.OnNext("a2");

Output:

b1 b2

Use it to ask two servers the same question and take whichever answers first. On a stream of streams, call Race().

Pairing values from two sources¶

Zip¶

Zip lines values up by position. It matches the first value from one side with the first from the other, then the second with the second, and so on.

Input: the left side sends a, b and completes. The right side sends 1, 2, 3.

var left = new Signal<string>();
var right = new Signal<int>();

left.Zip(right, (letter, number) => letter + number)
    .Subscribe(x => Console.Write($"{x} "), () => Console.WriteLine("| done"));

left.OnNext("a");
left.OnNext("b");
left.OnCompleted();

right.OnNext(1);
right.OnNext(2);
right.OnNext(3);

Output:

a1 b2 | done

Zip completed straight after b2. The left side had finished, so 3 had nothing to pair with. Zip stops as soon as no more pairs can form.

SyncLatest¶

SyncLatest keeps the newest value from each side. When either side sends a value, it calls your lambda with the newest value from both. It starts once both sides have sent something. The walkthrough at the top of this page shows it step by step.

SyncLatest with more sources¶

SyncLatest takes up to 16 streams. Your lambda receives the newest value from each one, in order.

Input: hours, minutes and seconds arrive separately

var hours = new Signal<int>();
var minutes = new Signal<int>();
var seconds = new Signal<int>();

hours.SyncLatest(minutes, seconds, (h, m, s) => $"{h:00}:{m:00}:{s:00}")
     .Subscribe(x => Console.Write($"{x} "));

hours.OnNext(9);
minutes.OnNext(30);
seconds.OnNext(0);    // prints 09:30:00
seconds.OnNext(1);    // prints 09:30:01
minutes.OnNext(31);   // prints 09:31:01

Output:

09:30:00 09:30:01 09:31:01

Nothing came out until all three had sent a value.

CombineLatest without a lambda¶

To get the newest values without writing a lambda, call CombineLatest. It works like SyncLatest, and hands you the newest values as a tuple with named elements First, Second, Third and so on.

Input: a name, then two ages

var name = new Signal<string>();
var age = new Signal<int>();

name.CombineLatest(age)
    .Subscribe(pair => Console.Write($"{pair.First} is {pair.Second}. "));

name.OnNext("Ada");
age.OnNext(36);
age.OnNext(37);

Output:

Ada is 36. Ada is 37.

This saves you writing a lambda when all you want is the values together. It takes up to 16 streams.

CombineLatest on a collection¶

When every stream carries the same type, CombineLatest on a collection hands you a list of the newest values. The list is in the same order as your collection.

Input: three int streams

var a = new Signal<int>();
var b = new Signal<int>();
var c = new Signal<int>();

new[] { a, b, c }.CombineLatest()
                 .Subscribe(values => Console.Write($"[{string.Join(",", values)}] "));

a.OnNext(1);
b.OnNext(2);
c.OnNext(3);    // prints [1,2,3]
b.OnNext(20);   // prints [1,20,3]

Output:

[1,2,3] [1,20,3]

Use this form when the number of streams is not known until your app runs, such as one stream per open document.

Pass a lambda to work on the list instead:

new[] { a, b, c }.CombineLatest(values => values.Sum())
                 .Subscribe(total => Console.Write($"{total} "));

Output, for the same pushes:

6 24

> [!IMPORTANT]

> To call the static form, pass an array. LinqExtensions.CombineLatest(new[] { a, b }) hands you a list.

> Writing LinqExtensions.CombineLatest(a, b) does not. With the streams listed one by one, C# picks the tuple

> form above instead, and you get a tuple where you expected a list.

Latch¶

Latch sends a value only when the left side sends. It pairs that value with the newest value the right side has sent. The right side sending on its own sends nothing.

Input: button clicks on the left, temperature readings on the right

var clicks = new Signal<string>();
var temperature = new Signal<int>();

clicks.Latch(temperature, (click, reading) => $"{click} at {reading}")
      .Subscribe(x => Console.WriteLine(x));

clicks.OnNext("click 1");        // no reading yet, dropped
temperature.OnNext(20);
temperature.OnNext(21);
clicks.OnNext("click 2");        // prints click 2 at 21
temperature.OnNext(22);
clicks.OnNext("click 3");        // prints click 3 at 22

Output:

click 2 at 21
click 3 at 22

click 1 was dropped, because the right side had not sent a reading yet. Use Latch when one stream says when and the other says what, such as saving the current form whenever the user clicks Save.

ForkJoin¶

ForkJoin waits for both sides to complete. Then it sends one value, built from the last value of each side.

Input: the left side sends a, b, c. The right side sends 1, 2. Both complete.

var left = new Signal<string>();
var right = new Signal<int>();

left.ForkJoin(right, (letter, number) => letter + number)
    .Subscribe(x => Console.Write($"{x} "), () => Console.WriteLine("| done"));

left.OnNext("a");
right.OnNext(1);
left.OnNext("b");
left.OnNext("c");
right.OnNext(2);
left.OnCompleted();
right.OnCompleted();

Output:

c2 | done

Use it to run two jobs at once and act when both have finished, such as loading a user and their settings before you show a page.

If either side completes without ever sending a value, ForkJoin has nothing to build from. It completes and sends nothing.

left.OnNext("a");
left.OnCompleted();
right.OnCompleted();   // right never sent anything

Output:

| done

Zip, SyncLatest, Latch and ForkJoin on one input¶

These four are the easiest to mix up. Here they all run on the same pushes, in the same order.

Pushes: the left sends a. The right sends 1. The left sends b, then c. The right sends 2. Then both complete.

The full output from each:

Read down one column to see one operator's rule at work:

• Zip waited at b and c because the right had only one value. When 2 arrived, it paired it with b, the next unmatched left value.
• SyncLatest sent something on every push after the first, because both sides had a value by then.
• Latch dropped a, because the right had sent nothing yet. It ignored 2, because only the left side triggers it.
• ForkJoin stayed silent the whole time, then sent one value at the end.

Every combination operator at a glance¶

Signal also offers several of these as factories that take the streams as arguments: Signal.Concat, Signal.Blend, Signal.Race, Signal.Pair, Signal.SyncLatest, Signal.PairLatest and Signal.ForkJoin. They behave the same as the operators.

The types behind these operators¶

LeadSignal<T>, PrependSignal<T>, StartWithEnumerableSignal<T>, ZipSignal<TLeft, TRight, TResult>, CombineLatestSignal<TLeft, TRight, TResult> and LatchSignal<TLeft, TRight, TResult> are public classes in ReactiveUI.Primitives.Advanced. Each takes its sources through the constructor.

using ReactiveUI.Primitives.Advanced;

// the operator
IObservable<string> a = left.Zip(right, (letter, number) => letter + number);

// the same thing, built directly
IObservable<string> b = new ZipSignal<string, int, string>(left, right, (letter, number) => letter + number);

Calling the operator is the normal path. Construct the type when you are writing an operator of your own and want to place one inside it.

The two package flavours¶

Every operator 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.*. They behave the same in both.