RxJava2で元の観測可能な情報を保存する方法

Question

私はコレクション（元）を返すREST呼び出しを持っていますが、このコレクションはフィルタリングされていますが、サブスクライブではonSuccessです。

この2番目の要素を「渡す」方法はわかりませんが、この演算子を使用してこの結果を取得する必要がありますか？

私は一つの方法は、ConnectableObservableである

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