kevalpatel2106

subscription.unsubscribe();

kevalpatel2106

public class MainActivity extends BaseActivity {
    private CompositeSubscription mSubscription = new CompositeSubscription();

    @Override
    protected void onCreate(Bundle savedInstanceState) {
      //...
      //...
      //...
      
      mSubscription.add(subscription1); //Add subscription 1

kevalpatel2106

Action1<Integer> onNextAction = new Action1<Integer>() {
    @Override
    public void call(Integer s) { //This is eqivelent to onNext()
        System.out.println(s);
    }
};

kevalpatel2106

android {
    buildToolsVersion ‘21.1.2’//Use the latest version at the time.
    defaultConfig {
      // Enable the experimental Jack build tools.
       jackOptions {
         enabled true
       }
    }
    ...
}

kevalpatel2106

android {
  defaultConfig {
    jackOptions {
      enabled true
    }
  }
  
  //Add support for java 8 features.
  compileOptions {
    sourceCompatibility JavaVersion.VERSION_1_8

kevalpatel2106

Observable.from(new Integer[]{1, 2, 3, 4, 5});

kevalpatel2106

Observable<Integer> observer1 = Observable.from(new Integer[]{1, 2, 3, 4, 5});  //Emit 1 to 5
Observable<Integer> observer2 = Observable.from(new Integer[]{6, 7, 8, 9, 10}); //Emit 6 to 10

Observable.concat(observer1, observer2) //Concat the output of both the operators.
        .subscribeOn(Schedulers.newThread())
        .observeOn(Schedulers.io())
        .subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer integer) {
                Log.d("Observer", "Output:" + integer);

kevalpatel2106

  Observable<Integer> observer1 = Observable.from(new Integer[]{1, 2, 3, 4, 5});
  Observable<Integer> observer2 = Observable.from(new Integer[]{6, 7, 8, 9, 10});

  Observable.merge(observer1, observer2)
          .subscribeOn(Schedulers.newThread())
          .observeOn(Schedulers.io())
          .subscribe(new Action1<Integer>() {
              @Override
              public void call(Integer integer) {
                  Log.d("Observer", "Output:" + integer);

kevalpatel2106

//Create observable 
Observable<String> observable = Observable
          .from(new Integer[]{1,2,3,4,5})
          .map(new Func1<Integer, String>() {
              @Override
              public String call(Integer integer) {
                  return integer + "ABC"; //Convert integer to string and emmit
              }
          });

kevalpatel2106

//Class that combines both data streams
class ZipObject {
    int number; 
    String alphabet;
}

Observable<Integer> observable1 = Observable.from(new Integer[]{1, 2, 3, 4, 5});  //Emits integers
Observable<String> observable2 = Observable.from(new String[]{"A", "B", "C", "D", "F"});  //Emits alphabets
Observable<ZipObject> observable = Observable.zip(observable1, observable2,   
    //Function that define how to zip outputs of both the stream into single object.