用例是,有两个数据源:
- 服务 1 - 从 source-1 获取
- 服务 2 - 从 source-2 获取
该应用程序应至少从 source-1 返回数据。如果 source-2 一切正常 - 数据将被“增强”,例如乘以 100。
服务 1 调用服务 2。
如果所有成功的用户都从服务 1 和服务 2 获取数据,如果服务 2 出现错误,则用户仅从服务 1 获取数据(至少)如果服务 1 出现错误 - 用户将收到错误。
有 hello-world-bench 代码可以模拟这种情况:
import io.reactivex.rxjava3.core.Observable;
import java.util.concurrent.TimeUnit;
class Response {
public Integer value;
public String warning;
public Response(Integer value) {
this.value = value;
}
@Override
public String toString() {
return "Response{" +
"value=" + value +
", warning='" + warning + '\'' +
'}';
}
}
class Service1 {
public Observable<Response> call(int arg) {
return Observable
.just(
new Response(1),
new Response(2),
new Response(3),
new Response(4))
.delay(100, TimeUnit.MILLISECONDS);
}
}
class Service2 {
public Observable<Response> call(int arg) {
if ( arg % 2 == 0) {
System.out.println("service 2: " + arg);
return Observable
.just(new Response(100 * arg)) // service 2 multiplies x 100 on the result it gets from the service 1
.delay(10, TimeUnit.MILLISECONDS);
} else {
System.out.println("service 2: " + arg);
return Observable.error(new RuntimeException("service 2 error"));
}
}
}
public class Step1 {
static Service1 service1 = new Service1();
static Service2 service2 = new Service2();
public static void main(String[] args) throws InterruptedException {
var oo1 = service1.call(1);
var oo3 = oo1.switchMapDelayError(x -> {
final Observable<Response> oo2 = service2.call(x.value);
return oo2
.onErrorReturn((ex) -> {
//System.out.println("Error handling..." + ex.getMessage() + " " + x);
x.warning = ex.getMessage();
return x; // returns at least service1 result
});
});
oo3.subscribe(x -> {
System.out.println(x);
});
Thread.sleep(100000);
}
}
这段代码的结果是:
service 2: 1
Response{value=1, warning='service 2 error'}
service 2: 2
service 2: 3
Response{value=3, warning='service 2 error'}
service 2: 4
Response{value=400, warning='null'}
问题是:没有预期:value=200
2*100
然而,如果我在 service2.call() //.delay(10, TimeUnit.MILLISECONDS) 评论延迟,那么它会得到预期的结果:
service 2: 1
Response{value=1, warning='service 2 error'}
service 2: 2
Response{value=200, warning='null'}
service 2: 3
Response{value=3, warning='service 2 error'}
service 2: 4
Response{value=400, warning='null'}
问题是:为什么.delay(10, TimeUnit.MILLISECONDS) on service2.call()
它不能产生 value=200 ?该解决方案有什么问题,我想念什么?
谢谢。