javascript 在RxJS中自动取消订阅同步执行

Posted 2021-05-03

const { Observable } = require('rxjs')

class ExecutableObservable extends Observable {
  execute(onError) {
    let returnValue

    const next = (
      value,
    ) => {
      returnValue = value
    }

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

    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