Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

ServerCalls: cancel only the request's Job #303

Merged
merged 1 commit into from
Nov 14, 2021
Merged

ServerCalls: cancel only the request's Job #303

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 3 additions & 5 deletions stub/src/main/java/io/grpc/kotlin/ServerCalls.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,15 +27,14 @@ import io.grpc.ServerMethodDefinition
import io.grpc.Status
import io.grpc.StatusException
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineName
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.async
import kotlinx.coroutines.cancel
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import java.util.concurrent.atomic.AtomicBoolean
Expand Down Expand Up @@ -236,8 +235,7 @@ object ServerCalls {
}
}

val rpcScope = CoroutineScope(context)
rpcScope.async {
val rpcJob = CoroutineScope(context).launch {
val mutex = Mutex()
val headersSent = AtomicBoolean(false) // enforces only sending headers once
val failure = runCatching {
Expand Down Expand Up @@ -272,7 +270,7 @@ object ServerCalls {
var isReceiving = true

override fun onCancel() {
rpcScope.cancel("Cancellation received from client")
rpcJob.cancel("Cancellation received from client")
}

override fun onMessage(message: RequestT) {
Expand Down
21 changes: 21 additions & 0 deletions stub/src/test/java/io/grpc/kotlin/ServerCallsTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,27 @@ class ServerCallsTest : AbstractCallsTest() {
cancelled.join()
}

@Test
fun unaryMethodCancellationContextWithJobPropagatedToServer() = runBlocking {
val completable = CompletableDeferred<Int>()
val requestReceived = Job()
val channel = makeChannel(
// Note that we use runBlocking's context here
ServerCalls.unaryServerMethodDefinition(coroutineContext, sayHelloMethod) {
requestReceived.complete()
suspendUntilCancelled {
completable.complete(42)
}
}
)

val stub = GreeterGrpc.newFutureStub(channel)
val future = stub.sayHello(helloRequest("Garnet"))
requestReceived.join()
future.cancel(true)
assertThat(completable.await()).isEqualTo(42)
}

@Test
fun unaryRequestHandledWithoutWaitingForHalfClose() = runBlocking {
val processingStarted = Job()
Expand Down