This page shows operators that perform mathematical or other operations over an entire sequence of items emitted by an Observable or Flowable. Because these operations must wait for the source Observable/Flowable to complete emitting items before they can construct their own emissions (and must usually buffer these items), these operators are dangerous to use on Observables and Flowables that may have very long or infinite sequences.

• Mathematical Operators
  • averageDouble
  • averageFloat
  • max
  • min
  • sumDouble
  • sumFloat
  • sumInt
  • sumLong
• Standard Aggregate Operators
  • count
  • reduce
  • reduceWith
  • collect
  • collectInto
  • toList
  • toSortedList
  • toMap
  • toMultimap

The operators in this section are part of the RxJava2Extensions project. You have to add the rxjava2-extensions module as a dependency to your project. It can be found at http://search.maven.org.

Note that unlike the standard RxJava aggregator operators, these mathematical operators return Observable and Flowable instead of the Single or Maybe.

The examples below assume that the MathObservable and MathFlowable classes are imported from the rxjava2-extensions module:

import hu.akarnokd.rxjava2.math.MathObservable;
import hu.akarnokd.rxjava2.math.MathFlowable;

Available in: Flowable, Observable, Maybe, Single, Completable

ReactiveX documentation: http://reactivex.io/documentation/operators/average.html

Calculates the average of Numbers emitted by an Observable and emits this average as a Double.

Observable<Integer> numbers = Observable.just(1, 2, 3);
MathObservable.averageDouble(numbers).subscribe((Double avg) -> System.out.println(avg));

// prints 2.0

Available in: Flowable, Observable, Maybe, Single, Completable

ReactiveX documentation: http://reactivex.io/documentation/operators/average.html

Calculates the average of Numbers emitted by an Observable and emits this average as a Float.

Observable<Integer> numbers = Observable.just(1, 2, 3);
MathObservable.averageFloat(numbers).subscribe((Float avg) -> System.out.println(avg));

// prints 2.0

Available in: Flowable, Observable, Maybe, Single, Completable

ReactiveX documentation: http://reactivex.io/documentation/operators/max.html

Emits the maximum value emitted by a source Observable. A Comparator can be specified that will be used to compare the elements emitted by the Observable.

Observable<Integer> numbers = Observable.just(4, 9, 5);
MathObservable.max(numbers).subscribe(System.out::println);

// prints 9

The following example specifies a Comparator to find the longest String in the source Observable:

final Observable<String> names = Observable.just("Kirk", "Spock", "Chekov", "Sulu");
MathObservable.max(names, Comparator.comparingInt(String::length))
        .subscribe(System.out::println);

// prints Chekov

Available in: Flowable, Observable, Maybe, Single, Completable

ReactiveX documentation: http://reactivex.io/documentation/operators/min.html

Emits the minimum value emitted by a source Observable. A Comparator can be specified that will be used to compare the elements emitted by the Observable.

Observable<Integer> numbers = Observable.just(4, 9, 5);
MathObservable.min(numbers).subscribe(System.out::println);

// prints 4

Available in: Flowable, Observable, Maybe, Single, Completable

ReactiveX documentation: http://reactivex.io/documentation/operators/sum.html

Adds the Doubles emitted by an Observable and emits this sum.

Observable<Double> numbers = Observable.just(1.0, 2.0, 3.0);
MathObservable.sumDouble(numbers).subscribe((Double sum) -> System.out.println(sum));

// prints 6.0

Available in: Flowable, Observable, Maybe, Single, Completable

ReactiveX documentation: http://reactivex.io/documentation/operators/sum.html

Adds the Floats emitted by an Observable and emits this sum.

Observable<Float> numbers = Observable.just(1.0F, 2.0F, 3.0F);
MathObservable.sumFloat(numbers).subscribe((Float sum) -> System.out.println(sum));

// prints 6.0

Available in: Flowable, Observable, Maybe, Single, Completable

ReactiveX documentation: http://reactivex.io/documentation/operators/sum.html

Adds the Integers emitted by an Observable and emits this sum.

Observable<Integer> numbers = Observable.range(1, 100);
MathObservable.sumInt(numbers).subscribe((Integer sum) -> System.out.println(sum));

// prints 5050

Available in: Flowable, Observable, Maybe, Single, Completable

ReactiveX documentation: http://reactivex.io/documentation/operators/sum.html

