refactor: improve type checking for Stream and Observable inputs across operators

Author: yiludege

src/observable.ts

@@ -1,5 +1,5 @@
 import { produce, createDraft, finishDraft } from 'limu'
-import { Stream } from './stream'
+import type { Stream } from './stream'
 import { safeCallback, isObjectLike, isAsyncFunction, isPromiseLike } from './utils'
 import {
   OnFulfilled,

src/operators/combine.ts

@@ -2,7 +2,7 @@ import { Observable } from '../observable'
 import { Stream } from '../stream'
 import { useUnsubscribeCallback } from '../utils'
 import { StreamTupleValues, PromiseStatus } from '../types'
-import { getGlobalFluthFactory } from '../utils'
+import { getGlobalFluthFactory, checkStreamOrObservableInput } from '../utils'
 
 /**
  * combine takes multiple streams or Observable, and return a stream that emits values from all the input streams.
@@ -12,23 +12,23 @@ import { getGlobalFluthFactory } from '../utils'
  * @returns {Stream}
  */
 export const combine = <T extends (Stream | Observable)[]>(...args$: T) => {
+  // check input type
+  if (!checkStreamOrObservableInput(args$, true)) {
+    throw new Error('combine operator only accepts Stream or Observable as input')
+  }
+
   const stream$ = (getGlobalFluthFactory()?.(
-    args$.map((arg$) => arg$?._getProtectedProperty?.('_v')) as StreamTupleValues<T>,
+    args$.map((arg$) => arg$._getProtectedProperty('_v')) as StreamTupleValues<T>,
   ) ||
     new Stream<StreamTupleValues<T>>(
-      args$.map((arg$) => arg$?._getProtectedProperty?.('_v')) as StreamTupleValues<T>,
+      args$.map((arg$) => arg$._getProtectedProperty('_v')) as StreamTupleValues<T>,
     )) as Stream<StreamTupleValues<T>>
   const payload: StreamTupleValues<T> = [] as any
   const promiseStatus = [...Array(args$.length)].map(() => PromiseStatus.PENDING)
   let finishCount = 0
   const { unsubscribeCallback } = useUnsubscribeCallback(stream$, args$.length)
   const completeCallback = () => (finishCount += 1)
 
-  // check input type
-  if (args$.some((arg$) => !(arg$ instanceof Stream) && !(arg$ instanceof Observable))) {
-    throw new Error('combine operator only accepts Stream or Observable as input')
-  }
-
   // if no input, return an empty stream
   if (args$.length === 0) {
     return stream$

src/operators/concat.ts

@@ -1,7 +1,7 @@
 import { Observable } from '../observable'
 import { Stream } from '../stream'
 import { StreamTupleValues, PromiseStatus } from '../types'
-import { getGlobalFluthFactory } from '../utils'
+import { getGlobalFluthFactory, checkStreamOrObservableInput } from '../utils'
 
 /**
  * concat takes multiple streams or Observable, and return a stream that emits values in the order of the input streams.
@@ -12,18 +12,18 @@ import { getGlobalFluthFactory } from '../utils'
  * @returns {Stream}
  */
 export const concat = <T extends (Stream | Observable)[]>(...args$: T) => {
-  const stream$ = (getGlobalFluthFactory()?.(args$?.[0]?._getProtectedProperty?.('_v')) ||
+  // check input type
+  if (!checkStreamOrObservableInput(args$, true)) {
+    throw new Error('concat operator only accepts Stream or Observable as input')
+  }
+
+  const stream$ = (getGlobalFluthFactory()?.(args$[0]?._getProtectedProperty('_v')) ||
     new Stream<StreamTupleValues<T>[number]>(
-      args$?.[0]?._getProtectedProperty?.('_v') as StreamTupleValues<T>[number],
+      args$[0]?._getProtectedProperty('_v') as StreamTupleValues<T>[number],
     )) as Stream<StreamTupleValues<T>[number]>
   const finishFlag = [...Array(args$.length)].map(() => false)
   const unsubscribeFlag = [...Array(args$.length)].map(() => false)
 
-  // check input type
-  if (args$.some((arg$) => !(arg$ instanceof Stream) && !(arg$ instanceof Observable))) {
-    throw new Error('concat operator only accepts Stream or Observable as input')
-  }
-
   // check input empty
   if (args$.length === 0) {
     return stream$

src/operators/finish.ts

@@ -2,7 +2,7 @@ import { Observable } from '../observable'
 import { Stream } from '../stream'
 import { useUnsubscribeCallback } from '../utils'
 import { StreamTupleValues, PromiseStatus } from '../types'
-import { getGlobalFluthFactory } from '../utils'
+import { getGlobalFluthFactory, checkStreamOrObservableInput } from '../utils'
 
 /**
  * @description
@@ -13,18 +13,18 @@ import { getGlobalFluthFactory } from '../utils'
  * @returns {Stream}
  */
 export const finish = <T extends (Stream | Observable)[]>(...args$: T) => {
+  // check input type
+  if (!checkStreamOrObservableInput(args$, true)) {
+    throw new Error('finish operator only accepts Stream or Observable as input')
+  }
+
   const stream$ = (getGlobalFluthFactory()?.() || new Stream<StreamTupleValues<T>>()) as Stream<
     StreamTupleValues<T>
   >
   const payload: StreamTupleValues<T> = [] as any
   let finishCount = 0
   let rejectFlag = false
 
-  // check input type
-  if (args$.some((arg$) => !(arg$ instanceof Stream) && !(arg$ instanceof Observable))) {
-    throw new Error('finish operator only accepts Stream or Observable as input')
-  }
-
   // if no input, return an empty stream
   if (args$.length === 0) {
     return stream$

src/operators/fork.ts

@@ -1,6 +1,6 @@
 import { Observable } from '../observable'
 import { Stream } from '../stream'
-import { getGlobalFluthFactory } from '../utils'
+import { getGlobalFluthFactory, checkStreamOrObservableInput } from '../utils'
 
 /**
  * fork takes a stream or Observable, and returns a stream that emits the same value as the input stream.
@@ -13,14 +13,15 @@ import { getGlobalFluthFactory } from '../utils'
  **/
 
 export const fork = <T>(arg$: Stream<T> | Observable<T>, autoUnsubscribe = true): Stream<T> => {
-  const stream$ = (getGlobalFluthFactory()?.() || new Stream<T>()) as Stream<T>
-  let finishFlag = false
-
   // check input type
-  if (!(arg$ instanceof Stream) && !(arg$ instanceof Observable)) {
+  if (!checkStreamOrObservableInput(arg$)) {
     throw new Error('fork operator only accepts Stream or Observable as input')
   }
 
+  const stream$ = (getGlobalFluthFactory()?.(arg$._getProtectedProperty('_v')) ||
+    new Stream<T>(arg$._getProtectedProperty('_v') as T)) as Stream<T>
+  let finishFlag = false
+
   // if arg$ is finished, should not fork
   if (arg$._getProtectedProperty('_finishFlag')) {
     return stream$

src/operators/merge.ts

@@ -1,6 +1,6 @@
 import { Observable } from '../observable'
 import { Stream } from '../stream'
-import { useUnsubscribeCallback } from '../utils'
+import { checkStreamOrObservableInput, useUnsubscribeCallback } from '../utils'
 import { StreamTupleValues } from '../types'
 import { getGlobalFluthFactory } from '../utils'
 
@@ -12,15 +12,15 @@ import { getGlobalFluthFactory } from '../utils'
  * @returns {Stream}
  */
 export const merge = <T extends (Stream | Observable)[]>(...args$: T) => {
-  const stream$ = (getGlobalFluthFactory()?.() ||
-    new Stream<StreamTupleValues<T>[number]>()) as Stream<StreamTupleValues<T>[number]>
-  let finishCount = 0
-
   // check input type
-  if (args$.some((arg$) => !(arg$ instanceof Stream) && !(arg$ instanceof Observable))) {
+  if (!checkStreamOrObservableInput(args$, true)) {
     throw new Error('merge operator only accepts Stream or Observable as input')
   }
 
+  const stream$ = (getGlobalFluthFactory()?.() ||
+    new Stream<StreamTupleValues<T>[number]>()) as Stream<StreamTupleValues<T>[number]>
+  let finishCount = 0
+
   // check input empty
   if (args$.length === 0) {
     return stream$

src/operators/partition.ts

@@ -2,7 +2,7 @@ import { Observable } from '../observable'
 import { Stream } from '../stream'
 import { finish } from './finish'
 import { PromiseStatus } from '../types'
-import { getGlobalFluthFactory } from '../utils'
+import { checkStreamOrObservableInput, getGlobalFluthFactory } from '../utils'
 
 /**
  * partition takes a stream or Observable, and a predicate function that takes value and index as arguments.
@@ -20,16 +20,16 @@ export const partition = <T>(
   predicate: (this: any, value: any, status: PromiseStatus, index: number) => boolean,
   thisArg?: any,
 ) => {
+  // check input type
+  if (!checkStreamOrObservableInput(stream$)) {
+    throw new Error('partition operator only accepts Stream or Observable as input')
+  }
+
   const selectedStream$ = (getGlobalFluthFactory()?.() || new Stream<T>()) as Stream<T>
   const unselectedStream$ = (getGlobalFluthFactory()?.() || new Stream<T>()) as Stream<T>
   let finishFlag = false
   let index = 1
 
-  // check input type
-  if (!(stream$ instanceof Stream) && !(stream$ instanceof Observable)) {
-    throw new Error('partition operator only accepts Stream or Observable as input')
-  }
-
   // check input finished
   if (stream$._getProtectedProperty('_finishFlag')) {
     finishFlag = true

src/operators/promiseAll.ts

@@ -1,8 +1,11 @@
 import { Observable } from '../observable'
 import { Stream } from '../stream'
 import { StreamTupleValues, PromiseStatus } from '../types'
-import { useUnsubscribeCallback } from '../utils'
-import { getGlobalFluthFactory } from '../utils'
+import {
+  getGlobalFluthFactory,
+  useUnsubscribeCallback,
+  checkStreamOrObservableInput,
+} from '../utils'
 
 /**
  * Internal implementation function for promiseAll variants
@@ -14,6 +17,11 @@ const promiseAllImpl = <T extends (Stream | Observable)[]>(
   args$: T,
   shouldAwait = true,
 ): Stream<StreamTupleValues<T>> => {
+  // check input type
+  if (!checkStreamOrObservableInput(args$, true)) {
+    throw new Error('promiseAll operator only accepts Stream or Observable as input')
+  }
+
   const stream$ = (getGlobalFluthFactory()?.() || new Stream<StreamTupleValues<T>>()) as Stream<
     StreamTupleValues<T>
   >
@@ -37,11 +45,6 @@ const promiseAllImpl = <T extends (Stream | Observable)[]>(
     }
   }
 
-  // check input type
-  if (!args$.every((arg$) => arg$ instanceof Stream || arg$ instanceof Observable)) {
-    throw new Error('promiseAll operator only accepts Stream or Observable as input')
-  }
-
   // check input empty
   if (args$.length === 0) {
     return stream$

src/operators/promiseRace.ts

@@ -1,7 +1,7 @@
 import { Observable } from '../observable'
 import { Stream } from '../stream'
 import { StreamTupleValues } from '../types'
-import { getGlobalFluthFactory } from '../utils'
+import { checkStreamOrObservableInput, getGlobalFluthFactory } from '../utils'
 
 /**
  * race takes multiple streams or Observable, and returns a stream that emits the first value of all the input streams.
@@ -11,17 +11,17 @@ import { getGlobalFluthFactory } from '../utils'
  * @returns {Stream}
  */
 export const promiseRace = <T extends (Stream | Observable)[]>(...args$: T) => {
+  // check input type
+  if (!checkStreamOrObservableInput(args$, true)) {
+    throw new Error('promiseRace operator only accepts Stream or Observable as input')
+  }
+
   const stream$ = (getGlobalFluthFactory()?.() ||
     new Stream<StreamTupleValues<T>[number]>()) as Stream<StreamTupleValues<T>[number]>
   let finishFlag = false
   let finishCount = 0
   let firstIndex: number | null = null
 
-  // check input type
-  if (!args$.every((arg$) => arg$ instanceof Stream || arg$ instanceof Observable)) {
-    throw new Error('promiseRace operator only accepts Stream or Observable as input')
-  }
-
   // check input empty
   if (args$.length === 0) {
     return stream$

src/utils.ts

@@ -1,5 +1,6 @@
 import { isObject, isMap, isSet } from 'limu/lib/support/util'
-import { Stream } from './stream'
+import type { Stream } from './stream'
+import { Observable } from './observable'
 
 export const safeCallback = (callback: any, errorHandler?: (error: any) => void) => {
   return (...args: any[]) => {
@@ -39,6 +40,14 @@ export const isAsyncFunction = (fn: any): fn is (...args: any[]) => PromiseLike<
   )
 }
 
+export const checkStreamOrObservableInput = (arg$: any, isArray = false) => {
+  if (isArray) {
+    return Array.isArray(arg$) && arg$.every((arg) => arg instanceof Observable)
+  } else {
+    return arg$ instanceof Observable
+  }
+}
+
 export const useUnsubscribeCallback = (stream$: Stream, length: number) => {
   let unsubscribeCount = 0
   const unsubscribeCallback = () => {