io.reactivex.exceptions.OnErrorNotImplementedException

[ankur-dev]

io.reactivex.exceptions.OnErrorNotImplementedException: The exception was not handled due to missing onError handler in the subscribe() method call.

I handle error in on subscribe method, but still getting this crash on subscribe of lifecycle event.

Whenever I am sending the wrong token to the server it gets crash

here is code

mStompClient = Stomp.over(
            Stomp.ConnectionProvider.OKHTTP,
            "ws" + UrlConstant.WEB_SOCKET_URL + "/websocket/tracker/websocket?access_token=" + token)
        mStompClient?.connect()

        mStompClient?.lifecycle()?.subscribe ({
            if (it.type != null) {
                when (it.type) {
                    LifecycleEvent.Type.OPENED -> {
                        Log.d("TAG", "Stomp connection opened")
                        subscribeTopic();
                        val randomNumber = mGenerator.nextInt(100)
                        randomNumberLiveData.postValue(randomNumber)
                    }
                    LifecycleEvent.Type.ERROR -> {
                        Log.d("TAG", "Error ${it.exception.message}")
                        connectToWebSocket()
                    }

                    LifecycleEvent.Type.CLOSED -> Log.d(
                        "TAG",
                        "Stomp connection closed"
                    )

                    LifecycleEvent.Type.FAILED_SERVER_HEARTBEAT -> Log.d(
                        "TAG",
                        "FAILED_SERVER_HEARTBEAT"
                    )
                }
            }
        },{
            Log.d("TAG", "Error on subscribe topic ${it.localizedMessage}")

        })

io.reactivex.exceptions.OnErrorNotImplementedException: The exception was not handled due to missing onError handler in the subscribe() method call. Further reading: https://github.com/ReactiveX/RxJava/wiki/Error-Handling | java.util.NoSuchElementException 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:77) at io.reactivex.internal.observers.BasicFuseableObserver.onError(BasicFuseableObserver.java:100) at io.reactivex.internal.operators.observable.ObservableDoOnEach$DoOnEachObserver.onError(ObservableDoOnEach.java:117) at io.reactivex.internal.observers.BasicFuseableObserver.onError(BasicFuseableObserver.java:100) at io.reactivex.internal.observers.BasicFuseableObserver.onError(BasicFuseableObserver.java:100) at io.reactivex.internal.observers.BasicFuseableObserver.fail(BasicFuseableObserver.java:110) at io.reactivex.internal.operators.observable.ObservableMap$MapObserver.onNext(ObservableMap.java:59) at io.reactivex.internal.operators.observable.ObservableConcatMap$ConcatMapDelayErrorObserver$DelayErrorInnerObserver.onNext(ObservableConcatMap.java:506) at io.reactivex.subjects.PublishSubject$PublishDisposable.onNext(PublishSubject.java:308) at io.reactivex.subjects.PublishSubject.onNext(PublishSubject.java:228) at ua.naiksoftware.stomp.provider.AbstractConnectionProvider.emitMessage(AbstractConnectionProvider.java:117) at ua.naiksoftware.stomp.provider.OkHttpConnectionProvider$1.onMessage(OkHttpConnectionProvider.java:67) at okhttp3.internal.ws.RealWebSocket.onReadMessage(RealWebSocket.java:322) at okhttp3.internal.ws.WebSocketReader.readMessageFrame(WebSocketReader.java:219) at okhttp3.internal.ws.WebSocketReader.processNextFrame(WebSocketReader.java:105) at okhttp3.internal.ws.RealWebSocket.loopReader(RealWebSocket.java:273) at okhttp3.internal.ws.RealWebSocket$1.onResponse(RealWebSocket.java:209) at okhttp3.RealCall$AsyncCall.execute(RealCall.java:174) at okhttp3.internal.NamedRunnable.run(NamedRunnable.java:32) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1167) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:641) at java.lang.Thread.run(Thread.java:919) Caused by: java.util.NoSuchElementException at java.util.Scanner.skip(Scanner.java:1755) at java.util.Scanner.skip(Scanner.java:1772) at ua.naiksoftware.stomp.dto.StompMessage.from(StompMessage.java:89) at ua.naiksoftware.stomp.-$$Lambda$Efbvy_1p5sTPPPDEnvuVVjpFgeg.apply(Unknown Source:2) at io.reactivex.internal.operators.observable.ObservableMap$MapObserver.onNext(ObservableMap.java:57) at io.reactivex.internal.operators.observable.ObservableConcatMap$ConcatMapDelayErrorObserver$DelayErrorInnerObserver.onNext(ObservableConcatMap.java:506)  at io.reactivex.subjects.PublishSubject$PublishDisposable.onNext(PublishSubject.java:308)  at io.reactivex.subjects.PublishSubject.onNext(PublishSubject.java:228)  at ua.naiksoftware.stomp.provider.AbstractConnectionProvider.emitMessage(AbstractConnectionProvider.java:117)  at ua.naiksoftware.stomp.provider.OkHttpConnectionProvider$1.onMessage(OkHttpConnectionProvider.java:67)  at okhttp3.internal.ws.RealWebSocket.onReadMessage(RealWebSocket.java:322)  at okhttp3.internal.ws.WebSocketReader.readMessageFrame(WebSocketReader.java:219)  at okhttp3.internal.ws.WebSocketReader.processNextFrame(WebSocketReader.java:105)  at okhttp3.internal.ws.RealWebSocket.loopReader(RealWebSocket.java:273)  at okhttp3.internal.ws.RealWebSocket$1.onResponse(RealWebSocket.java:209)  at okhttp3.RealCall$AsyncCall.execute(RealCall.java:174)  at okhttp3.internal.NamedRunnable.run(NamedRunnable.java:32)  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1167)  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:641)  at java.lang.Thread.run(Thread.java:919) 

[boginw]

I'm experiencing the same thing

[boginw]

In my case, ua.naiksoftware.stomp.dto.StompMessage.java:89 is failing on the following message:

ERROR
message:Failed to send message to ExecutorSubscribableChannel[clientInboundChannel]; nested exception is org.springframework.security.access.AccessDeniedException\c Access is denied
content-length:0

Breaking at line 89 shows that the header list is empty, which is bad because the reader expects "\n\n" next. The error seems to be in PATTERN_HEADER. Instead of "([^:\s]+)\s*:\s*([^:\s]+)" in probably should be "([^:\\s]+)\\s*:\\s*([^\\n]+)". That is, instead of capturing everything until a : in the second group, it should capture until an EOL. That should be in accordance with the STOMP protocol specification 1.2.

[boginw]

There are other errors on the current master branch, but I've created a  PR for this specific issue.There's no easy fix here, but if you really want it to work, you can follow the following steps:

Let's assume your project is stored at /home/project/

1. Clone this repository to /home/StompProtocolAndroid
2. In /home/StompProtocolAndroid, run ./gradlew :lib:build
3. Now, in your /home/project/app/build.gradle, add the following just below jcenter() and google()

       flatDir {
           dirs "$rootDir/../StompProtocolAndroid/lib/build/outputs/aar"
       }

This should allow you to continue working for now.