Vertx NetServer 控制读取流

Question

我正在尝试根据我必须使用的现有基础架构模拟 TCP 服务器以使用 Vertx 进行测试。

我正在模仿的服务器完全异步工作，并且根据缓冲区中指示请求长度的前标头知道传入缓冲区的长度。

我需要在连接到我的模拟 TCP 服务器的每个客户端套接字上读取传入请求的前 6 个字符。从这个 pre-header 中我读取了请求的实际长度（例如，对于 xx3018，我知道请求的全长是 3018）。

然后我需要根据长度读取缓冲区的其余部分，将其与响应映射匹配，并为请求返回正确的响应。

使用纯 java 的工作模拟服务器示例（快速实现，因此不会阻止其他开发:)）

public void run(String... args) throws Exception {
    log.info("Starting TCP Server");

    ServerSocket serverSocket = new ServerSocket(1750);

    while (true) {
        try {
            Socket socket = serverSocket.accept();

            CompletableFuture.runAsync(() -> {
                Exception e = null;
                while (e == null) {
                    try {
                        InputStream inputStream = socket.getInputStream();
                        OutputStream outputStream = socket.getOutputStream();

                        byte[] preHeader = new byte[6];
                        inputStream.read(preHeader);

                        String preHeaderValue = new String(preHeader);
                        log.info("Pre header: {}", preHeaderValue);

                        int length = Integer.valueOf(preHeaderValue.substring(2));
                        log.info("Request full length: {}", length);
                        byte[] request = new byte[length - 6];

                        inputStream.read(request);

                        String requestValue = new String(request);

                        log.info("Request: {}", requestValue);

                        String response = this.requestResponseProvider.getResponse(preHeaderValue + requestValue);
                        log.info("Response: {}", response);
                        outputStream.write(response.getBytes());
                    } catch (Exception ex) {
                        log.error("Encountered a problem: {}", e.getMessage());
                        e = ex;
                    }
                }
            });
        } catch (Exception e) {
            log.error("Encountered a problem: {}", e.getMessage());
        }
    }
}

我似乎找不到一种方法来控制输入流，就像我用普通 java 控制它一样。

Answer 1

在搁置这个问题很长时间之后，我决定尝试一下。

我记得在另一个项目中使用了以下模块：https://github.com/vert-x3/vertx-tcp-eventbus-bridge

我还记得在 tcp 桥的内部协议中，它会将有效负载的长度附加到通过 tcp 桥发送的缓冲区中，我查看了源代码以了解它如何处理块（也称为帧）

我发现了以下内容：https://github.com/vert-x3/vertx-tcp-eventbus-bridge/blob/master/src/main/java/io/vertx/ext/eventbus/bridge/tcp/impl/protocol/FrameParser.java 这正是我想要实现的目标：)

我对其进行了一些修改，将其转换为 Kotlin，并制作它以便我可以控制标头大小以及它提取有效负载长度的方式。

以下是使用 Vert.x NetServer 控制读取流的粗略快速示例：

suspend fun main() {
  val vertx = Vertx.vertx()
  initServer(vertx)
  initClient(vertx)
}

suspend fun initServer(vertx: Vertx) {
  val server = vertx.createNetServer(netServerOptionsOf(port = 8888, host = "localhost"))

  server
    .connectHandler { socket ->
      val parser = FrameParser(
        headerSize = 4,
        headerHandler = {
          it.getInt(0)
        },
        handler = {
          println(it.toString())
          println("---")
        }
      )
      socket.handler(parser)

      socket.exceptionHandler {
        it.printStackTrace()
        socket.close()
      }
    }
    .listenAwait()
}

suspend fun initClient(vertx: Vertx) {
  val client = vertx.createNetClient()
  val socket = client.connectAwait(port = 8888, host = "localhost")

  val message = "START|${"foobarfoobar".repeat(100)}|END"
  val length = message.length
  repeat(5) {
    repeat(100) {
      vertx.setPeriodic(10) {
        socket.write(
          Buffer.buffer()
            .appendInt(length)
            .appendString(message)
        )
      }
    }
    delay(1000)
  }
}

/**
 * Based on: https://github.com/vert-x3/vertx-tcp-eventbus-bridge/blob/master/src/main/java/io/vertx/ext/eventbus/bridge/tcp/impl/protocol/FrameParser.java
 */
class FrameParser(
  private val headerSize: Int,
  private val headerHandler: (Buffer) -> Int,
  private val handler: (Buffer) -> Unit
) : Handler<Buffer?> {

  private var _buffer: Buffer? = null
  private var _offset = 0

  override fun handle(buffer: Buffer?) {
    append(buffer)
    var offset: Int
    while (true) {
      // set a rewind point. if a failure occurs,
      // wait for the next handle()/append() and try again
      offset = _offset

      // how many bytes are in the buffer
      val remainingBytes = bytesRemaining()

      // at least expected header size
      if (remainingBytes < headerSize) {
        break
      }

      // what is the length of the message
      val length: Int = headerHandler(_buffer!!.getBuffer(_offset, _offset + headerSize))
      _offset += headerSize
      if (remainingBytes - headerSize >= length) {
        // we have a complete message
        handler(_buffer!!.getBuffer(_offset, _offset + length))
        _offset += length
      } else {
        // not enough data: rewind, and wait
        // for the next packet to appear
        _offset = offset
        break
      }
    }
  }

  private fun append(newBuffer: Buffer?) {
    if (newBuffer == null) {
      return
    }

    // first run
    if (_buffer == null) {
      _buffer = newBuffer
      return
    }

    // out of data
    if (_offset >= _buffer!!.length()) {
      _buffer = newBuffer
      _offset = 0
      return
    }

    // very large packet
    if (_offset > 0) {
      _buffer = _buffer!!.getBuffer(_offset, _buffer!!.length())
    }
    _buffer!!.appendBuffer(newBuffer)
    _offset = 0
  }

  private fun bytesRemaining(): Int {
    return if (_buffer!!.length() - _offset < 0) {
      0
    } else {
      _buffer!!.length() - _offset
    }
  }
}