Disable Flowable/Observable Buffer

Viewed 492

I'm having issues getting always the last value from combination obtained with a combineLatest operator.

I have 2 hot flowables (a, b) generating events at high frequency (an event every 100ms each):

Flowable<OrderBook> flowA = sourceA.getObservableOrderBook(value);
Flowable<OrderBook> flowB = sourceB.getObservableOrderBook(value);

combined with a combineLatest, which takes almost 300ms to do it's job.

Flowable<OrderBookCouple> combined = Flowable.combineLatest(flowA, flowB,        OrderBookCouple::new).observeOn(Schedulers.newThread());
combined.subscribe((bookCouple) -> {
                System.out.println("A timestamp: " + bookCouple.aOrderBook.getTimeStamp());
                System.out.println("B timestamp: " + bookCouple.bOrderBook.getTimeStamp());
                Thread.sleep(300);
            }

After one execution of the combiner, I would like to process the very last combination of event generated, meaning (lastA, lastB).

The default behaviour of the combined flow is to cache all combination of events in its own buffer, so that the combined flow is receveing combinations that are very old and this time gap is exploding.

How should I change my code to disable this buffer and receive always the very last combinaton?

1 Answers
Related