uniq

import rx.subject : SubjectObject;
import std.array : appender;

auto sub = new SubjectObject!int;
auto buf = appender!(int[]);

auto disposable = sub.uniq.subscribe(buf);
scope (exit)
    disposable.dispose();

.put(sub, [11, 11, 22, 22, 33]);

auto data = buf.data;
assert(data.length == 3);
assert(data[0] == 11);
assert(data[1] == 22);
assert(data[2] == 33);

import std.datetime : Date;
import rx.subject : SubjectObject;
import std.array : appender;

auto sub = new SubjectObject!Date;
auto buf = appender!(Date[]);

auto disposable = sub.uniq!"a.year == b.year".subscribe(buf);
scope (exit)
    disposable.dispose();

.put(sub, Date(2000, 1, 1));
.put(sub, Date(2000, 1, 2));
.put(sub, Date(2017, 3, 24));
.put(sub, Date(2017, 4, 24));
.put(sub, Date(2017, 4, 24));

auto data = buf.data;
assert(data.length == 2);
assert(data[0] == Date(2000, 1, 1));
assert(data[1] == Date(2017, 3, 24));

import std.datetime : Date;
import rx.subject : SubjectObject;
import std.array : appender;

auto sub = new SubjectObject!Date;
auto buf = appender!(Date[]);

auto disposable = sub.distinctUntilChanged!"a.year == b.year".subscribe(buf);
scope (exit)
    disposable.dispose();

.put(sub, Date(2000, 1, 1));
.put(sub, Date(2000, 1, 2));
.put(sub, Date(2017, 3, 24));
.put(sub, Date(2017, 4, 24));
.put(sub, Date(2017, 4, 24));

auto data = buf.data;
assert(data.length == 2);
assert(data[0] == Date(2000, 1, 1));
assert(data[1] == Date(2017, 3, 24));