close connection on RequestHandler termination (#74)

Author: mostroverkhov

build.gradle

@@ -41,9 +41,10 @@ subprojects {
         api 'io.netty:netty-buffer'
         api 'io.reactivex.rxjava2:rxjava'
         api 'org.jetbrains.kotlin:kotlin-stdlib-jdk7'
+        api 'org.slf4j:slf4j-api'
+
         compileOnly 'com.google.code.findbugs:jsr305'
 
-        testImplementation 'org.slf4j:slf4j-api'
         testImplementation 'junit:junit'
         testImplementation 'org.mockito:mockito-core'
         testImplementation 'org.hamcrest:hamcrest-library'

rsocket-core/gradle/dependency-locks/compileClasspath.lockfile

@@ -10,3 +10,4 @@ org.jetbrains.kotlin:kotlin-stdlib-jdk7:1.3.50
 org.jetbrains.kotlin:kotlin-stdlib:1.3.50
 org.jetbrains:annotations:13.0
 org.reactivestreams:reactive-streams:1.0.2
+org.slf4j:slf4j-api:1.7.28

rsocket-core/src/main/kotlin/io/rsocket/kotlin/internal/RSocketResponder.kt

@@ -61,6 +61,10 @@ internal class RSocketResponder(
         connection
                 .onClose()
                 .subscribe({ completion.complete() }, errorConsumer)
+
+        requestHandler
+                .onClose()
+                .subscribe({ completion.complete() }, errorConsumer)
     }
 
     override fun fireAndForget(payload: Payload): Completable {
@@ -291,7 +295,13 @@ internal class RSocketResponder(
 
         private fun completeOnce(err: Throwable) {
             if (completed.compareAndSet(false, true)) {
+
                 receiveDisposable.dispose()
+
+                connection
+                        .close()
+                        .subscribe({}, errorConsumer)
+
                 requestHandler
                         .close()
                         .subscribe({}, errorConsumer)

rsocket-core/src/test/kotlin/io/rsocket/kotlin/RSocketResponderTest.kt

@@ -39,6 +39,15 @@ class RSocketResponderTest {
     @get:Rule
     val rule = ServerSocketRule()
 
+    @Test(timeout = 2000)
+    fun testRequestHandlerCloseTerminatesRSocket() {
+        rule.acceptingSocket.close()
+                .andThen(
+                        rule.rsocket.onClose()
+                                .mergeWith(rule.conn.onClose())
+                ).blockingAwait()
+    }
+
     @Test(timeout = 2000)
     @Throws(Exception::class)
     fun testHandleResponseFrameNoError() {
@@ -110,7 +119,7 @@ class RSocketResponderTest {
 
         lateinit var sender: PublishProcessor<Frame>
         lateinit var receiver: PublishProcessor<Frame>
-        private lateinit var conn: LocalDuplexConnection
+        lateinit var conn: LocalDuplexConnection
         lateinit var errors: MutableList<Throwable>
         internal lateinit var rsocket: RSocketResponder
 
@@ -137,9 +146,10 @@ class RSocketResponderTest {
         }
 
         fun setAccSocket(acceptingSocket: RSocket) {
+            val cur = this.acceptingSocket
+            cur.close().subscribe()
+            cur.onClose().blockingAwait()
             this.acceptingSocket = acceptingSocket
-            acceptingSocket.close().subscribe()
-            acceptingSocket.onClose().blockingAwait()
             init()
         }
 

rsocket-transport-okhttp/gradle/dependency-locks/compileClasspath.lockfile

@@ -12,3 +12,4 @@ org.jetbrains.kotlin:kotlin-stdlib-jdk7:1.3.50
 org.jetbrains.kotlin:kotlin-stdlib:1.3.50
 org.jetbrains:annotations:13.0
 org.reactivestreams:reactive-streams:1.0.2
+org.slf4j:slf4j-api:1.7.28