Vertx plus Kotlin協程永遠掛起

Question

我使用Kotlin協程爲了學習目的重寫了一些Java Vertx異步代碼。但是，當我嘗試測試一個簡單的HTTP調用時，基於協程的測試永遠掛起，我真的不明白問題在哪裏。這裏是一個播放器：

@RunWith(VertxUnitRunner::class) 
class HelloWorldTest { 

    private val vertx: Vertx = Vertx.vertx() 

    @Before 
    fun setUp(context: TestContext) { 
     // HelloWorldVerticle is a simple http server that replies "Hello, World!" to whatever call 
     vertx.deployVerticle(HelloWorldVerticle::class.java!!.getName(), context.asyncAssertSuccess()) 
    } 

    // ORIGINAL ASYNC TEST HERE. IT WORKS AS EXPECTED 
    @Test 
    fun testAsync(context: TestContext) { 
     val atc = context.async() 
     vertx.createHttpClient().getNow(8080, "localhost", "/") { response -> 
      response.handler { body -> 
       context.assertTrue(body.toString().equals("Hello, World!")) 
       atc.complete() 
      } 
     } 
    } 

    // First attempt, it hangs forever, the response is never called 
    @Test 
    fun testSync1(context: TestContext) = runBlocking<Unit> { 
     val atc = context.async() 
     val body = await<HttpClientResponse> { 
      vertx.createHttpClient().getNow(8080, "localhost", "/", { response -> response.handler {it}}) 
     } 
     context.assertTrue(body.toString().equals("Hello, World!")) 
     atc.complete() 
    } 

    // Second attempt, it hangs forever, the response is never called 
    @Test 
    fun testSync2(context: TestContext) = runBlocking<Unit> { 
     val atc = context.async() 
     val response = await<HttpClientResponse> { 
       vertx.createHttpClient().getNow(8080, "localhost", "/", it) 
     } 
     response.handler { body -> 
      context.assertTrue(body.toString().equals("Hello, World!")) 
      atc.complete() 
     } 
    } 

    suspend fun <T> await(callback: (Handler<T>) -> Unit) = 
      suspendCoroutine<T> { cont -> 
       callback(Handler { result: T -> 
        cont.resume(result) 
       }) 
      } 
} 

大家都能弄清楚這個問題嗎？

Answer 1

在我看來，你的代碼有幾個問題：

1. 您可以運行測試的HTTP服務器得到了部署
2. 我相信，因爲你裏面runBlocking執行你的代碼卡住前事件循環完成請求。
3. 最後，我建議您使用HttpClienctResponse::bodyHandler方法而不是HttpClientResponse::handler，因爲處理程序可能會接收部分數據。

這裏是一個替代的解決方案工作正常：

import io.vertx.core.AbstractVerticle 
import io.vertx.core.Future 
import io.vertx.core.Handler 
import io.vertx.core.Vertx 
import io.vertx.core.buffer.Buffer 
import io.vertx.core.http.HttpClientResponse 
import kotlin.coroutines.experimental.Continuation 
import kotlin.coroutines.experimental.EmptyCoroutineContext 
import kotlin.coroutines.experimental.startCoroutine 
import kotlin.coroutines.experimental.suspendCoroutine 

inline suspend fun <T> await(crossinline callback: (Handler<T>) -> Unit) = 
     suspendCoroutine<T> { cont -> 
      callback(Handler { result: T -> 
       cont.resume(result) 
      }) 
     } 

fun <T : Any> async(code: suspend() -> T) = Future.future<T>().apply { 
    code.startCoroutine(object : Continuation<T> { 
     override val context = EmptyCoroutineContext 
     override fun resume(value: T) = complete() 
     override fun resumeWithException(exception: Throwable) = fail(exception) 
    }) 
} 

fun main(args: Array<String>) { 
    async { 
     val vertx: Vertx = Vertx.vertx() 

     //0. take the current context 
     val ctx = vertx.getOrCreateContext() 

     //1. deploy the http server 
     await<Unit> { cont -> 
      vertx.deployVerticle(object : AbstractVerticle() { 
       override fun start() { 
        vertx.createHttpServer() 
          .requestHandler { it.response().end("Hello World") } 
          .listen(7777) { ctx.runOnContext { cont.handle(Unit) } } 
        //note that it is important tp complete the handler in the correct context 
       } 
      }) 
     } 

     //2. send request 
     val response: HttpClientResponse = await { vertx.createHttpClient().getNow(7777, "localhost", "/", it) } 

     //3. await response 
     val body = await<Buffer> { response.bodyHandler(it) } 
     println("received $body") 
    } 
}