javascript RxJS中的同步执行

Posted 2021-05-03

const { Observable } = require('rxjs')

class ExecutableObservable extends Observable {
  execute(onError) {
    let hasExecuted = false
    let returnValue

    const next = (
      value,
    ) => {
      if (hasExecuted) {
        console
        .error(
          new Error(
            "Cannot return multiple values in `createExecutableObservable`. Make sure your pipeline returns only a single value and doesn't use asynchronous operators."
          ),
        )
      }

      returnValue = value
    }

    this
    .subscribe({
      complete: () => {},
      error: onError,
      next,
    })

    hasExecuted = true

    return returnValue
  }

  lift(operator) {
    const observable = new ExecutableObservable()

    observable
    .source = this

    observable
    .operator = operator

    return observable
  }
}

const createExecutableObservable = (
  source$,
) => {
  source$
  .__proto__ = new ExecutableObservable()

  return source$
}

module.exports = createExecutableObservable