RxJava Operator Series 3 (Part 1)

RxJava Operator Series Portal

• RxJava operator source code

https://github.com/xiehui999/fuseProgram

• RxJava Operator Series 1
• RxJava Operator Series 2

[[180583]]

Preface

In the previous article, we introduced some operators for creating Observables and converting data. Some of the operators for data conversion are still difficult to understand, but I believe that if you type the code once and modify the values ​​of various parameters and observe the execution log, it is still easy to understand. On the official website, each operator is given a legend. If you don’t understand the text clearly, you can also refer to the diagram to help yourself understand. In this article, we will introduce some common filter operators. Filter operators in RxJava are also relatively easy to understand. Okay, let’s continue our learning journey together.

Filter

This operator receives a Func1 parameter, in which we can use our own judgment conditions to judge the data we want to filter. When the data passes the judgment condition, it returns true to indicate that the data is emitted, otherwise it is not emitted, thus filtering out the data we want. As shown below, we filter out numbers that are not divisible by 2.

 Integer [] ints = {1, 2, 3, 4, 5, 6, 7, 8, 9}; 
 
       Observable observable = Observable. from (ints).filter(new Func1< Integer , Boolean>() { 
 
 @Override 
 
 public Boolean call( Integer   integer ) { 
 
 return   integer % 2 != 0; //Return true , it will not be filtered out, the data will be emitted, and the value that returns false will be filtered out 
 
 } 
 
 }); 
 
       Action1 action1 = new Action1< Integer >() { 
 
 @Override 
 
 public void call( Integer i) { 
 
               Log.e(TAG, "call: " +i ); 
 
 } 
 
 }; 
 
       observable.subscribe(action1);

Output log information

 call: 1 
 
 call: 3 
 
 call: 5 
 
 call: 7 
 
 call: 9

ofType

This operator is a special form of the filter operator. It filters an Observable to return only data of a specified type. For example, when the data source has string and int data, we can use this operator if we want to filter out the string, as shown in the following example code

 Observable.just(0, "one" , 6, 4, "two" , 8, "three" , 1, "four" , 0) 
 
 .ofType(String.class) 
 
 .subscribe(new Subscriber<String>() { 
 
 @Override 
 
 public void onCompleted() { 
 
 Log.e(TAG, "onCompleted:ofType " ); 
 
 } 
 
  
 
 @Override 
 
 public void onError(Throwable e) { 
 
 Log.e(TAG, "onError:ofType " ); 
 
 } 
 
  
 
 @Override 
 
 public void onNext(String string) { 
 
 Log.e(TAG, "onNext:ofType " + string); 
 
 } 
 
 });

Output log information

 onNext:ofType one 
 
 onNext:ofType two 
 
 onNext:ofType three 
 
 onNext:ofType four 
 
 onCompleted:ofType

Of course, in addition to filtering basic types of data, you can also filter custom types of data.

First

If we are only interested in the first item of data emitted by the Observable, or the first item of data that meets a certain condition, we can use the First operator.

 Observable.just(10, 11, 12, 13). first ().subscribe(new Action1() { 
 
 @Override 
 
 public void call( Integer   integer ) { 
 
                Log.e(TAG, integer + "" ); 
 
 } 
 
 });

The above log only prints a value of 10. Of course, we can also pass a parameter Fun1 to first and specify a condition as follows

 Observable.just(10, 11, 12, 13). first (new Func1< Integer , Boolean>() { 
 
 @Override 
 
 public Boolean call( Integer   integer ) { 
 
 return   integer > 12; 
 
 } 
 
        }).subscribe(new Action1 <Integer> () { 
 
 @Override 
 
 public void call( Integer   integer ) { 
 
                Log.e(TAG, integer + "" ); 
 
 } 
 
 });

The output information at this time is the first item of data 13 that satisfies integer > 12.

firstOrDefault

This operator is a variation of the first operator. It mainly emits a default value you specify in the parameter when no data is emitted. As shown below, it has two overloaded methods.

 Observable.just(11,12,13).firstOrDefault(10).subscribe(new Action1<Object>() { 
 
 @Override 
 
 public void call(Object o) { 
 
                Log.e(TAG, o.toString()); 
 
 } 
 
 });

