Shaping values¶

These helpers filter, reshape and pick values. Each one does a job you could build from Where, Select and friends, in one named call.

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

The examples push values by hand into a Signal<T>, a stream you send values into with OnNext.

Filtering¶

`WhereIsNotNull`¶

WhereIsNotNull drops null values and passes the rest through.

var names = new Signal<string?>();

names.WhereIsNotNull()
     .Subscribe(static name => Console.WriteLine(name));

names.OnNext("Ada");
names.OnNext(null);
names.OnNext("Grace");

Output:

Ada
Grace

The element type stays the same: on an IObservable<string?> you get an IObservable<string?> back, even though no null reaches your callback. So the compiler may still warn about null where you use the value. When you want an IObservable<string>, add a Select with the null-forgiving operator after it:

IObservable<string> knownNames = names.WhereIsNotNull().Select(static name => name!);

The async WhereIsNotNull hands back a stream whose type cannot be null, with no extra step.

`SkipWhileNull`¶

SkipWhileNull drops null values only until the first value that is not null. After that, it passes everything through, null included. Use it for a value that starts empty and fills in once, such as a loaded setting.

var selected = new Signal<string?>();

selected.SkipWhileNull()
        .Subscribe(static item => Console.WriteLine(item ?? "null"));

selected.OnNext(null);
selected.OnNext("apple");
selected.OnNext(null);
selected.OnNext("pear");

Output:

apple
null
pear

`WhereTrue`, `WhereFalse` and `Not`¶

These work on a stream of bool. WhereTrue passes only true. WhereFalse passes only false. Not flips each value.

var isBusy = new Signal<bool>();

isBusy.WhereTrue().Subscribe(static _ => Console.WriteLine("started"));
isBusy.WhereFalse().Subscribe(static _ => Console.WriteLine("finished"));
isBusy.Not().Subscribe(static canClick => Console.WriteLine($"button enabled: {canClick}"));

isBusy.OnNext(true);
isBusy.OnNext(false);

Output:

started
button enabled: False
finished
button enabled: True

`Filter`¶

Filter works on a stream of string. It passes the strings that match a regular expression. Give it a pattern, or a Regex you have already built. A pattern gets a 30 second match timeout, so a slow pattern cannot hang your app.

using System.Text.RegularExpressions;

var log = new Signal<string>();

log.Filter("^ERROR")
   .Subscribe(static line => Console.WriteLine($"error line: {line}"));

log.Filter(new Regex("warn", RegexOptions.IgnoreCase))
   .Subscribe(static line => Console.WriteLine($"warning line: {line}"));

log.OnNext("ERROR disk full");
log.OnNext("all good");
log.OnNext("WARN low memory");

Output:

error line: ERROR disk full
warning line: WARN low memory

Changing values¶

`SelectConstant`¶

SelectConstant replaces every value with the same value. It is Select(_ => constant) without the lambda.

var clicks = new Signal<int>();

clicks.SelectConstant("clicked")
      .Subscribe(static text => Console.WriteLine(text));

clicks.OnNext(1);
clicks.OnNext(2);

Output:

clicked
clicked

`AsSignal`¶

AsSignal replaces every value with RxVoid, a value that carries no data. Use it when you only care that something happened, not what it was. See RxVoid.

var saved = new Signal<string>();

IObservable<RxVoid> somethingSaved = saved.AsSignal();
somethingSaved.Subscribe(static _ => Console.WriteLine("saved"));

saved.OnNext("report.txt");

Output:

saved

`TrySelect`¶

TrySelect runs a lambda on each value and drops the results that are null. It filters and converts in one step, which suits parsing.

var input = new Signal<string>();

input.TrySelect(static text => int.TryParse(text, out var number) ? number : (int?)null)
     .Subscribe(static number => Console.WriteLine(number));

input.OnNext("1");
input.OnNext("one");
input.OnNext("3");

Output:

1
3

`WhereSelect`¶

WhereSelect takes a test and a lambda. It keeps the values that pass the test, then converts them. It is Where followed by Select, as one operator.

var numbers = new Signal<int>();

numbers.WhereSelect(static x => x % 2 == 0, static x => x * 10)
       .Subscribe(static x => Console.WriteLine(x));

numbers.OnNext(1);
numbers.OnNext(2);
numbers.OnNext(4);

Output:

20
40

`ScanWithInitial`¶

ScanWithInitial keeps a running total, like Fold. The difference is that it sends the starting value first, as soon as you subscribe. A display that shows the total gets something to show straight away.

var added = new Signal<int>();

added.ScanWithInitial(0, static (total, x) => total + x)
     .Subscribe(static total => Console.WriteLine(total));

added.OnNext(5);
added.OnNext(3);

Output:

0
5
8

`Pairwise`¶

