LollipopNetworkObservingStrategy implements NetworkObservingStrategy { @Override public Observable<Connectivity> observeNetworkConnectivity(final Context context) { final String service = Context.CONNECTIVITY_SERVICE; final ConnectivityManager manager = (ConnectivityManager) context.getSystemService(service); return Observable.create(new ObservableOnSubscribe<Connectivity>() { @Override public void subscribe(ObservableEmitter<Connectivity> subscriber) throws Exception { networkCallback = createNetworkCallback(subscriber, context); final NetworkRequest networkRequest = new NetworkRequest.Builder().build(); manager.registerNetworkCallback(networkRequest, networkCallback); } }).doOnDispose(new Action() { @Override public void run() { tryToUnregisterCallback(manager); } }).startWith(Connectivity.create(context)).distinctUntilChanged(); }  @Override Observable<Connectivity> observeNetworkConnectivity(final Context context); @Override void onError(final String message, final Exception exception);  }
@Test public void shouldObserveConnectivity() { final NetworkObservingStrategy strategy = new LollipopNetworkObservingStrategy(); final Context context = RuntimeEnvironment.application.getApplicationContext(); Connectivity connectivity = strategy.observeNetworkConnectivity(context).blockingFirst(); assertThat(connectivity.state()).isEqualTo(NetworkInfo.State.CONNECTED); }
@Test public void shouldStopObservingConnectivity() { final NetworkObservingStrategy strategy = new LollipopNetworkObservingStrategy(); final Application context = RuntimeEnvironment.application; final Observable<Connectivity> observable = strategy.observeNetworkConnectivity(context); final TestObserver<Connectivity> observer = new TestObserver<>(); observable.subscribe(observer); observer.dispose(); assertThat(observer.isDisposed()).isTrue(); }