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

             import { Cancellation } from "@implab/core-amd/Cancellation";
             import { ICancellation } from "@implab/core-amd/interfaces";
             import { isPromise } from "@implab/core-amd/safe";
             /**
              * The interface for the consumer of an observable sequence
              */
             export interface Observer<T> {
                 /**
                  * Called for the next element in the sequence
                  */
                 next?: (value: T) => void;
                 /**
                  * Called once when the error occurs in the sequence.
                  */
                 error?: (e: unknown) => void;
                 /**
                  * Called once at the end of the sequence.
                  */
                 complete?: () => void;
             }
             /**
              * The group of functions to feed an observable. These methods are provided to
              * the producer to generate a stream of events.
              */
             export type Sink<T> = {
                 /**
                  * Call to send the next element in the sequence
                  */
                 next: (value: T) => void;
                 /**
                  * Call to notify about the error occurred in the sequence.
                  */
                 error: (e: unknown) => void;
                 /**
                  * Call to signal the end of the sequence.
                  */
                 complete: () => void;
                 /**
                  * Checks whether the sink is accepting new elements. It's safe to
                  * send elements to the closed sink.
                  */
                 isClosed: () => boolean;
             };
             export type Producer<T> = (sink: Sink<T>) => (void | (() => void));
             export interface Unsubscribable {
                 unsubscribe(): void;
             }
             export const isUnsubscribable = (v: unknown): v is Unsubscribable =>
                 v !== null && v !== undefined && typeof (v as Unsubscribable).unsubscribe === "function";
             export const isSubscribable = <T = unknown>(v: unknown): v is Subscribable<T> =>
                 v !== null && v !== undefined && typeof (v as Subscribable<unknown>).subscribe === "function";
             export interface Subscribable<T> {
                 subscribe(consumer: Observer<T>): Unsubscribable;
             }
             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> {
                 /** Transforms elements of the sequence with the specified mapper
                  *
                  * @param mapper The mapper used to transform the values
                  */
                 map<T2>(mapper: (value: T) => T2): Observable<T2>;
                 /** Filters elements of the sequence. The resulting sequence will
                  * contain only elements which match the specified predicate.
                  *
                  * @param predicate The filter predicate.
                  */
                 filter(predicate: (value: T) => boolean): Observable<T>;
                 /** Completes the sequence once the condition is met.
                  * @param predicate The condition which should be met to complete the sequence
                  */
                 until(predicate: (value: T) => boolean): Observable<T>;
                 /** Keeps the sequence running while elements satisfy the condition.
                  *
                  * @param predicate The condition which should be met to continue.
                  */
                 while(predicate: (value: T) => boolean): Observable<T>;
                 /** Applies accumulator to each value in the sequence and
                  * emits the accumulated value for each source element
                  *
                  * @param accumulator
                  * @param initial
                  */
                 scan<A>(accumulator: AccumulatorFn<T, A>, initial: A): Observable<A>;
                 scan(accumulator: AccumulatorFn<T, T>): Observable<T>;
                 /** Applies accumulator to each value in the sequence and
                  * emits the accumulated value at the end of the sequence
                  *
                  * @param accumulator
                  * @param initial
                  */
                 reduce<A>(accumulator: AccumulatorFn<T, A>, initial: A): Observable<A>;
                 reduce(accumulator: AccumulatorFn<T, T>): Observable<T>;
                 /** Concatenates the specified sequences with this observable
                  *
                  * @param seq sequences to concatenate with the current observable
                  *
                  * The concatenation doesn't accumulate values from the specified sequences,
                  * The result of the concatenation is the new observable which will switch
                  * to the next observable after the previous one completes. Values emitted
                  * before the next observable being active are lost.
                  */
                 cat(...seq: Subscribable<T>[]): Observable<T>;
                 /** Pipes the specified operator to produce the new observable
                  * @param op The operator consumes this observable and produces a new one
                  *
                  * The operator is a higher order function which takes a source observable
                  * and returns a producer for the new observable.
                  *
                  * This function can be used to create a complex mapping between source and
                  * resulting observables. The operator may have a state (or a side effect)
                  * and can be connected to multiple observables.
                  */
                 pipe<U>(op: OperatorFn<T, U>): Observable<U>;
                 /** Waits for the next event to occur and returns a promise for the next value
                  * @param ct Cancellation token
                 */
                 next(ct?: ICancellation): Promise<T>;
                 /** Collects items of the sequence to the array. */
                 collect(ct?: ICancellation): Promise<T[]>;
             }
             const noop = () => { };
             const sink = <T>(consumer: Observer<T>) => {
                 const { next, error, complete } = consumer;
                 return {
                     next: next ? next.bind(consumer) : noop,
                     error: error ? error.bind(consumer) : noop,
                     complete: complete ? complete.bind(consumer) : noop,
                     isClosed: () => false
                 };
             };
             /** Wraps the producer to handle tear down logic and subscription management
              *
              * @param producer The producer to wrap
              * @returns The wrapper producer
              */
             const fuse = <T>(producer: Producer<T>) => ({ next, error, complete }: Sink<T>) => {
                 let done = false;
                 let cleanup = noop;
                 const _fin = <A extends unknown[]>(fn: (...args: A) => void) =>
                     (...args: A) => done ?
                         void (0) :
                         (done = true, cleanup(), fn(...args));
                 const _fin0 = () => done ? void (0) : (done = true, cleanup());
                 const safeSink = {
                     next: (value: T) => { !done && next(value); },
                     error: _fin(error),
                     complete: _fin(complete),
                     isClosed: () => done
                 };
                 cleanup = producer(safeSink) ?? noop;
                 return done ? cleanup() : _fin0;
             };
             const _observe = <T>(producer: Producer<T>): Observable<T> => ({
                 subscribe: (consumer: Observer<T>) => ({
                     unsubscribe: producer(sink(consumer)) ?? noop
                 }),
                 map: (mapper) => _observe(({ next, ...rest }) =>
                     producer({
                         next: next !== noop ? (v: T) => next(mapper(v)) : noop,
                         ...rest
                     })
                 ),
                 filter: (predicate) => _observe(({ next, ...rest }) =>
                     producer({
                         next: next !== noop ? (v: T) => predicate(v) ? next(v) : void (0) : noop,
                         ...rest
                     })
                 ),
                 until: predicate => _observe(({ next, complete, ...rest }) =>
                     producer({
                         next: v => predicate(v) ? complete() : next(v),
                         complete,
                         ...rest
                     })
                 ),
                 while: predicate => _observe(({ next, complete, ...rest }) =>
                     producer({
                         next: v => predicate(v) ? next(v) : complete(),
                         complete,
                         ...rest
                     })
                 ),
                 scan: <A>(...args: [AccumulatorFn<T, A>, A] | [AccumulatorFn<T, T>]) => _observe<T | A>(({ next, ...rest }) => {
                     if (args.length === 1) {
                         const [accumulator] = args;
                         let _acc: T;
                         let index = 0;
                         return producer({
                             next: next !== noop ? (v: T) => next(index++ === 0 ? _acc = v : _acc = accumulator(_acc, v)) : noop,
                             ...rest
                         });
                     } else {
                         const [accumulator, initial] = args;
                         let _acc = initial;
                         return producer({
                             next: next !== noop ? (v: T) => next(_acc = accumulator(_acc, v)) : noop,
                             ...rest
                         });
                     }
                 }),
                 reduce: <A>(...args: [AccumulatorFn<T, A>, A] | [AccumulatorFn<T, T>]) => _observe<T | A>(({ next, complete, error, ...rest }) => {
                     if (args.length === 1) {
                         const [accumulator] = args;
                         let _acc: T;
                         let index = 0;
                         return producer({
                             next: next !== noop ? (v: T) => {
                                 _acc = index++ === 0 ? v : accumulator(_acc, v);
                             } : noop,
                             complete: () => {
                                 if (index === 0) {
                                     error(new Error("The sequence can't be empty"));
                                 } else {
                                     next(_acc);
                                     complete();
                                 }
                             },
                             error,
                             ...rest
                         });
                     } else {
                         const [accumulator, initial] = args;
                         let _acc = initial;
                         return producer({
                             next: next !== noop ? (v: T) => {
                                 _acc = accumulator(_acc, v);
                             } : noop,
                             complete: () => {
                                 next(_acc);
                                 complete();
                             },
                             error,
                             ...rest
                         });
                     }
                 }),
                 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,
                     ({ next, complete, error, isClosed }) => ({
                         next: v => (next(v), complete()),
                         complete: () => error(new Error("The sequence is empty")),
                         error,
                         isClosed
                     })
                 ),
                 collect: collect(
                     producer,
                     ({ next, complete, ...rest }) => {
                         const data: T[] = [];
                         return {
                             next: v => data.push(v),
                             complete: () => (next(data), complete()),
                             ...rest
                         };
                     }
                 )
             });
             const collect = <T, U>(
                 producer: Producer<T>,
                 collector: (result: Sink<U>) => Sink<T>
             ) => (ct = Cancellation.none) => new Promise<U>((resolve, reject) => {
                 const fused = fuse<U>(({ next, complete, error, isClosed }) => {
                     const h = ct.register(error);
                     const cleanup = !isClosed() ?
                         producer(collector({ next, complete, error, isClosed })) ?? noop :
                         noop;
                     return () => {
                         h.destroy();
                         cleanup();
                     };
                 });
                 fused({
                     next: resolve,
                     error: reject,
                     complete: noop,
                     isClosed: () => false
                 });
             });
             export const observe = <T>(producer: Producer<T>) => _observe(fuse(producer));
             /** Converts an array to the observable sequence of its elements. */
             export const ofArray = <T>(items: T[]) => _observe<T>(
                 ({ next, complete }) => (
                     items.forEach(next),
                     complete()
                 )
             );
             /** Converts a subscribable to the observable */
             export const ofSubscribable = <T>(subscribable: Subscribable<T>) =>
                 observe(sink => {
                     const subscription = subscribable.subscribe(sink);
                     return () => subscription.unsubscribe();
                 });
             const of1 = <T>(item: T | PromiseLike<T>) => observe<T>(
                 ({ next, error, complete }) =>
                     isPromise(item) ?
                         void item.then(
                             v => (next(v), complete()),
                             error
                         ) :
                         (next(item), complete())
             );
             /** Converts a list of parameter values to the observable sequence. The
              * order of elements in the list will be preserved in the resulting sequence.
              */
             export const of = <T>(...items: (T | PromiseLike<T>)[]) => items.length === 1 ?
                 of1(items[0]) :
                 observe<T>(
                     ({ next, error, complete, isClosed }) => {
                         const n = items.length;
                         const _next = (start: number) => {
                             if (start > 0 && isClosed()) // when resumed
                                 return;
                             for (let i = start; i < n; i++) {
                                 const r = items[i];
                                 if (isPromise(r)) {
                                     r.then(v => (next(v), _next(i + 1)), error);
                                     return; // suspend
                                 } else {
                                     next(r);
                                 }
                             }
                             complete();
                         };
                         _next(0);
                     }
                 );
             export const empty = _observe<never>(({ complete }) => complete());

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";
             /**
              * Creates a buffer with the specified length. The buffer will immediately
              * subscribe to the source observable and start accumulating values.
              *
              * The resulting observable will repeat the buffered values for each new subscriber.
              *
              * @param length The number of elements to store.
              * @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)
                 let wpos = 0;
                 const buffer: T[] = [];
                 // writes the next value to the buffer
                 const write = (value: T) => {
                     buffer[wpos] = value;
                     wpos = (wpos + 1) % length;
                 };
                 // reads contents of the buffer
                 // cb will be called for each value in the buffer
                 const read = (cb: (item: T) => void) => {
                     const start = wpos + length - buffer.length;
                     const end = wpos + length;
                     for(let pos = start; pos < end; pos++ )
                         cb(buffer[pos % length]);
                 };
                 let status: Status = "active";
                 let lastError: unknown;
                 let subscribers: Sink<T>[] = [];
                 const scope = new Scope();
                 // cleanup method to release resources held by this subscription
                 const cleanup = (cb: (item: Sink<T>) => void) => {
                     scope.destroy();
                     const _subscribers = subscribers;
                     subscribers = [];
                     _subscribers.forEach(cb);
                 };
                 const sink: Sink<T> = {
                     isClosed: () => status !== "active",
                     complete: () => {
                         if (status === "active") {
                             status = "complete";
                             cleanup(s => s.complete());
                         }
                     },
                     error: e => {
                         if (status === "active") {
                             status = "error";
                             lastError = e;
                             cleanup(s => s.error(e));
                         }
                     },
                     next: v => {
                         if (status === "active") {
                             write(v);
                             const _subscribers = subscribers;
                             _subscribers.forEach(s => s.next(v));
                         }
                     }
                 };
                 if (ct.isRequested()) {
                     sink.error(new CancelledError("The operation was cancelled", ct));
                 } else {
                     scope.own(source.subscribe(sink));
                     scope.own(ct.register(e => sink.error(e)));
                 }
-                return (s: Sink<T>) => {
+                return observe( s => {
                     const _subscribers = subscribers;
                     read(s.next);
                     switch (status) {
                         case "active":
                             subscribers.push(s);
                             return () => {
                                 if (_subscribers === subscribers) {
                                     const pos = subscribers.indexOf(s);
                                     if (pos >= 0)
                                         subscribers.splice(pos, 1);
                                 }
                             };
                         case "complete":
                             s.complete();
                             break;
                         case "error":
                             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 = () => { };
             /** Joins multiple subscribers to the single one. The resulting subscriber
              * will be created when the first client subscribes and will be released
              * with the the last client unsubscribed.
              *
              * Use this wrapper to prevent spawning multiple producers.
              *
              * @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 };
                 // cleanup method to release resources held by this subscription
                 const cleanup = (cb: (item: Sink<T>) => void) => {
                     const _subscribers = subscribers;
                     subscribers = [];
                     _subscribers.forEach(cb);
                     subscription.unsubscribe();
                 };
                 const sink: Sink<T> = {
                     isClosed: () => false,
                     complete: () => cleanup(s => s.complete()),
                     error: e => cleanup(s => s.error(e)),
                     next: v => subscribers.forEach(s => s.next(v))
                 };
-                return client => {
+                return observe(client => {
                     const _subscribers = subscribers;
                     subscribers.push(client);
                     if (subscribers.length === 1)
                         subscription = source.subscribe(sink);
                     return () => {
                         if (_subscribers === subscribers) {
                             const pos = subscribers.indexOf(client);
                             if (pos >= 0)
                                 subscribers.splice(pos, 1);
                             if (!subscribers.length)
                                 subscription.unsubscribe();
                         }
                     };
-                };
+                });
             };

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

             import { PromiseOrValue } from "@implab/core-amd/interfaces";
-            import { isPromise } from "@implab/core-amd/safe";
+            import { isCancellable, isPromise } from "@implab/core-amd/safe";
-            import { observe, Observable } from "./observable";
+            import { observe, Observable, empty } from "./observable";
+            import { after } from "dojo/aspect";
+            import { subject } from "./operators/subject";
             export interface OrderedUpdate<T> {
                 /** The item is being updated */
                 readonly item: T;
                 /** The previous index of the item, -1 in case it is inserted */
                 readonly prevIndex: number;
                 /** The new index of the item, -1 in case it is deleted */
                 readonly newIndex: number;
             }
             export type QueryResults<T> = Observable<OrderedUpdate<T>>;
             interface DjObservableResults<T> {
                 /**
                  * Allows observation of results
                  */
                 observe(listener: (object: T, previousIndex: number, newIndex: number) => void, includeUpdates?: boolean): {
                     remove(): void;
                 };
             }
             interface Queryable<T, Q, O> {
                 query(query?: Q, options?: O): PromiseOrValue<T[]>;
             }
             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) =>
                 (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(processResults).then(undefined, error);
                             } else {
                                 processResults(results);
                             }
                             if (!isClosed() && (options?.observe !== false) && isDjObservableResults<T>(results)) {
                                 const h = results.observe((item, prevIndex, newIndex) => next({ item, prevIndex, newIndex }), includeUpdates);
                                 return () => h.remove();
                             } else {
                                 complete();
                             }
                         } catch (err) {
                             error(err);
                         }
                     });
                 };
