EmbeddedChannel 概述
特殊的EmbeddedChannel 方法
EmbeddedChannel 的数据流
使用EmbeddedChannel 测试ChannelHandler
- 测试入站消息
通过FixedLengthFrameDecoder 解码
这个特定的解码器将产生固定为3 字节大小的帧。因此，它可能会需要多个事件来提供足够的字节数以产生一个帧。最终，每个帧都会被传递给ChannelPipeline 中的下一个ChannelHandler。
1.定义FixedLengthFrameDecoder
public class FixedLengthFrameDecoder extends ByteToMessageDecoder { // 扩展ByteToMessageDecoder 以处理入站字节，并将它们解码为消息
private final int frameLength;
public FixedLengthFrameDecoder(int frameLength) { // 指定要生成的帧的长度
if (frameLength <= 0) {
throw new IllegalArgumentException("frameLength must be a positive integer: " + frameLength);
}
this.frameLength = frameLength;
}
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in,List<Object> out) throws Exception {
while (in.readableBytes() >= frameLength) { // 检查是否有足够的字节可以被读取，以生成下一个帧
ByteBuf buf = in.readBytes(frameLength); // 从ByteBuf 中读取一个新帧
out.add(buf); // 将该帧添加到已被解码的消息列表中
}
}
}
2.测试FixedLengthFrameDecoder
public class FixedLengthFrameDecoderTest {
@Test
public void testFramesDecoded() {
ByteBuf buf = Unpooled.buffer();
for (int i = 0; i < 9; i++) { // 创建一个ByteBuf，并存储9 字节
buf.writeByte(i);
}
ByteBuf input = buf.duplicate();
// 创建一个EmbeddedChannel，并添加一个FixedLengthFrameDecoder，其将以3 字节的帧长度被测试
EmbeddedChannel channel = new EmbeddedChannel(new FixedLengthFrameDecoder(3));
// write bytes
assertTrue(channel.writeInbound(input.retain()));
// 标记Channel为已完成状态
assertTrue(channel.finish());
// read messages
ByteBuf read = (ByteBuf) channel.readInbound();
// 读取所生成的消息，并且验证是否有3 帧（切片），其中每帧（切片）都为3 字节
assertEquals(buf.readSlice(3), read);
read.release();
read = (ByteBuf) channel.readInbound();
assertEquals(buf.readSlice(3), read);
read.release();
read = (ByteBuf) channel.readInbound();
assertEquals(buf.readSlice(3), read);
read.release();
assertNull(channel.readInbound());
buf.release();
}
@Test
public void testFramesDecoded2() {
ByteBuf buf = Unpooled.buffer();
for (int i = 0; i < 9; i++) {
buf.writeByte(i);
}
ByteBuf input = buf.duplicate();
EmbeddedChannel channel = new EmbeddedChannel(
new FixedLengthFrameDecoder(3));
assertFalse(channel.writeInbound(input.readBytes(2)));
assertTrue(channel.writeInbound(input.readBytes(7)));
assertTrue(channel.finish());
ByteBuf read = (ByteBuf) channel.readInbound();
assertEquals(buf.readSlice(3), read);
read.release();
read = (ByteBuf) channel.readInbound();
assertEquals(buf.readSlice(3), read);
read.release();
read = (ByteBuf) channel.readInbound();
assertEquals(buf.readSlice(3), read);
read.release();
assertNull(channel.readInbound());
buf.release();
}
}
入站ByteBuf 是通过两个步骤写入的。当writeInbound(input.readBytes(2))被调用时，返回了false。为什么呢？正如上面描述的，如果对readInbound()的后续调用将会返回数据，那么writeInbound()方法将会返回true。但是只有当有3 个或者更多的字节可供读取时，FixedLength-FrameDecoder 才会产生输出。该测试剩下的部分和testFramesDecoded()是相同的。
测试出站消息
通过AbsIntegerEncoder 编码
- 持有AbsIntegerEncoder 的EmbeddedChannel 将会以4 字节的负整数的形式写出站数据；
- 编码器将从传入的ByteBuf 中读取每个负整数，并将会调用Math.abs()方法来获取其绝对值；
- 编码器将会把每个负整数的绝对值写到ChannelPipeline 中。
- AbsIntegerEncoder
public class AbsIntegerEncoder extendsMessageToMessageEncoder<ByteBuf> {
@Override
protected void encode(ChannelHandlerContext channelHandlerContext,ByteBuf in, List<Object> out) throws Exception{
while (in.readableBytes() >= 4) {
int value = Math.abs(in.readInt());
out.add(value);
}
}
}
- 测试AbsIntegerEncoder
public class AbsIntegerEncoderTest {
@Test
public void testEncoded() {
ByteBuf buf = Unpooled.buffer();
for (int i = 1; i < 10; i++) {
buf.writeInt(i * -1);
}
EmbeddedChannel channel = new EmbeddedChannel(new AbsIntegerEncoder());
assertTrue(channel.writeOutbound(buf));
assertTrue(channel.finish());
// read bytes
for (int i = 1; i < 10; i++) {
assertEquals(i, channel.readOutbound());
}
assertNull(channel.readOutbound());
}
}
测试异常处理
通过FrameChunkDecoder 解码
最大的帧大小已经被设置为3 字节。如果一个帧的大小超出了该限制，那么程序将会丢弃它的字节，并抛出一个TooLongFrameException。位于ChannelPipeline 中的其他ChannelHandler 可以选择在exceptionCaught()方法中处理该异常或者忽略它。
- FrameChunkDecoder
public class FrameChunkDecoder extends ByteToMessageDecoder {
private final int maxFrameSize;
public FrameChunkDecoder(int maxFrameSize) {
this.maxFrameSize = maxFrameSize;
}
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in,List<Object> out) throws Exception {
int readableBytes = in.readableBytes();
if (readableBytes > maxFrameSize) {
// discard the bytes
in.clear();
throw new TooLongFrameException();
}
ByteBuf buf = in.readBytes(readableBytes);
out.add(buf);
}
}
- 测试FrameChunkDecoder
public class FrameChunkDecoderTest {
@Test
public void testFramesDecoded() {
ByteBuf buf = Unpooled.buffer();
for (int i = 0; i < 9; i++) {
buf.writeByte(i);
}
ByteBuf input = buf.duplicate();
EmbeddedChannel channel = new EmbeddedChannel(new FrameChunkDecoder(3));
assertTrue(channel.writeInbound(input.readBytes(2)));
try {
channel.writeInbound(input.readBytes(4));
Assert.fail(); // 如果上面没有抛出异常，那么就会到达这个断言，并且测试失败
} catch (TooLongFrameException e) {
// expected exception
}
assertTrue(channel.writeInbound(input.readBytes(3)));
assertTrue(channel.finish());
// Read frames
ByteBuf read = (ByteBuf) channel.readInbound(); // 读取字节
assertEquals(buf.readSlice(2), read);
read.release();
read = (ByteBuf) channel.readInbound();
assertEquals(buf.skipBytes(4).readSlice(3), read);
read.release();
buf.release();
}
}
网友评论