add stream error handlers to avoid global undeliverable error exceptions, which close app process if no appropriate global handler is set (#14)

Author: mostroverkhov

rsocket-core/src/main/java/io/rsocket/android/RSocketClient.kt

@@ -24,6 +24,8 @@ import io.reactivex.Completable
 import io.reactivex.Flowable
 import io.reactivex.Single
 import io.reactivex.disposables.Disposable
+import io.reactivex.functions.Action
+import io.reactivex.functions.Consumer
 import io.reactivex.processors.FlowableProcessor
 import io.reactivex.processors.PublishProcessor
 import io.reactivex.processors.UnicastProcessor
@@ -67,33 +69,29 @@ internal class RSocketClient @JvmOverloads constructor(
             val ackTimeoutMs = ackTimeout.toMillis
 
             this.keepAliveSendSub = completeOnStart
-                    .andThen(Flowable.interval(tickPeriod.toMillis,TimeUnit.MILLISECONDS))
+                    .andThen(Flowable.interval(tickPeriod.toMillis, TimeUnit.MILLISECONDS))
                     .doOnSubscribe({ _ -> timeLastTickSentMs = System.currentTimeMillis() })
                     .concatMap({ _ -> sendKeepAlive(ackTimeoutMs, missedAcks).toFlowable<Long>() })
-                    .doOnError({ t:Throwable ->
+                    .subscribe({},
+                            { t: Throwable ->
                                 errorConsumer(t)
-                                connection.close().subscribe()
+                                connection.close().subscribe({}, errorConsumer)
                             })
-                    .subscribe()
         }
 
         connection
                 .onClose()
                 .doFinally { cleanup() }
-                .doOnError(errorConsumer)
-                .subscribe()
+                .subscribe({}, errorConsumer)
 
         connection
                 .send(sendProcessor)
-                .doOnError { handleSendProcessorError(it) }
-                .subscribe()
+                .subscribe({}, { handleSendProcessorError(it) })
 
         connection
                 .receive()
                 .doOnSubscribe { started.onComplete() }
-                .doOnNext { handleIncomingFrames(it) }
-                .doOnError(errorConsumer)
-                .subscribe()
+                .subscribe({ handleIncomingFrames(it) },errorConsumer)
     }
 
     private fun handleSendProcessorError(t: Throwable) {
@@ -292,10 +290,12 @@ internal class RSocketClient @JvmOverloads constructor(
 
                                                         requestFrames
                                                                 .doOnNext { sendProcessor.onNext(it) }
-                                                                .doOnError { t ->
+                                                                .subscribe(
+                                                                {},
+                                                                { t ->
                                                                     errorConsumer(t)
                                                                     receiver.onError(CancellationException("Disposed"))
-                                                                }.subscribe()
+                                                                })
                                                     } else {
                                                         sendOneFrame(Frame.RequestN.from(streamId, l))
                                                     }
@@ -318,7 +318,7 @@ internal class RSocketClient @JvmOverloads constructor(
         }
     }
 
-    protected fun cleanup() {
+    private fun cleanup() {
 
         var subscribers: Collection<Subscriber<Payload>>
         var publishers: Collection<LimitableRequestPublisher<*>>

rsocket-core/src/main/java/io/rsocket/android/RSocketServer.kt

@@ -55,8 +55,9 @@ import org.reactivestreams.Subscription
 
         connection
                 .send(sendProcessor)
-                .doOnError { handleSendProcessorError(it) }
-                .subscribe()
+                .subscribe(
+                        {},
+                        { handleSendProcessorError(it) })
 
         this.receiveDisposable = connection
                 .receive()
@@ -68,18 +69,20 @@ import org.reactivestreams.Subscription
                                         Completable.complete()
                                     }).toFlowable<Void>()
                 }
-                .doOnError(errorConsumer)
                 .ignoreElements()
-                .subscribe()
+                .subscribe(
+                        {},
+                        errorConsumer)
 
         this.connection
                 .onClose()
-                .doOnError(errorConsumer)
                 .doFinally {
                     cleanup()
                     receiveDisposable.dispose()
                 }
-                .subscribe()
+                .subscribe(
+                        {},
+                        errorConsumer)
     }
 
     private fun handleSendProcessorError(t: Throwable) {
@@ -157,7 +160,7 @@ import org.reactivestreams.Subscription
         cleanUpSendingSubscriptions()
         cleanUpChannelProcessors()
 
-        requestHandler.close().subscribe()
+        requestHandler.close().subscribe({}, errorConsumer)
     }
 
     @Synchronized private fun cleanUpSendingSubscriptions() {
@@ -269,9 +272,8 @@ import org.reactivestreams.Subscription
                         })
                 .concatWith(Flowable.just(Frame.PayloadFrame.from(streamId, FrameType.COMPLETE)))
                 .onErrorResumeNext { t: Throwable -> Flowable.just(Frame.Error.from(streamId, t)) }
