Wie eine Liste nach groupBy in RxJava 2 bekommen?

Question

Ich habe eine Liste von Transaktionen. Jede Transaktion enthält unter anderem Informationen zu Währung und Betrag. Ich möchte eine Liste der Bestände erstellen, also den aktuellen Betrag nach Währung. Ich habe mit groupBy() angefangen und mit reduce fortgesetzt. Es scheint, ich habe zu abonnieren, bevor ich etwas mit den Ergebnissen tun, denn dies ist mir einen Fehler gibt:

Observable.fromIterable(transactions) 
      .groupBy(Transaction::getCurrency) 
      .flatMap(t -> t.reduce(new Holding(t.getKey()), (holding, transaction) -> holding.addTransaction(transaction.getAmount())) 

Er sagt: „keine Instanz des Typs Variable R vorhanden sein, damit Einzel < erstreckt R ObservableSource entspricht> ". wenn

Auf der anderen Seite versuche ich dies:

Observable.fromIterable(transactions) 
      .groupBy(Transaction::getCurrency) 
      .subscribe((GroupedObservable<String, Transaction> r) -> r.reduce(new Holding(r.getKey()), (holding, transaction) -> holding.addTransaction(transaction.getAmount())) 
        .toObservable() 
        .subscribe(t -> { 
           //t is a single Holding. 
          } 
        )); 

Ich kann nicht eine Liste bekommen, weil ich schon zu dem gruppierten Strom abonnierte. Ich könnte es hinzufügen, aber ich bin mir ziemlich sicher, dass es eine elegantere Lösung gibt, aber ich kann es nicht herausfinden.

Lösung basierend auf akarnokd Antwort:

Observable.fromIterable(transactions) 
      .groupBy(Transaction::getCurrency) 
      .flatMapSingle(Observable::toList) 
      .map(Holding::new) 
      .toList() 
      .subscribe(holdings -> { 
       whatever(holdings); 
      }); 

Answer 1

(von meinem Kommentar zum Beitrag):

flatMapSingle im oberen Fall versuchen. Auch innerhalb einer onNext Handler abonniert ist eine schlechte Praxis, wie Sie die Eigenschaften der Zusammensetzung von RxJava verlieren.

Answer 2

Da die Dokumentation sagt, die reduce Funktion

applies a function to each item emitted by an Observable, sequentially, and emit the final value.

Hiermit teilen wir Ihnen Möglichkeit, einen einzelnen Wert bekommen (eigentlich für jede Observable der Gruppe Du bekommst einen einzigen Gegenstand).

Sie können Ihre reduce Operation verschieben, nachdem Sie eine Liste bekommen. Sie könnten wahrscheinlich Ihre erste lange subscribe mit diesem ersetzen:

.subscribe(group -> group.toList() 

Sie dann einige Observable s auf der Anzahl der Gruppen basieren, die Sie haben, die jeweils eine einzelne List Ihrer vordefinierten Typs emittiert.

HINWEIS: nicht sicher, aber wahrscheinlich können Sie die erste subscribe mit einem flatMap ersetzen, die Ihren GroupedObservable in eine Observable, die eine Liste von Elementen emittieren verwandelt.

Answer 3

im oberen Fall

public Single<Map<Integer, List<Category>>> getSubCategoryListById(List<Category> categoryList) { 

    return Flowable.just(categoryList) 
     .flatMapIterable(new Function<List<Category>, Iterable<Category>>() { 
      @Override public Iterable<Category> apply(List<Category> categories) throws Exception { 
      return categories; 
      } 
     }) 
     .filter(new Predicate<Category>() { 
      @Override public boolean test(Category category) throws Exception { 
      return category.parent_id != 0; 
      } 
     }) 
     .groupBy(new Function<Category, Integer>() { 
      @Override public Integer apply(Category category) throws Exception { 
      return category.category_id; 
      } 
     }) 
     .flatMapSingle(new Function<GroupedFlowable<Integer, Category>, Single<List<Category>>>() { 
      @Override public Single<List<Category>> apply(
       GroupedFlowable<Integer, Category> integerCategoryGroupedFlowable) throws Exception { 
      return integerCategoryGroupedFlowable.toList(); 
      } 
     }) 
     .toMap(new Function<List<Category>, Integer>() { 
      @Override public Integer apply(List<Category> categories) throws Exception { 
      return categories.get(0).category_id; 
      } 
     }); 
    }