RxJava- How to backpressure a flatmap()

Question

Perhaps I'm overlooking a simple combination of operators (or an inherent cancellation behavior of RxJava altogether). But suppose I have a hot observable selectedItem that flatmaps to an RxJava-JDBC query.

@Test
public void testFlatMapBackPressure() { 
    Database db = null; //assign db

    BehaviorSubject<Integer> selectedItem = BehaviorSubject.create();

    //can I backpressure the queries so only the latest one is running, and any previous is cancelled?
    Observable<List<Integer>> currentValues = selectedItem.flatMap(i ->  db.select("SELECT VALUE FROM MY_TABLE WHERE ID =?")
            .parameter(i)
            .getAs(Integer.class)
            .toList());
}

How can I backpressure the flatMap() operator so it will ALWAYS only execute the latest query (and cancel any previous). I kind of want a backpressured flatMap operator to do something like this, where the "X" indicates a cancellation of the previous query

Is there a way to accomplish this? Or can it be accomplished already and I just do not see it?

akarnokd akarnokd · Accepted Answer · 2015-09-25T16:56:41

Sounds like you need switchMap() instead of flatMap().

Returns a new Observable by applying a function that you supply to each item emitted by the source Observable that returns an Observable, and then emitting the items emitted by the most recently emitted of these Observables.

RxJava- How to backpressure a flatmap()

1 Answers