How to get a list after groupBy in RxJava 2?

user_1067763 0
Viewed 2852

I have a list of transactions. Each transaction has currency and amount information among others. I want to create a list of holdings, so the current amount held by currency. I started with groupBy() and continued with reduce. It seems I have to subscribe before I can do anything with the results, because this gives me an error:

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

It says "no instance of type variable R exist so that Single conforms to ObservableSource< ? extends R>".

On the other hand if I try this:

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.
                            }
                    ));

I cannot get a list, because I already subscribed to the grouped stream. I could add it up, but I'm pretty sure there is a more elegant solution, but I cannot figure it out.

Solution based on akarnokd's answer:

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

This will work for sure

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;
          }
        });
  }
user_42511240
Related