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.