java - RxJava :How to do some process before subscribe

Question

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.过滤 block 在哪个上运行？ UI_Thread 还是 IO_Thread？

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

抱歉我的英语不好。

Answer 1

也许，您可以将doOnNext 和lift 与您的自定义运算符 一起使用。

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());