RxJava:How 在订阅前做一些处理

Question

RxJava:如何在订阅前做一些处理

我有一个 class 这样的：

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.should 我使用方法 filter() ?

2.which 过滤器块运行是否开启？ UI_Thread 或 IO_Thread ?

3.isCode3可以简单地改成Code4吗?

抱歉我的英语不好。

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

RxJava:How 在订阅前做一些处理

RxJava:How to do some process before subscribe

java

android

rx-java

rx-android