In the first chapter's reactive sum example, we were filtering the user input, based on a special pattern. The pattern was, for example, a: <number>. It is common to filter only interesting bits of data from the data stream. For example, it's useful to filter out <enter> key-down events only from all key-down events, or only lines containing a given expression from a file. That's why it is important to not only be able to transform our data but also to learn how to filter it.

There are many filtering operators in RxJava. The most important of these operators is filter(). Its marble diagram is very simple and is shown here:

It shows that the filter() operator filters the data by some property. In the diagram, it's the form of the elements: it filters only circles. Like all the other operators, filter() creates a new Observable instance from the source. This Observable instance emits only items that comply to the condition, defined by the filter() operator. The following piece of code illustrates that:

Observable<Integer> numbers = Observable
  .just(1, 13, 32, 45, 21, 8, 98, 103, 55);
Observable<Integer> filter = numbers
  .filter(n -> n % 2 == 0);
subscribePrint(filter, "Filter");

This will output only even numbers (32, 8, and 98), because of the filtering condition.

The filter() operator filters elements based on a user-defined function. There are quite a few additional filtering operators. In order to understand them, let's look at some simple examples:

Observable<Integer> numbers = Observable
  .just(1, 13, 32, 45, 21, 8, 98, 103, 55);
Observable<String> words = Observable
  .just(
    "One", "of", "the", "few", "of",
    "the", "crew", "crew"
  );
Observable<?> various = Observable
  .from(Arrays.asList("1", 2, 3.0, 4, 5L));

We define three Observable instances to use in our examples. The first one emits nine numbers. The second one emits all the words from a sentence, one by one. The third one emits elements of different types—strings, integers, doubles, and longs.

subscribePrint(numbers.takeLast(4), "Last 4");

The takeLast() operator returns a new Observable instance that emits only the last n items from the source Observable instance, only when it completes. This method has a few overloads. For example, there is one that emits the last N or less items from the source, emitted in a specified time window. Another one can receive a Scheduler instance in order to be executed on another thread.

In this example, only the last four items of the Observable instance will be filtered and output:

Last 4 : 8
Last 4 : 98
Last 4 : 103
Last 4 : 55
Last 4 ended!

Let's take a look at the following code snippet:

subscribePrint(numbers.last(), "Last");

The Observable instance created by the last() operator, which outputs only the last item emitted by the source Observable instance when it completes. If the source doesn't emit an item, a NoSuchElementException exception will be emitted as an OnError() notification. It has an overload that receives a predicate parameter of type T->Boolean. As a result, it emits only the last item emitted by the source, complying to the condition defined by the predicate. In this example, the output will be as follows:

Last : 55
Last ended!

The takeLastBuffer() method behaves much like the takeLast() method, but the Observable instance created by it emits only one item—a List instance containing the last N items from the source:

subscribePrint(
  numbers.takeLastBuffer(4), "Last buffer"
);

It has analogical overloads to the takeLast() method's. The output here is as follows:

Last buffer : [8, 98, 103, 55]
Last buffer ended!

The lastOrDefault() operator behaves like and has the same overload with a predicate as the last() operator:

subscribePrint(
  numbers.lastOrDefault(200), "Last or default"
);
subscribePrint(
  Observable.empty().lastOrDefault(200), "Last or default"
);

However, if the source doesn't emit anything, the lastOrDefault() operator emits the default value instead of the OnError notification. The output of this example is as follows:

Last or default : 55
Last or default ended!
Last or default : 200
Last or default ended!

The skipLast() operator is the exact opposite of the takeLast() method; it emits everything except the last N items from the source when it completes:

subscribePrint(numbers.skipLast(4), "Skip last 4");

It has similar overloads. The output of this example is as follows:

Skip last 4 : 1
Skip last 4 : 13

The skip() method is the same as the skipLast() method but skips the first N items instead of the last:

subscribePrint(numbers.skip(4), "Skip 4");

This means that the output of the example is as follows:

Skip 4 : 21
Skip 4 : 8
Skip 4 : 98
Skip 4 : 103
Skip 4 : 55
Skip 4 ended!

The take() operator is similar to the takeLast() operator, but instead of the last N items of the source, it emits the first N items.

subscribePrint(numbers.take(4), "First 4");

This is a commonly-used operator, cheaper than the takeLast() operator, because the takeLast() operator buffers its items and waits for the source to complete. This operator doesn't buffer its items but emits them when it receives them. It is very useful for limiting infinite Observable instances. The output of the preceding example is as follows:

First 4 : 1
First 4 : 13
First 4 : 32
First 4 : 45
First 4 ended!

Let's take a look at the following code snippet:

subscribePrint(numbers.first(), "First");

The first() operator is similar to the last() operator but emits only the first item emitted by the source. It emits the same OnError notification if there is no first item. Its predicate form has an alias— the takeFirst() operator. There is also a firstOrDefault() operator form of this operator. The output of this example is clear:

First : 1
First ended!

Let's take a look at the following code snippet:

subscribePrint(numbers.elementAt(5), "At 5");

The elementAt() operator is similar to the first() and last() operators but has no predicate form. There is an elementAtOrDefault() form though. It emits only the element at the specified index in the sequence of items, emitted by the source Observable instance. This example outputs the following:

At 5 : 8
At 5 ended!

Let's take a look at the following code snippet:

subscribePrint(words.distinct(), "Distinct");

The Observable instance produced by the distinct() operator emits the items from the source, excluding the repeated ones. There is an overload that can receive a function, returning a key or hash code value to be used to decide whether an item is distinct from another or not:

Distinct : One
Distinct : of
Distinct : the
Distinct : few
Distinct : crew
Distinct ended!

subscribePrint(
  words.distinctUntilChanged(), "Distinct until changed"
);

The distinctUntilChanged() operator is similar to the distinct() method, but the Observable instance that it returns emits all items emitted by the source Observable instance that are distinct from their immediate predecessors. So, in this example, it will emit every word, except the last one, crew.

subscribePrint( // (13)
  various.ofType(Integer.class), "Only integers"
);

The ofType() operator creates an Observable instance that emits only the items emitted by the source of a given type. It basically is a shortcut to this call: filter(v -> Class.isInstance(v)). In this example the output will be as follows:

Only integers : 2
Only integers : 4
Only integers ended!

These are the most commonly used filtering operators provided by RxJava. We'll be using some of them a lot in later examples.

The last operator we'll look at in this chapter is a transformational one, but a bit special. It can use previously accumulated states! Let's learn more about it.