implab/implabjs-djx Commit - r129:66546e709732

added observable.collect() method to collect a sequnce to the array...

cin -

r129:66546e709732 v1.8.0 default

parent child

Context file:

r129:66546e709732

Collapse all files

djx/readme.md +37 -6

             ## SYNOPSIS
-            ```tsx
+            ```jsx
             import { djbase, djclass, bind, prototype, AbstractConstructor } from "@implab/djx/declare";
             import { DjxWidgetBase } from "@implab/djx/tsx/DjxWidgetBase";
             This class extends `dijit/_WidgetBase` and contains logic from `_AttachMixin` thus
             it is capable to handle `data-dojo-attach-*` attributes from the rendered markup.
-            ```tsx
+            ```jsx
             @djclass
             export class MyFirstWidget extends djbase(DjxWidgetBase) {
                 render() {
                     "experimentalDecorators": true,
                     "jsxFactory": "createElement",
                     "jsx": "react",
-                    "target": "ES5", // minimal supported version
+                    "target": "ES5",
                     "lib": ["ES2015", "DOM"]
                 }
             }
             The typical implementation of this technique would look like
-            ```tsx
+            ```jsx
             import { createElement } from "@implab/djx/tsx";
             import {djclass, djbase, bind} from "@implab/djx/declare";
             This library helps you to get both goals with special trait `watch(...)`
-            ```tsx
+            ```jsx
             import { createElement } from "@implab/djx/tsx";
             import { djclass, djbase} from "@implab/djx/declare"
             The key feature of this approach that the rendering function within `watch` may
             return a complex rendition.
-            ```tsx
+            ```jsx
             // inside some widget
             render() {
                 return <section>
             * DOM Nodes and widgets are left intact,
             * any other kind of value will cause an error.
+            The watch method allows to observe a single value, for the large sets of data
+            this isn't suitable well and may lead to performance issues. Dojo provides
+            observable stores to being able to track individual changes. The library provides
+            `watchFor(observable, render)` method to render observable query results and
+            handle changes on per item basis.
+            ```jsx
+            // inside some widget
+            staff = new Observable(new Memory<Employee>()),
+            getStuff() {
+                return this.staff.query();
+            }
+            addEmployee(employee: Employee) {
+                this.staff.add(employee); // the rendition will update automatically
+            }
+            render() {
+                return <table>
+                    <thead>
+                        <tr><th>Name</th><th>Position</th><th>Salary</th></tr>
+                    </thead>
+                    <tbody>
+                    {watchFor(this.getStaff(), ({name, position, salary}) =>
+                        <tr><td>{name}</td><td>{position}</td><td>{salary}</td></tr>
+                    )}
+                    </tbody>
+                </table>
+            }
+            ```

djx/src/main/ts/observable.ts +82 -42

@@ -1,5 +1,6
1	import { Cancellation } from "@implab/core-amd/Cancellation";	1	import { Cancellation } from "@implab/core-amd/Cancellation";
2	import { ICancellation } from "@implab/core-amd/interfaces";	2	import { ICancellation } from "@implab/core-amd/interfaces";
		3	import { isPromise } from "@implab/core-amd/safe";
3		4
4	/**	5	/**
5	* The interface for the consumer of an observable sequence	6	* The interface for the consumer of an observable sequence
@@ -137,9 +138,12 export interface Observable<T> extends S
137	pipe<U>(op: OperatorFn<T, U>): Observable<U>;	138	pipe<U>(op: OperatorFn<T, U>): Observable<U>;
138		139
139	/** Waits for the next event to occur and returns a promise for the next value	140	/** Waits for the next event to occur and returns a promise for the next value
140	* @param ct Cancellation token to	141	* @param ct Cancellation token
141	*/	142	*/
142	next(ct?: ICancellation): Promise<T>;	143	next(ct?: ICancellation): Promise<T>;
		144
		145	/** Collects items of the sequence to the array. */
		146	collect(ct?: ICancellation): Promise<T[]>;
143	}	147	}
144		148
145	const noop = () => { };	149	const noop = () => { };
@@ -292,61 +296,97 const _observe = <T>(producer: Producer<
292		296
293	pipe: <U>(op: (source: Observable<T>) => Producer<U>) => observe(op(_observe(producer))),	297	pipe: <U>(op: (source: Observable<T>) => Producer<U>) => observe(op(_observe(producer))),
294		298
295	next: (ct?: ICancellation) => {	299	next: collect(
296	const _ct = ct ?? Cancellation.none;	300	producer,
297	return new Promise<T>((resolve, reject) => {	301	({ next, complete, error, isClosed }) => ({
298	// wrap the producer to handle only single event	302	next: v => (next(v), complete()),
299	const once = fuse<T>(({ next, complete, error, isClosed }) => {	303	complete: () => error(new Error("The sequence is empty")),
300	const h = _ct.register(error);	304	error,
		305	isClosed
		306	})
		307	),
		308
		309	collect: collect(
		310	producer,
		311	({ next, complete, ...rest }) => {
		312	const data: T[] = [];
		313	return {
		314	next: v => data.push(v),
		315	complete: () => (next(data), complete()),
		316	...rest
		317	};
		318	}
		319	)
		320	});
301		321
302	// is the _ct fires it will call error() and isClosed() will return true	322	const collect = <T, U>(
303	const cleanup = !isClosed() ?	323	producer: Producer<T>,
304	producer({	324	collector: (result: Sink<U>) => Sink<T>
305	next: v => (next(v), complete()),	325	) => (ct = Cancellation.none) => new Promise<U>((resolve, reject) => {
306	complete: () => error(new Error("The sequence is empty")),	326	const fused = fuse<U>(({ next, complete, error, isClosed }) => {
307	error,	327	const h = ct.register(error);
308	isClosed	328	const cleanup = !isClosed() ?
309	}) ?? noop :	329	producer(collector({ next, complete, error, isClosed })) ?? noop :
310	noop;	330	noop;
311		331
312	return () => {	332	return () => {
313	h.destroy();	333	h.destroy();
314	cleanup();	334	cleanup();
315	};	335	};
316	});	336	});
317		337
318	once({	338	fused({
319	next: resolve,	339	next: resolve,
320	error: reject,	340	error: reject,
321	complete: noop,	341	complete: noop,
322	isClosed: () => false	342	isClosed: () => false
323	});	343	});
324	});
325	}
326	});	344	});
327		345
328	export const observe = <T>(producer: Producer<T>) => _observe(fuse(producer));	346	export const observe = <T>(producer: Producer<T>) => _observe(fuse(producer));
329		347
330	export const ~~stream~~Array = <T>(items: T[]) => _observe<T>(	348	export const ofArray = <T>(items: T[]) => _observe<T>(
331	({ next, complete }) => (	349	({ next, complete }) => (
332	items.forEach(next),	350	items.forEach(next),
333	complete()	351	complete()
334	)	352	)
335	);	353	);
336		354
337	~~export~~ const ~~streamPromise~~ = <T>(~~promise~~: PromiseLike<T>) => observe<T>(	355	const of1 = <T>(item: T \| PromiseLike<T>) => observe<T>(
338	({ next, error, complete }) => ~~void~~ ~~promise~~.~~then~~(	356	({ next, error, complete }) =>
339	v => (next(v), complete()),	357	isPromise(item) ?
340	error	358	void item.then(
341	)	359	v => (next(v), complete()),
		360	error
		361	) :
		362	(next(item), complete())
342	);	363	);
343		364
344	export const of = <T>(...items: T[]) => ~~_observe~~<T>(	365	export const of = <T>(...items: (T \| PromiseLike<T>)[]) => items.length === 1 ?
345	({ next, complete }) => (	366	of1(items[0]) :
346	items.forEach(next),	367	observe<T>(
347	complete()	368	({ next, error, complete, isClosed }) => {
348	)	369	const n = items.length;
349	);	370
		371	const _next = (start: number) => {
		372	if (start > 0 && isClosed()) // when resumed
		373	return;
		374
		375	for (let i = start; i < n; i++) {
		376	const r = items[i];
		377	if (isPromise(r)) {
		378	r.then(v => (next(v), _next(i + 1)), error);
		379	return; // suspend
		380	} else {
		381	next(r);
		382	}
		383	}
		384	complete();
		385	};
		386
		387	_next(0);
		388	}
		389	);
