implab/implabjs-djx Commit - r148:63f3ad8e6cff

Merge

cin -

r148:63f3ad8e6cff v1.10.0 default

parent child

Context file:

r148:63f3ad8e6cff

Collapse all files

.hgtags +1 0

 b2c550c676ff6acf16f5bd56ee3804342f80b7 v1.9.0-rc3
 d1b83ebdfaa93599451175055a94f711c079f v1.9.0-rc4
 b8239b953c0384cf32cab46cf41dac97ea03b v1.9.0-rc5
+d91ae4b711eeb2145b9685240e1afada904 v1.9.0-rc6
             af4f8424e83d56e89a64f39e19514ca10dbd43c6 v1.9.0

djx/src/main/ts/observable.ts +20 -2

                 v !== null && v !== undefined && typeof (v as Subscribable<unknown>).subscribe === "function";
             export interface Subscribable<T> {
-                subscribe(consumer: Observer<T>): Unsubscribable;
+                /** Subscribes a consumer to events. If a consumer isn't specified
+                 * this method activates the producer to achieve side affects if any.
+                 */
+                subscribe(consumer?: Observer<T>): Unsubscribable;
             }
             export type AccumulatorFn<T, A> = (acc: A, value: T) => A;
                  */
                 map<T2>(mapper: (value: T) => T2): Observable<T2>;
