如何保留原有左右觀察到的信息對RxJava2

Question

我有一個REST調用返回集合（原件），這個集合進行過濾，但是，訂閱onSuccess我什麼同時獲得原始列表將過濾。

我不知道如何「通行證」這第二個元素，我應該使用哪個運營商來獲得這樣的結果？

我告訴我的代碼如下

Observable.fromCallable(new Callable<List<Integer>>() { 
     @Override public List<Integer> call() throws Exception { 
      // dynamic list obtained from REST call 
      // for simplicity here I return a list 
      return Arrays.asList(1, 2, 3, 4); 
     } 
    }) 
      .flatMap(new Function<List<Integer>, ObservableSource<Integer>>() { 
       @Override public ObservableSource<Integer> apply(List<Integer> integers) throws Exception { 
        return Observable.fromIterable(integers); 
       } 
      }) 
      .filter(new Predicate<Integer>() { 
       @Override public boolean test(Integer integer) throws Exception { 
        return integer > 2; 
       } 
      }) 
      .toList() 
      .subscribe(new SingleObserver<List<Integer>>() { 
       @Override public void onSubscribe(Disposable d) {} 

       @Override public void onSuccess(List<Integer> value) { 
        /////////////////// 
        // here I want both original and filtered list 
        /////////////////// 
       } 

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

Answer 1

一種方式的簡化版本與ConnectableObservable。您需要分享您的初始流的排放量。像這樣的東西

ConnectableObservable<List<Integer>> connectableObservable 
       = Observable.fromCallable(() -> { 
        // dynamic list obtained from REST call 
        // for simplicity here I return a list 
        return Arrays.asList(1, 2, 3, 4); 
       }).publish(); 

     Single.zip(connectableObservable.flatMapIterable(integers -> integers) 
       .filter(integer -> integer > 2) 
        .toList(), 
       connectableObservable.elementAtOrError(0), 
       (integers, lists) -> combine(integers, lists)) 
       .subscribe(o -> { 
        /////////////////// 
        // here you ll have a new object containing both the initial list and the filtered list 
        /////////////////// 
       }); 
     connectableObservable.connect();