350		390
351	export const empty = _observe<never>(({ complete }) => complete());	391	export const empty = _observe<never>(({ complete }) => complete());
352		392
@@ -474,7 +514,7 export const subject = <T>(producer: Pro
474	if (_subscribers === subscribers) {	514	if (_subscribers === subscribers) {
475	const pos = subscribers.indexOf(client);	515	const pos = subscribers.indexOf(client);
476	if (pos >= 0)	516	if (pos >= 0)
477	subscribers.splice(pos,1);	517	subscribers.splice(pos, 1);
478	if (!subscribers.length)	518	if (!subscribers.length)
479	cleanup();	519	cleanup();
480	}	520	}

djx/src/main/ts/store.ts +6 -3

             export const query = <T, Q, O>(store: Queryable<T, Q, O>, includeUpdates = true) =>
                 (query?: Q, options?: O & { observe?: boolean }) => {
                     return observe<OrderedUpdate<T>>(({ next, complete, error, isClosed }) => {
+                        const processResults = (items: T[]) =>
+                            items.forEach((item, newIndex) => next({ item, newIndex, prevIndex: -1 }));
                         try {
                             const results = store.query(query, options);
                             if (isPromise(results)) {
-                                results.then(items => items.forEach((item, newIndex) => next({ item, newIndex, prevIndex: -1 })))
+                                results.then(processResults).then(undefined, error);
-                                    .then(undefined, error);
                             } else {
-                                results.forEach((item, newIndex) => next({ item, newIndex, prevIndex: -1 }));
+                                processResults(results);
                             }
                             if (!isClosed() && (options?.observe !== false) && isDjObservableResults<T>(results)) {

djx/src/main/typings/index.d.ts +2 -1

             // eslint-disable-next-line @typescript-eslint/triple-slash-reference
             /// <reference path="./css-plugin.d.ts"/>
+            import _WidgetBase = require("dijit/_WidgetBase");
             import { Rendition } from "./tsx/traits";
             declare global {
                     type Ref<T> = ((value: T | undefined) => void);
-                    type Element = Rendition;
+                    type Element = Rendition | _WidgetBase;
                     interface DjxIntrinsicAttributes<E> {
                         /** alias for className */

djx/src/test/ts/observable-tests.ts +51 -28

-            import { empty, observe } from "./observable";
+            import { empty, observe, of } from "./observable";
-            import * as t from "tap";
+            import * as tap from "tap";
             import { Cancellation } from "@implab/core-amd/Cancellation";
+            import { delay } from "@implab/core-amd/safe";
             const subj1 = observe<number>(({ next, complete }) => {
                 next(1);
             };
             subj1.subscribe(consumer1);
-            t.equal(consumer1.sum, 1, "Should get only one value");
+            tap.equal(consumer1.sum, 1, "Should get only one value");
             subj1.subscribe(consumer1);
-            t.equal(consumer1.sum, 2, "Should get the value again");
+            tap.equal(consumer1.sum, 2, "Should get the value again");
             const consumer2 = {
                 value: 0,
             subj1
                 .map(v => {
-                    t.comment(`map1: ${v * 2}`);
+                    tap.comment(`map1: ${v * 2}`);
                     maps++;
                     return v * 2;
                 })
                 .map(v => {
-                    t.comment(`map2: ${v * 2}`);
+                    tap.comment(`map2: ${v * 2}`);
                     maps++;
                     return v * 2;
                 })
                 .map(v => {
-                    t.comment(`map3: ${v * 2}`);
+                    tap.comment(`map3: ${v * 2}`);
                     maps++;
                     return v * 2;
                 })
                 .subscribe(consumer2);
-            t.equal(consumer2.value, 8, "Should map");
+            tap.equal(consumer2.value, 8, "Should map");
-            t.equal(maps, 3, "The map chain should not be executed after completion");
+            tap.equal(maps, 3, "The map chain should not be executed after completion");
-            t.ok(consumer2.completed, "The completion signal should pass through");
+            tap.ok(consumer2.completed, "The completion signal should pass through");
             const subj2 = observe<number>(({ next, complete }) => {
                 [1, 2, 3, 4, 5].forEach(next);
                 complete();
                 return () => {
-                    t.comment("subj2: unsubscribe");
+                    tap.comment("subj2: unsubscribe");
                 };
             });
                 subscribed: 0,
                 unsubscribed: 0,
                 next(v: "even" | "odd") {
-                    this[v] ++;
+                    this[v]++;
                 },
                 complete() {
                     this.completed = true;
                     complete,
                     error
                 });
-                return () =>{
+                return () => {
                     consumer3.unsubscribe();
                     h.unsubscribe();
                 };
             subj3.subscribe(consumer3);
-            t.equal(consumer3.odd, 2, "Should get 2 odd elements");
+            tap.equal(consumer3.odd, 2, "Should get 2 odd elements");
-            t.equal(consumer3.even, 2, "Should get 2 even elements");
+            tap.equal(consumer3.even, 2, "Should get 2 even elements");
-            t.ok(consumer3.completed, "The sequence should completed");
+            tap.ok(consumer3.completed, "The sequence should completed");
-            t.equal(consumer3.subscribed, 1, "The subscription should be done once");
+            tap.equal(consumer3.subscribed, 1, "The subscription should be done once");
-            t.equal(consumer3.unsubscribed, 1, "The cleanup should be done after completion");
+            tap.equal(consumer3.unsubscribed, 1, "The cleanup should be done after completion");
-            subj2.reduce((a,b) => a + b).subscribe({
+            subj2.reduce((a, b) => a + b).subscribe({
-                next: val => t.comment("subj2: reduce =", val),
+                next: val => tap.comment("subj2: reduce =", val),
-                complete: () => t.comment("subj2: complete")
+                complete: () => tap.comment("subj2: complete")
             });
-            subj2.reduce((a,b) => a + b).next()
+            tap.test("of(...) tests", async t => {
-                .then(value => t.comment("subj2: next reduce=", value))
+                await subj2.reduce((a, b) => a + b).next()
-                .catch(() => {});
+                    .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"));
-            subj2.next().then(val => t.equal(val, 1, "Should peek the first element")).catch(() => {});
+                const rejected = Promise.reject("DIE!");
+                rejected.catch(() => { }); // SAFE AND SOUND
-            const cancelled = new Cancellation(cancel => cancel());
+                await t.resolves(
-            t.rejects(subj2.next(cancelled), "Cancelled next() method should fail").catch(() => {});
+                    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"
+                );
-            t.rejects(empty.next(), "Empty sequence should fail to get next element").catch(() => {});
  No newline at end of file
+                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");
+            }).catch(() => { });
  No newline at end of file

djx/src/test/ts/plan.ts +1 -1

             import "./declare-tests";
             import "./observable-tests";
             import "./state-tests";
-            import "./subject-tests";
  No newline at end of file
+            import "./subject-tests";

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