If the above code is written, the execution will have the same effect as the first one. Because the default value is used when no data is emitted, we change the above code as follows, using empty to create an Observable that does not emit any data but terminates normally.

 Observable.empty().firstOrDefault(10).subscribe(new Action1<Object>() { 
 
 @Override 
 
 public void call(Object o) { 
 
                Log.e(TAG, o.toString()); 
 
 } 
 
 });

It is found that the data 10 is output at this time. The operator also provides an overloaded method firstOrDefault(T defaultValue, Func1 super T, Boolean> predicate) with two parameters. We can add a condition. The following example

 Observable.just(10, 13,16).firstOrDefault(15, new Func1< Integer , Boolean>() { 
 
 @Override 
 
 public Boolean call( Integer   integer ) { 
 
 return   integer >20; 
 
 } 
 
        }).subscribe(new Action1 <Integer> () { 
 
 @Override 
 
 public void call( Integer   integer ) { 
 
                Log.e(TAG, "" + integer ); 
 
 } 
 
 });

At this time, data sources 10, 13, and 16 do not meet the requirement of being greater than 20, so the default value 15 will be output. If we increase the data source data by a value of 22, then the default value will no longer be output, but 22 will be output.

takeFirst

The difference between this operator and the first operator is that if the original Observable does not emit any data that meets the conditions, first will throw a NoSuchElementException and directly execute onError(), while takeFist will return an empty Observable (onNext() will not be called but onCompleted will be called)

As shown in the following sample code

 Observable.just(10,11).filter(new Func1< Integer , Boolean>() { 
 
 @Override 
 
 public Boolean call( Integer   integer ) { 
 
 return   integer >20; 
 
 } 
 
        }). first ().subscribe(new Subscriber<Object>() { 
 
  
 
 @Override 
 
 public void onCompleted() { 
 
                Log.e(TAG, "onCompleted: " ); 
 
 } 
 
  
 
 @Override 
 
 public void onError(Throwable e) { 
 
                Log.e(TAG, "onError: " +e.toString()); 
 
 } 
 
  
 
 @Override 
 
 public void onNext(Object o) { 
 
                Log.e(TAG, "onNext: " +o.toString()); 
 
 } 
 
 });

The output information after execution is as follows

 onError: java.util.NoSuchElementException: Sequence   contains   no elements

If you use takeFirst at this time

 Observable.just(10,11).takeFirst(new Func1< Integer , Boolean>() { 
 
 @Override 
 
 public Boolean call( Integer   integer ) { 
 
                Log.e(TAG, "call: takeFirst" ); 
 
 return   integer >30; 
 
 } 
 
        }).subscribe(new Subscriber<Object>() { 
 
  
 
 @Override 
 
 public void onCompleted() { 
 
                Log.e(TAG, "onCompleted: " ); 
 
 } 
 
  
 
 @Override 
 
 public void onError(Throwable e) { 
 
                Log.e(TAG, "onError: " +e.toString()); 
 
 } 
 
  
 
 @Override 
 
 public void onNext(Object o) { 
 
                Log.e(TAG, "onNext: " +o.toString()); 
 
 } 
 
 });

It is found that no exception occurs at this time, but onCompleted() is executed.

single

If the original Observable does not emit data exactly once before completion, it will throw a NoSuchElementException. In plain language, it can be understood that if the data sent is one item, the value of this item is output. If there are multiple data, an exception is thrown and the onError() method is executed.

The following code

 Observable.just(10, 11, 12, 13).single().subscribe(new Subscriber<Integer> ( ) { 
 
 @Override 
 
 public void onCompleted() { 
 
                 Log.e(TAG, "onCompleted" ); 
 
 } 
 
  
 
 @Override 
 
 public void onError(Throwable e) { 
 
                 Log.e(TAG, "onError" +e.toString()); 
 
 } 
 
  
 
 @Override 
 
 public void onNext( Integer   integer ) { 
 
                 Log.e(TAG, integer ); 
 
 } 
 
 });

Output information

 onError: java.util.NoSuchElementException: Sequence   contains   no elements

