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

             fc9f82c082ef432137da086a1fe9c37a12ba16a2 v1.0.0-rc2
 d80d7901b4c8ffe8728e4a7bf5f4b7e7a669bb5 v1.0.0-rc3
 a2c44d8e1f34dd30c2b50f92b7dc2e8f3247c43 v1.0.0-rc5
 c01fabe9ea9fb5e753fbeae8b0d2664e7072a66 v1.0.0-rc6
 e546fe36fdddc8324f1098ee950fa1a7ba19b93 v1.0.0-rc7
 f4d5e2c719a20ae6d65f1f4b5e2141ed765e975 v1.0.0-rc8
             a1ab2b5975ad4b19599fb61538e7aaf329fb528c v1.0.0-rc10
 b77ac3bf8f200876450ad50c308a9441a7f39c7 v1.0.0-rc11
 b72f33756d3d10553b743f0b9f4504148cf97d v1.0.0-rc12
             b88fac0e76c0e61e397e2995f468f7cf342afbc9 v1.0.0-rc13
             a46488b209e8aac583c1634043147d87740c63b4 v1.0.0-rc14
             1174538197f6796384e643f62100292f1377b137 v1.0.0-rc15
             e8012fdf09ae442094f3831abe70649f8520659e 1.0.0-rc16
             a1a1ef050ecc9d2a780e308cbffc2ad915a5c13d 1.0.0-rc17
 c6c7e16919cff4019a55661725789d287439b75 v1.0.0-rc18
 b6c4159c66cecf6c5957d33ad474919608489c6 v1.0.0
 b2dcc1ae97fe3673242c84fe8d4b50a55f0 v1.0.1
             ed3c20c09b000386b5204b483955eb61ee662eff v1.0.2
 ea350f98bb7a06e636251ea5cad403217e868 v1.0.3
 ba910a5425ed95f7a67ef4db1dbbf599e2cab v1.0.5
             deb0ed6fb68015d89ba1287f6b766d1f80b6e4ff v1.0.7
             dd0d589acfbbe938a5630547e83ad934ab597b64 v1.0.8
             f2499237b5bf0ac2d73a8e437b8f15fa8a6306a5 v1.0.9
             d4f0cdae9577caa605d9317d0ceb93f7b67895f3 v1.0.10
 a0018655d1c3dafc16ae50971cb535b8555e246 v1.1.0
             ff3695b0a48f00155ad6c3d7e000fe185ee6f95a v1.1.1
             c276b9b7fa833242d1bf75eeeaae9924feee51e8 v1.1.2
             fdde09e66c009f4950ec69a393a8e789725da758 v1.2.0
 c6055f20ce1a49bdca5e6dabce88ee8deed v1.2.1
             bc7556143fe536e3df374bc0070146311884284e v1.2.2
             e5bb5e80ce96fc4ca0fbbf0b5390443a616e99fa v1.2.3
             cc5be30e84f8a0d9a8e25447651576d9e46ab154 v1.2.4
 ab11174c7446830d91d5f1b652a18c6ecae5 v1.2.5
 a7b6319ebe24973fe10b8d82c19d3d86857b4e v1.2.6
 deb750dc18fcd9be83c28cb8371530fffd8 v1.2.7
 f8caa5bf8fd2d4a56a6cff40be31ab82a2727 v1.2.8
 c829aa08a2cf65cc074640199e79c5e49d2c93 v1.3.0
 a190b3a757dc650c2c612073cbffaaa720b6090 v1.4.0
 ccfaae984e9a458580664923c87f258cff9890f v1.4.4
             e9a9ed6d7647848e9c7f28e6e9f92d009259607d v1.5.0
             aac297dda27dac1e86bc93617ba172a2c92415f5 v1.6.0
             e07418577cbcaa6e31ccbf87c4495324ecbd88bd v1.6.1
             bc1b4dd8ca1a03af095ea1a61169c6025ef37fce v1.6.2
             fb2ea4d6aabac2b374f3b2bfddaca6a83dedf34f v1.6.3
             cede47727a1b0867d19be1d168aceb00b5e74e3f v1.7.0
 aad89415099d8e1942a8e67050c22e111d8b v1.7.1
 e70973248de01601aa364eadf11dd2f3a9f v1.8.0
             c7d9ad82b374c2a20710eba8fa40a9b1472ddc77 v1.8.1
             f139e2153e0da5649558b1ed9369da860d442c0d v1.9.0-rc1
 ce00ba2452d565f1ed3bb884643b4b88b9cde v1.9.0-rc2
 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

             import { id as mid} from "module";
             import { Cancellation } from "@implab/core-amd/Cancellation";
             import { ICancellation } from "@implab/core-amd/interfaces";
             import { TraceSource } from "@implab/core-amd/log/TraceSource";
             import { isPromise } from "@implab/core-amd/safe";
             const trace = TraceSource.get(mid);
             /**
              * 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));
             type FusedSink<T> = Omit<Sink<T>, "isClosed">;
             type FusedProducer<T> = (sink: FusedSink<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;
+                /** 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;
             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>;
+                /** 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.
                  *
                  * @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 errorFallback = (e: unknown) => trace.error("Unhandled observable error: {0}", e);
             const sink = <T>(consumer: Observer<T>) => {
                 // eslint-disable-next-line @typescript-eslint/unbound-method
                 const { next, error, complete } = consumer;
                 return {
                     next: next ? next.bind(consumer) : noop,
                     error: error ? error.bind(consumer) : errorFallback, // report unhandled errors
                     complete: complete ? complete.bind(consumer) : noop
                 };
             };
             /** Wraps the producer to handle tear down logic and subscription management
              *
              * The resulting producer will invoke cleanup logic on error or complete events
              * and will prevent calling of any method from the sink.
              *
              * @param producer The producer to wrap
              * @returns The wrapper producer
              */
             const fuse = <T>(producer: Producer<T>) => ({ next, error, complete }: FusedSink<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
                 };
                 // call the producer
                 cleanup = producer(safeSink) ?? noop;
                 // if the producer throws exception bypass it to the caller rather then to
                 // the sink. This is a feature.
                 // if the producer completed the sequence immediately call the cleanup in place
                 return done ? cleanup() : _fin0;
             };
             const _observe = <T>(producer: FusedProducer<T>): Observable<T> => ({
-                subscribe: (consumer: Observer<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
                     })
                 ),
