vertx-lang-kotlin-coroutines

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

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

  // 迭代直到 stream 被关闭
  // 非阻塞的
  for (msg in channel) {
    val temperature = msg.body()
    min = Math.min(min, temperature)
    max = Math.max(max, temperature)
  }

  // stream 现在被关闭了
}

vertx.createNetServer().connectHandler { socket ->

  // 记录解析器提供了一个以\r\n分隔的缓冲流
  val stream = RecordParser.newDelimited("\r\n", socket)

  // 将 stream 转换为一个 Kotlin channel
  val channel = stream.toChannel(vertx)

  // 启动协程
  launch {

    // 接收请求行
    // 非阻塞
    val line = channel.receive().toString().split(" ")
    val method = line[0]
    val uri = line[1]

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

    // 仍然需要解析标题和正文……
  }
}

// 接收 HTTP 头
val headers = HashMap<String, String>()
while (true) {

  // 非阻塞
  val header = channel.receive().toString()

  // 完成头解析
  if (header.isEmpty()) {
    break
  }

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

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

// 接收请求体
val transferEncoding = headers["transfer-encoding"]
val contentLength = headers["content-length"]

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

  // 处理分块编码，例如
  // 5\r\n
  // HELLO\r\n
  // 0\r\n
  // \r\n

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

    // 解析长度块
    // 非阻塞
    val len = channel.receive().toString().toInt(16)
    if (len == 0) {
      break
    }

    // 翻转stream以解析确切大小的块
    stream.fixedSizeMode(len + 2)

    // 接收数据块并添加到末尾
    // 非阻塞
    val chunk = channel.receive()
    body.appendBuffer(chunk, 0, chunk.length() - 2)

    // stream被翻转回\r\n分隔符以解析下一个块
    stream.delimitedMode("\r\n")
  }
} else if (contentLength != null) {

  // 翻转stream以解析确切大小的块
  stream.fixedSizeMode(contentLength.toInt())

  // 非阻塞
  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.toChannel(vertx)

  while (true) {
    val buffer = readBuffer()

    // 广播 temperature
    // 非阻塞但是可以被挂起
    channel.send(buffer)

    //等待1秒
    awaitEvent<Long> { vertx.setTimer(1000, it) }
  }
}

fun broadcastTemperature(httpResponse: HttpServerResponse) {
// 检查是否可以向 stream 中写入
  if (httpResponse.writeQueueFull()) {

    // 这时我们不能写入，所以我们设置了一个引流处理程序，当我们可以再次写的时候被调用
    httpResponse.drainHandler { broadcastTemperature(httpResponse) }
  } else {

    // 读入 temperature
    val temperature = readBuffer()

    // 将它写回 stream
    httpResponse.write(temperature)

    // 等待1秒
    vertx.setTimer(1000) {
      broadcastTemperature()
    }
  }
}