package com.wlf.netty.nettyapi.msgpack;import com.wlf.netty.nettyapi.constant.Delimiter;
import com.wlf.netty.nettyapi.javabean.Header;
import com.wlf.netty.nettyapi.javabean.NettyMessage;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;import java.util.List;public class NettyMessageDecoder extends ByteToMessageDecoder {/*** 消息体字节大小：分割符字段4字节+长度字段4字节+请求类型字典1字节+预留字段1字节=10字节*/private static final int HEAD_LENGTH = 10;@Overrideprotected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {while (true) {// 标记字节流开始位置byteBuf.markReaderIndex();// 若读取到分割标识，说明读取当前字节流开始位置了if (byteBuf.readInt() == Delimiter.DELIMITER) {break;}// 重置读索引为0byteBuf.resetReaderIndex();// 长度校验，字节流长度至少10字节，小于10字节则等待下一次字节流过来if (byteBuf.readableBytes() < HEAD_LENGTH) {byteBuf.resetReaderIndex();return;}}// 2、获取data的字节流长度int dataLength = byteBuf.readInt();// 校验数据包是否全部发送过来，总字节流长度(此处读取的是除去delimiter和length之后的总长度)-// type和reserved两个字节=data的字节流长度int totalLength = byteBuf.readableBytes();if ((totalLength - 2) < dataLength) {// 长度校验，字节流长度少于数据包长度，说明数据包拆包了，等待下一次字节流过来byteBuf.resetReaderIndex();return;}// 3、请求类型byte type = byteBuf.readByte();// 4、预留字段byte reserved = byteBuf.readByte();// 5、数据包内容byte[] data = null;if (dataLength > 0) {data = new byte[dataLength];byteBuf.readBytes(data);}NettyMessage nettyMessage = new NettyMessage();Header header = new Header();header.setDelimiter(Delimiter.DELIMITER);header.setLength(dataLength);header.setType(type);header.setReserved(reserved);nettyMessage.setHeader(header);nettyMessage.setData(data);list.add(nettyMessage);// 回收已读字节byteBuf.discardReadBytes();}
}

// channel.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024 * 1024 * 1024, 4, 4, 2, 0));

    /*** 构造PCM请求消息体** @return*/private byte[] buildPcmData() throws Exception {byte[] resultByte = longToBytes(System.currentTimeMillis());// 读取一个本地文件String AUDIO_PATH = "D:\\input\\test_1.pcm";try (RandomAccessFile raf = new RandomAccessFile(AUDIO_PATH, "r")) {int len = -1;byte[] content = new byte[1024];while((len = raf.read(content)) != -1){resultByte = addAll(resultByte, content);}}return resultByte;}

package com.wlf.netty.nettyserver;import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import org.junit.Assert;
import org.junit.Test;public class ByteBufTest {@Testpublic void byteBufTest() {ByteBuf byteBuf = Unpooled.buffer(10);byteBuf.writeInt(0xabef0101);byteBuf.writeInt(1024);byteBuf.writeByte((byte) 1);byteBuf.writeByte((byte) 0);// 开始读取printDelimiter(byteBuf);printLength(byteBuf);// 派生一个ByteBuf，取剩下2个字节，但读索引不动ByteBuf duplicatBuf = byteBuf.duplicate();printByteBuf(byteBuf);// 派生一个ByteBuf，取剩下2个字节，读索引动了ByteBuf sliceBuf = byteBuf.readSlice(2);printByteBuf(byteBuf);// 两个派生的对象其实是一样的Assert.assertEquals(duplicatBuf, sliceBuf);}private void printDelimiter(ByteBuf buf) {int newDelimiter = buf.readInt();System.out.printf("delimeter: %s\n", Integer.toHexString(newDelimiter));printByteBuf(buf);}private void printLength(ByteBuf buf) {int length = buf.readInt();System.out.printf("length: %d\n", length);printByteBuf(buf);}private void printByteBuf(ByteBuf buf) {System.out.printf("reader Index: %d, writer Index: %d, capacity: %d\n", buf.readerIndex(), buf.writerIndex(), buf.capacity());}
}

delimeter: abef0101
reader Index: 4, writer Index: 10, capacity: 10
length: 1024
reader Index: 8, writer Index: 10, capacity: 10
reader Index: 8, writer Index: 10, capacity: 10
reader Index: 10, writer Index: 10, capacity: 10