Pairwise sends each value together with the one before it, as a tuple of Previous and Current. The first value has nothing before it, so it produces no pair.

var temperatures = new Signal<int>();

temperatures.Pairwise()
            .Subscribe(static pair => Console.WriteLine($"{pair.Previous} -> {pair.Current}"));

temperatures.OnNext(20);
temperatures.OnNext(22);
temperatures.OnNext(21);

Output:

20 -> 22
22 -> 21

`LatestOrDefault`¶

LatestOrDefault sends a default value as soon as you subscribe. After that it sends the source's values, skipping any value equal to the one it sent last.

var score = new Signal<int>();

score.LatestOrDefault(-1)
     .Subscribe(static x => Console.WriteLine(x));

score.OnNext(5);
score.OnNext(5);
score.OnNext(6);

Output:

-1
5
6

Collections and text¶

`ForEach`¶

ForEach works on a stream of collections. It sends every item of each collection as a value of its own.

var pages = new Signal<IEnumerable<int>>();

pages.ForEach()
     .Subscribe(static item => Console.WriteLine(item));

pages.OnNext([1, 2]);
pages.OnNext([3]);

Output:

1
2
3

Pass a sequencer to send the items through it.

`FromArray`¶

FromArray turns any IEnumerable<T> into a stream that sends each item, then completes. It is an extension method on the collection, so it reads left to right.

new[] { 1, 2, 3 }.FromArray()
                 .Subscribe(static x => Console.WriteLine(x), static () => Console.WriteLine("completed"));

Output:

1
2
3
completed

Signal.FromEnumerable does the same job as a factory.

`Shuffle`¶

Shuffle works on a stream of arrays. It shuffles each array into a random order before passing it on. It shuffles the array you sent in place, so copy the array first if you still need the original order.

var decks = new Signal<int[]>();

decks.Shuffle()
     .Subscribe(static deck => Console.WriteLine(string.Join(", ", deck)));

decks.OnNext([1, 2, 3, 4]);   // prints the four numbers in a random order

`BufferUntil`¶

BufferUntil works on a stream of char. It collects characters from a start character to an end character, and sends them as one string that includes both. Characters outside a start and end pair are dropped. When the stream completes, it sends any unfinished string it holds.

var serial = new Signal<char>();

serial.BufferUntil('<', '>')
      .Subscribe(static message => Console.WriteLine(message));

foreach (var c in "noise<hello>more<bye")
{
    serial.OnNext(c);
}

serial.OnCompleted();

Output:

<hello>
<bye

`OnNext` with several values¶

This OnNext is an extension on any observer, including a signal. It sends several values in one call, in order.

var numbers = new Signal<int>();
numbers.Subscribe(static x => Console.WriteLine(x));

numbers.OnNext(1, 2, 3);

Output:

1
2
3

Splitting and picking¶

`Partition`¶

Partition splits one stream into two. Values that pass the test go to True, and the rest go to False.

var numbers = new Signal<int>();

var (evens, odds) = numbers.Partition(static x => x % 2 == 0);

evens.Subscribe(static x => Console.WriteLine($"even {x}"));
odds.Subscribe(static x => Console.WriteLine($"odd {x}"));

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

Output:

odd 1
even 2
odd 3

Both halves share one subscription to the source. A hot stream, one that sends values whether or not anyone is listening, such as a signal, works as above.

A cold stream starts its work when you subscribe, such as Signal.FromEnumerable. It can run to completion the moment the first half subscribes, so only that half gets values. Share it first with ShareLive, subscribe to both halves, then call Connect:

var cold = Signal.FromEnumerable([1, 2, 3, 4, 5]);

var (coldEvens, coldOdds) = cold.Partition(static x => x % 2 == 0);
coldEvens.Subscribe(static x => Console.WriteLine($"cold even {x}"));
coldOdds.Subscribe(static x => Console.WriteLine($"cold odd {x}"));

var shared = cold.ShareLive();
var (sharedEvens, sharedOdds) = shared.Partition(static x => x % 2 == 0);
sharedEvens.Subscribe(static x => Console.WriteLine($"shared even {x}"));
sharedOdds.Subscribe(static x => Console.WriteLine($"shared odd {x}"));
shared.Connect();

Output:

cold even 2
cold even 4
shared odd 1
shared even 2
shared odd 3
shared even 4
shared odd 5

Without sharing, the odd half got nothing: the even half had already used up the stream.

`TakeUntil` with a test¶

TakeUntil with a lambda passes values through until one passes the test. It sends that value too, then completes.

var progress = new Signal<int>();

progress.TakeUntil(static percent => percent >= 100)
        .Subscribe(static p => Console.WriteLine(p), static () => Console.WriteLine("completed"));

