找不到类 rx 的序列化程序.可使用 Jersey RxJava 客户端进行观察

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 的数组。