Используя один Наблюдаемый в часы другие Наблюдаемые

Question

У нас есть потребность в «часах» один горячий наблюдаемом на основе другого, например:

main:   --(1)-----(2)-----(3)-----(4)-------(5)----(6)------------| 
clock:   -------(X)------------(X)------(X)--------------(X)-------| 
______________________________________________________________________________ 

expected result: -------(1)(2)-----(3)-(3)-(4)--(4)--(5)----(6)--(6)-------| 

Более конкретно, когда «часы» излучающий элемент, последний пункт из другого наблюдаемого испускается снова. Выброс элемента задерживается до тех пор, пока оба наблюдаемых не испускают хотя бы один предмет. Прямо сейчас, мы достигаем это следующим образом:

<T, U> Observable<T> clock(Observable<T> main, Observable<U> clock) { 
    return Observable.combineLatest(main, clock, (mainItem, clockItem) -> mainItem); 
    } 

Это, кажется, несколько глупо, как мы применяем преобразование, а затем просто выбросить один из входов. Кроме того, у нас есть случайные проблемы с давлением, связанные с этим решением. Кажется, что существует существующий оператор Rx, который выполняет эту операцию, но до сих пор я не нашел подходящего API-метода, который делает именно это. Есть ли лучший, более идиоматический способ решения этой проблемы с RxJava?

Answer 1

Выглядит хорошо для меня. Это не значит, что вход clock не использовался, потому что, хотя значение не использовалось, время имеет.

Для обработки противодавления на горячем источнике вы можете применить один из операторов onBackpressureXXX. Например, если вы не хотите пропустить выходы, используйте .onBackpressureBuffer(). Другая стратегия заключается в использовании .throttle или .sample.