Rxjs - Observable 订阅观察一个变量，完成改变

Question

我正在尝试创建一个 observable 来监视我的应用程序中的变量，一旦变量变为 null，所有订阅者都应该得到通知并完成。我假设 takeUntil 属性最有效，但我似乎做错了什么。

export default class service {
    constructor($window) {
        'ngInject'
        this._currentObject = new BehaviorSubject(null)
    }

    open(config){
        this._currentObject.next(config)
        return this._currentObject.asObservable().takeUntil(null)
    }

    close(){
        this._currentObject.next(null)
    }
}   

所以对于任何调用 open 函数的方法，我想返回一个他们可以订阅的新 Observable，这样他们就知道对象何时变回 null。这将发生在应用程序中调用 close 方法

的其他地方

service.open({
            prop:'test',
            prop2: 'right',
        }).subscribe(s => {
            // fires when currentObject is set back to null 
            //automatically completes
        })

尝试此操作时，我在 takeUntil 上遇到错误(您在需要流的位置提供了“null”。您可以提供 Observable、Promise、Array 或 Iterable。)

我什至不确定这是一个不错的方法，但想知道是否有人可以看一下并给我建议。谢谢!

Answer 1

takeUntil接受一个可观察对象作为参数——而不是一个值——并接受直到该可观察对象发出一个值。

takeWhile更有可能是您需要的运算符，因为它需要一个谓词:

const source = new Rx.BehaviorSubject(null);
source
  .takeWhile((value, index) => (value !== null) || (index === 0))
  .subscribe(
    (value) => console.log(value),
    undefined,
    () => console.log("completed; changed back to null")
  );
  
source.next(1);
source.next(2);
source.next(3);
source.next(null);

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"></script>