+                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,
                         ...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 }) => {
                     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
                         });
                     } 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
                         });
                     }
                 }),
                 cat: (...seq) => _observe(({ next, complete: final, ...rest }) => {
                     let cleanup: () => void;
                     const len = seq.length;
                     const complete = (i: number) => i < len ?
                         () => {
                             const subscription = seq[i].subscribe({ next, complete: complete(i + 1), ...rest });
                             cleanup = subscription.unsubscribe.bind(subscription);
                         } : final;
                     cleanup = producer({ next, complete: complete(0), ...rest }) ?? noop;
                     return () => cleanup();
                 }),
                 pipe: <U>(op: OperatorFn<T, U>) => op(_observe(producer)),
                 next: collect(
                     producer,
                     ({ next, complete, error }) => ({
                         next: v => (next(v), complete()),
                         complete: () => error(new Error("The sequence is empty")),
                         error
                     })
                 ),
                 collect: collect(
                     producer,
                     ({ next, complete, error}) => {
                         const data: T[] = [];
                         return {
                             next: v => data.push(v),
                             complete: () => (next(data), complete()),
                             error
                         };
                     }
                 )
             });
             const collect = <T, U>(
                 producer: FusedProducer<T>,
                 collector: (result: FusedSink<U>) => FusedSink<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 })) ?? noop :
                         noop;
                     return () => {
                         h.destroy();
                         cleanup();
                     };
                 });
                 fused({
                     next: resolve,
                     error: reject,
                     complete: noop
                 });
             });
             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<T>(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/store.ts +39 -12

             import { PromiseOrValue } from "@implab/core-amd/interfaces";
             import { isCancellable, isPromise } from "@implab/core-amd/safe";
             import { observe, Observable, empty } from "./observable";
             import { after } from "dojo/aspect";
             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) =>
