propagate scope cancellation to target channel in outbound flow control

marcoferrer · Aug 11, 2019 · d99af22 · d99af22
1 parent 4da8aaf
commit d99af22
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 1 deletion.
diff --git a/...oroutines/src/main/kotlin/com/github/marcoferrer/krotoplus/coroutines/call/FlowControl.kt b/...oroutines/src/main/kotlin/com/github/marcoferrer/krotoplus/coroutines/call/FlowControl.kt
@@ -17,9 +17,11 @@
 package com.github.marcoferrer.krotoplus.coroutines.call
 
 import io.grpc.stub.CallStreamObserver
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineExceptionHandler
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.channels.ActorScope
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.actor
@@ -72,6 +74,14 @@ internal fun <T> CoroutineScope.applyOutboundFlowControl(
             targetChannel.close(e)
         }
     ) {
+        coroutineContext[Job]?.let { job ->
+            job.invokeOnCompletion {
+                if(job.isCancelled && !targetChannel.isClosedForSend){
+                    targetChannel.cancel(it as? CancellationException)
+                }
+            }
+        }
+
         for(handler in channel){
             if(isCompleted.get()) break
             handler(this)

diff --git a/...c/main/kotlin/com/github/marcoferrer/krotoplus/coroutines/client/ClientBidiCallChannel.kt b/...c/main/kotlin/com/github/marcoferrer/krotoplus/coroutines/client/ClientBidiCallChannel.kt
@@ -20,6 +20,7 @@ import com.github.marcoferrer.krotoplus.coroutines.call.FlowControlledInboundStr
 import com.github.marcoferrer.krotoplus.coroutines.call.applyOutboundFlowControl
 import io.grpc.stub.ClientCallStreamObserver
 import io.grpc.stub.ClientResponseObserver
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.ReceiveChannel
 import kotlinx.coroutines.channels.SendChannel
@@ -106,7 +107,7 @@ internal class ClientBidiCallChannelImpl<ReqT,RespT>(
 
     override fun onError(t: Throwable) {
         outboundChannel.close(t)
-        outboundChannel.cancel()
+        outboundChannel.cancel(CancellationException(t.message,t))
         inboundChannel.close(t)
     }
 }