Реактивный шаблон для расчесывания оператора, где он зависит от другого

Question

У меня есть приложение, которое потребляет услугу REST. Он делает запросы каждый раз, когда кто-то, например. нажимает кнопку. Однако сначала нужно получить токен, а затем обновлять каждые 20 минут. Вот мраморная схема, представляющая желаемое сочетание потоков ...

TOKEN SOURCE ---------A--------------B--------------C----------------D- 
REQUEST SOURCE ----1-----------2--3---------4-----------5-------6-------- 


RESULT SOURCE ---------A1-----A2-A3--------B4----------C5------C6------- 

Обычно источник запроса вызывает результат, который сочетает в себе с последним значением маркера источника, однако есть исключение, когда никакие пункты не имеют был испущен на токен-потоке - в этом случае запрос буферизуется до тех пор, пока не поступит первый токен, а затем он будет отправлен.

Оператор combineLatest почти присутствует, но он срабатывает при испускании потока. Мраморная диаграмма для оператора sample также кажется близкой, но она дросселирует выход на основе временного интервала, который не является тем, что я хочу.

Какой оператор/цепочка операторов будет работать для этого экземпляра?

Я прототипирую с RxJS, но мне нужно реализовать в RxSwift.

Answer 1

Вы были близки к раствору. Важной деталью является то, что Source (2) проектирует все свои элементы, в то время как Source (1) напрямую не проектируется, и наличие этого дополнительного бита информации может помочь нам отказаться, когда изменяется только Source (1).

s1.combineLatest(s2, (a, b) => [a, b]) 
    .distinctUntilChanged(v => v[1]) 

Вот схема тестирования:

var s1 = Rx.Observable.interval(997).take(10).map(i => String.fromCharCode(65 + i)) 
var s2 = Rx.Observable.interval(251).take(10).delay(500).skip(1) 

s1.combineLatest(s2, (a, b) => [a, b]) 
    .distinctUntilChanged(v => v[1]) 
    .subscribe(value => console.log(value)) 

Выход:

[ "А", 1] [ "А", 2] [ "А", 3] [ "А", 4] [ "Б", 5] [ "Б", 6] [ "Б", 7] [ "В", 8] [ "С", 9]

Answer 2

Поскольку вы хотите, чтобы ваше излучение данных срабатывало только на основе вашего REQUEST SOURCE, вероятно, это был бы лучший способ начать с ним этот поток, а затем переключить - или в этом случае flatMap - на TOKEN SOURCE и объединить эти два - не требуются продвинутые операторы:

(Примечание: Мое предположение, что ваш TOKEN SOURCE делает повторит последнюю фишку:

var token$ = Rx.Observable.interval(1000).take(10).map(() => Math.random()).publishReplay(1); // simulating the token source: new token every second and replaying the latest token 
var requests$ = Rx.Observable.interval(250).take(10); // simulating the request source: emitting a new request every 250ms 
token$.connect(); 

requests$ 
    .flatMap(request => token$ 
     .take(1) 
     .map(token => ({request, token})) 
    ) 
    .subscribe(function(requestWithToken) { 
     console.log("Matched: ", requestWithToken); 
    }); 

Вы можете проверить живую скрипку здесь: https://jsfiddle.net/rn8rzufg/