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 =>
              export const isSubscribable = <T = unknown>(v: unknown): v is Subscribable<T> =>
                  v !== null && v !== undefined && typeof (v as Subscribable<unknown>).subscribe === "function";
-             const 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;
-             };
-             export const observe = <T>(producer: Producer<T>): Observable<T> => new ObservableImpl<T>(fuse(producer));
+             export const observe = <T>(producer: Producer<T>): Observable<T> => new ObservableImpl<T>(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()
                      }
                  );
-             export const empty = new ObservableImpl<never>(({ complete }) => complete());
+             export const empty: Observable<never> = new ObservableImpl<never>(({ complete }) => complete());
              export type * from "./observable/interfaces";
  No newline at end of file

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 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;
                              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({
-                         next: consumer.next ? consumer.next.bind(consumer) : noop,
-                         error: err => finalize() && (consumer.error ? consumer.error(err) : errorFallback(err)),
-                         complete: () => finalize() && consumer.complete && consumer.complete(),
-                         isClosed
+                     teardown((0, this._producer)({
+                         next: next !== noop ? (v => isClosed() || next(v)) : noop,
+                         error,
+                         complete: () => finalize() && consumer.complete && consumer.complete()
                      }));
                      return { unsubscribe: finalize };
                  }
                  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,
-                             isClosed
+                             error: tapError ? (e => (tapError(e), error(e))) : error
                          }));
                  }
                      );
                  }
+                 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 }) => {
                      }
                  }
+                 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;
                                          complete();
                                      }
                                  },
-                                 error,
-                                 isClosed
+                                 error
                              });
                          });
                      } 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(_acc);
                                      complete();
                                  },
-                                 error,
-                                 isClosed
+                                 error
                              });
                          });
                      }
                      return new Promise<T>((resolve, reject) => {
                          ct.throwIfRequested();
-                         const { teardown, finalize, isClosed } = subscription();
-                         const error = (err: unknown) => finalize() && reject(err);
-                         const h = ct.register(error);
-                         teardown(() => h.destroy());
+                         const { teardown, finalize, isClosed, error } = subscription(reject, ct);
                          teardown((0, this._producer)({
                              next: (item: T) => finalize() && resolve(item),
                              complete: () => finalize() && reject(new Error("The sequence is empty")),
                          const acc: T[] = [];
-                         const { teardown, finalize, isClosed } = subscription();
-                         const error = (err: unknown) => finalize() && reject(err);
-                         const h = ct.register(error);
-                         teardown(() => h.destroy());
+                         const { teardown, finalize, isClosed, error } = subscription(reject, ct);
                          teardown((0, this._producer)({
                              next: (item: T) => isClosed() ? void (0) : acc.push(item),
                              complete: () => finalize() && resolve(acc),
                      });
                  }
+                 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
+                         }));
+                     });
+                 }
              }
  No newline at end of file

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

              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;
              }
                  /** 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>;
              }
  No newline at end of file

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

              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 = {

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