+                /** Injects the specified observer into the each producer to consumer chain.
+                 * The method is used to add side effect to the events processing.
+                 *
+                 * @param observer The consumer for the events
+                 */
+                tap(observer: Observer<T>): Observable<T>;
                 /** Filters elements of the sequence. The resulting sequence will
                  * contain only elements which match the specified predicate.
                  *
             };
             const _observe = <T>(producer: FusedProducer<T>): Observable<T> => ({
-                subscribe: (consumer: Observer<T>) => ({
+                subscribe: (consumer: Observer<T> = {}) => ({
                     unsubscribe: producer(sink(consumer)) ?? noop
                 }),
                     })
                 ),
+                tap: ({next: tapNext, complete: tapComplete, error: tapError}) => _observe(({next,complete, error}) =>
+                    producer({
+                        next: tapNext ? (v => (tapNext(v), next(v))) : next,
+                        complete: tapComplete ? (() => (tapComplete(), complete())): complete,
+                        error: tapError ? (e => (tapError(e), error(e))) : error
+                    })
+                ),
                 filter: (predicate) => _observe(({ next, ...rest }) =>
                     producer({
                         next: next !== noop ? (v: T) => predicate(v) ? next(v) : void (0) : noop,

djx/src/main/ts/store.ts +39 -12

             export const isDjObservableResults = <T>(v: object): v is DjObservableResults<T> =>
                 v && (typeof (v as { observe?: unknown; }).observe === "function");
-            export const query = <T, Q, O>(store: Queryable<T, Q, O>, includeUpdates = true) =>
+            export const query = <T, Q, O>(store: Queryable<T, Q, O>, includeUpdates = true) => {
-                (query?: Q, options?: O & { observe?: boolean }) => {
+                const q = queryEx(store, includeUpdates);
-                    return observe<OrderedUpdate<T>>(({ next, complete, error, isClosed }) => {
+                return (query?: Q, options?: O & { observe?: boolean }) => {
+                    const [data, updates] = q(query, options);
+                    return options?.observe === false ? data : data.cat(updates);
+                };
+            };
+            export const queryEx = <T, Q, O>(store: Queryable<T, Q, O>, includeUpdates = true) =>
+                (query?: Q, options?: O): [data: QueryResults<T>, updates: QueryResults<T>] => {
+                    const pending: T[] = [];
+                    let results: PromiseOrValue<T[]> = pending;
+                    const data = observe<OrderedUpdate<T>>(({ next, complete, error }) => {
                         const processResults = (items: T[]) =>
                             items.forEach((item, newIndex) => next({ item, newIndex, prevIndex: -1 }));
                         try {
-                            const results = store.query(query, options);
+                            if (results === pending)
+                                results = store.query(query, options);
                             if (isPromise(results)) {
-                                results.then(processResults).then(undefined, error);
+                                results.then(processResults).then(complete, error);
+                                if (isCancellable(results))
+                                    return results.cancel.bind(results);
                             } else {
                                 processResults(results);
+                                complete();
                             }
+                        } catch (e) {
+                            error(e);
+                        }
+                    });
-                            if (!isClosed() && (options?.observe !== false) && isDjObservableResults<T>(results)) {
+                    const updates = observe<OrderedUpdate<T>>(({ next, complete, error, isClosed }) => {
+                        try {
+                            if (!isClosed() && isDjObservableResults<T>(results)) {
                                 const h = results.observe((item, prevIndex, newIndex) => next({ item, prevIndex, newIndex }), includeUpdates);
                                 return () => h.remove();
                             } else {
                                 complete();
                             }
-                        } catch (err) {
+                        } catch (e) {
-                            error(err);
+                            error(e);
                         }
                     });
+                    return [data, updates];
                 };
             interface IndexedStore<T> {
                 get(id: string | number): PromiseLike<T> | T | null | undefined;
             }
                 observe?: boolean;
             }
-            type Change<T> = [item: NonNullable<T>, id: string | number | undefined] |
+            export type ItemUpdate<T> = [item: NonNullable<T>, id: string | number | undefined] |
             [item: undefined | null, id: string | number];
             const filterItem = (itemId: string | number) =>
-                <T>(source: Observable<Change<T>>) =>
+                <T>(source: Observable<ItemUpdate<T>>) =>
                     observe<T>(({ next, complete, error }) => {
                         const subscription = source
                             .filter(([, id]) => id === itemId)
             export const get = <T>(store: IndexedStore<T>) => {
                 const changes = hasNotifications<T>(store) ?
-                    observe<Change<T>>(({ next }) => {
+                    observe<ItemUpdate<T>>(({ next }) => {
-                        const handle = after(store, "notify", (...args: Change<T>) => next(args), true);
+                        const handle = after(store, "notify", (...args: ItemUpdate<T>) => next(args), true);
                         return () => handle.remove();
                     }) : empty;

djx/src/test/ts/observable-store-tests.ts +25 -1

@@ -1,6 +1,6
1	import Memory = require("dojo/store/Memory");	1	import Memory = require("dojo/store/Memory");
2	import Observerable = require("dojo/store/Observable");	2	import Observerable = require("dojo/store/Observable");
3	import { get } from "./store";	3	import { get, queryEx } from "./store";
4	import tap = require("tap");	4	import tap = require("tap");
5		5
6	interface Person {	6	interface Person {
@@ -61,5 +61,29 tap.test("store::get(...) tests", async
61	samSubscription.unsubscribe();	61	samSubscription.unsubscribe();
62	}	62	}
63		63
		64	store.put({ id: samId, name: "Sam", age: 29});
		65
		66	const [data, updates] = queryEx(store)({ age: 29}, { sort: [{attribute: "id"}] });
		67
		68	const dump: string[] = [];
		69
		70	const subscription = data
		71	.tap({
		72	complete: () => dump.push("eof")
		73	})
		74	.cat(updates)
		75	.tap({
		76	next: ({item: {id}}) => dump.push(id),
		77	complete: () => dump.push("eof")
		78	})
		79	.subscribe({});
		80
		81	t.same(dump, ["id:peter", "id:sam", "eof"]);
		82
		83	store.put({ id: "id:mary", name: "Mary", age: 29});
		84
		85	t.same(dump, ["id:peter", "id:sam", "eof", "id:mary"]);
		86
		87	subscription.unsubscribe();
64		88
65	}).catch(() => { }); No newline at end of file	89	}).catch(() => { });

djx/src/test/ts/observable-tests.ts +24 -1

@@ -147,4 +147,27 tap.test("of(...) tests", async t => {
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(() => { }); No newline at end of file	150	}).catch(() => { });
		151
		152	tap.test(".tap() tests", async t => {
		153	const side: number[] = [];
		154
		155	of(1,2)
		156	.tap({next: v => side.push(v), complete: () => side.push(0)})
		157	.tap({next: v => side.push(v*v)})
		158	.subscribe({});
		159
		160	t.same(side, [1,1,2,4,0], ".tap() should be called in the order of registration");
		161
		162	side.length = 0;
		163
		164	await new Promise<void>(resolve => {
		165	of(1,2,delay(1).then(() => 3))
		166	.tap({next: v => side.push(v)})
		167	.tap({ next: v => v === 1 && resolve()})
		168	.subscribe({});
		169	});
		170
		171	t.same(side, [1,2], ".tap() should be processed synchronously");
		172
		173	}).catch(() => {}); No newline at end of file

General Comments 0

Write
Preview

You need to be logged in to leave comments. Login now

No TODOs yet

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages