feat(rstream): #281 update Subscription error/teardown logic

- replace old `Subscription` class w/ what was recently `Sub2` (removed) - update/fix done(), subscribe()/unsubscribe() logic - update related constructs (Stream, StreamSync, MetaStream, etc.) - update Stream ctor (and factory fns) to support error handler opts arg - update Timeout error dispatch - fix typehints
thi-ng · Mar 10, 2021 · a9e4040 · a9e4040
1 parent db0ab34
commit a9e4040
Show file tree

Hide file tree

Showing 14 changed files with 183 additions and 498 deletions.
diff --git a/packages/rstream/src/forkjoin.ts b/packages/rstream/src/forkjoin.ts
@@ -129,8 +129,8 @@ export const forkJoin = <IN, MSG, RES, OUT>(
                 String(id),
                 opts.src
                     .transform(map((x: IN) => opts.fork(id, numWorkers, x)))
-                    .subscribe(
-                        tunnel<MSG, RES>({
+                    .subscribe<RES>(
+                        tunnel({
                             src: opts.worker,
                             transferables: opts.transferables,
                             interrupt: opts.interrupt === true,

diff --git a/packages/rstream/src/from/promise.ts b/packages/rstream/src/from/promise.ts
@@ -1,4 +1,4 @@
-import { CloseMode, CommonOpts, State } from "../api";
+import { CloseMode, State, WithErrorHandlerOpts } from "../api";
 import { Stream } from "../stream";
 import { optsWithID } from "../utils/idgen";
 
@@ -13,7 +13,10 @@ import { optsWithID } from "../utils/idgen";
  * @param src -
  * @param opts -
  */
-export const fromPromise = <T>(src: Promise<T>, opts?: Partial<CommonOpts>) => {
+export const fromPromise = <T>(
+    src: Promise<T>,
+    opts?: Partial<WithErrorHandlerOpts>
+) => {
     let canceled = false;
     let isError = false;
     let err: any = {};

diff --git a/packages/rstream/src/from/promises.ts b/packages/rstream/src/from/promises.ts
@@ -1,5 +1,5 @@
 import { mapcat } from "@thi.ng/transducers";
-import type { CommonOpts, ISubscription } from "../api";
+import type { ISubscription, WithErrorHandlerOpts } from "../api";
 import { optsWithID } from "../utils/idgen";
 import { fromPromise } from "./promise";
 
@@ -38,10 +38,11 @@ import { fromPromise } from "./promise";
  * ```
  *
  * @param promises -
+ * @param opts -
  */
 export const fromPromises = <T>(
     promises: Iterable<T | PromiseLike<T>>,
-    opts?: Partial<CommonOpts>
+    opts?: Partial<WithErrorHandlerOpts>
 ): ISubscription<T[], T> =>
     fromPromise(Promise.all(promises), optsWithID("promises", opts)).transform(
         mapcat((x: T[]) => x)

diff --git a/packages/rstream/src/from/worker.ts b/packages/rstream/src/from/worker.ts
@@ -1,9 +1,9 @@
-import { CommonOpts, LOGGER } from "../api";
+import { LOGGER, WithErrorHandlerOpts } from "../api";
 import { Stream } from "../stream";
 import { optsWithID } from "../utils/idgen";
 import { makeWorker } from "../utils/worker";
 
-export interface FromWorkerOpts extends CommonOpts {
+export interface FromWorkerOpts extends WithErrorHandlerOpts {
     /**
      * If true, the worker will be terminated when the stream
      * is being closed.
@@ -43,17 +43,17 @@ export const fromWorker = <T>(
     const _worker = makeWorker(worker);
     opts = optsWithID("worker", opts);
     return new Stream<T>((stream) => {
-        const ml = (e: MessageEvent) => {
+        const msgListener = (e: MessageEvent) => {
             stream.next(e.data);
         };
-        const el = (e: MessageEvent) => {
+        const errListener = (e: MessageEvent) => {
             stream.error(e.data);
         };
-        _worker.addEventListener("message", ml);
-        _worker.addEventListener("error", <EventListener>el);
+        _worker.addEventListener("message", msgListener);
+        _worker.addEventListener("error", <EventListener>errListener);
         return () => {
-            _worker.removeEventListener("message", ml);
-            _worker.removeEventListener("error", <EventListener>el);
+            _worker.removeEventListener("message", msgListener);
+            _worker.removeEventListener("error", <EventListener>errListener);
             if (opts!.terminate !== false) {
                 LOGGER.info("terminating worker", _worker);
                 _worker.terminate();

diff --git a/packages/rstream/src/index.ts b/packages/rstream/src/index.ts
@@ -6,7 +6,6 @@ export * from "./stream";
 export * from "./stream-merge";
 export * from "./stream-sync";
 export * from "./subscription";
-export * from "./sub2";
 export * from "./trigger";
 export * from "./tween";
 

diff --git a/packages/rstream/src/metastream.ts b/packages/rstream/src/metastream.ts
@@ -160,7 +160,7 @@ export class MetaStream<A, B> extends Subscription<A, B> {
     }
 
     unsubscribe(sub?: Subscription<B, any>) {
-        if (this.stream && (!sub || this.subs.length === 1)) {
+        if (this.stream && (!sub || this.subs.size === 1)) {
             this.detach(!sub);
         }
         return super.unsubscribe(sub);

diff --git a/packages/rstream/src/stream-sync.ts b/packages/rstream/src/stream-sync.ts
@@ -6,13 +6,7 @@ import {
     partitionSync,
     PartitionSync,
 } from "@thi.ng/transducers";
-import {
-    ISubscribable,
-    ISubscription,
-    LOGGER,
-    State,
-    TransformableOpts,
-} from "./api";
+import { ISubscribable, ISubscription, LOGGER, TransformableOpts } from "./api";
 import { Subscription } from "./subscription";
 import { isFirstOrLastInput } from "./utils/checks";
 import { optsWithID } from "./utils/idgen";
@@ -271,10 +265,10 @@ export class StreamSync<
 
     unsubscribe(sub?: Subscription<B, any>) {
         if (!sub) {
+            LOGGER.debug(this.id, "unsub sources");
             for (let s of this.sources.values()) {
                 s.unsubscribe();
             }
-            this.state = State.DONE;
             this.sources.clear();
             this.psync.clear();
             this.realSourceIDs.clear();

diff --git a/packages/rstream/src/stream.ts b/packages/rstream/src/stream.ts
@@ -4,10 +4,12 @@ import {
     CommonOpts,
     IStream,
     ISubscriber,
+    ISubscription,
     LOGGER,
     StreamCancel,
     StreamSource,
     TransformableOpts,
+    WithErrorHandlerOpts,
 } from "./api";
 import { Subscription } from "./subscription";
 import { optsWithID } from "./utils/idgen";
@@ -97,27 +99,32 @@ export class Stream<T> extends Subscription<T, T> implements IStream<T> {
     protected _cancel: StreamCancel | undefined;
     protected _inited: boolean;
 
-    constructor(opts?: Partial<CommonOpts>);
-    constructor(src: StreamSource<T>, opts?: Partial<CommonOpts>);
+    constructor(opts?: Partial<WithErrorHandlerOpts>);
+    constructor(src: StreamSource<T>, opts?: Partial<WithErrorHandlerOpts>);
     constructor(
-        src?: StreamSource<T> | Partial<CommonOpts>,
-        opts?: Partial<CommonOpts>
+        src?: StreamSource<T> | Partial<WithErrorHandlerOpts>,
+        opts?: Partial<WithErrorHandlerOpts>
     ) {
-        const [_src, _opts] = isFunction(src) ? [src, opts] : [undefined, src];
-        super(undefined, optsWithID("stream", _opts));
+        const [_src, _opts] = isFunction(src)
+            ? [src, opts || {}]
+            : [undefined, src || {}];
+        super(
+            _opts.error ? { error: _opts.error } : undefined,
+            optsWithID("stream", _opts)
+        );
         this.src = _src;
         this._inited = false;
     }
 
-    subscribe<C>(sub: Subscription<T, C>): Subscription<T, C>;
+    subscribe<C>(sub: ISubscription<T, C>): ISubscription<T, C>;
     subscribe(
         sub: Partial<ISubscriber<T>>,
         opts?: Partial<CommonOpts>
-    ): Subscription<T, T>;
+    ): ISubscription<T, T>;
     subscribe<C>(
         sub: Partial<ISubscriber<C>>,
         opts?: Partial<TransformableOpts<T, C>>
-    ): Subscription<T, C>;
+    ): ISubscription<T, C>;
     subscribe(
         sub: Partial<ISubscriber<any>>,
         opts: Partial<TransformableOpts<any, any>> = {}
@@ -130,12 +137,12 @@ export class Stream<T> extends Subscription<T, T> implements IStream<T> {
         return $sub;
     }
 
-    unsubscribe(sub?: Subscription<T, any>) {
+    unsubscribe(sub?: ISubscription<T, any>) {
         const res = super.unsubscribe(sub);
         if (
             res &&
             (!sub ||
-                ((!this.subs || !this.subs.length) &&
+                ((!this.subs || !this.subs.size) &&
                     this.closeOut !== CloseMode.NEVER))
         ) {
             this.cancel();
@@ -151,7 +158,7 @@ export class Stream<T> extends Subscription<T, T> implements IStream<T> {
     }
 
     error(e: any) {
-        super.error(e);
+        if (super.error(e)) return true;
         this.cancel();
         return false;
     }