+            interface IndexedStore<T> {
+                get(id: string | number): PromiseLike<T> | T | null | undefined;
+            }
+            interface Notifications<T> {
+                notify(item: T | undefined, id: string | number | undefined): void;
+            }
+            const hasNotifications = <T>(x: unknown): x is Notifications<T> =>
+                typeof x === "object" && x !== null && (typeof (x as Notifications<T>).notify === "function");
+            interface GetOpts {
+                observe?: boolean;
+            }
+            type Change<T> = [item: NonNullable<T>, id: string | number | undefined] |
+            [item: undefined | null, id: string | number];
+            const filterItem = (itemId: string | number) =>
+                <T>(source: Observable<Change<T>>) =>
+                    observe<T>(({ next, complete, error }) => {
+                        const subscription = source
+                            .filter(([, id]) => id === itemId)
+                            .subscribe({
+                                next: ([item]) => item !== null && item !== undefined ? next(item) : complete(),
+                                complete,
+                                error
+                            });
+                        return () => subscription.unsubscribe();
+                    });
+            export const get = <T>(store: IndexedStore<T>) => {
+                const changes = hasNotifications<T>(store) ?
+                    observe<Change<T>>(({ next }) => {
+                        const handle = after(store, "notify", (...args: Change<T>) => next(args), true);
+                        return () => handle.remove();
+                    }).pipe(subject) : empty;
+                return (id: string | number, opts: GetOpts = {}) =>
+                    observe<T>(({ next, complete, error }) => {
+                        const result = store.get(id);
+                        const handle = (x: T | null | undefined) => {
+                            if (x !== null && x !== undefined)
+                                next(x);
+                            complete();
+                        };
+                        if (isPromise(result)) {
+                            result.then(handle, error);
+                            if (isCancellable(result))
+                                return () => result.cancel();
+                        } else {
+                            handle(result);
+                        }
+                    }).cat(opts.observe !== false ? changes.pipe(filterItem(id)) : empty);
+            };
  No newline at end of file

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

             import { empty, observe, of } from "./observable";
             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);
                 complete();
                 next(2);
             });
             const consumer1 = {
                 sum: 0,
                 next(v: number) {
                     this.sum += v;
                 }
             };
             subj1.subscribe(consumer1);
             tap.equal(consumer1.sum, 1, "Should get only one value");
             subj1.subscribe(consumer1);
             tap.equal(consumer1.sum, 2, "Should get the value again");
             const consumer2 = {
                 value: 0,
                 completed: false,
                 next(v: number) { this.value = v; },
                 complete() { this.completed = true; }
             };
             let maps = 0;
             subj1
                 .map(v => {
                     tap.comment(`map1: ${v * 2}`);
                     maps++;
                     return v * 2;
                 })
                 .map(v => {
                     tap.comment(`map2: ${v * 2}`);
                     maps++;
                     return v * 2;
                 })
                 .map(v => {
                     tap.comment(`map3: ${v * 2}`);
                     maps++;
                     return v * 2;
                 })
                 .subscribe(consumer2);
             tap.equal(consumer2.value, 8, "Should map");
             tap.equal(maps, 3, "The map chain should not be executed after completion");
             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 () => {
                     tap.comment("subj2: unsubscribe");
                 };
             });
             const consumer3 = {
                 even: 0,
                 odd: 0,
                 completed: false,
                 subscribed: 0,
                 unsubscribed: 0,
                 next(v: "even" | "odd") {
                     this[v]++;
                 },
                 complete() {
                     this.completed = true;
                 },
                 subscribe() {
                     this.subscribed++;
                 },
                 unsubscribe() {
                     this.unsubscribed++;
                 }
             };
