War*_*lax 5 java reactive-programming rx-java reactive rx-java2
我有两个Observables,我们称它们为PeanutButter和Jelly。我想将它们组合到一个Sandwich Observable. 我可以使用:
Observable<PeanutButter> peanutButterObservable = ...;
Observable<Jelly> jellyObservable = ...;
Observable<Sandwich> sandwichObservable = Observable.combineLatest(
peanutButterObservable,
jellyObservable,
(pb, j) -> makeSandwich(pb, j))
Run Code Online (Sandbox Code Playgroud)
问题是 RX在发射第一个组合之前等待第一个PeanutButter和第一个Jelly发射,Sandwich但Jelly可能永远不会发射,这意味着我永远不会得到第一个Sandwich。
我想组合这两个提要,以便在发出任一提要的第一个项目时立即发出组合项目,无论另一个提要是否尚未发出任何内容,我该如何在 RxJava 中执行此操作?
一种可能的方法是使用startWith运算符在订阅时触发每个流的已知值的发射。combineLatest()如果任一流发出值,则这种方式将触发。你只需要留意onNext消费者的初始/信号值。
像这样的东西...:
@Test
public void sandwiches() {
final Observable<String> peanutButters = Observable.just("chunky", "smooth")
.startWith("--initial--");
final Observable<String> jellies = Observable.just("strawberry", "blackberry", "raspberry")
.startWith("--initial--");
Observable.combineLatest(peanutButters, jellies, (peanutButter, jelly) -> {
return new Pair<>(peanutButter, jelly);
})
.subscribe(
next -> {
final String peanutButter = next.getFirst();
final String jelly = next.getSecond();
if(peanutButter.equals("--initial--") && jelly.equals("--initial--")) {
// initial emissions
} else if(peanutButter.equals("--initial--")) {
// jelly emission
} else if(jelly.equals("--initial--")) {
// peanut butter emission
} else {
// peanut butter + jelly emissions
}
},
error -> {
System.err.println("## onError(" + error.getMessage() + ")");
},
() -> {
System.out.println("## onComplete()");
}
);
}
Run Code Online (Sandbox Code Playgroud)
| 归档时间: |
|
| 查看次数: |
2524 次 |
| 最近记录: |