scan

1 import rx.subject : SubjectObject;
2 
3 auto subject = new SubjectObject!int;
4 
5 auto sum = subject.scan!((a, b) => a + b)(0);
6 static assert(isObservable!(typeof(sum), int));
7 
8 import std.array : appender;
9 
10 auto buf = appender!(int[]);
11 auto disposable = sum.subscribe(buf);
12 scope (exit)
13     disposable.dispose();
14 
15 foreach (_; 0 .. 5)
16 {
17     subject.put(1);
18 }
19 
20 auto result = buf.data;
21 assert(result.length == 5);
22 import std.algorithm : equal;
23 
24 assert(equal(result, [1, 2, 3, 4, 5]));