+            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;
             }
             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] |
+            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)
                             .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 }) => {
+                    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;
                 return (id: string | number, opts: GetOpts = {}) =>
                     observe<T>(({ next, complete, error }) => {
                         try {
                             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).then(undefined, error);
                                 if (isCancellable(result))
                                     return () => result.cancel();
                             } else {
                                 handle(result);
                             }
                         } catch (e) {
                             error(e);
                         }
                     }).cat(opts.observe !== false ? changes.pipe(filterItem(id)) : empty);
             };

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

             import Memory = require("dojo/store/Memory");
             import Observerable = require("dojo/store/Observable");
-            import { get } from "./store";
+            import { get, queryEx } from "./store";
             import tap = require("tap");
             interface Person {
                 id: string;
                 name: string;
                 age: number;
             }
             tap.test("store::get(...) tests", async t => {
                 const store = new Observerable(new Memory<Person>());
                 const getPerson = get(store);
                 const peterId = "id:peter";
                 const samId = "id:sam";
                 const peter = getPerson(peterId);
                 const sam = getPerson(samId);
                 const seq1 = await getPerson(peterId, { observe: false }).collect();
                 t.ok(seq1.length === 0, "Should be empty sequence");
                 let peterChangeCount = 0;
                 let samChangeCount = 0;
                 let peterDeleted = 0;
                 const peterSubscription = peter.subscribe({
                     next: () => peterChangeCount++,
                     complete: () => peterDeleted++
                 });
                 const samSubscription = sam.subscribe({
                     next: () => samChangeCount++
                 });
                 try {
                     t.equal(peterChangeCount, 0, "Should be no changes recorded");
                     store.put({id: peterId, name: "Peter", age: 30 });
                     t.equal(peterChangeCount, 1, "Should record 1 object change");
                     t.equal(samChangeCount, 0, "Should not record other object changes");
                     store.remove(peterId);
                     t.equal(peterDeleted, 1, "Should complete sequence");
                     t.equal(peterChangeCount, 1, "Should not record remove operations");
                     store.put({id: peterId, name: "Peter", age: 29});
                     t.equal(peterChangeCount, 1, "Should not record changes after completion");
                 } finally {
                     peterSubscription.unsubscribe();
                     samSubscription.unsubscribe();
                 }
+                store.put({ id: samId, name: "Sam", age: 29});
+                const [data, updates] = queryEx(store)({ age: 29}, { sort: [{attribute: "id"}] });
+                const dump: string[] = [];
+                const subscription = data
+                    .tap({
+                        complete: () => dump.push("eof")
+                    })
+                    .cat(updates)
+                    .tap({
+                        next: ({item: {id}}) => dump.push(id),
+                        complete: () => dump.push("eof")
+                    })
+                    .subscribe({});
+                t.same(dump, ["id:peter", "id:sam", "eof"]);
+                store.put({ id: "id:mary", name: "Mary", age: 29});
+                t.same(dump, ["id:peter", "id:sam", "eof", "id:mary"]);
+                subscription.unsubscribe();
             }).catch(() => { });

djx/src/test/ts/observable-tests.ts +23 0

             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 => 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(() => { });
+            tap.test(".tap() tests", async t => {
+                const side: number[] = [];
+                of(1,2)
+                    .tap({next: v => side.push(v), complete: () => side.push(0)})
+                    .tap({next: v => side.push(v*v)})
+                    .subscribe({});
+                t.same(side, [1,1,2,4,0], ".tap() should be called in the order of registration");
+                side.length = 0;
+                await new Promise<void>(resolve => {
+                    of(1,2,delay(1).then(() => 3))
+                        .tap({next: v => side.push(v)})
+                        .tap({ next: v => v === 1 && resolve()})
+                        .subscribe({});
+                });
+                t.same(side, [1,2], ".tap() should be processed synchronously");
             }).catch(() => {});

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