implab/implabjs-djx Commit - r136:435ce00ba245

added store::get method to wrap up dojo/store/get

cin -

r136:435ce00ba245 v1.9.0-rc2 default

parent child

Context file:

r136:435ce00ba245

Collapse all files

djx/src/test/ts/observable-store-tests.ts created 644 +65 0

@@ -0,0 +1,65
	1	import Memory = require("dojo/store/Memory");
	2	import Observerable = require("dojo/store/Observable");
	3	import { get } from "./store";
	4	import tap = require("tap");
	5
	6	interface Person {
	7	id: string;
	8
	9	name: string;
	10
	11	age: number;
	12	}
	13
	14	tap.test("store::get(...) tests", async t => {
	15	const store = new Observerable(new Memory<Person>());
	16
	17	const getPerson = get(store);
	18
	19	const peterId = "id:peter";
	20
	21	const samId = "id:sam";
	22
	23	const peter = getPerson(peterId);
	24	const sam = getPerson(samId);
	25
	26	const seq1 = await getPerson(peterId, { observe: false }).collect();
	27
	28	t.ok(seq1.length === 0, "Should be empty sequence");
	29
	30	let peterChangeCount = 0;
	31	let samChangeCount = 0;
	32	let peterDeleted = 0;
	33
	34	const peterSubscription = peter.subscribe({
	35	next: () => peterChangeCount++,
	36	complete: () => peterDeleted++
	37	});
	38	const samSubscription = sam.subscribe({
	39	next: () => samChangeCount++
	40	});
	41
	42	try {
	43	t.equal(peterChangeCount, 0, "Should be no changes recorded");
	44
	45	store.put({id: peterId, name: "Peter", age: 30 });
	46
	47	t.equal(peterChangeCount, 1, "Should record 1 object change");
	48	t.equal(samChangeCount, 0, "Should not record other object changes");
	49
	50	store.remove(peterId);
	51
	52	t.equal(peterDeleted, 1, "Should complete sequence");
	53	t.equal(peterChangeCount, 1, "Should not record remove operations");
	54
	55	store.put({id: peterId, name: "Peter", age: 29});
	56
	57	t.equal(peterChangeCount, 1, "Should not record changes after completion");
	58
	59	} finally {
	60	peterSubscription.unsubscribe();
	61	samSubscription.unsubscribe();
	62	}
	63
	64
	65	}).catch(() => { }); No newline at end of file

djx/src/main/ts/observable.ts +8 -13

             export type AccumulatorFn<T, A> = (acc: A, value: T) => A;
