@@ -54,3 +54,4 894b8239b953c0384cf32cab46cf41dac97ea03b | |||||
54 | 63215d91ae4b711eeb2145b9685240e1afada904 v1.9.0-rc6 |
|
54 | 63215d91ae4b711eeb2145b9685240e1afada904 v1.9.0-rc6 | |
55 | af4f8424e83d56e89a64f39e19514ca10dbd43c6 v1.9.0 |
|
55 | af4f8424e83d56e89a64f39e19514ca10dbd43c6 v1.9.0 | |
56 | 63f3ad8e6cffcf32fc5a555e55e8544227293d4c v1.10.0 |
|
56 | 63f3ad8e6cffcf32fc5a555e55e8544227293d4c v1.10.0 | |
|
57 | d9c99ae7dec87ef47aded05def6a094a892df0e6 v1.10.1 |
@@ -42,19 +42,32 export const query = <T, Q, O>(store: Qu | |||||
42 | }; |
|
42 | }; | |
43 | }; |
|
43 | }; | |
44 |
|
44 | |||
|
45 | /** | |||
|
46 | * Wraps the query method of the store, the resulting method takes a query | |||
|
47 | * expression and returns two observable sequences. The first sequence represents | |||
|
48 | * the results of the query, the second sequence provides the updates to the | |||
|
49 | * query results. | |||
|
50 | * | |||
|
51 | * @param store The store used to query data | |||
|
52 | * @param includeUpdates The flag to include item updates not only additions and | |||
|
53 | * deletions. By default this flag is set to true. | |||
|
54 | * @returns Two observable sequences | |||
|
55 | */ | |||
45 | export const queryEx = <T, Q, O>(store: Queryable<T, Q, O>, includeUpdates = true) => |
|
56 | export const queryEx = <T, Q, O>(store: Queryable<T, Q, O>, includeUpdates = true) => | |
46 | (query?: Q, options?: O): [data: QueryResults<T>, updates: QueryResults<T>] => { |
|
57 | (query?: Q, options?: O): [data: QueryResults<T>, updates: QueryResults<T>] => { | |
47 |
|
58 | |||
48 | const pending: T[] = []; |
|
59 | /** count active observers */ | |
49 |
|
60 | let listeners = 0; | ||
50 |
let results: PromiseOrValue<T[]> = |
|
61 | let results: PromiseOrValue<T[]> = []; | |
51 |
|
62 | |||
52 | const data = observe<OrderedUpdate<T>>(({ next, complete, error }) => { |
|
63 | const data = observe<OrderedUpdate<T>>(({ next, complete, error }) => { | |
53 | const processResults = (items: T[]) => |
|
64 | const processResults = (items: T[]) => | |
54 | items.forEach((item, newIndex) => next({ item, newIndex, prevIndex: -1 })); |
|
65 | items.forEach((item, newIndex) => next({ item, newIndex, prevIndex: -1 })); | |
55 |
|
66 | |||
56 | try { |
|
67 | try { | |
57 | if (results === pending) |
|
68 | // is there are no active observers here, we need to query actual | |
|
69 | // data from the store. | |||
|
70 | if (listeners === 0) | |||
58 | results = store.query(query, options); |
|
71 | results = store.query(query, options); | |
59 |
|
72 | |||
60 | if (isPromise(results)) { |
|
73 | if (isPromise(results)) { | |
@@ -74,8 +87,14 export const queryEx = <T, Q, O>(store: | |||||
74 | const updates = observe<OrderedUpdate<T>>(({ next, complete, error, isClosed }) => { |
|
87 | const updates = observe<OrderedUpdate<T>>(({ next, complete, error, isClosed }) => { | |
75 | try { |
|
88 | try { | |
76 | if (!isClosed() && isDjObservableResults<T>(results)) { |
|
89 | if (!isClosed() && isDjObservableResults<T>(results)) { | |
|
90 | // subscribe fot the changes | |||
|
91 | listeners++; | |||
77 | const h = results.observe((item, prevIndex, newIndex) => next({ item, prevIndex, newIndex }), includeUpdates); |
|
92 | const h = results.observe((item, prevIndex, newIndex) => next({ item, prevIndex, newIndex }), includeUpdates); | |
78 |
return () => |
|
93 | return () => { | |
|
94 | // unsubscribe from changes | |||
|
95 | listeners--; | |||
|
96 | h.remove(); | |||
|
97 | }; | |||
79 | } else { |
|
98 | } else { | |
80 | complete(); |
|
99 | complete(); | |
81 | } |
|
100 | } |
@@ -144,30 +144,48 tap.test("of(...) tests", async t => { | |||||
144 | "of(...) should terminate with error when a parameter is rejected" |
|
144 | "of(...) should terminate with error when a parameter is rejected" | |
145 | ); |
|
145 | ); | |
146 |
|
146 | |||
147 | t.same(await of(1,2,3).collect(), [1,2,3], ".collect() should return the collected sequence"); |
|
147 | t.same(await of(1, 2, 3).collect(), [1, 2, 3], ".collect() should return the collected sequence"); | |
148 | await t.rejects(of(1,2,3).collect(cancelled), ".collect() should support cancellation"); |
|
148 | await t.rejects(of(1, 2, 3).collect(cancelled), ".collect() should support cancellation"); | |
149 |
|
149 | |||
150 | }).catch(() => { }); |
|
150 | }).catch(() => { }); | |
151 |
|
151 | |||
152 | tap.test(".tap() tests", async t => { |
|
152 | tap.test(".tap() tests", async t => { | |
153 | const side: number[] = []; |
|
153 | const side: number[] = []; | |
154 |
|
154 | |||
155 | of(1,2) |
|
155 | of(1, 2) | |
156 | .tap({next: v => side.push(v), complete: () => side.push(0)}) |
|
156 | .tap({ next: v => side.push(v), complete: () => side.push(0) }) | |
157 | .tap({next: v => side.push(v*v)}) |
|
157 | .tap({ next: v => side.push(v * v) }) | |
158 | .subscribe({}); |
|
158 | .subscribe({}); | |
159 |
|
159 | |||
160 | t.same(side, [1,1,2,4,0], ".tap() should be called in the order of registration"); |
|
160 | t.same(side, [1, 1, 2, 4, 0], ".tap() should be called in the order of registration"); | |
161 |
|
161 | |||
162 | side.length = 0; |
|
162 | side.length = 0; | |
163 |
|
163 | |||
164 | await new Promise<void>(resolve => { |
|
164 | await new Promise<void>(resolve => { | |
165 | of(1,2,delay(1).then(() => 3)) |
|
165 | of(1, 2, delay(1).then(() => 3)) | |
166 | .tap({next: v => side.push(v)}) |
|
166 | .tap({ next: v => side.push(v) }) | |
167 | .tap({ next: v => v === 1 && resolve()}) |
|
167 | .tap({ next: v => v === 1 && resolve() }) | |
168 | .subscribe({}); |
|
168 | .subscribe({}); | |
169 | }); |
|
169 | }); | |
170 |
|
170 | |||
171 | t.same(side, [1,2], ".tap() should be processed synchronously"); |
|
171 | t.same(side, [1, 2], ".tap() should be processed synchronously"); | |
|
172 | ||||
|
173 | }).catch(() => { }); | |||
|
174 | ||||
|
175 | tap.test(".while() tests", async t => { | |||
|
176 | ||||
|
177 | const seq = of(1, 2, 3, 4).while(v => v <= 2); | |||
|
178 | ||||
|
179 | t.same(await seq.collect(), [1, 2], "Should collect only taken elements"); | |||
172 |
|
180 | |||
173 | }).catch(() => {}); No newline at end of file |
|
181 | const data: number[] = []; | |
|
182 | let complete = 0; | |||
|
183 | seq.subscribe({ | |||
|
184 | next: v => data.push(v), | |||
|
185 | complete: () => complete++ | |||
|
186 | }); | |||
|
187 | ||||
|
188 | t.same(data, [1, 2], "Should receive only taken elements"); | |||
|
189 | t.equal(complete, 1, "Complete should run once"); | |||
|
190 | ||||
|
191 | }).catch(() => { }); No newline at end of file |
General Comments 0
You need to be logged in to leave comments.
Login now