debounce

1 import rx.subject : SubjectObject;
2 import core.thread : Thread;
3 import core.time : dur;
4 
5 auto obs = new SubjectObject!int;
6 
7 import std.array : appender;
8 
9 auto buf = appender!(int[]);
10 auto d = obs.debounce(dur!"msecs"(100), new TaskPoolScheduler).doSubscribe(buf);
11 scope (exit)
12     d.dispose();
13 
14 .put(obs, 1);
15 Thread.sleep(dur!"msecs"(200));
16 .put(obs, 2);
17 .put(obs, 3);
18 Thread.sleep(dur!"msecs"(200));
19 
20 assert(buf.data.length == 2);
21 assert(buf.data[0] == 1);
22 assert(buf.data[1] == 3);