implab/implabjs-djx Commit - r144:63215d91ae4b

added tap() method to observable...

cin -

r144:63215d91ae4b v1.9.0-rc6 default

parent child

Context file:

r144:63215d91ae4b

Collapse all files

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