netty使用EmbeddedChannel对channel的出入站进行单元测试

一种特殊的Channel实现----EmbeddedChannel，它是Netty专门为改进针对ChannelHandler的单元测试而提供的。

测试入站消息

public class FixedLengthFrameDecoder extends ByteToMessageDecoder { private final int frameLength; public FixedLengthFrameDecoder(int frameLength) { if (frameLength <= 0) { throw new IllegalArgumentException("frameLength must be 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); out.add(buf); } } }

public class FixedLengthFrameDecoderTest { @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 FixedLengthFrameDecoder(3)); Assert.assertTrue(channel.writeInbound(input.retain())); Assert.assertTrue(channel.finish()); ByteBuf read = channel.readInbound(); Assert.assertEquals(buf.readSlice(3), read); read.release(); read = channel.readInbound(); Assert.assertEquals(buf.readSlice(3), read); read.release(); read = channel.readInbound(); Assert.assertEquals(buf.readSlice(3), read); read.release(); Assert.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)); Assert.assertFalse(channel.writeInbound(input.readBytes(2))); Assert.assertTrue(channel.writeInbound(input.readBytes(7))); Assert.assertTrue(channel.finish()); ByteBuf read = channel.readInbound(); Assert.assertEquals(buf.readSlice(3), read); read.release(); read = channel.readInbound(); Assert.assertEquals(buf.readSlice(3), read); read.release(); read = channel.readInbound(); Assert.assertEquals(buf.readSlice(3), read); read.release(); Assert.assertNull(channel.readInbound()); buf.release(); } }

测试出站消息

public class AbsIntegerEncoder extends MessageToMessageEncoder<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); } } }

public class AbsIntegerEncoderTest { @Test public void testEncoded() { ByteBuf buf = Unpooled.buffer(); for (int i = 0; i < 10; i++) { buf.writeInt(i * -1); } EmbeddedChannel channel = new EmbeddedChannel(new AbsIntegerEncoder()); Assert.assertTrue(channel.writeOutbound(buf)); Assert.assertTrue(channel.finish()); for (int i = 0; i < 10; i++) { Assert.assertEquals(Integer.valueOf(i), channel.readOutbound()); } Assert.assertNull(channel.readOutbound()); } }

测试异常处理

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) { in.clear(); throw new TooLongFrameException(); } ByteBuf buf = in.readBytes(readableBytes); out.add(buf); } }

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)); Assert.assertTrue(channel.writeInbound(input.readBytes(2))); try { channel.writeInbound(input.readBytes(4)); Assert.fail(); } catch (TooLongFrameException e) { } Assert.assertTrue(channel.writeInbound(input.readBytes(3))); Assert.assertTrue(channel.finish()); ByteBuf read = channel.readInbound(); Assert.assertEquals(buf.readSlice(2), read); read.release(); read = channel.readInbound(); Assert.assertEquals(buf.skipBytes(4).readSlice(3), read); read.release(); buf.release(); } }