/**
  * 获取内容
  * @param fragment
  * @param param
  * @param cacheKey
  * @return
  */
 public Maybe<ContentModel> getContent(Fragment fragment, ContentParam param, String cacheKey) {

  if (apiService == null) {
   apiService = RetrofitManager.get().apiService();
  }

  return apiService.loadContent(param)
    .retryWhen(new RetryWithDelay(3,1000))
    .compose(RxLifecycle.bind(fragment).<ContentModel>toLifecycleTransformer())
    .compose(RxUtils.<ContentModel>toCacheTransformer(cacheKey));
 }

import com.safframework.log.L;

import org.reactivestreams.Publisher;

import java.util.concurrent.TimeUnit;

import io.reactivex.Flowable;
import io.reactivex.annotations.NonNull;
import io.reactivex.functions.Function;

/**
 * 重试机制
 * Created by tony on 2017/11/6.
 */

public class RetryWithDelay implements Function<Flowable<? extends Throwable>, Publisher<?>> {

 private final int maxRetries;
 private final int retryDelayMillis;
 private int retryCount;

 public RetryWithDelay(final int maxRetries, final int retryDelayMillis) {
  this.maxRetries = maxRetries;
  this.retryDelayMillis = retryDelayMillis;
  this.retryCount = 0;
 }

 @Override
 public Publisher<?> apply(@NonNull Flowable<? extends Throwable> attempts) throws Exception {

  return attempts.flatMap(new Function<Throwable, Publisher<?>>() {
   @Override
   public Publisher<?> apply(Throwable throwable) throws Exception {
    if (++retryCount <= maxRetries) {

     L.i("RetryWithDelay", "get error, it will try after " + retryDelayMillis
       + " millisecond, retry count " + retryCount);
     // When this Observable calls onNext, the original
     // Observable will be retried (i.e. re-subscribed).
     return Flowable.timer(retryDelayMillis, TimeUnit.MILLISECONDS);

    } else {

     // Max retries hit. Just pass the error along.
     return Flowable.error(throwable);
    }
   }
  });
 }
}

RetrofitManager.get()
    .adService()
    .vmw(param)
    .compose(RxLifecycle.bind(fragment).<VMWModel>toLifecycleTransformer())
    .subscribeOn(Schedulers.io())
    .onErrorReturn(new Function<Throwable, VMWModel>() {
     @Override
     public VMWModel apply(Throwable throwable) throws Exception {
      return new VMWModel();
     }
    });

import android.accounts.NetworkErrorException
import android.content.Context

import com.safframework.log.L
import io.reactivex.observers.DisposableMaybeObserver
import java.net.ConnectException
import java.net.SocketTimeoutException
import java.net.UnknownHostException

/**
 * Created by Tony Shen on 2017/8/8.
 */
abstract class BaseMaybeObserver<T> : DisposableMaybeObserver<T>() {

 internal var mAppContext: Context

 init {
  mAppContext = AppUtils.getApplicationContext()
 }

 override fun onSuccess(data: T) {
  onMaybeSuccess(data)
 }

 abstract fun onMaybeSuccess(data: T)

 override fun onError(e: Throwable) {
  var message = e.message
  L.e(message)

  when(e) {

   is ConnectException -> message = mAppContext.getString(R.string.connect_exception_error)
   is SocketTimeoutException -> message = mAppContext.getString(R.string.timeout_error)
   is UnknownHostException -> message = mAppContext.getString(R.string.network_error)
   is NetworkErrorException -> message = mAppContext.getString(R.string.network_error)
   else -> message = mAppContext.getString(R.string.something_went_wrong)
  }

  RxBus.get().post(FailedEvent(message))
 }

 override fun onComplete() {}
}

import android.accounts.NetworkErrorException
import android.content.Context

import com.safframework.log.L
import io.reactivex.observers.ResourceCompletableObserver
import java.net.ConnectException
import java.net.SocketTimeoutException
import java.net.UnknownHostException

/**
 * Created by Tony Shen on 2017/8/8.
 */
abstract class BaseCompletableObserver : ResourceCompletableObserver() {

 internal var mAppContext: Context

 init {
  mAppContext = AppUtils.getApplicationContext()
 }

 override fun onComplete() {
  onSuccess()
 }

 abstract fun onSuccess()

 override fun onError(e: Throwable) {
  var message = e.message
  L.e(message)

  when(e) {

   is ConnectException -> message = mAppContext.getString(R.string.connect_exception_error)
   is SocketTimeoutException -> message = mAppContext.getString(R.string.timeout_error)
   is UnknownHostException -> message = mAppContext.getString(R.string.network_error)
   is NetworkErrorException -> message = mAppContext.getString(R.string.network_error)
   else -> message = mAppContext.getString(R.string.something_went_wrong)
  }

  RxBus.get().post(FailedEvent(message))
 }
}

    model.getContent(VideoFragment.this,param, cacheKey)
      .compose(RxJavaUtils.<ContentModel>maybeToMain())
      .doFinally(new Action() {
       @Override
       public void run() throws Exception {
        refreshlayout.finishRefresh();
       }
      })
      .subscribe(new BaseMaybeObserver<ContentModel>(){

     @Override
     public void onMaybeSuccess(ContentModel data) {
      adpter.addDataToFront(data);
     }
    });