如何以安全的方式替换字节码头？

Question

我设计了以下协议。

Header的长度	标头	内容	定界符（END标志）
0x00000101	Hello	JSON	### End ###

客户端发送一条消息，其中包含“标头的长度”，“标头”，“内容”，“内容”和“分界符”。应该注意的是，客户不会一次发送完整的消息，“竞争”字段将被多次发送。当客户端终点发送内容字段时，它将连接“ ### End ###”。服务器的作业是获取“标题”字段和“内容”字段。

业务逻辑是处理“标题”和“内容”。因此，我需要从Bytebuf获得这些。我想到的第一件事是使用解码器扩展要获得“标头”。当我获得第一个解码框架（即“标头”）时，我将通过将解码器替换为另一个来更改点线。最后，我将使用新处理程序来处理剩下的消息。

我试图根据

Answer 1

您的协议格式比在解码器中的简单构建要复杂得多。这意味着您必须编写自己的解码器。

查看您的格式，我们可以识别以下状态：

• 读取标题长度：准确阅读4个字节
• 读取标题：从最后一个值中读取x量
• 读取消息，直到您看到定界符

在使用时实现这样的状态开关不太困难 replayingdecoder 。

例如：

enum FrameDecoderState {
    FRAME_HEADER_LENGTH,
    FRAME_HEADER,
    FRAME_BODY
}

public class FrameDecoder extends ReplayingDecoder<FrameDecoderState> {
    private int headerLength;
    private ByteBuf header;
    private int detectedMessageLength;
    private final ByteBuf delimiter;

    public FrameDecoder(ByteBuf delimiter) {
        super(FRAME_HEADER_LENGTH);
        this.delimiter = Objects.requireNonNull(delimiter, "delimiter");
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        super.channelUnregistered(ctx);
        if (header != null) {
            header.release();
        }
        delimiter.release();
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> out) throws Exception {
        switch(this.state()) {
            default:
                throw new AssertionError("Shouldn't reach here.");
            case FRAME_HEADER_LENGTH:
                headerLength = buf.readInt();
                checkpoint(FRAME_HEADER);
            case FRAME_HEADER:
                header = buf.readBytes(headerLength);
                detectedMessageLength = 0;
                checkpoint(FRAME_BODY);
            case FRAME_BODY:
                int bodyStart = buf.readerIndex();
                while(true) {
                    if (findDelimiter(bodyStart + detectedMessageLength, buf)) {
                        out.add(new FrameDecoderFrame(header, buf.readBytes(detectedMessageLength)));
                        buf.readerIndex(bodyStart + detectedMessageLength + delimiter.capacity());
                        checkpoint(FRAME_HEADER_LENGTH);
                        break;
                    }
                    detectedMessageLength++;
                }
        }
    }

    private boolean findDelimiter(int bufIndex, ByteBuf buf) {
        for (int delimiterIndex = 0; delimiterIndex < delimiter.capacity(); delimiterIndex++) {
            if (buf.getByte(bufIndex) != delimiter.getByte(delimiterIndex)) {
                return false;
            } else {
                bufIndex++;
            }
        }
        return true;
    }
}

class FrameDecoderFrame extends AbstractReferenceCounted {
    private final ByteBuf header;
    private final ByteBuf body;

    public FrameDecoderFrame(ByteBuf header, ByteBuf body) {
        this.header = header;
        this.body = body;
    }

    public ByteBuf getHeader() {
        return header;
    }

    public ByteBuf getBody() {
        return body;
    }

    @Override
    public int hashCode() {
        int hash = 3;
        hash = 41 * hash + Objects.hashCode(this.header);
        hash = 41 * hash + Objects.hashCode(this.body);
        return hash;
    }

    @Override
    public boolean equals(Object obj) {
        if (this == obj) {
            return true;
        }
        if (obj == null) {
            return false;
        }
        if (getClass() != obj.getClass()) {
            return false;
        }
        final FrameDecoderFrame other = (FrameDecoderFrame) obj;
        if (!Objects.equals(this.header, other.header)) {
            return false;
        }
        if (!Objects.equals(this.body, other.body)) {
            return false;
        }
        return true;
    }

    @Override
    public String toString() {
        return "FrameDecoderFrame{" + "header=" + header + ", body=" + body + '}';
    }

    @Override
    protected void deallocate() {
        header.release();
        body.release();
    }

    @Override
    public FrameDecoderFrame touch(Object o) {
        header.touch(o);
        body.touch(o);
        return this;
    }
}

编写此类编码后，请确保对其进行测试，包括有部分数据，或者当同一消息中有多个帧时，例如：

@org.junit.jupiter.api.Test
public void testBasic() {
    var channel = new EmbeddedChannel(new FrameDecoder(Unpooled.wrappedBuffer("###end###".getBytes(US_ASCII))));

    channel.writeInbound(Unpooled.buffer().writeInt(4));
    assertNull(channel.readInbound());
    channel.writeInbound(Unpooled.wrappedBuffer("te".getBytes(US_ASCII)));
    assertNull(channel.readInbound());
    channel.writeInbound(Unpooled.wrappedBuffer("st{\"test\":".getBytes(US_ASCII)));
    assertNull(channel.readInbound());
    channel.writeInbound(Unpooled.wrappedBuffer("true}".getBytes(US_ASCII)));
    assertNull(channel.readInbound());
    channel.writeInbound(Unpooled.wrappedBuffer("###".getBytes(US_ASCII)));
    assertNull(channel.readInbound());
    channel.writeInbound(Unpooled.wrappedBuffer("end".getBytes(US_ASCII)));
    assertNull(channel.readInbound());
    channel.writeInbound(Unpooled.buffer().writeBytes("###".getBytes(US_ASCII)).writeInt(0).writeBytes("[]###end###".getBytes(US_ASCII)));
    assertEquals(
        new FrameDecoderFrame(
            Unpooled.wrappedBuffer("test".getBytes(US_ASCII)),
            Unpooled.wrappedBuffer("{\"test\":true}".getBytes(US_ASCII))
        ),
        channel.readInbound()
    );
    assertEquals(
        new FrameDecoderFrame(
            Unpooled.EMPTY_BUFFER,
            Unpooled.wrappedBuffer("[]".getBytes(US_ASCII))
        ),
        channel.readInbound()
    );
    assertNull(channel.readInbound());
}