PreLollipopNetworkObservingStrategy implements NetworkObservingStrategy { @Override public Observable<Connectivity> observeNetworkConnectivity(final Context context) { final IntentFilter filter = new IntentFilter(); filter.addAction(ConnectivityManager.CONNECTIVITY_ACTION); return Observable.create(new ObservableOnSubscribe<Connectivity>() { @Override public void subscribe(final ObservableEmitter<Connectivity> emitter) throws Exception { final BroadcastReceiver receiver = new BroadcastReceiver() { @Override public void onReceive(Context context, Intent intent) { emitter.onNext(Connectivity.create(context)); } }; context.registerReceiver(receiver, filter); Disposable disposable = disposeInUiThread(new Action() { @Override public void run() { tryToUnregisterReceiver(context, receiver); } }); emitter.setDisposable(disposable); } }).defaultIfEmpty(Connectivity.create()); }  @Override Observable<Connectivity> observeNetworkConnectivity(final Context context); @Override void onError(final String message, final Exception exception);  }
@Test @SuppressWarnings("CheckReturnValue") public void shouldObserveConnectivity() { final NetworkObservingStrategy strategy = new PreLollipopNetworkObservingStrategy(); final Context context = RuntimeEnvironment.application.getApplicationContext(); strategy.observeNetworkConnectivity(context).subscribe(new Consumer<Connectivity>() { @Override public void accept(Connectivity connectivity) throws Exception { assertThat(connectivity.state()).isEqualTo(NetworkInfo.State.CONNECTED); } }); }
@Test public void shouldStopObservingConnectivity() { final NetworkObservingStrategy strategy = new PreLollipopNetworkObservingStrategy(); final Context context = RuntimeEnvironment.application.getApplicationContext(); final Observable<Connectivity> observable = strategy.observeNetworkConnectivity(context); final TestObserver<Connectivity> observer = new TestObserver<>(); observable.subscribe(observer); observer.dispose(); assertThat(observer.isDisposed()).isTrue(); }