Getting Started With RxAndroid observable Observer Subscription

RxJava is a Java VM implementation of ReactiveX (Reactive Extensions) developed by Netflix, and it is one of the most discussed libraries for enabling Reactive Programming in Android development .functional Reactive Programming is hard when you come from an imperative world but once you understand it, it’s so awesome! It is lot like Observer Design Pattern .

Basics

The basic building blocks of reactive code are observable and observers , When working with ReactiveX we will be using this terms extensively.

Observable : Observable is an object which emits data .
Observer : Observer is the counter part of Observable which listens to the data emitted by Observable by subscribing.

Setting Up RxAndroid

To use RxAndroid in an Android Studio project, add it as a compile dependency in the app module’s build.gradle.

compile 'io.reactivex:rxandroid:1.1.0'
compile 'io.reactivex:rxjava:1.1.0'

Creating Observable Observer

a. Observable

The Observable class has many static methods to create Observable objects.The following code shows you how to use the just operator to create a very simple Observable that emits a single String.

Observable<String> observable = Observable.just("hello");

b. Observer

The Observer (Interface) is a counterpart of Observable which listens to the data emitted by it.

Observer<String> observer = new Observer<String>() {
    @Override
    public void onCompleted() {
    }

    @Override
    public void onError(Throwable e) {
    }

    @Override
    public void onNext(String s) {
        textView.setText(s);
    }
};

c. Subscription

c.1 Subscribe : Using Observable object call subscribe() method to attach observer , which takes observer as a parameter to it.

Subscription subscription;

subscription=  observable.subscribe(observer);

c.2 Unsubscribe : Using subscription object call unsubscribe method to detach observable from observer.

if (subscription != null && !subscription.isUnsubscribed()) {
    subscription.unsubscribe();
}

When using RxJava in Android application Perform the Subscribing and UnSubscribing of obsevables and obsevers in the Activity/fragment lifecyle i.e when you resume the screen subscribe in onResume() method callback and when you leave the screen unsubscribe either in onStop() or onDestroy() .

There are several static method in Observable class for creating Observable object lets see few of them

Just.
from.
fromCallable.
create.
single.

1 Just

Observable

Observable<String> observable = Observable.just("hello");

Subscribing Observale with observer

Subscription subscription;

subscription = observable.subscribe(new Observer<String>() {
    @Override
    public void onCompleted() {
    }

    @Override
    public void onError(Throwable e) {
    }

    @Override
    public void onNext(String s) {
        textView.setText(s);
    }
});

Unsubscribing Observale with observer

if (subscription != null && !subscription.isUnsubscribed()) {
    subscription.unsubscribe();
}

code: https://github.com/hdpavan/RxAndroidSample/blob/master/app/src/main/java/com/tutorialsbuzz/rxandroidsample/Examples/FragmentExample1.java

2 from

Observable

Observable<Integer> observable = 
         Observable.from(new Integer[]{1, 2, 3, 4, 5, 6});

Subscribing Observale with observer

private Subscription subscription;

subscription = observable.subscribe(new Observer<Integer>() {
    @Override
    public void onCompleted() {
    }

    @Override
    public void onError(Throwable e) {
    }

    @Override
    public void onNext(Integer integer) {
        textView.append(String.valueOf(integer) + "\n");
    }
});

Unsubscribing Observale with observer

if (subscription != null && !subscription.isUnsubscribed()) {
    subscription.unsubscribe();
}

code:https://github.com/hdpavan/RxAndroidSample/blob/master/app/src/main/java/com/tutorialsbuzz/rxandroidsample/Examples/FragmentExample3.java

Asynchronous

Asynchronously loading data is probably the most common use of RxJava in Android. lets see how to create .

observable.subscribeOn(Schedulers.io())
 .observeOn(AndroidSchedulers.mainThread())
 .subscribe(...)

Schedulers.io() observable.subscribeOn(Schedulers.io()) : Runs Obseverable piece of code in IO Thread .

AndroidSchedules.mainThread() observable.observeOn(AndroidSchedulers.mainThread()) : observer the data emitted by Observable in mainThread , i.e. we want our onNext() callback to be called on the UI thread

3 fromCallable

Observable

Observable<List<String>> observable = 
     Observable.fromCallable(new Callable<List<String>>() {
        @Override
        public List<String> call() {
           return mRestClient.getFavoriteTvShows();
       }
});

Subscribing Observale with observer

private Subscription mSubscription;

mSubscription = observable
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer<List<String>>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(List<String> strings) {
                displayData(strings);
            }
        });

Unsubscribing Observale with observer

if (mSubscription != null && !mSubscription.isUnsubscribed()) {
    mSubscription.unsubscribe();
}

code:https://github.com/hdpavan/RxAndroidSample/blob/master/app/src/main/java/com/tutorialsbuzz/rxandroidsample/Examples/FragmentExample4.java

4 Create

Observable

Observable<List<String>> fetchdata = 
    Observable.create(new Observable.OnSubscribe<List<String>>() {
        @Override
        public void call(Subscriber<? super List<String>> subscriber) {
          List<String> data = mRestClient.getFavoriteTvShows();
          subscriber.onNext(data);
          subscriber.onCompleted();
    }
});

Subscribing Observale with observer

mSubscription = fetchdata
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer<List<String>>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(List<String> strings) {
                displayData(strings);
            }
        });

Unsubscribing Observale with observer

if (mSubscription != null && !mSubscription.isUnsubscribed()) {
    mSubscription.unsubscribe();
}

code:https://github.com/hdpavan/RxAndroidSample/blob/master/app/src/main/java/com/tutorialsbuzz/rxandroidsample/Examples/FragmentExample5.java

5 Single

Observable

Single<List<String>> tvShowSingle = 
  Single.fromCallable(new Callable<List<String>>() {
       @Override
       public List<String> call() throws Exception {
          return mRestClient.getFavoriteTvShows();
      }
});

Subscribing Observale with observer

mSubscription = tvShowSingle.subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new SingleSubscriber<List<String>>() {
            @Override
            public void onSuccess(List<String> value) {
                displayData(value);
            }

            @Override
            public void onError(Throwable error) {

            }
        });

Unsubscribing Observale with observer

if (mSubscription != null && !mSubscription.isUnsubscribed()) {
    mSubscription.unsubscribe();
}

code:https://github.com/hdpavan/RxAndroidSample/blob/master/app/src/main/java/com/tutorialsbuzz/rxandroidsample/Examples/FragmentExample6.java