Netty 框架学习 —— 单元测试

EmbeddedChannel 概述

ChannelHandler 是 Netty 程序的关键元素，所以彻底地测试它们应该是你的开发过程中的一个标准部分，EmbeddedChannel 是 Netty 专门为改进针对 ChannelHandler 的单元测试而提供的。Netty 提供了它所谓的 Embedded 传输，这个传输是 EmbeddedChannel 的功能，提供了通过 ChannelPipeline 传播事件的简便方法

这个方法是：将入站数据或者出站数据写入到 EmbeddedChannel 中，然后检查是否有任何东西到达 CHannelPipeline 的尾端。通过这种方式，你可以确定消息是否已经被编码或者解码过，以及是否触发了任何 ChannelHandler 动作

下表列出了 EmbeddedChannel 的相关方法

入站数据由 ChannelInboundHandler 处理，代表从远程节点读取的数据。出站数据由 ChannelOutboundHandler 处理，代表将要写到远程节点的数据。根据你要测试的 ChannelHandler，你可以使用 Inbound() 或者 Outbound() 方法对，或者兼而有之

下图展示了使用 EmbeddedChannel 的方法，数据是如何流经 ChannelPipeline 的。 你可以使用 writeOutbound()方法将消息写到 Channel 中，并通过 ChannelPipeline 沿 着出站的方向传递。随后，你可以使用 readOutbound()方法来读取已被处理过的消息，以确 定结果是否和预期一样。类似地，对于入站数据，你需要使用 writeInbound()和 readInbound() 方法

使用 EmbeddedChannel 测试 ChannelHandler

1. 测试入站消息

下述代码展示了一个简单的 ByteToMessageDecoder 实现，给定足够的数据，这个实现将产生固定大小的帧。如果没有足够的数据可供读取，它将等待下一个数据块的到来，并将再次检查是否能够产生一个新的帧

public class FixedLengthFrameDecoder extends 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 中读取一个新帧
            ByteBuf buf = in.readBytes(frameLength);
            //将该帧添加到已被解码的消息列表中
            out.add(buf);
        }
    }
}

下述代码展示了使用 EmbeddedChannel 的对于前面代码的测试

public class FixedLengthFrameDecoderTest {
    
    @Test
    public void testFrameDecoded() {
        //创建一个 ByteBuf，并存储 9 字节
        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));
        //将数据写入 EmbeddedChannel
        System.out.println(channel.writeInbound(input.retain()));//true
        //标记 Channel 为已完成状态
        System.out.println(channel.finish());//true

        //读取所生成的消息，并且验证是否有 3 帧，其中每帧都为 3 字节
        ByteBuf read = channel.readInbound();
        System.out.println(buf.readSlice(3).equals(read));// true

        read = channel.readInbound();
        System.out.println(buf.readSlice(3).equals(read));// true
        read.release();

        read = channel.readInbound();
        System.out.println(buf.readSlice(3).equals(read));// true
        read.release();

        System.out.println(channel.readInbound() == null);// true
        buf.release();
    }

    @Test
    public void testFramesDescode2() {
        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));
        //返回 false，因为没有一个完整的可供读取的帧
        System.out.println(channel.writeInbound(input.readBytes(2)));// false
        System.out.println(channel.writeInbound(input.readBytes(7)));// true

        System.out.println(channel.finish());// true
        ByteBuf read = channel.readInbound();
        System.out.println(buf.readSlice(3) == read);// false
        read.release();

        read = channel.readInbound();
        System.out.println(buf.readSlice(3) == read);// false
        read.release();

        read = channel.readInbound();
        System.out.println(buf.readSlice(3) == read);// false
        read.release();

        System.out.println(channel.readInbound() == null);// true
        buf.release();
    }
}

2. 测试入站消息

测试出站消息的处理过程和刚才所看到的类似，在下面的例子中，我们将会展示如何使用 EmbeddedChannel 来测试另一个编码器形式的 ChannelOutboundHandler，编码器是一种将一种消息格式转换为另一种的组件

