##// END OF EJS Templates
implab » implabjs-djx
RSS
added reduce() and next() methods to observable...
cin -
r116:aac297dda27d v1.6.0 default
parent child
Show More
Show file before | Show file after | Hide comments
@@ -0,0 +1,19
1 {
2 // Используйте IntelliSense, чтобы узнать о возможных атрибутах.
3 // Наведите указатель мыши, чтобы просмотреть описания существующих атрибутов.
4 // Для получения дополнительной информации посетите: https://go.microsoft.com/fwlink/?linkid=830387
5 "version": "0.2.0",
6 "configurations": [
7 {
8 "type": "node",
9 "request": "launch",
10 "name": "Launch tests",
11 "skipFiles": [
12 "<node_internals>/**"
13 ],
14 "program": "${workspaceFolder}/djx/build/test/index.js",
15 "cwd": "${workspaceFolder}/djx/build/test",
16 "console": "integratedTerminal"
17 }
18 ]
19 } No newline at end of file
Show file before | Show file after | Hide comments
@@ -0,0 +1,58
1 import { PromiseOrValue } from "@implab/core-amd/interfaces";
2 import { isPromise } from "@implab/core-amd/safe";
3 import { observe, Observable } from "./observable";
4
5 export interface OrderedUpdate<T> {
6 /** The item is being updated */
7 readonly item: T;
8
9 /** The previous index of the item, -1 in case it is inserted */
10 readonly prevIndex: number;
11
12 /** The new index of the item, -1 in case it is deleted */
13 readonly newIndex: number;
14
15 }
16
17 export type QueryResults<T> = Observable<OrderedUpdate<T>>;
18
19 interface DjObservableResults<T> {
20 /**
21 * Allows observation of results
22 */
23 observe(listener: (object: T, previousIndex: number, newIndex: number) => void, includeUpdates?: boolean): {
24 remove(): void;
25 };
26 }
27
28 interface Queryable<T, A extends unknown[]> {
29 query(...args: A): PromiseOrValue<T[]>;
30 }
31
32 export const isObservableResults = <T>(v: object): v is DjObservableResults<T> =>
33 v && (typeof (v as { observe?: unknown; }).observe === "function");
34
35 export const query = <T, A extends unknown[]>(store: Queryable<T, A>, includeUpdates = true) =>
36 (...args: A) => {
37 return observe<OrderedUpdate<T>>(({ next, complete, error, isClosed }) => {
38 try {
39 const results = store.query(...args);
40 if (isPromise(results)) {
41 results.then(items => items.forEach((item, newIndex) => next({ item, newIndex, prevIndex: -1 })))
42 .then(undefined, error);
43 } else {
44 results.forEach((item, newIndex) => next({ item, newIndex, prevIndex: -1 }));
45 }
46
47 if (!isClosed() && isObservableResults<T>(results)) {
48 const h = results.observe((item, prevIndex, newIndex) => next({ item, prevIndex, newIndex }), includeUpdates);
49 return () => h.remove();
50 } else {
51 complete();
52 }
53 } catch (err) {
54 error(err);
55 }
56 });
57
58 };
Show file before | Show file after | Hide comments
@@ -9,7 +9,7
9 9 "version": "0.0.1-dev",
10 10 "license": "BSD-2-Clause",
11 11 "devDependencies": {
12 "@implab/core-amd": "^1.4.0",
12 "@implab/core-amd": "^1.4.6",
13 13 "@implab/dojo-typings": "1.0.3",
14 14 "@types/chai": "4.1.3",
15 15 "@types/requirejs": "2.1.31",
@@ -31,7 +31,7
31 31 "yaml": "~1.7.2"
32 32 },
33 33 "peerDependencies": {
34 "@implab/core-amd": "^1.4.0",
34 "@implab/core-amd": "^1.4.6",
35 35 "dojo": "^1.10.0"
36 36 }
37 37 },
@@ -453,10 +453,13
453 453 "dev": true
454 454 },
455 455 "node_modules/@implab/core-amd": {
456 "version": "1.4.0",
457 "resolved": "https://registry.npmjs.org/@implab/core-amd/-/core-amd-1.4.0.tgz",
458 "integrity": "sha512-gaJX1mhri7YpmXDTAYELZnmTznzXYpk2AI7Decsttdi6xY+bqGgH24q0AFcKrx8RY2jfsFXxDdf0fITz2HpBbw==",
459 "dev": true
456 "version": "1.4.6",
457 "resolved": "https://registry.npmjs.org/@implab/core-amd/-/core-amd-1.4.6.tgz",
458 "integrity": "sha512-I1RwUAxeiodePpiBzveoHaehMSAyk7NFPPPEvDqfphHBC8yXoXWAaUrp7EcOKEzjXAs7lJQVhNpmjCjIqoj6BQ==",
459 "dev": true,
460 "peerDependencies": {
461 "dojo": "^1.10.0"
462 }
460 463 },
461 464 "node_modules/@implab/dojo-typings": {
462 465 "version": "1.0.3",
@@ -7040,10 +7043,11
7040 7043 "dev": true
7041 7044 },
7042 7045 "@implab/core-amd": {
7043 "version": "1.4.0",
7044 "resolved": "https://registry.npmjs.org/@implab/core-amd/-/core-amd-1.4.0.tgz",
7045 "integrity": "sha512-gaJX1mhri7YpmXDTAYELZnmTznzXYpk2AI7Decsttdi6xY+bqGgH24q0AFcKrx8RY2jfsFXxDdf0fITz2HpBbw==",
7046 "dev": true
7046 "version": "1.4.6",
7047 "resolved": "https://registry.npmjs.org/@implab/core-amd/-/core-amd-1.4.6.tgz",
7048 "integrity": "sha512-I1RwUAxeiodePpiBzveoHaehMSAyk7NFPPPEvDqfphHBC8yXoXWAaUrp7EcOKEzjXAs7lJQVhNpmjCjIqoj6BQ==",
7049 "dev": true,
7050 "requires": {}
7047 7051 },
7048 7052 "@implab/dojo-typings": {
7049 7053 "version": "1.0.3",
Show file before | Show file after | Hide comments
@@ -15,11 +15,11
15 15 "access": "public"
16 16 },
17 17 "peerDependencies": {
18 "@implab/core-amd": "^1.4.0",
18 "@implab/core-amd": "^1.4.6",
19 19 "dojo": "^1.10.0"
20 20 },
21 21 "devDependencies": {
22 "@implab/core-amd": "^1.4.0",
22 "@implab/core-amd": "^1.4.6",
23 23 "@types/chai": "4.1.3",
24 24 "@types/requirejs": "2.1.31",
25 25 "@types/yaml": "1.2.0",
Show file before | Show file after | Hide comments
@@ -1,5 +1,5
1 import { PromiseOrValue } from "@implab/core-amd/interfaces";
2 import { isPromise } from "@implab/core-amd/safe";
1 import { Cancellation } from "@implab/core-amd/Cancellation";
2 import { ICancellation } from "@implab/core-amd/interfaces";
3 3
4 4 /**
5 5 * The interface for the consumer of an observable sequence
@@ -29,23 +29,23 export type Sink<T> = {
29 29 /**
30 30 * Call to send the next element in the sequence
31 31 */
32 next: (value: T) => void;
32 next: (value: T) => void;
33
34 /**
35 * Call to notify about the error occurred in the sequence.
36 */
37 error: (e: unknown) => void;
33 38
34 /**
35 * Call to notify about the error occurred in the sequence.
36 */
37 error: (e: unknown) => void;
38
39 /**
40 * Call to signal the end of the sequence.
41 */
42 complete: () => void;
39 /**
40 * Call to signal the end of the sequence.
41 */
42 complete: () => void;
43 43
44 /**
45 * Checks whether the sink is accepting new elements. It's safe to
46 * send elements to the closed sink.
47 */
48 isClosed: () => boolean;
44 /**
45 * Checks whether the sink is accepting new elements. It's safe to
46 * send elements to the closed sink.
47 */
48 isClosed: () => boolean;
49 49 };
50 50
51 51 export type Producer<T> = (sink: Sink<T>) => (void | (() => void));
@@ -64,6 +64,8 export interface Subscribable<T> {
64 64 subscribe(consumer: Partial<Observer<T>>): Unsubscribable;
65 65 }
66 66
67 export type AccumulatorFn<T, A> = (acc: A, value: T) => A;
68
67 69 /** The observable source of items. */
68 70 export interface Observable<T> extends Subscribable<T> {
69 71 /** Transforms elements of the sequence with the specified mapper
@@ -85,11 +87,33 export interface Observable<T> extends S
85 87 * @param accumulator
86 88 * @param initial
87 89 */
88 scan<A>(accumulator: (acc: A, value: T) => A, initial: A): Observable<A>;
90 scan<A>(accumulator: AccumulatorFn<T, A>, initial: A): Observable<A>;
91 scan(accumulator: AccumulatorFn<T, T>): Observable<T>;
89 92
93 /** Applies accumulator to each value in the sequence and
94 * emits the accumulated value at the end of the sequence
95 *
96 * @param accumulator
97 * @param initial
98 */
99 reduce<A>(accumulator: AccumulatorFn<T, A>, initial: A): Observable<A>;
100 reduce(accumulator: AccumulatorFn<T, T>): Observable<T>;
101
102 /** Concatenates the specified sequences with this observable
103 *
104 * @param seq sequences to concatenate with the current observable
105 */
90 106 cat(...seq: Subscribable<T>[]): Observable<T>;
91 107
92 pipe<U>(f: (source: Observable<T>) => Producer<U>): Observable<U>;
108 /** Pipes the specified operator to produce the new observable
109 * @param op The operator which consumes this observable and produces a new one
110 */
111 pipe<U>(op: (source: Observable<T>) => Producer<U>): Observable<U>;
112
113 /** Waits for the next event to occur and returns a promise for the next value
114 * @param ct Cancellation token to
115 */
116 next(ct?: ICancellation): Promise<T>;
93 117 }
94 118
95 119 const noop = () => { };
@@ -134,24 +158,75 const _observe = <T>(producer: Producer<
134 158 subscribe: (consumer: Partial<Observer<T>>) => ({
135 159 unsubscribe: producer(sink(consumer)) ?? noop
136 160 }),
161
137 162 map: (mapper) => _observe(({ next, ...rest }) =>
138 163 producer({
139 164 next: next !== noop ? (v: T) => next(mapper(v)) : noop,
140 165 ...rest
141 166 })
142 167 ),
168
143 169 filter: (predicate) => _observe(({ next, ...rest }) =>
144 170 producer({
145 171 next: next !== noop ? (v: T) => predicate(v) ? next(v) : void (0) : noop,
146 172 ...rest
147 173 })
148 174 ),
149 scan: (accumulator, initial) => _observe(({ next, ...rest }) => {
150 let _acc = initial;
151 return producer({
152 next: next !== noop ? (v: T) => next(_acc = accumulator(_acc, v)) : noop,
153 ...rest
154 });
175
176 scan: <A>(...args: [AccumulatorFn<T, A>, A] | [AccumulatorFn<T, T>]) => _observe<T | A>(({ next, ...rest }) => {
177 if (args.length === 1) {
178 const [accumulator] = args;
179 let _acc: T;
180 let index = 0;
181 return producer({
182 next: next !== noop ? (v: T) => next(index++ === 0 ? _acc = v : _acc = accumulator(_acc, v)) : noop,
183 ...rest
184 });
185 } else {
186 const [accumulator, initial] = args;
187 let _acc = initial;
188 return producer({
189 next: next !== noop ? (v: T) => next(_acc = accumulator(_acc, v)) : noop,
190 ...rest
191 });
192 }
193 }),
194
195 reduce: <A>(...args: [AccumulatorFn<T, A>, A] | [AccumulatorFn<T, T>]) => _observe<T | A>(({ next, complete, error, ...rest }) => {
196 if (args.length === 1) {
197 const [accumulator] = args;
198 let _acc: T;
199 let index = 0;
200 return producer({
201 next: next !== noop ? (v: T) => {
202 _acc = index++ === 0 ? v : accumulator(_acc, v);
203 } : noop,
204 complete: () => {
205 if (index === 0) {
206 error(new Error("The sequence can't be empty"));
207 } else {
208 next(_acc);
209 complete();
210 }
211 },
212 error,
213 ...rest
214 });
215 } else {
216 const [accumulator, initial] = args;
217 let _acc = initial;
218 return producer({
219 next: next !== noop ? (v: T) => {
220 _acc = accumulator(_acc, v);
221 } : noop,
222 complete: () => {
223 next(_acc);
224 complete();
225 },
226 error,
227 ...rest
228 });
229 }
155 230 }),
156 231
157 232 cat: (...seq) => _observe(({ next, complete: final, ...rest }) => {
@@ -173,61 +248,59 const _observe = <T>(producer: Producer<
173 248 return () => cleanup();
174 249 }),
175 250
176 pipe: <U>(f: (source: Observable<T>) => Producer<U>) => observe(f(_observe(producer)))
177 });
178
179 export interface OrderUpdate<T> {
180 /** The item is being updated */
181 item: T;
251 pipe: <U>(op: (source: Observable<T>) => Producer<U>) => observe(op(_observe(producer))),
182 252
183 /** The previous index of the item, -1 in case it is inserted */
184 prevIndex: number;
185
186 /** The new index of the item, -1 in case it is deleted */
187 newIndex: number;
188 }
253 next: (ct?: ICancellation) => {
254 const _ct = ct ?? Cancellation.none;
255 return new Promise<T>((resolve, reject) => {
256 // wrap the producer to handle only single event
257 const once = fuse<T>(({ next, complete, error, isClosed }) => {
258 const h = _ct.register(error);
189 259
190 interface ObservableResults<T> {
191 /**
192 * Allows observation of results
193 */
194 observe(listener: (object: T, previousIndex: number, newIndex: number) => void, includeUpdates?: boolean): {
195 remove(): void;
196 };
197 }
260 // is the _ct fires it will call error() and isClosed() will return true
261 const cleanup = !isClosed() ?
262 producer({
263 next: v => (next(v), complete()),
264 complete: () => error(new Error("The sequence is empty")),
265 error,
266 isClosed
267 }) ?? noop :
268 noop;
198 269
199 interface Queryable<T, A extends unknown[]> {
200 query(...args: A): PromiseOrValue<T[]>;
201 }
270 return () => {
271 h.destroy();
272 cleanup();
273 };
274 });
202 275
203 export const isObservableResults = <T>(v: object): v is ObservableResults<T> =>
204 v && (typeof (v as { observe?: unknown; }).observe === "function");
276 once({
277 next: resolve,
278 error: reject,
279 complete: noop,
280 isClosed: () => false
281 });
282 });
283 }
284 });
205 285
206 286 export const observe = <T>(producer: Producer<T>) => _observe(fuse(producer));
207 287
208 export const empty = observe<never>(({ complete }) => complete());
288 export const streamArray = <T>(items: T[]) => _observe<T>(
289 ({ next, complete }) => (
290 items.forEach(next),
291 complete()
292 )
293 );
209 294
210 export const query = <T, A extends unknown[]>(store: Queryable<T, A>) =>
211 (...args: A) => {
212 return observe<OrderUpdate<T>>(({ next, complete, error }) => {
213 try {
214 const results = store.query(...args);
215 if (isPromise(results)) {
216 results.then(items => items.forEach((item, newIndex) => next({ item, newIndex, prevIndex: -1 })))
217 .then(undefined, error);
218 } else {
219 results.forEach((item, newIndex) => next({ item, newIndex, prevIndex: -1 }));
220 }
295 export const streamPromise = <T>(promise: PromiseLike<T>) => observe<T>(
296 ({next, error, complete}) => void promise.then(v => (next(v), complete()), error)
297 );
221 298
222 if (isObservableResults<T>(results)) {
223 const h = results.observe((item, prevIndex, newIndex) => next({ item, prevIndex, newIndex }));
224 return () => h.remove();
225 } else {
226 complete();
227 }
228 } catch (err) {
229 error(err);
230 }
231 });
299 export const of = <T>(...items: T[]) => _observe<T>(
300 ({ next, complete }) => (
301 items.forEach(next),
302 complete()
303 )
304 );
232 305
233 };
306 export const empty = _observe<never>(({ complete }) => complete()); No newline at end of file
Show file before | Show file after | Hide comments
@@ -7,10 +7,11 import Stateful = require("dojo/Stateful
7 7 import _WidgetBase = require("dijit/_WidgetBase");
8 8 import { DjxWidgetBase } from "./tsx/DjxWidgetBase";
9 9 import { WatchRendition } from "./tsx/WatchRendition";
10 import { Observable, observe, OrderUpdate, Subscribable } from "./observable";
10 import { Observable, observe, Subscribable } from "./observable";
11 11 import djAttr = require("dojo/dom-attr");
12 12 import djClass = require("dojo/dom-class");
13 13 import { AnimationAttrs, WatchForRendition } from "./tsx/WatchForRendition";
14 import { OrderedUpdate } from "./store";
14 15
15 16 export function createElement<T extends Constructor | string | ((props: object) => Element)>(elementType: T, ...args: unknown[]): Rendition {
16 17 if (typeof elementType === "string") {
@@ -101,7 +102,7 export function watch(
101 102 }
102 103 }
103 104
104 export const watchFor = <T>(source: T[] | Subscribable<OrderUpdate<T>>, render: (item: T, index: number) => unknown, opts: AnimationAttrs = {}) => {
105 export const watchFor = <T>(source: T[] | Subscribable<OrderedUpdate<T>>, render: (item: T, index: number) => unknown, opts: AnimationAttrs = {}) => {
105 106 return new WatchForRendition({
106 107 ...opts,
107 108 subject: source,
Show file before | Show file after | Hide comments
@@ -9,7 +9,8 import { collectNodes, destroy as safeDe
9 9 import { IDestroyable } from "@implab/core-amd/interfaces";
10 10 import { play } from "../play";
11 11 import * as fx from "dojo/fx";
12 import { isObservableResults, isSubsribable, OrderUpdate, Subscribable } from "../observable";
12 import { isSubsribable, Subscribable } from "../observable";
13 import { isObservableResults, OrderedUpdate } from "../store";
13 14
14 15 const trace = TraceSource.get(mid);
15 16
@@ -21,7 +22,7 interface ItemRendition {
21 22 destroy(): void;
22 23 }
23 24
24 interface RenderTask<T> extends OrderUpdate<T> {
25 interface RenderTask<T> extends OrderedUpdate<T> {
25 26 animate: boolean;
26 27 }
27 28
@@ -34,7 +35,7 export interface AnimationAttrs {
34 35 }
35 36
36 37 export interface WatchForRenditionAttrs<T> extends AnimationAttrs {
37 subject: T[] | Subscribable<OrderUpdate<T>>;
38 subject: T[] | Subscribable<OrderedUpdate<T>>;
38 39
39 40 component: (arg: T, index: number) => unknown;
40 41 }
@@ -60,7 +61,7 export class WatchForRendition<T> extend
60 61
61 62 private readonly _itemRenditions: ItemRendition[] = [];
62 63
63 private readonly _subject: T[] | Subscribable<OrderUpdate<T>>;
64 private readonly _subject: T[] | Subscribable<OrderedUpdate<T>>;
64 65
65 66 private readonly _renderTasks: RenderTask<T>[] = [];
66 67
@@ -97,7 +98,7 export class WatchForRendition<T> extend
97 98 const result = this._subject;
98 99
99 100 if (result) {
100 if (isSubsribable<OrderUpdate<T>>(result)) {
101 if (isSubsribable<OrderedUpdate<T>>(result)) {
101 102 let animate = false;
102 103 const subscription = result.subscribe({
103 104 next: ({ item, prevIndex, newIndex }) => this._onItemUpdated({ item, prevIndex, newIndex, animate })
Show file before | Show file after | Hide comments
@@ -1,5 +1,6
1 import { observe } from "./observable";
1 import { empty, observe } from "./observable";
2 2 import * as t from "tap";
3 import { Cancellation } from "@implab/core-amd/Cancellation";
3 4
4 5 const subj1 = observe<number>(({ next, complete }) => {
5 6 next(1);
@@ -54,29 +55,73 t.ok(consumer2.completed, "The completio
54 55 const subj2 = observe<number>(({ next, complete }) => {
55 56 [1, 2, 3, 4, 5].forEach(next);
56 57 complete();
57 }).pipe<string>(self => ({ next, complete, error }) => {
58 t.comment("subj2: subscribe");
58 return () => {
59 t.comment("subj2: unsubscribe");
60 };
61 });
62
63 const consumer3 = {
64 even: 0,
65 odd: 0,
66 completed: false,
67 subscribed: 0,
68 unsubscribed: 0,
69 next(v: "even" | "odd") {
70 this[v] ++;
71 },
72 complete() {
73 this.completed = true;
74 },
75 subscribe() {
76 this.subscribed++;
77 },
78 unsubscribe() {
79 this.unsubscribed++;
80 }
81 };
82
83
84 const subj3 = subj2.pipe<"even" | "odd">(self => ({ next, complete, error }) => {
85 consumer3.subscribe();
86 let count = 0;
59 87 const h = self.subscribe({
60 88 next: val => {
61 89 if (val % 2 === 0)
62 90 next("odd");
63 91 else
64 92 next("even");
93 if (++count === 4)
94 complete();
65 95 },
66 96 complete,
67 97 error
68 98 });
69 99 return () =>{
70 t.comment("subj2: unsubscribe");
100 consumer3.unsubscribe();
71 101 h.unsubscribe();
72 102 };
73 103 });
74 104
75 subj2.subscribe({
76 next: val => t.comment("subj2: ", val),
105 subj3.subscribe(consumer3);
106
107 t.equal(consumer3.odd, 2, "Should get 2 odd elements");
108 t.equal(consumer3.even, 2, "Should get 2 even elements");
109 t.ok(consumer3.completed, "The sequence should completed");
110 t.equal(consumer3.subscribed, 1, "The subscription should be done once");
111 t.equal(consumer3.unsubscribed, 1, "The cleanup should be done after completion");
112
113 subj2.reduce((a,b) => a + b).subscribe({
114 next: val => t.comment("subj2: reduce =", val),
77 115 complete: () => t.comment("subj2: complete")
78 116 });
79 subj2.subscribe({
80 next: val => t.comment("subj2: ", val),
81 complete: () => t.comment("subj2: complete")
82 }); No newline at end of file
117
118 subj2.reduce((a,b) => a + b).next()
119 .then(value => t.comment("subj2: next reduce=", value))
120 .catch(() => {});
121
122 subj2.next().then(val => t.equal(val, 1, "Should peek the first element")).catch(() => {});
123
124 const cancelled = new Cancellation(cancel => cancel());
125 t.rejects(subj2.next(cancelled), "Cancelled next() method should fail").catch(() => {});
126
127 t.rejects(empty.next(), "Empty sequence should fail to get next element").catch(() => {}); No newline at end of file
Show file before | Show file after | Hide comments
@@ -9,6 +9,7
9 9 "target": "ES5",
10 10 "jsx": "react",
11 11 "lib": ["es5", "es2015.promise", "es2015.symbol", "es2015.iterable", "dom", "scripthost"],
12 "noUnusedLocals": true
12 "noUnusedLocals": true,
13 "downlevelIteration": true
13 14 }
14 15 } No newline at end of file
Show file before | Show file after | Hide comments
@@ -33,7 +33,7
33 33 "dev": true,
34 34 "license": "BSD-2-Clause",
35 35 "peerDependencies": {
36 "@implab/core-amd": "^1.4.0",
36 "@implab/core-amd": "^1.4.6",
37 37 "dojo": "^1.10.0"
38 38 }
39 39 },
Show file before | Show file after | Hide comments
@@ -1,4 +1,5
1 1 import { Contact } from "./Contact";
2 import { QueryResults } from "@implab/djx/store";
2 3
3 4 export type AppointmentRole = "organizer" | "speaker" | "participant";
4 5
@@ -16,5 +17,5 export interface Appointment {
16 17 /** Duration in minutes */
17 18 duration: number;
18 19
19 getMembers(role?: AppointmentRole): Member[];
20 getMembers(role?: AppointmentRole): QueryResults<Member>;
20 21 } No newline at end of file
Show file before | Show file after | Hide comments
@@ -3,9 +3,9 import Observable = require("dojo/store/
3 3 import { Appointment, AppointmentRole, Member } from "./Appointment";
4 4 import { Contact } from "./Contact";
5 5 import { Uuid } from "@implab/core-amd/Uuid";
6 import { query } from "@implab/djx/observable";
7 6 import { IDestroyable } from "@implab/core-amd/interfaces";
8 7 import { delay } from "@implab/core-amd/safe";
8 import { query } from "@implab/djx/store";
9 9
10 10 type AppointmentRecord = Omit<Appointment, "getMembers"> & { id: string };
11 11
@@ -41,8 +41,12 export class MainContext implements IDes
41 41 );
42 42 }
43 43
44 private readonly _queryAppointmentsRx = query(this._appointments);
45
46 private readonly _queryMembersRx = query(this._members);
47
44 48 queryAppointments({ dateFrom, dateTo }: { dateFrom?: Date; dateTo?: Date; } = {}) {
45 return query(this._appointments)(({ startAt }) =>
49 return this._queryAppointmentsRx(({ startAt }) =>
46 50 (!dateFrom || dateFrom <= startAt) &&
47 51 (!dateTo || startAt <= dateTo)
48 52 ).map(item(this._mapAppointment));
@@ -61,7 +65,7 export class MainContext implements IDes
61 65 title,
62 66 startAt,
63 67 duration,
64 getMembers: (role?: AppointmentRole) => this._members.query(role ? { appointmentId: id, role } : { appointmentId: id })
68 getMembers: (role?: AppointmentRole) => this._queryMembersRx(role ? { appointmentId: id, role } : { appointmentId: id })
65 69 });
66 70
67 71 destroy() {
Show file before | Show file after | Hide comments
@@ -1,7 +1,8
1 1 import { id as mid } from "module";
2 2 import { BehaviorSubject, Observer, Unsubscribable } from "rxjs";
3 3 import { IDestroyable } from "@implab/core-amd/interfaces";
4 import { OrderUpdate, Observable } from "@implab/djx/observable";
4 import { Observable } from "@implab/djx/observable";
5 import { OrderedUpdate } from "@implab/djx/store";
5 6 import { Appointment, Member } from "./Appointment";
6 7 import { MainContext } from "./MainContext";
7 8 import { LocalDate } from "@js-joda/core";
@@ -11,7 +12,7 import { TraceSource } from "@implab/cor
11 12 const trace = TraceSource.get(mid);
12 13
13 14 export interface State {
14 appointments: Observable<OrderUpdate<Appointment>>;
15 appointments: Observable<OrderedUpdate<Appointment>>;
15 16
16 17 dateTo: LocalDate;
17 18
Show file before | Show file after | Hide comments
@@ -2,7 +2,8 import { djbase, djclass } from "@implab
2 2 import { DjxWidgetBase } from "@implab/djx/tsx/DjxWidgetBase";
3 3 import { bind, createElement, prop, watch, watchFor } from "@implab/djx/tsx";
4 4 import MainModel from "../model/MainModel";
5 import { OrderUpdate, Observable } from "@implab/djx/observable";
5 import { Observable } from "@implab/djx/observable";
6 import { OrderedUpdate } from "@implab/djx/store";
6 7 import { Appointment } from "../model/Appointment";
7 8 import { LocalDate } from "@js-joda/core";
8 9 import Button = require("dijit/form/Button");
@@ -10,7 +11,7 import Button = require("dijit/form/Butt
10 11 @djclass
11 12 export default class MainWidget extends djbase(DjxWidgetBase) {
12 13
13 appointments?: Observable<OrderUpdate<Appointment>>;
14 appointments?: Observable<OrderedUpdate<Appointment>>;
14 15
15 16 model: MainModel;
16 17
General Comments 0
You need to be logged in to leave comments. Login now