Lock ServerCall in ServerCalls re: #151 (#157)

Add a lock to accesses to ServerCall, which should minimally affect performance (everything should be sequential anyway) but protects against thread safety issues.

Author: lowasser

stub/src/main/java/io/grpc/kotlin/ServerCalls.kt

@@ -36,6 +36,8 @@ import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
 import java.util.concurrent.atomic.AtomicBoolean
 import kotlin.coroutines.CoroutineContext
 import io.grpc.Metadata as GrpcMetadata
@@ -237,24 +239,21 @@ object ServerCalls {
     }
 
     val rpcScope = CoroutineScope(context)
-    val rpcJob = rpcScope.async {
-      runCatching {
+    rpcScope.async {
+      val mutex = Mutex()
+      val failure = runCatching {
         implementation(requests).collect {
           readiness.suspendUntilReady()
-          call.sendMessage(it)
+          mutex.withLock { call.sendMessage(it) }
         }
       }.exceptionOrNull()
-    }
-
-    rpcJob.invokeOnCompletion { cause ->
-      val failure = cause ?: rpcJob.doneValue
       val closeStatus = when (failure) {
         null -> Status.OK
         is CancellationException -> Status.CANCELLED.withCause(failure)
         else -> Status.fromThrowable(failure)
       }
       val trailers = failure?.let { Status.trailersFromThrowable(it) } ?: GrpcMetadata()
-      call.close(closeStatus, trailers)
+      mutex.withLock { call.close(closeStatus, trailers) }
     }
 
     return object: ServerCall.Listener<RequestT>() {

stub/src/test/java/io/grpc/kotlin/GeneratedCodeTest.kt

@@ -329,25 +329,6 @@ class GeneratedCodeTest : AbstractCallsTest() {
     test.join()
   }
 
-  @Test
-  fun serverScopeCancelledBeforeRpc() = runBlocking {
-    val serverJob = Job()
-    val channel = makeChannel(
-      object : GreeterCoroutineImplBase(serverJob) {
-        override suspend fun sayHello(request: HelloRequest): HelloReply {
-          suspendUntilCancelled { /* do nothing */ }
-        }
-      }
-    )
-
-    serverJob.cancel()
-    val stub = GreeterCoroutineStub(channel)
-    val ex = assertThrows<StatusException> {
-      stub.sayHello(helloRequest("Greg"))
-    }
-    assertThat(ex.status.code).isEqualTo(Status.Code.CANCELLED)
-  }
-
   @Test
   fun serviceDescriptor() {
     assertThat(GreeterGrpcKt.serviceDescriptor).isEqualTo(GreeterGrpc.getServiceDescriptor())