Отфильтровать наблюдаемое, используя значения из другого наблюдаемого

Вы можете использовать withLatestFrom. .

source.withLatestFrom(checkboxes, (data, checkbox) => ({data, checkbox}))
  .filter(({data, checkbox}) => ...)

Здесь checkboxes - это наблюдаемое, представляющее список входов флажков. source - это наблюдаемое, представляющее поток событий, поступающих с сервера. В функции фильтра вы можете проверить, действительны ли данные по сравнению с настройками флажка, и пропустить их.

Обратите внимание: важно, чтобы checkboxes испускал по крайней мере 1 значение, прежде чем поток сможет что-либо испустить.

Пс. Что касается других ответов, это решение работает, даже если источник холодный.

Чтобы фильтровать поток A с использованием значений потока B, вам необходимо наблюдать за потоком B и использовать самые последние значения для фильтрации потока A.

Используйте switch() для преобразования B наблюдаемый к наблюдаемому, производящий значения от наблюдаемого.

checkedInputValuesSource
    .map(function (options) {
        return dataSource
            .filter(function (value) {
                return options.indexOf(value) !== -1;
            });
    })
    .switch()
    .subscribe(function (x) {
        console.log('out: ' + x);
    });

Использование switch() предполагает, что dataSource является горячая наблюдаемая.

Пример использования interval() для создания фиктивных данных:

var input,
    checkedInputValuesSource,
    dataSource;

input = document.querySelectorAll('input');

// Generate source describing the current filter.
checkedInputValuesSource = Rx.Observable
    .fromEvent(input, 'change')
    .map(function () {
        var inputs = document.querySelectorAll('input'),
            checkedInputValues = [];
        
        [].forEach.call(inputs, function (e) {
            if (e.checked) {
                checkedInputValues.push(e.value);
            }
        });
        
        return checkedInputValues;
    })
    .startWith([]);

// Generate random data source (hot).
dataSource = Rx.Observable
    .interval(500)
    .map(function () {
        var options = ['a', 'b', 'c'];
    
        return options[Math.floor(Math.floor(Math.random() * options.length))];
    })
    .do(function (x) {
        console.log('in: ' + x);
    })
    .share();

checkedInputValuesSource
    .map(function (options) {
        return dataSource
            .filter(function (value) {
                return options.indexOf(value) !== -1;
            });
    })
    .switch()
    .subscribe(function (x) {
        console.log('out: ' + x);
    });

<script src='https://rawgit.com/Reactive-Extensions/RxJS/v.2.5.3/dist/rx.all.js'></script>

<input type='checkbox' value='a'>
<input type='checkbox' value='b'>
<input type='checkbox' value='c'>

В этом примере будет получен результат, подобный следующему:

in: c
in: a
out: a
in: b
in: c
out: a
in: b
in: a

Где in отражает весь сгенерированный ввод, а b данные, прошедшие фильтр. Фильтр настраивается путем установки флажков напротив входов, отражающих значения «a», «b» и «c».

Видимо, мне нужна была комбинация select, filter и switchLatest. Я написал небольшой тестовый пример, демонстрирующий это: https://gist.github.com/igstan/d5b8db7b43f49dd87382#file-observable-filter-observable-js-L36-L45

Расширяя ответ от @Dorus ... В Котлине это можно сделать так:

val observable: Observable<Data> = ...
val filter: Observable<Checkbox> = ...
val filtered: Observable<Data> =
        observable.filterWithLatestFrom(filter) { checkbox -> checkbox.isSelected }

Используя функцию расширения:

/**
 * Return an [Observable] with type [T1] that is filtered using the last event emitted by the [other] observable.
 */
fun <T1 : Any, T2 : Any> Observable<T1>.filterWithLatestFrom(other: Observable<T2>, filterFunction: (T2) -> Boolean)
: Observable<T1> {
    return this.withLatestFrom(other) { obs1, obs2 -> Pair(obs1, obs2) }
        .filter { (_, obs2) -> filterFunction.invoke(obs2) }
        .map { (obs1, _) -> obs1}
}