该示例将会按照下列方式工作：

• 持有 AbsIntegerEncoder 的 EmbeddedChannel 将会以 4 字节的负整数的形式写出站数据
• 编码器将从传入的 ByteBuf 中读取每个负整数，并将会调用 Math.abs() 方法来获取其绝对值
• 编码器将会把每个负整数的绝对值写到 ChannelPipeline 中

下述代码展示了这个逻辑

public class AbsIntegerEncoder extends MessageToMessageEncoder<ByteBuf> {
    @Override
    protected void encode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {
        while (msg.readableBytes() >= 4) {
            //从输入的 ByteBuf 中读取下一个整数，并且计算其绝对值
            int value = Math.abs(msg.readInt());
            //将该整数写入到编码消息的 List 中
            out.add(value);
        }
    }
}

使用 EmbeddedChannel 来测试代码

public class AbsIntegerEncoderTest {
    @Test
    public void testEncoded() {
        ByteBuf buf = Unpooled.buffer();
        for (int i = 1; i < 10; i++) {
            buf.writeInt(i * -1);
        }
        // 创建一个 EmbeddedChanel，并安装一个要测试的 AbsIntegerEncoder
        EmbeddedChannel channel = new EmbeddedChannel(new AbsIntegerEncoder());
        // 写入 ByteBuf，调用 readOutbound() 方法将会产生数据
        System.out.println(channel.writeOutbound(buf));
        System.out.println(channel.finish());

        channel.readOutbound();
        for (int i = 1; i < 10; i++) {
            int temp = channel.readOutbound();
            System.out.println(temp);
        }
        System.out.println(channel.readOutbound() == null);
    }
}

下面是代码中执行的步骤。

• 将 4 字节的负整数写到一个新的 ByteBuf 中
• 创建一个 EmbeddedChannel，并为它分配一个 AbsIntegerEncoder
• 调用 EmbeddedChannel 上的 writeOutbound()方法来写入该 ByteBuf
• 标记该 Channel 为已完成状态
• 从 EmbeddedChannel 的出站端读取所有的整数，并验证是否只产生了绝对值

测试异常处理

应用程序通常需要执行比转换数据更加复杂的任务。例如，你可能需要处理格式不正确的输 入或者过量的数据。在下一个示例中，如果所读取的字节数超出了某个特定的限制，我们将会抛出一个 TooLongFrameException，这是一种经常用来防范资源被耗尽的方法

实现的代码如下

// 扩展 ByteToMessageDecoder 以将入站字节码为消息
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) {
            // 如果该帧太大，则丢弃它并抛出一个 TooLongFrameException
            in.clear();
            throw new TooLongFrameException();
        }
        // 否则，从 ByteBuf 中读取一个新的帧
        ByteBuf buf = in.readBytes(readableBytes);
        // 该帧添加到解码消息的List中
        out.add(buf);
    }
}

再使用 EmbeddedChannel 来测试这段代码

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，并向其安装一个帧大小为 3 字节的 FixedLengthFrameDecoder
        EmbeddedChannel channel = new EmbeddedChannel(new FrameChunkDecoder(3));

        System.out.println(channel.writeInbound(input.readBytes(2)));
        try {
            // 写入一个 4 字节大小的帧，并捕获预期的异常
            channel.writeInbound(input.readBytes(4));
        } catch (TooLongFrameException e) {
            e.printStackTrace();
        }

        // 写入剩余的 2 字节，会产生一个有效帧
        System.out.println(channel.writeInbound(input.readBytes(3)));//true
        System.out.println(channel.finish());

        // 读取产生的消息，并且验证值
        ByteBuf read = channel.readInbound();
        System.out.println(read.equals(buf.readSlice(2)));//true
        read.release();

        read = channel.readInbound();
        System.out.println(read.equals(buf.skipBytes(4).readSlice(3)));//true
        read.release();
        buf.release();

    }
}