refactor(rstream): update types/generics

- remove default generics from ISubscription due to inference troubles - add ITransformable.map() - fix unsubscribe() arg types in various classes - update tests (generics only)
thi-ng · Mar 12, 2021 · c982288 · c982288
1 parent 36bff90
commit c982288
Show file tree

Hide file tree

Showing 9 changed files with 25 additions and 17 deletions.
diff --git a/packages/rstream/src/api.ts b/packages/rstream/src/api.ts
@@ -134,7 +134,7 @@ export interface ISubscriber<T> {
     /**
      * Internal use only. Do not use.
      */
-    __owner?: ISubscription<any>;
+    __owner?: ISubscription<any, any>;
     [id: string]: any;
 }
 
@@ -210,9 +210,13 @@ export interface ITransformable<B> {
     transform<C>(
         opts: WithTransform<B, C> & Partial<WithErrorHandlerOpts>
     ): ISubscription<B, C>;
+    map<C>(
+        fn: Fn<B, C>,
+        opts?: Partial<WithErrorHandlerOpts>
+    ): ISubscription<B, C>;
 }
 
-export interface ISubscription<A = any, B = A>
+export interface ISubscription<A, B>
     extends IDeref<B | undefined>,
         ISubscriber<A>,
         ISubscribable<B>,

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

diff --git a/packages/rstream/src/stream-merge.ts b/packages/rstream/src/stream-merge.ts
@@ -136,7 +136,7 @@ export class StreamMerge<A, B> extends Subscription<A, B> {
         return ok;
     }
 
-    unsubscribe(sub?: Subscription<B, any>) {
+    unsubscribe(sub?: ISubscription<B, any>) {
         if (!sub) {
             for (let s of this.sources.values()) {
                 s.unsubscribe();

diff --git a/packages/rstream/src/stream-sync.ts b/packages/rstream/src/stream-sync.ts
@@ -263,7 +263,7 @@ export class StreamSync<
         return <A>res;
     }
 
-    unsubscribe(sub?: Subscription<B, any>) {
+    unsubscribe(sub?: ISubscription<B, any>) {
         if (!sub) {
             LOGGER.debug(this.id, "unsub sources");
             for (let s of this.sources.values()) {

diff --git a/packages/rstream/src/subs/asidechain.ts b/packages/rstream/src/subs/asidechain.ts
@@ -12,7 +12,7 @@ export abstract class ASidechain<A, S, B> extends Subscription<A, B> {
         super(undefined, opts);
     }
 
-    unsubscribe(sub?: Subscription<any, any>) {
+    unsubscribe(sub?: ISubscription<B, any>) {
         const res = super.unsubscribe(sub);
         if (!sub || !this.subs.length) {
             this.sideSub.unsubscribe();

diff --git a/packages/rstream/src/subscription.ts b/packages/rstream/src/subscription.ts
@@ -85,7 +85,7 @@ export class Subscription<A, B> implements ISubscription<A, B> {
     closeIn: CloseMode;
     closeOut: CloseMode;
     parent?: ISubscription<any, A>;
-    __owner?: ISubscription<any>;
+    __owner?: ISubscription<any, any>;
 
     protected xform?: Reducer<B[], A>;
     protected cacheLast: boolean;
@@ -228,7 +228,7 @@ export class Subscription<A, B> implements ISubscription<A, B> {
         return this.transform(map(fn), opts || {});
     }
 
-    unsubscribe(sub?: Partial<ISubscription<B>>) {
+    unsubscribe(sub?: ISubscription<B, any>) {
         return sub ? this.unsubscribeChild(sub) : this.unsubscribeSelf();
     }
 
@@ -240,7 +240,7 @@ export class Subscription<A, B> implements ISubscription<A, B> {
         return true;
     }
 
-    protected unsubscribeChild(sub: Partial<ISubscription<B>>) {
+    protected unsubscribeChild(sub: ISubscription<B, any>) {
         LOGGER.debug(this.id, "unsub child", sub.id);
         const idx = this.subs.indexOf(sub);
         if (idx >= 0) {

diff --git a/packages/rstream/test/from-iterable.ts b/packages/rstream/test/from-iterable.ts
@@ -45,7 +45,7 @@ describe("fromIterable", function () {
     });
 
     it("finishes", (_done) => {
-        let sub: ISubscription = src.subscribe({
+        let sub: ISubscription<any, any> = src.subscribe({
             next() {},
             done() {
                 assert.strictEqual(src.getState(), State.DONE, "src not done");

diff --git a/packages/rstream/test/subscription.ts b/packages/rstream/test/subscription.ts
@@ -30,7 +30,7 @@ describe("Subscription", function () {
         });
         let state2 = State.IDLE;
         let err: any;
-        let sub2: ISubscription;
+        let sub2: ISubscription<any, any>;
         sub2 = src.subscribe({
             next() {
                 throw 1;

diff --git a/packages/rstream/test/utils.ts b/packages/rstream/test/utils.ts
@@ -1,16 +1,20 @@
 import * as assert from "assert";
 import { ISubscription, State } from "../src";
 
-export const assertState = (x: ISubscription, state: State) =>
+export const assertState = (x: ISubscription<any, any>, state: State) =>
     assert.strictEqual(x.getState(), state, `${x.id} != ${State[state]}`);
 
-export const assertIdle = (x: ISubscription) => assertState(x, State.IDLE);
+export const assertIdle = (x: ISubscription<any, any>) =>
+    assertState(x, State.IDLE);
 
-export const assertActive = (x: ISubscription) => assertState(x, State.ACTIVE);
+export const assertActive = (x: ISubscription<any, any>) =>
+    assertState(x, State.ACTIVE);
 
-export const assertDone = (x: ISubscription) => assertState(x, State.DONE);
+export const assertDone = (x: ISubscription<any, any>) =>
+    assertState(x, State.DONE);
 
-export const assertUnsub = (x: ISubscription) =>
+export const assertUnsub = (x: ISubscription<any, any>) =>
     assertState(x, State.UNSUBSCRIBED);
 
-export const assertError = (x: ISubscription) => assertState(x, State.ERROR);
+export const assertError = (x: ISubscription<any, any>) =>
+    assertState(x, State.ERROR);