refactor atomic server tests

marcoferrer · May 20, 2019 · ab3ee29 · ab3ee29
1 parent e376fe0
commit ab3ee29
Show file tree

Hide file tree

Showing 6 changed files with 82 additions and 56 deletions.
diff --git a/.travis.yml b/.travis.yml
@@ -1,6 +1,7 @@
 language: java
 jdk:
   - oraclejdk8
+dist: trusty
 sudo: false
 addons:
   apt:
@@ -16,7 +17,11 @@ before_cache:
   - rm -rf $HOME/.gradle/caches/*/fileHashes/
 jobs:
   include:
-  - stage: "Tests"
-    script: ./gradlew build && cd example-project && ./gradlew test
+    - name: Gradle Check
+      install:
+        - ./gradlew assemble
+      script:
+        - ./gradlew check
+        - cd example-project && ./gradlew test
 after_success:
   - bash <(curl -s https://codecov.io/bash)
diff --git a/...kotlin/com/github/marcoferrer/krotoplus/coroutines/server/ServerCallBidiStreamingTests.kt b/...kotlin/com/github/marcoferrer/krotoplus/coroutines/server/ServerCallBidiStreamingTests.kt
@@ -35,6 +35,7 @@ import io.mockk.coVerify
 import io.mockk.spyk
 import io.mockk.verify
 import io.mockk.verifyOrder
+import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.cancel
@@ -46,9 +47,9 @@ import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.yield
 import org.junit.Rule
 import org.junit.Test
-import java.util.concurrent.atomic.AtomicReference
 import kotlin.coroutines.CoroutineContext
 import kotlin.coroutines.coroutineContext
 import kotlin.test.assertEquals
@@ -384,23 +385,29 @@ class ServerCallBidiStreamingTests {
 
     @Test
     fun `Server method is at least invoked before being cancelled`(){
-        val respChannel = AtomicReference<SendChannel<HelloReply>?>()
-        val serverCtx = AtomicReference<CoroutineContext?>()
+        val deferredRespChannel = CompletableDeferred<SendChannel<HelloReply>>()
+        val deferredCtx = CompletableDeferred<CoroutineContext>()
+
         grpcServerRule.serviceRegistry.addService(object : GreeterCoroutineGrpc.GreeterImplBase() {
-            override val initialContext: CoroutineContext = Dispatchers.Unconfined
+            override val initialContext: CoroutineContext = Dispatchers.Default
             override suspend fun sayHelloStreaming(
                 requestChannel: ReceiveChannel<HelloRequest>,
                 responseChannel: SendChannel<HelloReply>
             ) {
-                respChannel.set(spyk(responseChannel))
-                serverCtx.set(coroutineContext)
+                val respChan = spyk(responseChannel)
+                deferredCtx.complete(coroutineContext.apply {
+                    get(Job)!!.invokeOnCompletion {
+                        deferredRespChannel.complete(respChan)
+                    }
+                })
                 // Need to receive message since
                 // cancellation occurs in client
                 // half close.
                 requestChannel.receive()
-                delay(10)
+                delay(100)
+                yield()
                 repeat(3){
-                    respChannel.get()!!.send { message = "response" }
+                    respChan.send { message = "response" }
                 }
             }
         })
@@ -413,12 +420,14 @@ class ServerCallBidiStreamingTests {
         reqObserver.onCompleted()
 
         runBlocking {
-            do { delay(100) } while(serverCtx.get() == null)
-            assert(serverCtx.get()?.get(Job)!!.isCompleted){ "Server job should be completed" }
-            assert(serverCtx.get()?.get(Job)!!.isCancelled){ "Server job should be cancelled" }
-            assert(respChannel.get()!!.isClosedForSend){ "Abandoned response channel should be closed" }
-            verify(exactly = 1) { responseObserver.onError(matchStatus(Status.CANCELLED, "CANCELLED")) }
-            coVerify(exactly = 0) { respChannel.get()!!.send(any()) }
+            val respChannel = deferredRespChannel.await()
+            assert(respChannel.isClosedForSend){ "Abandoned response channel should be closed" }
+            verify(exactly = 1) { responseObserver.onError(matchStatus(Status.CANCELLED, "CANCELLED: test")) }
+            coVerify(exactly = 0) { respChannel.send(any()) }
+
+            val serverCtx = deferredCtx.await()
+            assert(serverCtx[Job]!!.isCompleted){ "Server job should be completed" }
+            assert(serverCtx[Job]!!.isCancelled){ "Server job should be cancelled" }
         }
     }
 }
diff --git a/...tlin/com/github/marcoferrer/krotoplus/coroutines/server/ServerCallClientStreamingTests.kt b/...tlin/com/github/marcoferrer/krotoplus/coroutines/server/ServerCallClientStreamingTests.kt
@@ -32,6 +32,7 @@ import io.grpc.stub.StreamObserver
 import io.grpc.testing.GrpcServerRule
 import io.mockk.spyk
 import io.mockk.verify
+import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.cancel
@@ -42,10 +43,10 @@ import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.yield
 import org.junit.Rule
 import org.junit.Test
 import java.util.concurrent.atomic.AtomicBoolean
-import java.util.concurrent.atomic.AtomicReference
 import kotlin.coroutines.CoroutineContext
 import kotlin.coroutines.coroutineContext
 import kotlin.test.assertEquals
@@ -254,20 +255,22 @@ class ServerCallClientStreamingTests {
 
     @Test
     fun `Server method is at least invoked before being cancelled`(){
-        val serverMethodExecuted = AtomicBoolean()
         val serverMethodCompleted = AtomicBoolean()
-        val serverCtx = AtomicReference<CoroutineContext?>()
-
+        val deferredCtx = CompletableDeferred<CoroutineContext>()
         grpcServerRule.serviceRegistry.addService(object : GreeterCoroutineGrpc.GreeterImplBase() {
-            override val initialContext: CoroutineContext = Dispatchers.Unconfined
+            override val initialContext: CoroutineContext = Dispatchers.Default
             override suspend fun sayHelloClientStreaming(requestChannel: ReceiveChannel<HelloRequest>): HelloReply {
-                serverMethodExecuted.set(true)
-                serverCtx.set(coroutineContext)
+                coroutineContext.let { ctx ->
+                    ctx[Job]!!.invokeOnCompletion {
+                        deferredCtx.complete(ctx)
+                    }
+                }
                 // Need to receive message since
                 // cancellation occurs in client
                 // half close.
                 requestChannel.receive()
-                delay(10)
+                delay(100)
+                yield()
                 serverMethodCompleted.set(true)
                 return expectedResponse
             }
@@ -282,12 +285,11 @@ class ServerCallClientStreamingTests {
         reqObserver.onCompleted()
 
         runBlocking {
-            do { delay(100) } while(serverCtx.get() == null)
-
-            verify(exactly = 1) { responseObserver.onError(matchStatus(Status.CANCELLED, "CANCELLED")) }
+            val serverCtx = deferredCtx.await()
+            verify(exactly = 1) { responseObserver.onError(matchStatus(Status.CANCELLED, "CANCELLED: test")) }
 
-            assert(serverCtx.get()?.get(Job)!!.isCompleted){ "Server job should be completed" }
-            assert(serverCtx.get()?.get(Job)!!.isCancelled){ "Server job should be cancelled" }
+            assert(serverCtx[Job]!!.isCompleted){ "Server job should be completed" }
+            assert(serverCtx[Job]!!.isCancelled){ "Server job should be cancelled" }
             assertFalse(serverMethodCompleted.get(),"Server method should not complete")
         }
     }

diff --git a/...tlin/com/github/marcoferrer/krotoplus/coroutines/server/ServerCallServerStreamingTests.kt b/...tlin/com/github/marcoferrer/krotoplus/coroutines/server/ServerCallServerStreamingTests.kt
@@ -36,6 +36,7 @@ import io.mockk.coVerify
 import io.mockk.spyk
 import io.mockk.verify
 import io.mockk.verifyOrder
+import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
@@ -48,7 +49,6 @@ import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.yield
 import org.junit.Rule
 import org.junit.Test
-import java.util.concurrent.atomic.AtomicReference
 import kotlin.coroutines.CoroutineContext
 import kotlin.coroutines.coroutineContext
 import kotlin.test.assertEquals
@@ -221,38 +221,44 @@ class ServerCallServerStreamingTests {
 
     @Test
     fun `Server method is at least invoked before being cancelled`(){
-        val respChannel = AtomicReference<SendChannel<HelloReply>?>()
-        val serverCtx = AtomicReference<CoroutineContext?>()
-
+        val deferredRespChannel = CompletableDeferred<SendChannel<HelloReply>>()
+        val deferredCtx = CompletableDeferred<CoroutineContext>()
         grpcServerRule.serviceRegistry.addService(object : GreeterCoroutineGrpc.GreeterImplBase() {
-            override val initialContext: CoroutineContext = Dispatchers.Unconfined
+            override val initialContext: CoroutineContext = Dispatchers.Default
             override suspend fun sayHelloServerStreaming(
                 request: HelloRequest,
                 responseChannel: SendChannel<HelloReply>
             ) {
-                respChannel.set(spyk(responseChannel))
-                serverCtx.set(coroutineContext)
-                delay(5)
+                val respChan = spyk(responseChannel)
+                deferredCtx.complete(coroutineContext.apply {
+                    get(Job)!!.invokeOnCompletion {
+                        deferredRespChannel.complete(respChan)
+                    }
+                })
+                delay(100)
+                yield()
                 repeat(3){
-                    respChannel.get()!!.send { message = "response" }
+                    respChan.send { message = "response" }
                 }
             }
         })
 
         val stub = GreeterGrpc.newBlockingStub(grpcServerRule.channel)
             .withInterceptors(CancellingClientInterceptor)
 
-        assertFailsWithStatus(Status.CANCELLED,"CANCELLED"){
+        assertFailsWithStatus(Status.CANCELLED,"CANCELLED: test"){
             val iter = stub.sayHelloServerStreaming(HelloRequest.getDefaultInstance())
             while(iter.hasNext()){}
         }
 
         runBlocking {
-            do { delay(100) } while(serverCtx.get() == null)
-            assert(serverCtx.get()?.get(Job)!!.isCompleted){ "Server job should be completed" }
-            assert(serverCtx.get()?.get(Job)!!.isCancelled){ "Server job should be cancelled" }
-            assert(respChannel.get()!!.isClosedForSend){ "Abandoned response channel should be closed" }
-            coVerify(exactly = 0) { respChannel.get()!!.send(any()) }
+            val respChannel = deferredRespChannel.await()
+            assert(respChannel.isClosedForSend){ "Abandoned response channel should be closed" }
+            coVerify(exactly = 0) { respChannel.send(any()) }
+
+            val serverCtx = deferredCtx.await()
+            assert(serverCtx[Job]!!.isCompleted){ "Server job should be completed" }
+            assert(serverCtx[Job]!!.isCancelled){ "Server job should be cancelled" }
         }
     }
 }
diff --git a/...rc/test/kotlin/com/github/marcoferrer/krotoplus/coroutines/server/ServerCallUnaryTests.kt b/...rc/test/kotlin/com/github/marcoferrer/krotoplus/coroutines/server/ServerCallUnaryTests.kt
@@ -33,17 +33,18 @@ import io.grpc.stub.StreamObserver
 import io.grpc.testing.GrpcServerRule
 import io.mockk.spyk
 import io.mockk.verify
+import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.yield
 import org.junit.Rule
 import org.junit.Test
 import java.util.concurrent.atomic.AtomicBoolean
-import java.util.concurrent.atomic.AtomicReference
 import kotlin.coroutines.CoroutineContext
 import kotlin.coroutines.coroutineContext
 import kotlin.test.assertEquals
@@ -159,16 +160,18 @@ class ServerCallUnaryTests {
 
     @Test
     fun `Server method is at least invoked before being cancelled`(){
-        val serverMethodExecuted = AtomicBoolean()
         val serverMethodCompleted = AtomicBoolean()
-        val serverCtx = AtomicReference<CoroutineContext?>()
-
+        val deferredCtx = CompletableDeferred<CoroutineContext>()
         grpcServerRule.serviceRegistry.addService(object : GreeterCoroutineGrpc.GreeterImplBase() {
-            override val initialContext: CoroutineContext = Dispatchers.Unconfined
+            override val initialContext: CoroutineContext = Dispatchers.Default
             override suspend fun sayHello(request: HelloRequest): HelloReply {
-                serverMethodExecuted.set(true)
-                serverCtx.set(coroutineContext)
-                delay(10)
+                coroutineContext.let { ctx ->
+                    ctx[Job]!!.invokeOnCompletion {
+                        deferredCtx.complete(ctx)
+                    }
+                }
+                delay(100)
+                yield()
                 serverMethodCompleted.set(true)
                 return expectedResponse
             }
@@ -178,14 +181,14 @@ class ServerCallUnaryTests {
             .newBlockingStub(grpcServerRule.channel)
             .withInterceptors(CancellingClientInterceptor)
 
-        assertFailsWithStatus(Status.CANCELLED,"CANCELLED"){
+        assertFailsWithStatus(Status.CANCELLED,"CANCELLED: test"){
             stub.sayHello(HelloRequest.getDefaultInstance())
         }
 
         runBlocking {
-            do { delay(100) } while(serverCtx.get() == null)
-            assert(serverCtx.get()?.get(Job)!!.isCompleted){ "Server job should be completed" }
-            assert(serverCtx.get()?.get(Job)!!.isCancelled){ "Server job should be cancelled" }
+            val serverCtx = deferredCtx.await()
+            assert(serverCtx[Job]!!.isCompleted){ "Server job should be completed" }
+            assert(serverCtx[Job]!!.isCancelled){ "Server job should be cancelled" }
             assertFalse(serverMethodCompleted.get(),"Server method should not complete")
         }
     }

diff --git a/protoc-gen-kroto-plus/generator-tests/build.gradle b/protoc-gen-kroto-plus/generator-tests/build.gradle
@@ -2,6 +2,7 @@ apply plugin: 'com.google.protobuf'
 
 compileTestKotlin {
     kotlinOptions {
+        jvmTarget = "1.8"
         freeCompilerArgs += [
                 "-Xuse-experimental=kotlin.Experimental"
         ]