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

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.过滤块在哪个上运行？UI_Thread或者IO_Thread？

3.是 Code3 可以简单地转为 Code4 吗？

并为我糟糕的英语感到抱歉。

Answer 1

也许，你可以使用doOnNextand 。liftyour 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());