observable-tests.ts
193 lines
| 5.1 KiB
| video/mp2t
|
TypeScriptLexer
cin
|
r129 | import { empty, observe, of } from "./observable"; | |
| import * as tap from "tap"; | |||
|
cin
|
r116 | import { Cancellation } from "@implab/core-amd/Cancellation"; | |
|
cin
|
r129 | import { delay } from "@implab/core-amd/safe"; | |
|
cin
|
r102 | ||
|
cin
|
r157 | const subj1 = observe<number>(({ next, error, complete }) => { | |
|
cin
|
r102 | next(1); | |
| complete(); | |||
| next(2); | |||
|
cin
|
r157 | next(3); | |
| error(new Error("This error should be ignored")); | |||
| next(4); | |||
|
cin
|
r102 | }); | |
| const consumer1 = { | |||
| sum: 0, | |||
| next(v: number) { | |||
| this.sum += v; | |||
| } | |||
|
cin
|
r108 | }; | |
|
cin
|
r102 | ||
| subj1.subscribe(consumer1); | |||
|
cin
|
r129 | tap.equal(consumer1.sum, 1, "Should get only one value"); | |
|
cin
|
r102 | ||
| subj1.subscribe(consumer1); | |||
|
cin
|
r129 | tap.equal(consumer1.sum, 2, "Should get the value again"); | |
|
cin
|
r102 | ||
| const consumer2 = { | |||
| value: 0, | |||
| completed: false, | |||
| next(v: number) { this.value = v; }, | |||
| complete() { this.completed = true; } | |||
| }; | |||
| let maps = 0; | |||
| subj1 | |||
| .map(v => { | |||
|
cin
|
r129 | tap.comment(`map1: ${v * 2}`); | |
|
cin
|
r102 | maps++; | |
| return v * 2; | |||
| }) | |||
|
cin
|
r114 | .map(v => { | |
|
cin
|
r129 | tap.comment(`map2: ${v * 2}`); | |
|
cin
|
r102 | maps++; | |
| return v * 2; | |||
| }) | |||
| .map(v => { | |||
|
cin
|
r129 | tap.comment(`map3: ${v * 2}`); | |
|
cin
|
r102 | maps++; | |
|
cin
|
r108 | return v * 2; | |
|
cin
|
r102 | }) | |
| .subscribe(consumer2); | |||
|
cin
|
r129 | tap.equal(consumer2.value, 8, "Should map"); | |
| tap.equal(maps, 3, "The map chain should not be executed after completion"); | |||
| tap.ok(consumer2.completed, "The completion signal should pass through"); | |||
|
cin
|
r114 | ||
| const subj2 = observe<number>(({ next, complete }) => { | |||
| [1, 2, 3, 4, 5].forEach(next); | |||
| complete(); | |||
|
cin
|
r116 | return () => { | |
|
cin
|
r129 | tap.comment("subj2: unsubscribe"); | |
|
cin
|
r116 | }; | |
| }); | |||
| const consumer3 = { | |||
| even: 0, | |||
| odd: 0, | |||
| completed: false, | |||
| subscribed: 0, | |||
| unsubscribed: 0, | |||
| next(v: "even" | "odd") { | |||
|
cin
|
r129 | this[v]++; | |
|
cin
|
r116 | }, | |
| complete() { | |||
| this.completed = true; | |||
| }, | |||
| subscribe() { | |||
| this.subscribed++; | |||
| }, | |||
| unsubscribe() { | |||
| this.unsubscribed++; | |||
| } | |||
| }; | |||
|
cin
|
r136 | const subj3 = subj2.pipe<"even" | "odd">(self => observe(({ next, complete, error }) => { | |
|
cin
|
r116 | consumer3.subscribe(); | |
| let count = 0; | |||
|
cin
|
r114 | const h = self.subscribe({ | |
| next: val => { | |||
| if (val % 2 === 0) | |||
| next("odd"); | |||
| else | |||
| next("even"); | |||
|
cin
|
r116 | if (++count === 4) | |
| complete(); | |||
|
cin
|
r114 | }, | |
| complete, | |||
| error | |||
| }); | |||
|
cin
|
r129 | return () => { | |
|
cin
|
r116 | consumer3.unsubscribe(); | |
|
cin
|
r114 | h.unsubscribe(); | |
| }; | |||
|
cin
|
r136 | })); | |
|
cin
|
r114 | ||
|
cin
|
r116 | subj3.subscribe(consumer3); | |
|
cin
|
r129 | tap.equal(consumer3.odd, 2, "Should get 2 odd elements"); | |
| tap.equal(consumer3.even, 2, "Should get 2 even elements"); | |||
| tap.ok(consumer3.completed, "The sequence should completed"); | |||
| tap.equal(consumer3.subscribed, 1, "The subscription should be done once"); | |||
| tap.equal(consumer3.unsubscribed, 1, "The cleanup should be done after completion"); | |||
|
cin
|
r116 | ||
|
cin
|
r129 | subj2.reduce((a, b) => a + b).subscribe({ | |
| next: val => tap.comment("subj2: reduce =", val), | |||
| complete: () => tap.comment("subj2: complete") | |||
|
cin
|
r114 | }); | |
|
cin
|
r116 | ||
|
cin
|
r129 | tap.test("of(...) tests", async t => { | |
| await subj2.reduce((a, b) => a + b).next() | |||
| .then(value => t.comment("subj2: next reduce=", value)); | |||
| await subj2.next().then(val => t.equal(val, 1, "Should peek the first element")); | |||
| const cancelled = new Cancellation(cancel => cancel()); | |||
| await t.rejects(subj2.next(cancelled), "Cancelled next() method should fail"); | |||
| await t.rejects(empty.next(), "Empty sequence should fail to get next element"); | |||
| await of(delay(1).then(() => 1), Promise.resolve(2), 3) | |||
| .reduce<number[]>((a, x) => [...a, x], []) | |||
| .next() | |||
| .then(res => t.same(res, [1, 2, 3], "of(...) should keep the order")); | |||
|
cin
|
r116 | ||
|
cin
|
r129 | const rejected = Promise.reject("DIE!"); | |
| rejected.catch(() => { }); // SAFE AND SOUND | |||
|
cin
|
r116 | ||
|
cin
|
r129 | await t.resolves( | |
| of(Promise.resolve(1), rejected).next(), | |||
| "of(...) should emit non-rejected items" | |||
| ); | |||
| await t.rejects( | |||
| of(1, Promise.reject("DIE!")).reduce((a) => a).next(), | |||
| "of(...) should terminate with error when a parameter is rejected" | |||
| ); | |||
|
cin
|
r116 | ||
|
cin
|
r152 | t.same(await of(1, 2, 3).collect(), [1, 2, 3], ".collect() should return the collected sequence"); | |
| await t.rejects(of(1, 2, 3).collect(cancelled), ".collect() should support cancellation"); | |||
|
cin
|
r129 | ||
|
cin
|
r144 | }).catch(() => { }); | |
| tap.test(".tap() tests", async t => { | |||
| const side: number[] = []; | |||
|
cin
|
r152 | of(1, 2) | |
| .tap({ next: v => side.push(v), complete: () => side.push(0) }) | |||
| .tap({ next: v => side.push(v * v) }) | |||
|
cin
|
r144 | .subscribe({}); | |
|
cin
|
r152 | t.same(side, [1, 1, 2, 4, 0], ".tap() should be called in the order of registration"); | |
|
cin
|
r144 | ||
| side.length = 0; | |||
| await new Promise<void>(resolve => { | |||
|
cin
|
r152 | of(1, 2, delay(1).then(() => 3)) | |
| .tap({ next: v => side.push(v) }) | |||
| .tap({ next: v => v === 1 && resolve() }) | |||
|
cin
|
r144 | .subscribe({}); | |
| }); | |||
|
cin
|
r152 | t.same(side, [1, 2], ".tap() should be processed synchronously"); | |
| }).catch(() => { }); | |||
| tap.test(".while() tests", async t => { | |||
| const seq = of(1, 2, 3, 4).while(v => v <= 2); | |||
| t.same(await seq.collect(), [1, 2], "Should collect only taken elements"); | |||
|
cin
|
r144 | ||
|
cin
|
r152 | const data: number[] = []; | |
| let complete = 0; | |||
| seq.subscribe({ | |||
| next: v => data.push(v), | |||
| complete: () => complete++ | |||
| }); | |||
| t.same(data, [1, 2], "Should receive only taken elements"); | |||
| t.equal(complete, 1, "Complete should run once"); | |||
| }).catch(() => { }); |