progress.OnNext(50);
progress.OnNext(100);
progress.OnNext(120);

Output:

50
100
completed

TakeUntil with another stream or a CancellationToken stops on a signal from outside instead.

`WaitUntil`¶

WaitUntil drops values until one passes the test. It sends that one value, then completes.

var status = new Signal<int>();

status.WaitUntil(static code => code == 200)
      .Subscribe(static code => Console.WriteLine(code), static () => Console.WriteLine("completed"));

status.OnNext(404);
status.OnNext(200);
status.OnNext(500);

Output:

200
completed

`SwitchIfEmpty`¶

SwitchIfEmpty watches a stream. If it completes without sending anything, SwitchIfEmpty subscribes to a fallback stream instead.

Signal.Empty<int>()
      .SwitchIfEmpty(Signal.Emit(42))
      .Subscribe(static x => Console.WriteLine(x));

Signal.Emit(1)
      .SwitchIfEmpty(Signal.Emit(42))
      .Subscribe(static x => Console.WriteLine(x));

Output:

42
1

`FirstMatchFromCandidates`¶

FirstMatchFromCandidates tries a list of options in order and stops at the first one that works. For each option it runs a lambda that returns a stream, converts the stream's value, and tests the result. An option whose stream fails is skipped. If no option passes, it sends a fallback value.

It suits "look in each place until you find it", such as a list of config files.

string[] files = ["user.json", "team.json", "default.json"];

files.FirstMatchFromCandidates(
        project: static file => LoadText(file),
        transform: static text => text.Length,
        predicate: static length => length > 0,
        fallback: -1)
     .Subscribe(static length => Console.WriteLine(length));

static IObservable<string> LoadText(string file) =>
    file == "user.json"
        ? Signal.Fail<string>(new FileNotFoundException(file))
        : Signal.Emit($"{{ \"from\": \"{file}\" }}");

Output:

user.json failed, so it was skipped. team.json loaded, and its length passed the test.

Combining¶

`GetMax` and `GetMin`¶

GetMax and GetMin watch several streams of numbers. Once every stream has sent a value, they send the largest or smallest of the latest values, and send again whenever one changes. They work on any struct that implements IComparable<T>, such as int, double or DateTime.

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

a.GetMax(b).Subscribe(static x => Console.WriteLine($"max {x}"));

a.OnNext(3);    // b has no value yet
b.OnNext(7);
a.OnNext(9);

Output:

max 7
max 9

`CombineLatestValuesAreAllTrue` and `CombineLatestValuesAreAllFalse`¶

These work on a collection of bool streams. Once every stream has sent a value, they send whether all the latest values are true, or all false. A form's Save button is a good fit: enable it when every field is valid.

var nameValid = new Signal<bool>();
var emailValid = new Signal<bool>();

new[] { nameValid, emailValid }
    .CombineLatestValuesAreAllTrue()
    .Subscribe(static canSave => Console.WriteLine($"can save: {canSave}"));

nameValid.OnNext(true);
emailValid.OnNext(false);
emailValid.OnNext(true);

Output:

can save: False
can save: True

CombineLatestValuesAreAllFalse answers the opposite question: whether every latest value is false. Use it, for example, to show a hint while no field is valid yet.

var firstValid = new Signal<bool>();
var secondValid = new Signal<bool>();

new[] { firstValid, secondValid }
    .CombineLatestValuesAreAllFalse()
    .Subscribe(static nothingValid => Console.WriteLine($"nothing valid: {nothingValid}"));

firstValid.OnNext(false);
secondValid.OnNext(false);
secondValid.OnNext(true);

Output:

nothing valid: True
nothing valid: False

`SelectManyThen`¶

SelectManyThen runs two steps that each return a stream, one after the other, for each value. It is SelectMany twice, as one operator.

Signal.Emit(1)
      .SelectManyThen(
          static id => Signal.Emit(id * 10),       // look up an order
          static order => Signal.Emit(order + 1))  // then its invoice
      .Subscribe(static x => Console.WriteLine(x));

Output:

`RunAll`¶

RunAll works on a list of IObservable<RxVoid> steps. It subscribes to them one at a time, in order, waiting for each to complete before starting the next. When the last completes, it sends one RxVoid and completes.

IObservable<RxVoid>[] steps =
[
    Signal.Lazy(static () => { Console.WriteLine("step 1"); return Signal.Emit(RxVoid.Default); }),
    Signal.Lazy(static () => { Console.WriteLine("step 2"); return Signal.Emit(RxVoid.Default); }),
];

steps.RunAll()
     .Subscribe(static _ => Console.WriteLine("all done"));

Output:

step 1
step 2
all done

Every helper on this page at a glance¶

