Crash OnErrorNotImplementedException

[chihung93]

io.reactivex.exceptions.OnErrorNotImplementedException: 
  at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:704)
  at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:701)
  at io.reactivex.internal.observers.LambdaObserver.onError(LambdaObserver.java:74)
  at io.reactivex.internal.observers.LambdaObserver.onNext(LambdaObserver.java:64)
  at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.drainNormal(ObservableObserveOn.java:200)
  at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.run(ObservableObserveOn.java:252)
  at io.reactivex.android.schedulers.HandlerScheduler$ScheduledRunnable.run(HandlerScheduler.java:109)
  at android.os.Handler.handleCallback(Handler.java:751)
  at android.os.Handler.dispatchMessage(Handler.java:95)
  at android.os.Looper.loop(Looper.java:154)
  at android.app.ActivityThread.main(ActivityThread.java:6692)
  at java.lang.reflect.Method.invoke(Native Method:0)
  at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:1468)
  at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:1358)
Caused by: android.view.WindowManager$BadTokenException: 
  at android.view.ViewRootImpl.setView(ViewRootImpl.java:890)
  at android.view.WindowManagerGlobal.addView(WindowManagerGlobal.java:342)
  at android.view.WindowManagerImpl.addView(WindowManagerImpl.java:97)
  at android.app.Dialog.show(Dialog.java:538)
  at com.hung.nc.MyApplication$1.accept(MyApplication.java:93)
  at com.hung.nc.MyApplication$1.accept(MyApplication.java:75)
  at io.reactivex.internal.observers.LambdaObserver.onNext(LambdaObserver.java:60)

private void initConnectivity() {
        internetDisposable = ReactiveNetwork.observeInternetConnectivity(5000, "www.google.com", 80, 10000)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Boolean>() {
                    @Override
                    public void accept(Boolean isConnectedToInternet) {
                        if (isConnectedToInternet) {
                            if (!FavoriteManager.getInstance().hasSynchronized()) {
                                FavoriteManager.getInstance().synchronizeFavorites(true, null);
                            }
                            if (appStateMonitor.isAppInForeground()) {
                                if (errorInternet != null) {
                                    if (errorInternet.isShowing()) {
                                        errorInternet.dismiss();
                                    }
                                }
                            }
                        } else {
                            if (appStateMonitor.isAppInForeground() && ContextHelper.getMainActivity() != null) {
                                if (errorInternet == null) {
                                    errorInternet = AlertUtil.dialogError(ContextHelper.getMainActivity(), ResourceUtil.getString(R.string.error_internet), null);
                                    errorInternet.show();
                                } else if (!errorInternet.isShowing()) {
                                    errorInternet.show();
                                }
                            }
                        }
                    }
                });
    }

 @Override
    public void onAppDidEnterBackground() {
        safelyDispose(internetDisposable);
    }

    private void safelyDispose(Disposable... disposables) {
        for (Disposable subscription : disposables) {
            if (subscription != null && !subscription.isDisposed()) {
                subscription.dispose();
            }
        }
    }

Tks .

[pwittchen]

Hi,

Thanks for reporting this issue. You're getting that error because you haven't implemented an error handling in your subscription and pinging www.google.com fails (error occurs).

In order to handle this error, you should use Observer interface instead of Consumer interface as follows:

internetDisposable = ReactiveNetwork
    .observeInternetConnectivity(5000, "www.google.com", 80, 10000)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Observer<Boolean>() {
      @Override
      public void onSubscribe(final Disposable d) {
        // this will be invoked before operation is started
      }

      @Override
      public void onNext(final Boolean aBoolean) {
        // do your action, when you're connected to the internet
      }

      @Override
      public void onError(final Throwable e) {
        // handle an error here
      }

      @Override
      public void onComplete() {
        // this will be invoked when operation is completed
      }
    });

Both Consumer and Observer are RxJava2 operators.

Regards,
Piotr

[chihung93]

I don't think so.
I think that issue is a bug.
accept just return true or false for the network.
Why I must handle error while I just know network state.
Tks.

[Bhavdip]

If you want to observe the network connectivity is true or false then you should use default implementation But If you have your own custom details for checking Internet connectivity then you should Subscribe an Observer as @pwittchen mentioned. Using that way you can listen true and false in onNext method simply ignore other methods.

 ReactiveNetwork.observeInternetConnectivity()
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Consumer<Connectivity>() {
          @Override public void accept(Boolean isConnectedToInternet) {
            // do something with isConnectedToInternet value
          }
        });

[pwittchen]

There may occur another error not related with the boolean state of the connection. For example, connection timeout, local network problem or something like that. In such cases, you'll receive an error.

When I was testing this, I haven't encountered such issues and it worked fine with just Consumer, but I can imagine that in particular cases error may be thrown and in real apps, it's good to handle it.

[chihung93]

thank you so much @pwittchen @Bhavdip.

[pwittchen]

We discussed this issue, so I'm considering it as clarified and resolved. In the case of further problems, you can re-open this issue or create a new one.

[toe-pyae-sone-oo]

add second parameter in .subscribe(new Consumer<>(), new Consumer<Throwable>() {})
like that.

[javierpe]

If I don't want to know about any error and only execute .subscribe()?

[pwittchen]

@javierpe You can do it, but while checking Internet connectivity many exceptions may occur. If you won't handle them, they will crash your app. You don't have to do anything in onError method implementation, but logging information about error and showing appropriate message to the user if necessary is a good practice.

[Harmeetkaur01]

@pwittchen I m facing the same issue when I try to connected to 5G network otherwise it is working fine

[pwittchen]

@Harmeetkaur01 for some reason, your device is crashing while trying to connect to 5G network. This issue depends on a particular device - not the library. In order to handle this error gracefully, implement error handler in the way @toe-pyae-sone-oo suggested in the comment above.

[XProfessorJ]

I got this problem as well, and i think the fatal reason causes this problem is that Consumer only have accept, it seems like it have no catch error when some problem throws out. But Observer have onError() as we all know.
So just like write a try catch for the Consumer as it forget to catch for itself when doing the internet transaction.
Here is my solution, hoping will do help to you.

.subscribe(new Consumer<String>() { @Override public void accept(String s) throws Exception { } }, new Consumer<Throwable>() { @Override public void accept(Throwable throwable) throws Exception { } });

[mbobiosio]

I got this problem as well, and i think the fatal reason causes this problem is that Consumer only have accept, it seems like it have no catch error when some problem throws out. But Observer have onError() as we all know.
So just like write a try catch for the Consumer as it forget to catch for itself when doing the internet transaction.
Here is my solution, hoping will do help to you.

.subscribe(new Consumer<String>() { @Override public void accept(String s) throws Exception { } }, new Consumer<Throwable>() { @Override public void accept(Throwable throwable) throws Exception { } });

This is exactly what I'd been trying to achieve. Worked flawlessly. Thanks