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

2.哪個過濾塊運行？ UI_Thread或IO_Thread？

3.is Code3可以簡單地代碼4？

對不起，我的英語不好。

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