Helper	What it does
`WhereIsNotNull`	Drops `null` values.
`SkipWhileNull`	Drops `null` values until the first value that is not `null`.
`WhereTrue` / `WhereFalse`	Passes only `true`, or only `false`.
`Not`	Flips each `bool`.
`Filter`	Passes strings that match a regular expression.
`SelectConstant`	Replaces every value with one value.
`AsSignal`	Replaces every value with `RxVoid`.
`TrySelect`	Converts each value and drops `null` results.
`WhereSelect`	Filters, then converts.
`ScanWithInitial`	A running total that sends the starting value first.
`Pairwise`	Sends each value with the one before it.
`LatestOrDefault`	Sends a default first, then values that differ from the last one sent.
`ForEach`	Sends each item of each collection.
`FromArray`	Turns a collection into a stream.
`Shuffle`	Shuffles each array in place.
`BufferUntil`	Collects characters between a start and an end character.
`OnNext(params T[])`	Sends several values to an observer.
`Partition`	Splits a stream in two by a test.
`TakeUntil(predicate)`	Passes values up to and including the first that passes a test.
`WaitUntil`	Sends the first value that passes a test, then completes.
`SwitchIfEmpty`	Uses a fallback stream when the source completes empty.
`FirstMatchFromCandidates`	Tries options in order and sends the first result that passes a test.
`GetMax` / `GetMin`	The largest or smallest latest value across streams.
`CombineLatestValuesAreAllTrue` / `...AllFalse`	Whether every latest `bool` is `true`, or `false`.
`SelectManyThen`	Two stream-returning steps in a row.
`RunAll`	Runs `RxVoid` steps one after another.

The types behind these helpers¶

Most helpers here have a public class in ReactiveUI.Primitives.Extensions.Operators, such as WhereIsNotNullObservable<T>, PairwiseObservable<T>, PartitionObservable<T> and TrySelectObservable<T, TOut>. ScanWithInitialObservable<T, TAccumulate> is in ReactiveUI.Primitives.Extensions. Each takes its source through the constructor. Call the helper in normal code, and construct the class when you write an operator of your own.

SelectManyThenCoordinator<TSource, TMid, TResult>, in ReactiveUI.Primitives.Extensions.Operators, is the witness behind SelectManyThen. It takes the downstream witness and both steps. Subscribe it to the source, and hand it back as the subscription:

using ReactiveUI.Primitives.Extensions.Operators;

new InvoiceLookup(Signal.Emit(1))
    .Subscribe(static x => Console.WriteLine(x), static () => Console.WriteLine("completed"));

public sealed class InvoiceLookup(IObservable<int> orderIds) : IObservable<int>
{
    public IDisposable Subscribe(IObserver<int> witness)
    {
        var coordinator = new SelectManyThenCoordinator<int, int, int>(
            witness,
            static id => Signal.Emit(id * 10),
            static order => Signal.Emit(order + 1));

        orderIds.Subscribe(coordinator);
        return coordinator;
    }
}

Output:

11
completed

Observables.Return(value) and SingleValueSignal<T> in ReactiveUI.Primitives.Extensions build a stream that sends one value and completes, like Signal.Emit.

Shaping values¶

Filtering¶

WhereIsNotNull¶

SkipWhileNull¶

WhereTrue, WhereFalse and Not¶

Filter¶

Changing values¶

SelectConstant¶

AsSignal¶

TrySelect¶

WhereSelect¶

ScanWithInitial¶

Pairwise¶

LatestOrDefault¶

Collections and text¶

ForEach¶

FromArray¶

Shuffle¶

BufferUntil¶

OnNext with several values¶

Splitting and picking¶

Partition¶

TakeUntil with a test¶

WaitUntil¶

SwitchIfEmpty¶

FirstMatchFromCandidates¶

Combining¶

GetMax and GetMin¶

CombineLatestValuesAreAllTrue and CombineLatestValuesAreAllFalse¶

SelectManyThen¶

RunAll¶

Every helper on this page at a glance¶

The types behind these helpers¶

`WhereIsNotNull`¶

`SkipWhileNull`¶

`WhereTrue`, `WhereFalse` and `Not`¶

`Filter`¶

`SelectConstant`¶

`AsSignal`¶

`TrySelect`¶

`WhereSelect`¶

`ScanWithInitial`¶

`Pairwise`¶

`LatestOrDefault`¶

`ForEach`¶

`FromArray`¶

`Shuffle`¶

`BufferUntil`¶

`OnNext` with several values¶

`Partition`¶

`TakeUntil` with a test¶

`WaitUntil`¶

`SwitchIfEmpty`¶

`FirstMatchFromCandidates`¶

`GetMax` and `GetMin`¶

`CombineLatestValuesAreAllTrue` and `CombineLatestValuesAreAllFalse`¶

`SelectManyThen`¶

`RunAll`¶