压缩可观察对象的错误处理

Question

我的用例是：我得到一个永久链接列表，并且需要为每个永久链接发出两个 REST 请求以获取它们的部分数据。当两个请求都返回时，我想将它们的信息合并在一起并对其进行处理（此处 - 打印出来）。我想使用 zip 运算符用代码来完成。这是我当前的代码（连同我正在使用的库的模拟）：

public class Main {

    public static void main(String[] args) {
        ContentManager cm = new ContentManager();

        Observable
                .from(cm.getPermalinks(10))
                .flatMap(permalink -> Observable.zip(
                        Observable.<Content>create(subscriber -> cm.getDataByPermalink(permalink, new SubscribingRestCallback(subscriber))),
                        Observable.<Content>create(subscriber -> cm.getStreamByPermalink(permalink, new SubscribingRestCallback(subscriber))),
                        (dataContent, streamUrlContent) -> {
                            if (dataContent == null || streamUrlContent == null) {
                                System.err.println("not zipping " + dataContent + " and " + streamUrlContent);
                                return Observable.empty();
                            }

                            return new Content(dataContent.permalink, dataContent.logoUrl, streamUrlContent.streamUrl);
                        }))
                .subscribe(System.out::println);
    }
}

class SubscribingRestCallback implements RestCallback {

    private final Subscriber<? super Content> subscriber;

    public SubscribingRestCallback(Subscriber<? super Content> subscriber) {
        this.subscriber = subscriber;
    }

    @Override
    public void onSuccess(Content content) {
        subscriber.onNext(content);
        subscriber.onCompleted();
    }

    @Override
    public void onFailure(int code, String message) {
        System.err.println(message);
        subscriber.onNext(null);
        subscriber.onCompleted();
    }
}

public class Content {

    public final String permalink;

    public final String logoUrl;

    public final String streamUrl;

    public Content(String permalink, String logoUrl, String streamUrl) {
        this.permalink = permalink;
        this.logoUrl = logoUrl;
        this.streamUrl = streamUrl;
    }

    @Override
    public String toString() {
        return String.format("Content [%s, %s, %s]", permalink, logoUrl, streamUrl);
    }
}

public interface RestCallback {

    void onSuccess(Content content);

    void onFailure(int code, String message);
}

class ContentManager {

    private final Random random = new Random();

    public List<String> getPermalinks(int n) {
        List<String> permalinks = new ArrayList<>(n);
        for (int i = 1; i <= n; ++i) {
            permalinks.add("perma_" + i);
        }

        return permalinks;
    }

    public void getDataByPermalink(String permalink, RestCallback callback) {
        getByPermalink(permalink, callback, false);
    }

    public void getStreamByPermalink(String permalink, RestCallback callback) {
        getByPermalink(permalink, callback, true);
    }

    private void getByPermalink(String permalink, RestCallback callback, boolean stream) {
        // simulate network latency and unordered results
        new Thread(() -> {
            try {
                Thread.sleep(random.nextInt(1000) + 200);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if (random.nextInt(100) < 95) {
                String logoUrl;
                String streamUrl;
                if (stream) {
                    logoUrl = null;
                    streamUrl = "http://" + permalink + "/stream";
                } else {
                    logoUrl = "http://" + permalink + "/logo.png";
                    streamUrl = null;
                }
                callback.onSuccess(new Content(permalink, logoUrl, streamUrl));
            } else {
                callback.onFailure(-1, permalink + " data failure");
            }
        }).start();
    }
}

总的来说，它可以工作，但我不喜欢这个实现中的错误处理。基本上，REST 请求可能会失败，在这种情况下，onFailure 方法会调用 subscriber.onNext(null)，这样 zip 方法总能处理一些事情（一个请求可能失败，但另一个请求失败可能没有，我不知道哪个失败了）。然后，在 zip 函数中，我需要一个 if 来检查两者是否都不是 null（如果部分 Content 中的任何一个是 [=17=，我的代码将崩溃]).

如果可能的话，我希望能够在某处使用 filter 运算符过滤掉 null。或者也许有比为失败案例发出 null 值更好的方法，但它仍然可以与 zip 函数一起使用？

Answer 1

首先，通知 Subscriber 错误的正确方法是调用 subscriber.onError 方法：

class SubscribingRestCallback implements RestCallback {
    private final Subscriber<? super Content> subscriber;

    public SubscribingRestCallback(Subscriber<? super Content> subscriber) {
        this.subscriber = subscriber;
    }

    @Override
    public void onSuccess(Content content) {
        subscriber.onNext(content);
        subscriber.onCompleted();
    }

    @Override
    public void onFailure(int code, String message) {
        subscriber.onError(new Exception(message));
    }
}

即使你不希望整个流失败，你仍然需要调用一个subscriber.onError()方法。还有一些其他方法可以浅化错误。其中之一是 onErrorResumeNext 运算符：

Observable
        .from(cm.getPermalinks(10))
        .flatMap(permalink -> Observable.zip(
                Observable.<Content>create(subscriber -> cm.getDataByPermalink(permalink, new SubscribingRestCallback(subscriber))),
                Observable.<Content>create(subscriber -> cm.getStreamByPermalink(permalink, new SubscribingRestCallback(subscriber))),
                (dataContent, streamUrlContent) -> {
                    return new Content(dataContent.permalink, dataContent.logoUrl, streamUrlContent.streamUrl);
                }).onErrorResumeNext(Observable.empty()))
        .subscribe(System.out::println);

编辑

I have one last question: if you notice my zipper functions, I return Observable.empty() if the two objects cannot be zipped, and once I return Content. This seems wrong. How should I handle such error conditions in the zipper function?

是的，返回Observable.empty()是完全错误的。从 zip 函数中抛出异常似乎是最好的解决方案：

Observable
        .from(cm.getPermalinks(10))
        .flatMap(permalink -> Observable.zip(
                Observable.<Content>create(subscriber -> cm.getDataByPermalink(permalink, new SubscribingRestCallback(subscriber))),
                Observable.<Content>create(subscriber -> cm.getStreamByPermalink(permalink, new SubscribingRestCallback(subscriber))),
                (dataContent, streamUrlContent) -> {
                    if (!isDataValid(dataContent, streamUrlContent)) {
                        throw new RuntimeException("Something went wrong.");
                    }
                    return new Content(dataContent.permalink, dataContent.logoUrl, streamUrlContent.streamUrl);
                }).onErrorResumeNext(Observable.empty()))
        .subscribe(System.out::println);

压缩可观察对象的错误处理

Error handling for zipped observables

rx-java