Adds the Longs emitted by an Observable and emits this sum.

Observable<Long> numbers = Observable.rangeLong(1L, 100L);
MathObservable.sumLong(numbers).subscribe((Long sum) -> System.out.println(sum));

// prints 5050

Note that these standard aggregate operators return a Single or Maybe because the number of output items is always know to be at most one.

Available in: Flowable, Observable, Maybe, Single, Completable

ReactiveX documentation: http://reactivex.io/documentation/operators/count.html

Counts the number of items emitted by an Observable and emits this count as a Long.

Observable.just(1, 2, 3).count().subscribe(System.out::println);

// prints 3

Available in: Flowable, Observable, Maybe, Single, Completable

ReactiveX documentation: http://reactivex.io/documentation/operators/reduce.html

Apply a function to each emitted item, sequentially, and emit only the final accumulated value.

Observable.range(1, 5)
    .reduce((product, x) -> product * x)
    .subscribe(System.out::println);

// prints 120

Available in: Flowable, Observable, Maybe, Single, Completable

ReactiveX documentation: http://reactivex.io/documentation/operators/reduce.html

Apply a function to each emitted item, sequentially, and emit only the final accumulated value.

Observable.just(1, 2, 2, 3, 4, 4, 4, 5)
        .reduceWith(TreeSet::new, (set, x) -> {
            set.add(x);
            return set;
        })
        .subscribe(System.out::println);

// prints [1, 2, 3, 4, 5]

Available in: Flowable, Observable, Maybe, Single, Completable

ReactiveX documentation: http://reactivex.io/documentation/operators/reduce.html

Collect items emitted by the source Observable into a single mutable data structure and return an Observable that emits this structure.

Observable.just("Kirk", "Spock", "Chekov", "Sulu")
        .collect(() -> new StringJoiner(" \uD83D\uDD96 "), StringJoiner::add)
        .map(StringJoiner::toString)
        .subscribe(System.out::println);

// prints Kirk 🖖 Spock 🖖 Chekov 🖖 Sulu

Available in: Flowable, Observable, Maybe, Single, Completable

ReactiveX documentation: http://reactivex.io/documentation/operators/reduce.html

Collect items emitted by the source Observable into a single mutable data structure and return an Observable that emits this structure.

Note: the mutable value that will collect the items (here the StringBuilder) will be shared between multiple subscribers.

Observable.just('R', 'x', 'J', 'a', 'v', 'a')
        .collectInto(new StringBuilder(), StringBuilder::append)
        .map(StringBuilder::toString)
        .subscribe(System.out::println);

// prints RxJava

Available in: Flowable, Observable, Maybe, Single, Completable

ReactiveX documentation: http://reactivex.io/documentation/operators/to.html

Collect all items from an Observable and emit them as a single List.

Observable.just(2, 1, 3)
        .toList()
        .subscribe(System.out::println);

// prints [2, 1, 3]

Available in: Flowable, Observable, Maybe, Single, Completable

ReactiveX documentation: http://reactivex.io/documentation/operators/to.html

Collect all items from an Observable and emit them as a single, sorted List.

Observable.just(2, 1, 3)
        .toSortedList(Comparator.reverseOrder())
        .subscribe(System.out::println);

// prints [3, 2, 1]

Available in: Flowable, Observable, Maybe, Single, Completable

ReactiveX documentation: http://reactivex.io/documentation/operators/to.html

Convert the sequence of items emitted by an Observable into a Map keyed by a specified key function.

Observable.just(1, 2, 3, 4)
        .toMap((x) -> {
            // defines the key in the Map
            return x;
        }, (x) -> {
            // defines the value that is mapped to the key
            return (x % 2 == 0) ? "even" : "odd";
        })
        .subscribe(System.out::println);

// prints {1=odd, 2=even, 3=odd, 4=even}

Available in: Flowable, Observable, Maybe, Single, Completable

ReactiveX documentation: http://reactivex.io/documentation/operators/to.html

Convert the sequence of items emitted by an Observable into a Collection that is also a Map keyed by a specified key function.

Observable.just(1, 2, 3, 4)
        .toMultimap((x) -> {
            // defines the key in the Map
            return (x % 2 == 0) ? "even" : "odd";
        }, (x) -> {
            // defines the value that is mapped to the key
            return x;
        })
        .subscribe(System.out::println);

// prints {even=[2, 4], odd=[1, 3]}