Skip to content

Mathematical and Aggregate Operators

Jump to bottom Edit New page
DavidMGross edited this page Dec 26, 2013 · 33 revisions

This section explains operators that perform mathematical or other operations over an entire sequence of items emitted by an Observable. Because these operations must wait for the source Observable 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 that may have very long or infinite sequences.

  • average( ) — calculates the average of Integers emitted by an Observable and emits this average
  • averageLongs( ) — calculates the average of Longs emitted by an Observable and emits this average
  • averageFloats( ) — calculates the average of Floats emitted by an Observable and emits this average
  • averageDoubles( ) — calculates the average of Doubles emitted by an Observable and emits this average
  • count( ) and longCount( ) — counts the number of items emitted by an Observable and emits this count
  • max( ) — emits the maximum value emitted by a source Observable
  • maxBy( ) — emits the item emitted by the source Observable that has the maximum key value
  • min( ) — emits the minimum value emitted by a source Observable
  • minBy( ) — emits the item emitted by the source Observable that has the minimum key value
  • sum( ) — adds the Integers emitted by an Observable and emits this sum
  • sumLongs( ) — adds the Longs emitted by an Observable and emits this sum
  • sumFloats( ) — adds the Floats emitted by an Observable and emits this sum
  • sumDoubles( ) — adds the Floats emitted by an Observable and emits this sum

average( )

calculates the average of numbers emitted by an Observable and emits this average

The average( ) method returns an Observable that calculates the average of the Integers emitted by a source Observable and then emits this average as an Integer, as shown in the following sample code:

def myObservable = Observable.create({ anObserver ->
  anObserver.onNext(4);
  anObserver.onNext(3);
  anObserver.onNext(2);
  anObserver.onNext(1);
  anObserver.onCompleted();
});

myObservable.average().subscribe(
  { println(it); },                  // onNext
  { println("Error encountered"); }, // onError
  { println("Sequence complete"); }  // onCompleted
);
2
Sequence complete

There are also specialized "average" methods for Longs, Floats, and Doubles (averageLongs( ), averageFloats( ), and averageDoubles( )).

Note that these methods will fail with an IllegalArgumentException if the source Observable does not emit any items.

see also:

count( ) and longCount( )

counts the number of items emitted by an Observable and emits this count

The count( ) method returns an Observable that emits a single item: an Integer that represents the total number of items emitted by the source Observable, as shown in the following sample code:

def myObservable = Observable.create({ anObserver ->
  anObserver.onNext('Three');
  anObserver.onNext('Two');
  anObserver.onNext('One');
  anObserver.onCompleted();
});

myObservable.count().subscribe(
   { println(it); },                          // onNext
   { println("Error: " + it.getMessage()); }, // onError
   { println("Sequence complete"); }          // onCompleted
);
3
Sequence complete

longCount( ) is essentially the same, but emits its item as a Long rather than an Integer.

see also:

max( )

emits the maximum value emitted by a source Observable

The max( ) operator waits until the source Observable completes, and then emits the item emitted by the source Observable that had the highest value, before itself completing. If more than one item has this maximum value, max( ) emits the last such item. You may optionally pass in a comparator that max( ) will use to determine the maximum of two emitted items.

see also:

maxBy( )

emits the item emitted by the source Observable that has the maximum key value

The maxBy( ) operator is similar to max( ) but instead of emitting the maximum item emitted by the source Observable, it emits the last item from the source Observable that has the maximum key, where that key is generated by a function applied to each item. You supply this function.

see also:

min( )

emits the minimum value emitted by a source Observable

The min( ) operator waits until the source Observable completes, and then emits the item emitted by the source Observable that had the lowest value, before itself completing. If more than one item has this minimum value, min( ) emits the last such item. You may optionally pass in a comparator that min( ) will use to determine the minimum of two emitted items.

see also:

minBy( )

emits the item emitted by the source Observable that has the minimum key value

The minBy( ) operator is similar to min( ) but instead of emitting the minimum item emitted by the source Observable, it emits the last item from the source Observable that has the minimum key, where that key is generated by a function applied to each item. You supply this function.

see also:

sum( )

adds the numbers emitted by an Observable and emits this sum

The sum( ) method returns an Observable that adds the Integers emitted by a source Observable and then emits this sum as an Integer, as shown in the following sample code:

def myObservable = Observable.create({ anObserver ->
  anObserver.onNext(4);
  anObserver.onNext(3);
  anObserver.onNext(2);
  anObserver.onNext(1);
  anObserver.onCompleted();
});

myObservable.sum().subscribe(
  { println(it); },                  // onNext
  { println("Error encountered"); }, // onError
  { println("Sequence complete"); }  // onCompleted
);
10
Sequence complete

There are also specialized "sum" methods for Longs, Floats, and Doubles (sumLongs( ), sumFloats( ), and sumDoubles( )).

see also:

A Netflix Original Production
Tech Blog | Twitter @NetflixOSS | Twitter @RxJava | Jobs

Clone this wiki locally