RxJava is basically a library that helps the programmers to work on the asynchronous applications. In this post we will be seeing about Reactive programming and how it can be implemented with the help of RxJava. First of all let us see what is reactive programming.
Reactive programming is a method of programming where you define a source and a consumer. The source and receiver is then connected through. The source generates data required and then the consumer needs the data to process, this is when RxJava comes into the picture. Once the consumer and the source are connected then the library will take care of pushing the data and henceforth.
1) Observable – source
2) Subscriber – consumer
3) Subscribe – connect
Below we will see in detail about the source, consumer and the connect using subscribe.
Observable and subscriber class:
Observable is a stream abstraction in RxJava. Observable class emits a stream of data or events. Subscriber class on the other hand acts upon the emitted items. Observable is to emit one or more items, and then completed successfully or with an error.
Observable can have multiple subscribers, and each item emitted by the observable, the item will be sent to the Subscriber.onNext() method to be handled. Once Observable finished emitting items, is call the Subscriber.onCompleted() method, else call the error the observable will call the Subscriber.onError() method.
Observable integerObservable = Observable.create(new Observable.OnSubscribe() {
@Override
public void call(Subscriber subscriber) {
subscriber.onNext(1);
subscriber.onNext(2);
subscriber.onNext(3);
subscriber.onCompleted();
}
});
Observable has emits the integers 1, 2, and 3 and then completes. Now we need to create a subscriber so we can act upon the stream of emissions.
Subscriber integerSubscriber = new Subscriber() {
@Override
public void onCompleted() {
Log.d(“Subscriber”, “onCompleted: ” + true);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer value) {
Log.d(“Subscriber”, “onNext: ” + value);
}
};
Subscriber will simply print out any items emitted and notify us upon completion. Once you have an Observable and a Subscriber you can connect them with the Observable.subscribe() method.
integerObservable.subscribe(integerSubscriber);
// Outputs:
// onNext: 1
// onNext: 2
// onNext: 3
// Complete!
Observable code can be simplified by using the Observable.just() method to create an observable to emit only the values defined, and changing our subscriber to an anonymous inner class, we get the following.
Observable.just(1, 2 ,3).subscribe(new Subscriber() {
@Override
public void onCompleted() {
Log.d(“Just”,”Completed”);
}
@Override
public void onError(Throwable e) {}
@Override
public void onNext(Integer value) {
Log.d(“Just”,”onNext: ” + value);
}
});
Operators:
Creating a subscribing to Observable is simple enough, and may not seem overly useful, but that is just the beginning of what is possible with RxJava
Observable can have its output transformed by what is called an Operator.
Multiple Operators can be chained onto Observable.
Observable.just(1, 2, 3, 4, 5, 6, 7, 8) // add more numbers
.filter(new Func1() {
@Override
public Boolean call(Integer value) {
return value % 2 == 1;
}
})
.subscribe(new Subscriber() {
@Override
public void onCompleted() {
System.out.println(“Complete!”);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer value) {
System.out.println(“onNext: ” + value);
}
});
// Outputs:
// onNext: 1
// onNext: 3
// onNext: 5
/ onNext: 7
// Complete!
In filter() operator defines a function that will take in our emitted Integer values, and return true for all odd number, else false for even number. the event number return false from our filter() function are not emitted to the subscriber.
Thus we have seen a sample program of how to define a source and consumer with observable and subscriber class and how to to connect them with the help of Subscribe function. This post will give you a clear idea about the way of incorporating reactive programming with the help of RxJava on Android platform.