'RxJava- CombineLatest but only fire for one Observable's emission?

Let's say I have two infinite Observables that can emit values at any moment. They combine to create a Observable<ProcessFileEvent>.

Observable<Integer>  selectedFileId= ...
Observable<MouseClick> buttonClick = ...

Observable<ProcessFileEvent> `processFileEvent` = Observable.combineLatest(selectedFileId, buttonClick, (s,b) -> {
    //create ProcessFileEvent here
});

The problem is I only want the processFileEvent to emit when buttonClick emits something, not selectedFileId. It's definitely not the behavior a user expects when a file ID is inputted and it kicks off a ProcessFileEvent. How do I combine but only emit when the buttonClick emits?

javareactive-programmingrx-java


Solution 1:[1]

Use .distinctUntilChanged() on the MouseClick object. That way you'll only get events when the MouseClick changes.

Create a class that contains both fileId and mouseClick:

static class FileMouseClick {
    final int fileId;
    final MouseClick mouseClick;

    FileMouseClick(int fileId, MouseClick mouseClick) {
        this.fileId = fileId;
        this.mouseClick = mouseClick;
    }
}

Then

Observable.combineLatest(selectedFileId, buttonClick, 
                         (s,b) -> new FileMouseClick(s,b))
    .distinctUntilChanged(f -> f.mouseClick)
    .map(toProcessFileEvent())

Solution 2:[2]

Use withLatestFrom:

Observable<Integer>  selectedFileId= ...
Observable<MouseClick> buttonClick = ...

Observable<ProcessFileEvent> processFileEvent = buttonClick.withLatestFrom(selectedFieldId, (b,s) -> {
    //create ProcessFileEvent here
});

It only emits with when the first Observable buttonClick emits.

Solution 3:[3]

You can use Observable.Join to do this. Pay special attention to this paragraph:

However, what could I do to make sure that these windows did not overlap- so that, once a second value was produced I would no longer see the first value? Well, if we returned the left sequence from the leftDurationSelector, that could do the trick. But wait, when we return the sequence left from the leftDurationSelector, it would try to create another subscription and that may introduce side effects. The quick answer to that is to Publish and RefCount the left sequence. If we do that, the results look more like this.

left  |-0-1-2-3-4-5|
right |---A---B---C|
result|---1---3---5
          A   B   C

That marble diagram is what you want, where selectedFileId is the left sequence and buttonClick is the right sequence.

Solution 4:[4]

this is what worked for me, the missing piece was startWith() operators. Once you add it into the observables it will start working working.

Sample

         Observable.combineLatest(
                    firstObservable.observeDoors("1").startWith(emptyList<Door>()),
                    secondObservable.observeUnits("2").startWith( emptyList<Door>())
                ) { doors, units ->
                    ArrayList<Door>().apply {
                        addAll(units)
                        addAll(doors)
                    }
                }.map { 
                    //do something 
                }

it's because combine needs all the sources to start to emit at least one value and it will initialize the stream and after that just listen to it for changes.

Sources

This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.

Source: Stack Overflow

Solution Source
Solution 1
Solution 2 paulpdaniels
Solution 3 Jerry Federspiel
Solution 4 vikas kumar

Related Questions