-            export type OperatorFn<T, U> = (source: Observable<T>) => Producer<U>;
+            export type OperatorFn<T, U> = (source: Observable<T>) => Observable<U>;
             /** The observable source of items. */
             export interface Observable<T> extends Subscribable<T> {
                 cat: (...seq) => _observe(({ next, complete: final, ...rest }) => {
                     let cleanup: () => void;
-                    const complete = () => {
+                    const len = seq.length;
-                        const continuation = seq.shift();
+                    const complete = (i: number) => i < len ?
-                        if (continuation) {
+                        () => {
-                            // if we have a next sequence, subscribe to it
+                            const subscription = seq[i].subscribe({ next, complete: complete(i + 1), ...rest });
-                            const subscription = continuation.subscribe({ next, complete, ...rest });
                             cleanup = subscription.unsubscribe.bind(subscription);
-                        } else {
+                        } : final;
-                            // otherwise notify the consumer about completion
-                            final();
-                    };
-                    cleanup = producer({ next, complete, ...rest }) ?? noop;
+                    cleanup = producer({ next, complete: complete(0), ...rest }) ?? noop;
                     return () => cleanup();
                 }),
-                pipe: <U>(op: OperatorFn<T,U>) => observe(op(_observe(producer))),
+                pipe: <U>(op: OperatorFn<T, U>) => op(_observe(producer)),
                 next: collect(
                     producer,

djx/src/main/ts/operators/buffer.ts +4 -4

             import { Cancellation } from "@implab/core-amd/Cancellation";
             import { CancelledError } from "@implab/core-amd/CancelledError";
-            import { Producer, Sink, Subscribable } from "../observable";
+            import { Observable, Sink, Subscribable, observe } from "../observable";
             import { Scope } from "../tsx/Scope";
             /**
              * @param ct Cancellation token to unsubscribe from the original observable.
              *
              */
-            export const buffer = (length: number, ct = Cancellation.none) => <T>(source: Subscribable<T>): Producer<T> => {
+            export const buffer = (length: number, ct = Cancellation.none) => <T>(source: Subscribable<T>): Observable<T> => {
                 type Status = "active" | "complete" | "error";
                 // ring-buffer, wpos will rotate in range (0...length-1)
                     scope.own(ct.register(e => sink.error(e)));
                 }
-                return (s: Sink<T>) => {
+                return observe( s => {
                     const _subscribers = subscribers;
                     read(s.next);
                     switch (status) {
                             s.error(lastError);
                             break;
                     }
-                };
+                });
             };

djx/src/main/ts/operators/subject.ts +4 -4

-            import { Producer, Sink, Subscribable } from "../observable";
+            import { Observable, Sink, Subscribable, observe } from "../observable";
             const noop = () => { };
              * @param source The source observable
              * @returns The wrapped producer
              */
-            export const subject = <T>(source: Subscribable<T>): Producer<T> => {
+            export const subject = <T>(source: Subscribable<T>): Observable<T> => {
                 let subscribers: Sink<T>[] = [];
                 let subscription = { unsubscribe: noop };
                     next: v => subscribers.forEach(s => s.next(v))
                 };
-                return client => {
+                return observe(client => {
                     const _subscribers = subscribers;
                     subscribers.push(client);
                     if (subscribers.length === 1)
                                 subscription.unsubscribe();
                         }
                     };
+                });
             };
-            };

djx/src/main/ts/store.ts +64 -2

@@ -1,6 +1,8
1	import { PromiseOrValue } from "@implab/core-amd/interfaces";	1	import { PromiseOrValue } from "@implab/core-amd/interfaces";
2	import { isPromise } from "@implab/core-amd/safe";	2	import { isCancellable, isPromise } from "@implab/core-amd/safe";
3	import { observe, Observable } from "./observable";	3	import { observe, Observable, empty } from "./observable";
		4	import { after } from "dojo/aspect";
		5	import { subject } from "./operators/subject";
4		6
5	export interface OrderedUpdate<T> {	7	export interface OrderedUpdate<T> {
6	/** The item is being updated */	8	/** The item is being updated */
@@ -59,3 +61,63 export const query = <T, Q, O>(store: Qu
59	});	61	});
60		62
61	};	63	};
		64
		65	interface IndexedStore<T> {
		66	get(id: string \| number): PromiseLike<T> \| T \| null \| undefined;
		67	}
		68
		69	interface Notifications<T> {
		70	notify(item: T \| undefined, id: string \| number \| undefined): void;
		71	}
		72
		73	const hasNotifications = <T>(x: unknown): x is Notifications<T> =>
		74	typeof x === "object" && x !== null && (typeof (x as Notifications<T>).notify === "function");
		75
		76	interface GetOpts {
		77	observe?: boolean;
		78	}
		79
		80	type Change<T> = [item: NonNullable<T>, id: string \| number \| undefined] \|
		81	[item: undefined \| null, id: string \| number];
		82
		83	const filterItem = (itemId: string \| number) =>
		84	<T>(source: Observable<Change<T>>) =>
		85	observe<T>(({ next, complete, error }) => {
		86	const subscription = source
		87	.filter(([, id]) => id === itemId)
		88	.subscribe({
		89	next: ([item]) => item !== null && item !== undefined ? next(item) : complete(),
		90	complete,
		91	error
		92	});
		93	return () => subscription.unsubscribe();
		94	});
		95
		96	export const get = <T>(store: IndexedStore<T>) => {
		97	const changes = hasNotifications<T>(store) ?
		98	observe<Change<T>>(({ next }) => {
		99	const handle = after(store, "notify", (...args: Change<T>) => next(args), true);
		100	return () => handle.remove();
		101	}).pipe(subject) : empty;
		102
		103
		104	return (id: string \| number, opts: GetOpts = {}) =>
		105	observe<T>(({ next, complete, error }) => {
		106	const result = store.get(id);
		107
		108	const handle = (x: T \| null \| undefined) => {
		109	if (x !== null && x !== undefined)
		110	next(x);
		111	complete();
		112	};
		113
		114	if (isPromise(result)) {
		115	result.then(handle, error);
		116
		117	if (isCancellable(result))
		118	return () => result.cancel();
		119	} else {
		120	handle(result);
		121	}
		122	}).cat(opts.observe !== false ? changes.pipe(filterItem(id)) : empty);
		123	}; No newline at end of file

djx/src/test/ts/observable-tests.ts +2 -2

             };
-            const subj3 = subj2.pipe<"even" | "odd">(self => ({ next, complete, error }) => {
+            const subj3 = subj2.pipe<"even" | "odd">(self => observe(({ next, complete, error }) => {
                 consumer3.subscribe();
                 let count = 0;
                 const h = self.subscribe({
                     consumer3.unsubscribe();
                     h.unsubscribe();
                 };
-            });
+            }));
             subj3.subscribe(consumer3);

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

             import "./observable-tests";
             import "./state-tests";
             import "./subject-tests";
+            import "./observable-store-tests";

playground/src/main/ts/view/NewAppointment.tsx +1 -2

             import { id as mid } from "module";
             import { djbase, djclass } from "@implab/djx/declare";
-            import { attach, bind, createElement, prop } from "@implab/djx/tsx";
+            import { bind, createElement, prop } from "@implab/djx/tsx";
             import { DjxWidgetBase } from "@implab/djx/tsx/DjxWidgetBase";
             import Form from "@implab/djx/form/Form";
-            import { LocalDateTime } from "@js-joda/core";
             import { TraceSource } from "@implab/core-amd/log/TraceSource";
             import DateTextBox = require("dijit/form/DateTextBox");
             import Button = require("dijit/form/Button");

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