Kotlin onError not Implemented Problem

[WrathChaos]

Hello, I've been using this library since 1.5 year. Thank you so much for this awesome library. After I converted my code to Kotlin, a crush is appeared. (I am still not sure about problem is caused by Kotlin or not) Here is what i got after some seconds I connected and received the data from stomp socket. Gist: https://gist.github.com/WrathChaos/e36b315118b9e80fcd7ab45cc8689da8

Code :

java.lang.IllegalStateException: Exception thrown on Scheduler.Worker thread. AddonError handling. at rx.android.schedulers.LooperScheduler$ScheduledAction.run(LooperScheduler.java:112) at android.os.Handler.handleCallback(Handler.java:815) at android.os.Handler.dispatchMessage(Handler.java:104) at android.os.Looper.loop(Looper.java:194) at android.app.ActivityThread.main(ActivityThread.java:5624) at java.lang.reflect.Method.invoke(Native Method) at java.lang.reflect.Method.invoke(Method.java:372) at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:959) at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:754) Caused by: rx.exceptions.OnErrorNotImplementedException at rx.internal.util.InternalObservableUtils$ErrorNotImplementedAction.call(InternalObservableUtils.java:386) at rx.internal.util.InternalObservableUtils$ErrorNotImplementedAction.call(InternalObservableUtils.java:383) at rx.internal.util.ActionSubscriber.onError(ActionSubscriber.java:44) at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:153) at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:115) at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.checkTerminated(OperatorObserveOn.java:273) at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.call(OperatorObserveOn.java:216) at rx.android.schedulers.LooperScheduler$ScheduledAction.run(LooperScheduler.java:107) at android.os.Handler.handleCallback(Handler.java:815)  at android.os.Handler.dispatchMessage(Handler.java:104)  at android.os.Looper.loop(Looper.java:194)  at android.app.ActivityThread.main(ActivityThread.java:5624)  at java.lang.reflect.Method.invoke(Native Method)  at java.lang.reflect.Method.invoke(Method.java:372)  at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:959)  at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:754)  Caused by: rx.exceptions.MissingBackpressureException at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.onNext(OperatorObserveOn.java:160) at rx.internal.operators.OperatorSubscribeOn$SubscribeOnSubscriber.onNext(OperatorSubscribeOn.java:74) at rx.observers.Subscribers$5.onNext(Subscribers.java:235) at ua.naiksoftware.stomp.client.StompClient.callSubscribers(StompClient.java:143) at ua.naiksoftware.stomp.client.StompClient.lambda$connect$1(StompClient.java:109) at ua.naiksoftware.stomp.client.StompClient$$Lambda$3.call(Unknown Source) at rx.internal.util.ActionSubscriber.onNext(ActionSubscriber.java:39) at rx.observers.SafeSubscriber.onNext(SafeSubscriber.java:134) at rx.internal.operators.OnSubscribeMap$MapSubscriber.onNext(OnSubscribeMap.java:77) at rx.observers.Subscribers$5.onNext(Subscribers.java:235) at ua.naiksoftware.stomp.WebSocketsConnectionProvider.emitMessage(WebSocketsConnectionProvider.java:157) at ua.naiksoftware.stomp.WebSocketsConnectionProvider.access$300(WebSocketsConnectionProvider.java:30) at ua.naiksoftware.stomp.WebSocketsConnectionProvider$1.onMessage(WebSocketsConnectionProvider.java:102) at org.java_websocket.client.WebSocketClient.onWebsocketMessage(WebSocketClient.java:292) at org.java_websocket.WebSocketImpl.decodeFrames(WebSocketImpl.java:417) at org.java_websocket.WebSocketImpl.decode(WebSocketImpl.java:170) at org.java_websocket.client.WebSocketClient.run(WebSocketClient.java:229) at java.lang.Thread.run(Thread.java:818)

I got Caused by: rx.exceptions.OnErrorNotImplementedException.. I checked your library and you already implemented the onError but it is not working at all.. I could not get error on Lifecycle.

Here is my code of using StompProtocolAndroid :

` mStompClient = Stomp.over(WebSocket::class.java, "ws://" + wsURL + "gateway/websocket")

    mStompClient!!.lifecycle()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe { lifecycleEvent ->
                when (lifecycleEvent.type) {
                    LifecycleEvent.Type.OPENED -> Log.e("STOMP", "Stomp connection is opened!")
                    LifecycleEvent.Type.ERROR -> Log.e("STOMP", "Stomp connection error", lifecycleEvent.exception)
                    LifecycleEvent.Type.CLOSED -> Log.e("STOMP", "SSMActivity Stomp connection is closed")
                    else -> Log.e("Stomp", "STOMP Lifecycle booom")
                }
            }

    // Receive Data
    mStompClient!!.topic("/topic/" ....")
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe { topicMessage ->
                Log.d("Stomp", "Received " + topicMessage.payload)
                // My logic
            }

    mStompClient!!.connect()`

Please help me ASAP, thank you :)