implab/implabjs-djx Commit - r157:f9518367061a

WIP observables

cin -

r157:f9518367061a default

parent child

Context file:

r157:f9518367061a

Collapse all files

djx/src/main/ts/observable.ts +4 -40

             import { isPromise } from "@implab/core-amd/safe";
-            import { Unsubscribable, Producer, FusedSink, Observable, Subscribable } from "./observable/interfaces";
+            import { Unsubscribable, Producer, Observable, Subscribable } from "./observable/interfaces";
             import { ObservableImpl } from "./observable/ObservableImpl";
             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";
-            const noop = () => { };
+            export const observe = <T>(producer: Producer<T>): Observable<T> => new ObservableImpl<T>(producer);
-            /** 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;
-            };
-            export const observe = <T>(producer: Producer<T>): Observable<T> => new ObservableImpl<T>(fuse(producer));
             /** Converts an array to the observable sequence of its elements. */
-            export const ofArray = <T>(items: T[]) => new ObservableImpl<T>(
+            export const ofArray = <T>(items: T[]): Observable<T> => new ObservableImpl<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 = new ObservableImpl<never>(({ complete }) => complete());
+            export const empty: Observable<never> = new ObservableImpl<never>(({ complete }) => complete());
             export type * from "./observable/interfaces";

djx/src/main/ts/observable/ObservableImpl.ts +69 -31

             import { id as mid } from "module";
-            import { AccumulatorFn, Observable, Observer, OperatorFn, Producer, Subscribable } from "./interfaces";
+            import { AccumulatorFn, Observable, Observer, OperatorFn, Producer, Sink, Subscribable, Unsubscribable } from "./interfaces";
             import { TraceSource } from "@implab/core-amd/log/TraceSource";
             import { Predicate } from "@implab/core-amd/interfaces";
             import { Cancellation } from "@implab/core-amd/Cancellation";
             const trace = TraceSource.get(mid);
             const noop = () => { };
             const errorFallback = (e: unknown) => trace.error("Unhandled observable error: {0}", e);
-            const subscription = () => {
+            const subscription = (reject: (err: unknown) => void, ct = Cancellation.none) => {
                 let done = false;
                 let cleanup = noop;
                 const finalize = () => done ? false : (cleanup(), done = true);
                 const isClosed = () => done;
                 const teardown = (cb: void | (() => void)) => {
                     if (cb) {
                         if (done) {
                             cb();
                         } else {
                             const _prev = cleanup;
-                            cleanup = () => (_prev(), cb());
+                            cleanup = _prev === noop ? cb : () => (_prev(), cb());
                         }
                     }
                 };
-                return { finalize, isClosed, teardown };
+                const error = (err: unknown) => finalize() && reject(err);
+                if (ct.isSupported()) {
+                    const h = ct.register(error);
+                    teardown(() => h.destroy());
+                }
+                return { finalize, isClosed, teardown, error };
             };
+            type FusedSink<T> = Omit<Sink<T>, "isClosed">;
+            export const fuse = <T>(producer: Producer<T>) => (consumer: FusedSink<T>) => {
+                const reject = (err: unknown) => consumer.error ? consumer.error(err) : errorFallback(err);
+                const { teardown, isClosed, finalize, error } = subscription(reject);
+                const next = consumer.next ? consumer.next.bind(consumer) : noop;
+                teardown(producer({
+                    next: next !== noop ? (v => isClosed() || next(v)) : noop,
+                    error,
+                    complete: () => finalize() && consumer.complete && consumer.complete(),
+                    isClosed
+                }));
+                return { unsubscribe: finalize };
+            };
+            type FusedProducer<T> = (consumer: FusedSink<T>) => void | (() => void);
             export class ObservableImpl<T> implements Observable<T> {
-                private readonly _producer: Producer<T>;
+                private readonly _producer: FusedProducer<T>;
-                constructor(producer: Producer<T>) {
+                constructor(producer: FusedProducer<T>) {
                     this._producer = producer;
                 }
                 subscribe(consumer: Observer<T> = {}) {
-                    const { teardown, isClosed, finalize } = subscription();
+                    const reject = (err: unknown) => consumer.error ? consumer.error(err) : errorFallback(err);
+                    const next = consumer.next ? consumer.next.bind(consumer) : noop;
+                    const { teardown, isClosed, finalize, error } = subscription(reject);
-                    teardown(this._producer({
+                    teardown((0, this._producer)({
-                        next: consumer.next ? consumer.next.bind(consumer) : noop,
+                        next: next !== noop ? (v => isClosed() || next(v)) : noop,
-                        error: err => finalize() && (consumer.error ? consumer.error(err) : errorFallback(err)),
+                        error,
-                        complete: () => finalize() && consumer.complete && consumer.complete(),
+                        complete: () => finalize() && consumer.complete && consumer.complete()
-                        isClosed
                     }));
                     return { unsubscribe: finalize };
                 }
                 map<T2>(mapper: (value: T) => T2) {
                     return new ObservableImpl<T2>(({ next, ...rest }) =>
                         this._producer({
                             next: next !== noop ? (v: T) => next(mapper(v)) : noop,
                             ...rest
                         }));
                 }
                 tap({ next: tapNext, complete: tapComplete, error: tapError }: Observer<T>) {
-                    return new ObservableImpl<T>(({ next, complete, error, isClosed }) =>
+                    return new ObservableImpl<T>(({ next, complete, error }) =>
                         this._producer({
                             next: tapNext ? (v => (tapNext(v), next(v))) : next,
                             complete: tapComplete ? (() => (tapComplete(), complete())) : complete,
-                            error: tapError ? (e => (tapError(e), error(e))) : error,
+                            error: tapError ? (e => (tapError(e), error(e))) : error
-                            isClosed
                         }));
                 }
                 filter(predicate: Predicate<T>) {
                     return new ObservableImpl<T>(({ next, ...rest }) =>
                         this._producer({
                             next: next !== noop ? (v: T) => predicate(v) ? next(v) : void (0) : noop,
                             ...rest
                         })
                     );
                 }
                 until(predicate: Predicate<T>) {
                     return new ObservableImpl<T>(({ next, complete, ...rest }) =>
                         this._producer({
                             next: v => predicate(v) ? complete() : next(v),
                             complete,
                             ...rest
                         })
                     );
                 }
                 while(predicate: Predicate<T>) {
                     return new ObservableImpl<T>(({ next, complete, ...rest }) =>
                         this._producer({
                             next: v => predicate(v) ? next(v) : complete(),
                             complete,
                             ...rest
                         })
                     );
                 }
+                scan<A>(accumulator: AccumulatorFn<T, A>, initial: A): Observable<A>;
+                scan(accumulator: AccumulatorFn<T, T>): Observable<T>;
                 scan<A>(...args: [AccumulatorFn<T, A>, A] | [AccumulatorFn<T, T>]) {
                     if (args.length === 1) {
                         return new ObservableImpl<T | A>(({ next, ...rest }) => {
                             const [accumulator] = args;
                             let _acc: T;
                             let index = 0;
                             return this._producer({
                                 next: next !== noop ? (v: T) => next(index++ === 0 ? _acc = v : _acc = accumulator(_acc, v)) : noop,
                                 ...rest
                             });
                         });
                     } else {
                         return new ObservableImpl<T | A>(({ next, ...rest }) => {
                             const [accumulator, initial] = args;
                             let _acc = initial;
                             return this._producer({
                                 next: next !== noop ? (v: T) => next(_acc = accumulator(_acc, v)) : noop,
                                 ...rest
                             });
                         });
                     }
                 }
+                reduce<A>(accumulator: AccumulatorFn<T, A>, initial: A): Observable<A>;
+                reduce(accumulator: AccumulatorFn<T, T>): Observable<T>;
                 reduce<A>(...args: [AccumulatorFn<T, A>, A] | [AccumulatorFn<T, T>]) {
                     if (args.length === 1) {
-                        return new ObservableImpl<T | A>(({ next, complete, error, isClosed }) => {
+                        return new ObservableImpl<T | A>(({ next, complete, error }) => {
                             const [accumulator] = args;
                             let _acc: T;
                             let index = 0;
                             return this._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,
+                                error
-                                isClosed
                             });
                         });
                     } else {
-                        return new ObservableImpl<T | A>(({ next, complete, error, isClosed }) => {
+                        return new ObservableImpl<T | A>(({ next, complete, error }) => {
                             const [accumulator, initial] = args;
                             let _acc = initial;
                             return this._producer({
                                 next: next !== noop ? (v: T) => {
                                     _acc = accumulator(_acc, v);
                                 } : noop,
                                 complete: () => {
                                     next(_acc);
                                     complete();
                                 },
-                                error,
+                                error
-                                isClosed
                             });
                         });
                     }
                 }
                 cat(...seq: Subscribable<T>[]) {
                     return new ObservableImpl<T>(({ 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 = this._producer({ next, complete: complete(0), ...rest }) ?? noop;
                         return () => cleanup();
                     });
                 }
                 pipe<U>(op: OperatorFn<T, U>): Observable<U> {
                     return op(this);
                 }
                 next(ct = Cancellation.none) {
                     return new Promise<T>((resolve, reject) => {
                         ct.throwIfRequested();
-                        const { teardown, finalize, isClosed } = subscription();
+                        const { teardown, finalize, isClosed, error } = subscription(reject, ct);
-                        const error = (err: unknown) => finalize() && reject(err);
-                        const h = ct.register(error);
-                        teardown(() => h.destroy());
                         teardown((0, this._producer)({
                             next: (item: T) => finalize() && resolve(item),
                             complete: () => finalize() && reject(new Error("The sequence is empty")),
                             error,
                             isClosed
                         }));
                     });
                 }
                 collect(ct = Cancellation.none) {
                     return new Promise<T[]>((resolve, reject) => {
                         ct.throwIfRequested();
                         const acc: T[] = [];
-                        const { teardown, finalize, isClosed } = subscription();
+                        const { teardown, finalize, isClosed, error } = subscription(reject, ct);
-                        const error = (err: unknown) => finalize() && reject(err);
-                        const h = ct.register(error);
-                        teardown(() => h.destroy());
                         teardown((0, this._producer)({
                             next: (item: T) => isClosed() ? void (0) : acc.push(item),
                             complete: () => finalize() && resolve(acc),
                             error,
                             isClosed
                         }));
                     });
                 }
+                forEach(next: (item: T) => void, ct = Cancellation.none): Promise<void> {
+                    return new Promise<void>((resolve, reject) => {
+                        ct.throwIfRequested();
+                        const { teardown, finalize, isClosed, error } = subscription(reject, ct);
+                        teardown((0, this._producer)({
+                            next: item => !isClosed() && next(item),
+                            complete: () => finalize() && resolve(),
+                            error,
+                            isClosed
+                        }));
+                    });
+                }
             }

djx/src/main/typings/observable/interfaces.d.ts +12 -4

             import { ICancellation } from "@implab/core-amd/interfaces";
             /**
              * 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.
                  *
                  * This method is useful to notify a producer while it's emitting the series
                  * of synchronous events.
                  */
                 isClosed: () => boolean;
             };
             export type Producer<T> = (sink: Sink<T>) => (void | (() => void));
-            export type FusedSink<T> = Omit<Sink<T>, "isClosed">;
-            export type FusedProducer<T> = (sink: FusedSink<T>) => (void | (() => void));
             export interface Unsubscribable {
                 unsubscribe(): void;
             }
             export interface Subscribable<T> {
                 /** 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[]>;
+                /**
+                 * Iterates through the elements in this observable until the end of the
+                 * sequence is reached or the operation is cancelled.
+                 *
+                 * @param next The callback for the next item in the sequence
+                 * @param ct The optional cancellation token for this operation
+                 *
+                 * @returns A promise which will be fulfilled when the operation is completed.
+                 *   In case of the cancellation this promise will be rejected.
+                 */
+                forEach(next: (item: T) => void, ct?: ICancellation): Promise<void>;
             }

djx/src/test/ts/observable-tests.ts +4 -1

             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 }) => {
+            const subj1 = observe<number>(({ next, error, complete }) => {
                 next(1);
                 complete();
                 next(2);
+                next(3);
+                error(new Error("This error should be ignored"));
+                next(4);
             });
             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(() => { });
             tap.test(".while() tests", async t => {
                 const seq = of(1, 2, 3, 4).while(v => v <= 2);
                 t.same(await seq.collect(), [1, 2], "Should collect only taken elements");
                 const data: number[] = [];
                 let complete = 0;
                 seq.subscribe({
                     next: v => data.push(v),
                     complete: () => complete++
                 });
                 t.same(data, [1, 2], "Should receive only taken elements");
                 t.equal(complete, 1, "Complete should run once");
             }).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