compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
compile 'io.reactivex.rxjava2:rxjava:2.0.7'
compile 'io.reactivex:rxandroid:1.2.0'
compile 'io.reactivex:rxjava:1.1.5'

compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
//compile 'io.reactivex.rxjava2:rxjava:2.0.7'
compile 'io.reactivex:rxandroid:1.2.0'
//compile 'io.reactivex:rxjava:1.1.5'

compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
compile 'io.reactivex.rxjava2:rxjava:2.0.7'

Because RxAndroid releases are few and far between, it is recommended you also
explicitly depend on RxJava's latest version for bug fixes and new features.

  Observable.create(new Observable.OnSubscribe<String>() {

   @Override
   public void call(Subscriber<? super String> subscriber) {
    subscriber.onNext("hello");
   }

  }).subscribe(new Action1<String>() {

   @Override
   public void call(String s) {
    System.out.println(s);
   }
  });

  Observable.create(new ObservableOnSubscribe<String>() {

   @Override
   public void subscribe(ObservableEmitter<String> e) throws Exception {
    e.onNext("hello");
   }

  }).subscribe(new Consumer<String>() {

   @Override
   public void accept(String s) {
    System.out.println(s);
   }
  });

 /**
  * 跟compose()配合使用,比如ObservableUtils.wrap(obj).compose(toMain())
  * @param <T>
  * @return
  */
 public static <T> Observable.Transformer<T, T> toMain() {
  return new Observable.Transformer<T, T>() {
   @Override
   public Observable<T> call(Observable<T> tObservable) {
    return tObservable
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread());
   }
  };
 }

 /**
  * 跟compose()配合使用,比如ObservableUtils.wrap(obj).compose(toMain())
  * @param <T>
  * @return
  */
 public static <T> ObservableTransformer<T, T> toMain() {

  return new ObservableTransformer<T, T>() {

   @Override
   public ObservableSource<T> apply(Observable<T> upstream) {
    return upstream.subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread());
   }
  };
 }

 public static <T> FlowableTransformer<T, T> toMain() {

  return new FlowableTransformer<T, T>() {

   @Override
   public Publisher<T> apply(Flowable<T> upstream) {
    return upstream.subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread());
   }
  };
 }

ConnectableObservable<Data> connectableObservable = Observable
   .concat(Observable.from(list))
   .first(new Func1<Data, Boolean>() {
    @Override
    public Boolean call(Data data) {
     return DataUtils.isAvailable(data);
    }
   }).publish();

ConnectableObservable<Data> connectableObservable = Observable
   .concat(Observable.fromIterable(list))
   .filter(new Predicate<Data>() {

    @Override
    public boolean test(@NonNull Data data) throws Exception {
     return DataUtils.isAvailable(data);
    }
   }).firstElement().toObservable().publish();

import rx.Observable;

/**
 * 使用方法:
 *   String s = null;
 *   Optional.ofNullable(s).orElse("default")); // 如果s为null,则显示default,否则显示s的值
 * @author Tony Shen
 *
 */
public class Optional<T> {

 Observable<T> obs;

 public Optional(Observable<T> obs) {
  this.obs = obs;
 }

 public static <T> Optional<T> of(T value) {
  if (value == null) {
   throw new NullPointerException();
  } else {
   return new Optional<T>(Observable.just(value));
  }
 }

 public static <T> Optional<T> ofNullable(T value) {
  if (value == null) {
   return new Optional<T>(Observable.<T>empty());
  } else {
   return new Optional<T>(Observable.just(value));
  }
 }

 public T get() {
  return obs.toBlocking().single();
 }

 public T orElse(T defaultValue) {
  return obs.defaultIfEmpty(defaultValue).toBlocking().single();
 }
}

import io.reactivex.Observable;

/**
 * 使用方法:
 *   String s = null;
 *   Optional.ofNullable(s).orElse("default"); // 如果s为null,则显示default,否则显示s的值
 * @author Tony Shen
 *
 */
public class Optional<T> {

 Observable<T> obs;

 public Optional(Observable<T> obs) {
  this.obs = obs;
 }

 public static <T> Optional<T> of(T value) {
  if (value == null) {
   throw new NullPointerException();
  } else {
   return new Optional<T>(Observable.just(value));
  }
 }

 public static <T> Optional<T> ofNullable(T value) {
  if (value == null) {
   return new Optional<T>(Observable.<T>empty());
  } else {
   return new Optional<T>(Observable.just(value));
  }
 }

 public T get() {

  return obs.blockingSingle();
 }

 public T orElse(T defaultValue) {

  return obs.defaultIfEmpty(defaultValue).blockingSingle();
 }
}