RxJava：订阅前如何做一些处理答案

【问题标题】：RxJava:How to do some process before subscribeRxJava：订阅前如何做一些处理
【发布时间】：2015-12-03 01:24:30
【问题描述】：

RxJava：订阅前如何处理

我有这样的课：

Code1:

class Container {
    boolean    success;
    List<Book> books;
}

我要改代码：

Code2:

  Observable.just(createContainer())
          .subscribeOn(Schedulers.io())
          .observeOn(AndroidSchedulers.mainThread())
          .subscribe(container -> {
              if (container.success) {
                  doSomethingWhenSuccess();

                  for (Book book : container.books) {
                      doSomethingForBook(book);
                  }
              } else {
                  doSomethingWhenFail();
              }
          });

到

Code3:

    Observable.just(createContainer())
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Subscriber<Container>() {
                @Override
                public void onCompleted() {

                }

                @Override
                public void onError(Throwable e) {
                    doSomethingWhenFail();
                }

                @Override
                public void onNext(Container container) {
                    if (container.success) {
                        doSomethingWhenSuccess();

                        Observable.from(container.books)
                                .subscribeOn(Schedulers.io())
                                .observeOn(AndroidSchedulers.mainThread())
                                .subscribe(book -> {
                                    doSomethingForBook(book);
                                });
                    } else {
                        doSomethingWhenFail();
                    }

                }
            });

这是正确的做法吗？

Code4:

    Observable.just(createContainer())
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .filter(container -> {                  //step 2
                if (container.success) {
                    doSomethingWhenSuccess();
                } else {
                    doSomethingWhenFail();
                }
                return container.success;
            })
            .flatMap(container -> Observable.from(container.books))
            .subscribe(book -> {
                doSomethingForBook(book);
            });

问题是：

1.我应该使用filter()的方法吗？

2.过滤器块在哪个上运行？ UI_Thread 或 IO_Thread ?

3.是Code3可以简单到Code4吗？

对不起我糟糕的英语。

【问题讨论】：

很不清楚你在问什么，什么不起作用。

标签： java android rx-java rx-android

【解决方案1】：

也许，您可以将doOnNext 和lift 与your custom operator 一起使用。

class ContainerCheck implements Observable.Operator<Container, Container> {
    @Override
    public Subscriber<? super Container> call(Subscriber<? super Container> o) {
        return new Subscriber<Container>() {
            @Override
            public void onCompleted() {
                if (o.isUnsubscribed()) return;
                o.onCompleted();
            }

            @Override
            public void onError(Throwable e) {
                if (o.isUnsubscribed()) return;
                o.onError(e);
            }

            @Override
            public void onNext(Container container) {
                if (o.isUnsubscribed()) return;
                if (container.success) {
                    o.onNext(container);
                } else {
                    o.onError(new Throwable("your costom exception"));
                }
            }
        };
    }
}

将ContainerCheck 与lift 一起使用。

Observable.just(createContainer(true))
        .subscribeOn(Schedulers.io())
        .lift(new ContainerCheck())
        .observeOn(AndroidSchedulers.mainThread())
        .doOnNext(container -> doSomethingWhenSuccess())
        .map(container -> container.books)
        .flatMap(Observable::from)
        .subscribe(book -> {
            doSomethingForBook(book);
        }, e -> doSomethingWhenFail());

【讨论】：