combineLatest

import rx : SubjectObject, CounterObserver;

auto hello = new SubjectObject!string;
auto world = new SubjectObject!string;

auto message = combineLatest!((a, b) => a ~ ", " ~ b ~ "!")(hello, world);

auto observer = new CounterObserver!string;
message.doSubscribe(observer);

.put(hello, "Hello");
.put(world, "world");

assert(observer.putCount == 1);
assert(observer.lastValue == "Hello, world!");

.put(world, "D-man");
assert(observer.putCount == 2);
assert(observer.lastValue == "Hello, D-man!");

import rx : SubjectObject, CounterObserver, uniq;

auto count1 = new SubjectObject!int;
auto count2 = new SubjectObject!int;
auto count3 = new SubjectObject!int;

import std.algorithm : max;

alias pickMax = combineLatest!max;
auto observable = pickMax(count1, count2, count3).uniq();

auto observer = new CounterObserver!int;
observable.doSubscribe(observer);

.put(count1, 0);
.put(count2, 0);
.put(count3, 0);

assert(observer.putCount == 1);
assert(observer.lastValue == 0);

.put(count1, 10);
assert(observer.putCount == 2);
assert(observer.lastValue == 10);

.put(count2, 10);
assert(observer.putCount == 2);

.put(count3, 11);
assert(observer.putCount == 3);
assert(observer.lastValue == 11);