-                .doOnNext({ sendProcessor.onNext(it) })
                 .doFinally { removeSubscription(streamId) }
-                .subscribe()
+                .subscribe({ sendProcessor.onNext(it) })
 
         return Completable.complete()
     }
@@ -363,25 +365,8 @@ import org.reactivestreams.Subscription
             }
         }
 
-        private class Subs(val s: Subscription) : DisposableSubscription() {
-            @Volatile private var isDisposed = false
-
-            override fun isDisposed(): Boolean = isDisposed
-
-            override fun dispose() {
-                isDisposed = true
-                cancel()
-            }
-
-            override fun cancel() = s.cancel()
-
-            override fun request(n: Long) = s.request(n)
-        }
-
         companion object {
             fun disposable(d: Disposable): DisposableSubscription = Disp(d)
-
-            fun subscription(s: Subscription): DisposableSubscription = Subs(s)
         }
     }
 

rsocket-core/src/main/java/io/rsocket/android/internal/ClientServerInputMultiplexer.kt

@@ -75,30 +75,22 @@ class ClientServerInputMultiplexer(val source: DuplexConnection, plugins: Plugin
                     }
                     type
                 }
-                .subscribe { group ->
+                .subscribe({ group ->
                     when (group.key) {
                         Type.STREAM_ZERO -> streamZero.onNext(group)
                         Type.SERVER -> server.onNext(group)
                         Type.CLIENT -> client.onNext(group)
                     }
-                }
+                }, { /*noop - errors are handled by demuxed frame streams*/ })
     }
 
-    fun asServerConnection(): DuplexConnection {
-        return serverConnection
-    }
+    fun asServerConnection(): DuplexConnection = serverConnection
 
-    fun asClientConnection(): DuplexConnection {
-        return clientConnection
-    }
+    fun asClientConnection(): DuplexConnection = clientConnection
 
-    fun asStreamZeroConnection(): DuplexConnection {
-        return streamZeroConnection
-    }
+    fun asStreamZeroConnection(): DuplexConnection = streamZeroConnection
 
-    fun close(): Completable {
-        return source.close()
-    }
+    fun close(): Completable = source.close()
 
     private class InternalDuplexConnection(private val source: DuplexConnection,
                                            p: Flowable<Flowable<Frame>>) : DuplexConnection {
@@ -122,25 +114,17 @@ class ClientServerInputMultiplexer(val source: DuplexConnection, plugins: Plugin
 
         override fun receive(): Flowable<Frame> {
             return processor.flatMapPublisher { f ->
-                if (debugEnabled) {
-                    return@flatMapPublisher f.doOnNext { frame -> LOGGER.debug("receiving -> " + frame.toString()) }
-                } else {
-                    return@flatMapPublisher f
-                }
+                if (debugEnabled)
+                    f.doOnNext { frame -> LOGGER.debug("receiving -> " + frame.toString()) }
+                 else f
             }
         }
 
-        override fun close(): Completable {
-            return source.close()
-        }
+        override fun close(): Completable = source.close()
 
-        override fun onClose(): Completable {
-            return source.onClose()
-        }
+        override fun onClose(): Completable = source.onClose()
 
-        override fun availability(): Double {
-            return source.availability()
-        }
+        override fun availability(): Double = source.availability()
     }
 
     companion object {