Rx 运算符：只要来自其他 Observable 的最新值为 true，就从 Observable 中获取

Question

我正在寻找具有以下语义的 Rx 运算符：

Observable<int> inputValues = …;
Observable<bool> gate = …;

inputValues
    .combineLatest(gate)
    .filter((pair<int, bool> pair) -> { return pair.second; })
    .map((pair<int, bool> pair) -> { return pair.first; });

它从第一个 Observable 发出值，而第二个 Observable 的最新值是true。使用combineLatest，当第二个Observable变为true时，我们也会得到一个值。如果我们不想这样，我们可以使用withLatestFrom 代替combineLatest。

这个操作符是否存在（在任何 Rx 实现中）？

Answer 1

我很抱歉给你 C# 代码，但这里是如何编写的。希望有人可以帮我翻译。

使用.Switch 运算符。它将IObservable<IObservable<T>> 转换为IObservable<T>，只返回由外部可观察对象返回的最新可观察对象产生的值。

var _gate = new Subject<bool>();
var _inputValues = new Subject<int>();

IObservable<int> inputValues = _inputValues;
IObservable<bool> gate = _gate;

IObservable<int> query =
    gate
        .StartWith(true)
        .Select(g => g ? inputValues : Observable.Never<int>())
        .Switch();

query
    .Subscribe(v => Console.WriteLine(v));

_inputValues.OnNext(1);
_inputValues.OnNext(2);
_inputValues.OnNext(3);
_gate.OnNext(false);
_inputValues.OnNext(4);
_inputValues.OnNext(5);
_gate.OnNext(true);
_inputValues.OnNext(6);

这会产生：

1 2 3 6

如果inputValues 是ReplaySubject，那么您可以这样做以使查询正常工作：

IObservable<int> query =
    inputValues
        .Publish(i =>
            gate
                .StartWith(true)
                .Select(g => g ? i : Observable.Never<int>())
                .Switch());