如果我理解您的意思正确,您需要像这样封装您现有的
callExternalUrl
。
static Observable<String> callExternalUrlAsync(String url, String json, HttpMethod method)
{
return Observable.fromCallable(() -> callExternalUrl(url, json, method))
.subscribeOn(Schedulers.io())
.flatMap(re -> {
if (re.hasBody())
return Observable.just(re.getBody());
else
return Observable.error(new RuntimeException("Bad response status " + re.getStatusCode()));
},
e -> Observable.error(e),
(Func0<Observable<? extends String>>) (() -> Observable.empty()))
.observeOn(Schedulers.computation());
}
代码简述:
- 它在
Schedulers.io
上安排现有的 callExternalUrl
的执行。
- 对
ResponseEntity<T>
进行最小转换,以成功的 T
和错误情况。同样在 io
调度程序上进行,但由于非常短暂,所以不重要。(如果在 callExternalUrl
中出现异常,则原样传递。)
- 使订阅者的结果在
Schedulers.computation
上执行。
注意事项:
- 您可能想为
subscribeOn
和 observeOn
使用自定义调度程序。
- 您可能希望在传递给
flatMap
的第一个 lambda 中使用更好的逻辑来区分成功和错误,并且绝对需要一些更具体的异常类型。
高阶魔法
如果您愿意使用高阶函数并为减少代码重复而牺牲一点性能,可以像这样做:
static <T> Observable<T> wrapCallExternalAsAsync(Func3<String, String, HttpMethod, ResponseEntity<T>> externalCall, String url, String json, HttpMethod method)
{
return Observable.fromCallable(() -> externalCall.call(url, json, method))
.subscribeOn(Schedulers.io())
.flatMap(re -> {
if (re.hasBody())
return Observable.just(re.getBody());
else
return Observable.error(new RuntimeException("Bad response status " + re.getStatusCode()));
},
e -> Observable.error(e),
(Func0<Observable<? extends T>>) (() -> Observable.empty()))
.observeOn(Schedulers.computation());
}
static Observable<String> callExternalUrlAsync_HigherOrder(String url, String json, HttpMethod method)
{
return wrapCallExternalAsAsync(MyClass::callExternalUrl, url, json, method);
}
MyClass
是您的 callExternalUrl
所在的位置。
更新(仅限异步调用)
private static RxClient httpClient = Rx.newClient(RxObservableInvoker.class); // 在此处您可能需要传递自定义的 ExecutorService
private <T> Observable<String> executeHttpAsync(String url, String httpMethod, Entity<T> entity) {
return httpClient.target(url)
.request()
.headers(httpHeaders)
.rx()
.method(httpMethod, entity)
.map(resp -> {
if (200 != resp.getStatus()) {
throw new RuntimeException("Bad status code " + resp.getStatus());
} else {
if (!resp.hasEntity()) {
throw new RuntimeException("Empty response");
} else {
try {
return resp.readEntity(String.class);
} catch (Exception ex) {
throw new RuntimeException(ex);
}
}
}
})
.observeOn(Schedulers.computation());
}
private Observable<String> executeGetAsync(String url) {
return executeHttpAsync(url, "GET", null);
}
private Observable<String> executePostAsync(String url, String json) {
return executeHttpAsync(url, "POST", Entity.json(json));
}
同样适用类似的警告:
- 您可能想要为
newClient
调用和observeOn
都使用自定义调度程序。
- 您可能希望有一些更好的错误处理逻辑,而不仅仅是检查它是否为HTTP 200,而且肯定需要一些更具体的异常类型。但这都是业务逻辑特定的,所以由您决定。
此外,从您的示例中无法清楚地了解请求正文(HttpEntity
)的实际构建方式,以及您是否总是希望将String
作为响应,就像在原始示例中一样。尽管如此,我只是按原样复制了您的逻辑。如果您需要其他内容,您可能应该参考https://jersey.java.net/documentation/2.25/media.html#json上的文档。
Content-Length
作为POST请求的一部分发送。我认为你需要展示你当前的实际代码,可能甚至需要创建一个新的问题,因为它对于评论来说太复杂了。没有最小、完整和可验证的示例,很难猜测。 - SergGr