Retry with Connection

BroadcastObservable.java

import android.content.BroadcastReceiver;
import android.content.Context;
import android.content.Intent;
import android.content.IntentFilter;
import android.net.ConnectivityManager;
import android.net.NetworkInfo;
import android.os.Looper;

import rx.Observable;
import rx.Scheduler;
import rx.Subscriber;
import rx.Subscription;
import rx.android.schedulers.AndroidSchedulers;
import rx.functions.Action0;
import rx.subscriptions.Subscriptions;

public class BroadcastObservable implements Observable.OnSubscribe<Boolean> {

	private final Context context;

	public static Observable<Boolean> fromConnectivityManager(Context context) {
		return Observable.create(new BroadcastObservable(context))
				.share();
	}

	public BroadcastObservable(Context context) {
		this.context = context;
	}

	@Override
	public void call(Subscriber<? super Boolean> subscriber) {
		BroadcastReceiver receiver = new BroadcastReceiver() {
			@Override
			public void onReceive(Context context, Intent intent) {
				subscriber.onNext(isConnectedToInternet());
			}
		};

		context.registerReceiver(receiver, new IntentFilter(ConnectivityManager.CONNECTIVITY_ACTION));

		subscriber.add(unsubscribeInUiThread(() -> context.unregisterReceiver(receiver)));
	}

	private boolean isConnectedToInternet() {
		ConnectivityManager manager = (ConnectivityManager) context.getSystemService(Context.CONNECTIVITY_SERVICE);
		NetworkInfo networkInfo = manager.getActiveNetworkInfo();
		return networkInfo != null && networkInfo.isConnected();
	}

	private static Subscription unsubscribeInUiThread(final Action0 unsubscribe) {
		return Subscriptions.create(() -> {
			if (Looper.getMainLooper() == Looper.myLooper()) {
				unsubscribe.call();
			} else {
				final Scheduler.Worker inner = AndroidSchedulers.mainThread().createWorker();
				inner.schedule(() -> {
					unsubscribe.call();
					inner.unsubscribe();
				});
			}
		});
	}

}

RetryWithConnectivityIncremental.java

import android.content.Context;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import rx.Observable;
import rx.functions.Func1;

public class RetryWithConnectivityIncremental implements Func1<Observable<? extends Throwable>, Observable<?>> {
	private final int maxTimeout;
	private final TimeUnit timeUnit;
	private final Observable<Boolean> isConnected;
	private final int startTimeOut;
	private int timeout;

	public RetryWithConnectivityIncremental(Context context, int startTimeOut, int maxTimeout, TimeUnit timeUnit) {
		this.startTimeOut = startTimeOut;
		this.maxTimeout = maxTimeout;
		this.timeUnit = timeUnit;
		this.timeout = startTimeOut;
		isConnected = getConnectedObservable(context);
	}

	@Override
	public Observable<?> call(Observable<? extends Throwable> observable) {
		return observable.flatMap((Throwable throwable) -> {
			if (throwable instanceof RetrofitError && ((RetrofitError) throwable).getKind() == RetrofitError.Kind.NETWORK) {
				return isConnected;
			} else {
				return Observable.error(throwable);
			}
		}).compose(attachIncementalTimeout());
	}

	private Observable.Transformer<Boolean, Boolean> attachIncementalTimeout() {
		return observable -> observable.timeout(timeout, timeUnit)
				.doOnError(throwable -> {
					if (throwable instanceof TimeoutException) {
						timeout = timeout > maxTimeout ? maxTimeout : timeout + startTimeOut;
					}
				});
	}

	private Observable<Boolean> getConnectedObservable(Context context) {
		return BroadcastObservable.fromConnectivityManager(context)
				.distinctUntilChanged()
				.filter(isConnected -> isConnected);
	}

}