java - 使用 Jersey RxJava 客户端未找到类 rx.Observable 的序列化器

Question

我正在创建一个 API 资源，该资源使用 Jersey RxJava 客户端将其他资源聚合到单个响应中。但是，我收到一个返回的错误，这有点令人费解。返回的对象是一个 JSONArray 但我得到了这个:

No serializer found for class rx.Observable and no properties discovered to create BeanSerializer (to avoid exception, disable SerializationFeature.FAIL_ON_EMPTY_BEANS) ) (through reference chain: org.json.simple.JSONArray[0])

及方法:

@Override
public void getUsersDashboard(String token, String userId, @Suspended final AsyncResponse async) {
    List<JSONObject> list = //query persistence layer

    final Queue<String> errors = new ConcurrentLinkedQueue<>();

    final List<Observable> observables = list.stream()
            .map(jsonObject -> (String) jsonObject.get("href"))
            .map(link -> dashboard(token, link, errors))
            .collect(Collectors.toList());

    Observable.just(new JSONArray())
            .zipWith(observables, (jsonArray, resultFromObservable) -> {
                jsonArray.add(resultFromObservable);
                return jsonArray;
            })
            .subscribe(async::resume, async::resume);
}

有什么想法会导致这种情况吗？

Answer 1

所以，你的observables列表的类型为List<Observable> 。由此我假设 dashboard方法返回 Observable<?> ;进一步假设它返回 Observable<T extends JSONNode> ，我会重写你的方法如下:

@Override
public void getUsersDashboard(String token, String userId, @Suspended final AsyncResponse async) {
    List<JSONObject> list = //query persistence layer

    final Queue<String> errors = new ConcurrentLinkedQueue<>();

    Observable
    .fromIterable(list)
    .map(jsonObject -> (String) jsonObject.get("href"))
    .flatMap(link -> dashboard(token, link, errors))
    .collect(JSONArray::new, JSONArray::add)
    .subscribe(async::resume, async::resume);
}

如果您使用以下行，zip 方法也可以工作:

.zipWith(Observable.merge(observables), (jsonArray, resultFromObservable) -> {

但是，zip()从每个可观察值中取出一个元素，这意味着您最终会得到一个大小为 1 的数组。