implab/implabjs-djx Commit - r138:98b2c550c676

Added a fallback error handler to observables, it will report unhandled errors

cin -

r138:98b2c550c676 v1.9.0-rc3 default

parent child

Context file:

r138:98b2c550c676

Collapse all files

djx/src/main/ts/observable.ts +7 -1

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

djx/src/main/ts/store.ts +6 -4

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

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