RxJava- CombineLatest 但仅针对一个 Observable 的发射触发？

Question

假设我有两个可以随时发出值的无限 Observable。他们结合起来创造了一个 Observable<ProcessFileEvent>.

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

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

问题是我只希望 processFileEvent 在 buttonClick 发出某些东西时发出，而不是 selectedFileId。当输入文件 ID 并启动 ProcessFileEvent 时，这绝对不是用户期望的行为。如何组合但仅在 buttonClick 发出时发出？

Answer 1

您可以使用 Observable.Join 来做到这一点。特别注意这段：

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

那个弹珠图就是你想要的，其中selectedFileId是左边的序列，buttonClick是右边的序列。

Answer 2

在 MouseClick 对象上使用 .distinctUntilChanged()。这样，您只会在 MouseClick 更改时收到事件。

创建一个包含 fileId 和 mouseClick 的 class:

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

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

然后

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

Answer 3

使用withLatestFrom:

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

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

它仅在第一个 Observable buttonClick 发出时发出。

Answer 4

这对我有用，缺少的部分是 startWith() 运算符。一旦将其添加到可观察对象中，它将开始工作。

样本

         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 
                }

这是因为 combine 需要所有源开始发出至少一个值，它会初始化流，然后只监听它的变化。

RxJava- CombineLatest 但仅针对一个 Observable 的发射触发？

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

java

reactive-programming

rx-java