-            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({
                     next: val => {
                         if (val % 2 === 0)
                             next("odd");
                         else
                             next("even");
                         if (++count === 4)
                             complete();
                     },
                     complete,
                     error
                 });
                 return () => {
                     consumer3.unsubscribe();
                     h.unsubscribe();
                 };
-            });
+            }));
             subj3.subscribe(consumer3);
             tap.equal(consumer3.odd, 2, "Should get 2 odd elements");
             tap.equal(consumer3.even, 2, "Should get 2 even elements");
             tap.ok(consumer3.completed, "The sequence should completed");
             tap.equal(consumer3.subscribed, 1, "The subscription should be done once");
             tap.equal(consumer3.unsubscribed, 1, "The cleanup should be done after completion");
             subj2.reduce((a, b) => a + b).subscribe({
                 next: val => tap.comment("subj2: reduce =", val),
                 complete: () => tap.comment("subj2: complete")
             });
             tap.test("of(...) tests", async t => {
                 await subj2.reduce((a, b) => a + b).next()
                     .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"));
                 const rejected = Promise.reject("DIE!");
                 rejected.catch(() => { }); // SAFE AND SOUND
                 await t.resolves(
                     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.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(() => { });

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

             import "./declare-tests";
             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");
             import ValidationTextBox = require("dijit/form/ValidationTextBox");
             const trace = TraceSource.get(mid);
             interface NewAppointmentProps {
                 title: string;
                 startAt?: Date;
             }
             @djclass
             export default class NewAppointment extends djbase(DjxWidgetBase) {
                 value: NewAppointmentProps = {title: "Appointment 1"};
                 render() {
                     return <section>
                         <Form<NewAppointmentProps> ref={bind("value", prop(this, "value"))}
                             method="DIALOG"
                             onSubmit={this._onSubmit}
                             onReset={this._onReset}
                             onValidStateChange={this._onValidStateChange}
                             onValueChange={this._onValueChange}
                         >
                             <p><label>Title: <ValidationTextBox required name="title"/></label></p>
                             <p><label>Appointment date: <DateTextBox required name="startAt"/></label></p>
                             <footer>
                                 <Button type="submit">Send</Button>
                                 <Button type="reset">Reset</Button>
                             </footer>
                         </Form>
                     </section>;
                 }
                 private readonly _onSubmit = (evt: Event) => {
                     trace.debug("onSubmit");
                 };
                 private readonly _onValidStateChange = (isValid?: boolean) => {
                     trace.debug("isValid={0}", isValid);
                 };
                 private readonly _onValueChange = (value: NewAppointmentProps) => {
                     trace.debug("valueChange={0}", value);
                 };
                 private readonly _onReset = (evt: Event) => {
                     trace.debug("onReset");
                     //evt.preventDefault();
                 };
             }

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