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 将 observable 作为参数 - 而不是值 - 直到 observable 发出值为止。

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>