vertx-lang-kotlin-coroutines

suspend fun handleTemperatureStream() {
  val stream = vertx.eventBus().consumer<Double>("temperature")
  val channel = stream.toReceiveChannel(vertx)

  var min = Double.MAX_VALUE
  var max = Double.MIN_VALUE

  // Iterate until the stream is closed
  // Non-blocking
  for (msg in channel) {
    val temperature = msg.body()
    min = min(min, temperature)
    max = max(max, temperature)
  }

  // The stream is now closed
}

vertx.createNetServer().connectHandler { socket ->

  // The record parser provides a stream of buffers delimited by \r\n
  val stream = RecordParser.newDelimited("\r\n", socket)

  // Convert the stream to a Kotlin channel
  val channel = stream.toReceiveChannel(vertx)

  // Run the coroutine
  launch {

    // Receive the request-line
    // Non-blocking
    val line = channel.receive().toString().split(" ")
    val method = line[0]
    val uri = line[1]

    println("Received HTTP request ($method, $uri)")

    // Still need to parse headers and body...
  }
}

// Receive HTTP headers
val headers = HashMap<String, String>()
while (true) {

  // Non-blocking
  val header = channel.receive().toString()

  // Done with parsing headers
  if (header.isEmpty()) {
    break
  }

  val pos = header.indexOf(':')
  headers[header.substring(0, pos).lowercase(Locale.getDefault())] = header.substring(pos + 1).trim()
}

println("Received HTTP request ($method, $uri) with headers ${headers.keys}")

// Receive the request body
val transferEncoding = headers["transfer-encoding"]
val contentLength = headers["content-length"]

val body: Buffer?
if (transferEncoding == "chunked") {

  // Handle chunked encoding, e.g
  // 5\r\n
  // HELLO\r\n
  // 0\r\n
  // \r\n

  body = Buffer.buffer()
  while (true) {

    // Parse length chunk
    // Non-blocking
    val len = channel.receive().toString().toInt(16)
    if (len == 0) {
      break
    }

    // The stream is flipped to parse a chunk of the exact size
    stream.fixedSizeMode(len + 2)

    // Receive the chunk and append it
    // Non-blocking
    val chunk = channel.receive()
    body.appendBuffer(chunk, 0, chunk.length() - 2)

    // The stream is flipped back to the \r\n delimiter to parse the next chunk
    stream.delimitedMode("\r\n")
  }
} else if (contentLength != null) {

  // The stream is flipped to parse a body of the exact size
  stream.fixedSizeMode(contentLength.toInt())

  // Non-blocking
  body = channel.receive()
} else {
  body = null
}

val bodySize = body?.length() ?: 0
println("Received HTTP request ($method, $uri) with headers ${headers.keys} and body with size $bodySize")

suspend fun sendChannel(httpResponse : HttpServerResponse) {
  val channel = httpResponse.toSendChannel(vertx)

  while (true) {
    val buffer = readBuffer()

    // Broadcast the temperature
    // Non-blocking but could be suspended
    channel.send(buffer)

    // Wait for one second
    awaitEvent<Long> { vertx.setTimer(1000, it) }
  }
}

suspend fun sendChannel(httpResponse : HttpServerResponse) {
  val channel = httpResponse.toSendChannel(vertx)

  while (true) {
    val buffer = readBuffer()

    // Broadcast the temperature
    // Non-blocking but could be suspended
    channel.send(buffer)

    // Wait for one second
    awaitEvent<Long> { vertx.setTimer(1000, it) }
  }
}

val bus = vertx.eventBus()
coroutineEventBus {
  bus.coConsumer<String>("some-address") {
    computeSomethingWithSuspendingFunction()
    it.reply("done")
  }
}

val router = Router.router(vertx)
coroutineRouter {
  // Route.coRespond is similar to Route.respond but using a suspending function
  router.get("/my-resource").coRespond {
    // similar to Route.respond but using a suspending function
    val response = computeSomethingWithSuspendingFunction()
    response // sent by Vert.x to the client
  }
  // Router.coErrorHandler is similar to Router.errorHandler but using a suspending function
  router.coErrorHandler(404) { rc ->
    val html = computeHtmlPageWithSuspendingFunction()
    rc.response().setStatusCode(404).putHeader(CONTENT_TYPE, TEXT_HTML).end(html)
  }
}