If you make a simple change to the above code

 Observable.just(10, 11, 12, 13).filter(new Func1< Integer , Boolean>() { 
 
 @Override 
 
 public Boolean call( Integer   integer ) { 
 
 return   integer > 12; 
 
 } 
 
        }).subscribe(new Subscriber <Integer> () { 
 
 @Override 
 
 public void onCompleted() { 
 
                 Log.e(TAG, "onCompleted" ); 
 
 } 
 
  
 
 @Override 
 
 public void onError(Throwable e) { 
 
                 Log.e(TAG, "onError" +e.toString()); 
 
 } 
 
  
 
 @Override 
 
 public void onNext( Integer   integer ) { 
 
                 Log.e(TAG, integer ); 
 
 } 
 
 });

At this time, the data 13 will be output, because there is only one data after passing the filter. single also has two variants: singleOrDefault(T) and singleOrDefault(T,Func1). You can test the difference with your own code.

Last

This operator has the opposite meaning of first. It is used when we are only interested in the last data emitted by Observable or the last data that meets a certain condition.

Sample Code

 Observable.just(10, 11, 12, 13) .last ().subscribe(new Action1<Integer> ( ) { 
 
 @Override 
 
 public void call( Integer   integer ) { 
 
               Log.e(TAG, "call: " + integer ); 
 
 } 
 
 });

After execution, the output is 13. It has an overloaded method that can specify conditions and obtain the last data that meets the conditions. Modify the above code as follows

 Observable.just(10, 11, 12, 13). last (new Func1< Integer , Boolean>() { 
 
 @Override 
 
 public Boolean call( Integer   integer ) { 
 
 return   integer < 12; 
 
 } 
 
        }).subscribe(new Action1 <Integer> () { 
 
 @Override 
 
 public void call( Integer   integer ) { 
 
                Log.e(TAG, "call: " + integer ); 
 
 } 
 
 });

At this time, the final output data is 11. This operator has several variants like first, such as lastOrDefault and TakeLast. You can test the specific effect yourself.

Skip

This operator skips the first few items of data and then emits the data.

 Observable.range(1, 10).skip(6).subscribe(new Action1<Integer> ( ) { 
 
 @Override 
 
 public void call( Integer   integer ) { 
 
               Log.e(TAG, "call: " + integer ); 
 
 } 
 
 });

Output log information

 call: 7 
 
 call: 8 
 
 call: 9 
 
 call: 10

There are two overloaded methods for skip.skip(long time, TimeUnit unit) is executed on the computation scheduler by default. If you want to update the UI, you need to specify it as AndroidSchedulers.mainThread() through the observeOn method. Of course, there is also an overloaded method skip(long time, TimeUnit unit, Scheduler scheduler) to specify the scheduler. One thing to note is that the first parameter of these two overloaded methods is not the number of data to be skipped, but the time.

 Observable.interval(500, TimeUnit.MILLISECONDS) 
 
                .skip(2, TimeUnit.SECONDS) 
 
                .observeOn(AndroidSchedulers.mainThread()) 
 
                .subscribe(new Subscriber<Long>() { 
 
 @Override 
 
 public void onCompleted() { 
 
  
 
 } 
 
  
 
 @Override 
 
 public void onError(Throwable e) { 
 
  
 
 } 
 
  
 
 @Override 
 
 public void onNext(Long aLong) { 
 
                        tv.append( "\n" + aLong); 
 
                        if (aLong > 10) { 
 
                            this.unsubscribe(); 
 
 } 
 
 } 
 
 });

As shown in the above code, interval generates a data every 500 milliseconds, skip is set to 2 seconds, and the subscription is unsubscribed when the data is greater than 10.

skipLast

Just the opposite of skip, ignoring the n data items generated by ***

 Observable.range(1, 10).skipLast(6).subscribe(new Action1<Integer> ( ) { 
 
 @Override 
 
 public void call( Integer   integer ) { 
 
               Log.e(TAG, "call: " + integer ); 
 
 } 
 
 });

Output log information

 call: 1 
 
 call: 2 
 
 call: 3 
 
 call: 4

Continue