Retrofit und RxJava: Kombinierbar fließbar mit beobachtbar, um das Ergebnis zu erhalten

Question

Ich möchte ein Flowable mit einem Observable kombinieren, um ein Modell aus diesen Quellen zu erstellen.

Von Zimmer bekomme ich CartModel Artikel dieses Modell enthält eine ID von Produkt und Menge.

@Override 
     public Flowable<List<CartModel>> getCartItems() { 
      return appLocalDataStore.getCartItems(); 
     } 

Danach habe ich eine Funktion, die für jede ID von oben Funktion ein Paar Menge und Produktdetails erstellen.

@Override 
public Observable<List<Pair<WsProductDetails, Integer>>> getCartWithProductData() { 
    return appLocalDataStore.getCartItems().toObservable().flatMap(new Function<List<CartModel>, ObservableSource<CartModel>>() { 
     @Override 
     public ObservableSource<CartModel> apply(List<CartModel> cartModels) throws Exception { 
      return Observable.fromIterable(cartModels); 
     } 
    }).flatMap(new Function<CartModel, ObservableSource<Pair<WsProductDetails, Integer>>>() { 
     @Override 
     public ObservableSource<Pair<WsProductDetails, Integer>> apply(final CartModel cartModel) throws Exception { 
      return appRemoteDataStore.getProductBySKU(cartModel.getId()).map(new Function<WsProductDetails, Pair<WsProductDetails, Integer>>() { 
       @Override 
       public Pair<WsProductDetails, Integer> apply(WsProductDetails wsProductDetails) throws Exception { 
        Log.d("TestCartItems", "SKU" + wsProductDetails.getSku() + " quantity" + cartModel.getQuantity()); 
        return new Pair<>(wsProductDetails, cartModel.getQuantity()); 
       } 
      }); 
     } 
    }).toList().toObservable(); 
} 

@GET(PATH_PRODUCT_DETAILS_BY_SKY) 
    Observable<WsProductDetails> getProductBySKU(@Path(PATH_CODE) String cod); 

Die obige Funktion funktioniert ok, weil die Protokolle corect sind.

Aber in My Presenter OnNext() || OnError() || onComplete() -Funktion wird nie aufgerufen. Warum passiert das?

@Override 
    public void getCartItems() { 
     Observable<List<Pair<WsProductDetails, Integer>>> source = appDataStore.getCartWithProductData(); 
     mCompositeDisposable.add(source 
       .observeOn(AndroidSchedulers.mainThread()) 
       .subscribeOn(Schedulers.io()) 
       .subscribeWith(new DisposableObserver<List<Pair<WsProductDetails, Integer>>>() { 
        @Override 
        public void onComplete() { 
         Log.d(TAG, "onComplete()"); 
        } 

        @Override 
        public void onNext(List<Pair<WsProductDetails, Integer>> pairs) { 
         Log.d(TAG, "onError()"); 
         view.showCartItems(pairs); 
        } 

        @Override 
        public void onError(Throwable e) { 
         Log.d(TAG, "onError()"); 
        } 
       })); 

    } 
} 

Answer 1

Sie haben die Lookup flatMap auf die innere fromIterable anzuwenden, damit es endlich bekommt:

@Override 
public Observable<List<Pair<WsProductDetails, Integer>>> getCartWithProductData() { 
    return appLocalDataStore.getCartItems().toObservable().flatMap(new Function<List<CartModel>, ObservableSource<CartModel>>() { 
     @Override 
     public ObservableSource<CartModel> apply(List<CartModel> cartModels) throws Exception { 
      return Observable.fromIterable(cartModels) 
       .flatMap(new Function<CartModel, ObservableSource<Pair<WsProductDetails, Integer>>>() { 
        @Override 
        public ObservableSource<Pair<WsProductDetails, Integer>> apply(final CartModel cartModel) throws Exception { 
         return appRemoteDataStore.getProductBySKU(cartModel.getId()) 
          .map(new Function<WsProductDetails, Pair<WsProductDetails, Integer>>() { 
          @Override 
          public Pair<WsProductDetails, Integer> apply(WsProductDetails wsProductDetails) throws Exception { 
           Log.d("TestCartItems", "SKU" + wsProductDetails.getSku() + " quantity" + cartModel.getQuantity()); 
           return new Pair<>(wsProductDetails, cartModel.getQuantity()); 
          } 
         }); 
        } 
       }).toList().toObservable(); 
     } 
    }); 
}