RxNetworkConnectivity.java

public class NetworkConnectivity {

  @VisibleForTesting public static final String SYSTEM_SERVICE = Context.CONNECTIVITY_SERVICE;
  @VisibleForTesting public Subscription networkSubscription = Subscriptions.empty();

  public void addConnectivityCallback(Context context, ConnectivityCallback callback) {
    networkSubscription =
        createConnectivityObserver(context).observeOn(new MainThread().getScheduler())
            .subscribe(callback);
  }

  private Observable<Void> createConnectivityObserver(Context context) {
    ConnectivityManager connectivityManager = getConnectivityManager(context);

    return Observable.create(emitter -> {
      ConnectivityManager.NetworkCallback networkCallback =
          new ConnectivityManager.NetworkCallback() {

            @Override public void onAvailable(Network network) {
              emitter.onCompleted();
            }
          };

      connectivityManager.registerNetworkCallback(new NetworkRequest.Builder().build(),
          networkCallback);

      emitter.setCancellation(() -> connectivityManager.unregisterNetworkCallback(networkCallback));
    }, Emitter.BackpressureMode.BUFFER);
  }

  public void removeConnectivityCallback() {
    if (!networkSubscription.isUnsubscribed()) {
      networkSubscription.unsubscribe();
    }
  }

  private ConnectivityManager getConnectivityManager(Context context) {
    return (ConnectivityManager) context.getSystemService(SYSTEM_SERVICE);
  }

  public abstract static class ConnectivityCallback extends Subscriber<Void> {

    public abstract void onConnected();

    @Override public void onCompleted() {
      unsubscribe();
      onConnected();
    }

    @Override public void onError(Throwable e) {
      unsubscribe();
    }

    @Override public void onNext(Void aVoid) {
      // Empty events
    }
  }
}