content: package com.alibaba.otter.canal.protocol;
  
  import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
  import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
  import com.alibaba.otter.canal.protocol.CanalEntry.Header;
  import com.alibaba.otter.canal.protocol.CanalPacket.Compression;
  import com.alibaba.otter.canal.protocol.CanalPacket.Messages;
  import com.alibaba.otter.canal.protocol.CanalPacket.Packet;
  import com.alibaba.otter.canal.protocol.CanalPacket.PacketType;
  import com.alibaba.otter.canal.protocol.Message;
  import com.alibaba.otter.canal.protocol.exception.CanalClientException;
  import com.google.protobuf.ByteString;
  import com.google.protobuf.CodedOutputStream;
  import com.google.protobuf.WireFormat;
  import org.junit.Test;
  
  import java.io.IOException;
  import java.util.Arrays;
  import java.util.List;
  
  public class ProtocolTest {
  
      @Test(expected = CanalClientException.class)
      public void testSimple() throws IOException {
          Header.Builder headerBuilder = Header.newBuilder();
          headerBuilder.setLogfileName("mysql-bin.000001");
          headerBuilder.setLogfileOffset(1024);
          headerBuilder.setExecuteTime(1024);
          Entry.Builder entryBuilder = Entry.newBuilder();
          entryBuilder.setHeader(headerBuilder.build());
          entryBuilder.setEntryType(EntryType.ROWDATA);
          Entry entry = entryBuilder.build();
          Message message = new Message(3, true, Arrays.asList(entry.toByteString()));
  
          byte[] body = buildData(message);
          Packet packet = Packet.parseFrom(body);
          switch (packet.getType()) {
              case MESSAGES: {
                  if (!packet.getCompression().equals(Compression.NONE)) {
                      throw new CanalClientException("compression is not supported in this connector");
                  }
  
                  Messages messages = Messages.parseFrom(packet.getBody());
                  Message result = new Message(messages.getBatchId());
                  for (ByteString byteString : messages.getMessagesList()) {
                      result.addEntry(Entry.parseFrom(byteString));
                  }
  
                  System.out.println(result);
                  break;
              }
              default: {
                  throw new CanalClientException("unexpected packet type: " + packet.getType());
              }
          }
      }
  
      @SuppressWarnings("deprecation")
      private byte[] buildData(Message message) throws IOException {
          List<ByteString> rowEntries = message.getRawEntries();
          // message size
          int messageSize = 0;
          messageSize += com.google.protobuf.CodedOutputStream.computeInt64Size(1, message.getId());
  
          int dataSize = 0;
          for (ByteString rowEntry : rowEntries) {
              dataSize += CodedOutputStream.computeBytesSizeNoTag(rowEntry);
          }
          messageSize += dataSize;
          messageSize += 1 * rowEntries.size();
          // packet size
          int size = 0;
          size += com.google.protobuf.CodedOutputStream.computeEnumSize(3, PacketType.MESSAGES.getNumber());
          size += com.google.protobuf.CodedOutputStream.computeTagSize(5)
                  + com.google.protobuf.CodedOutputStream.computeRawVarint32Size(messageSize) + messageSize;
          // TODO recyle bytes[]
          byte[] body = new byte[size];
          CodedOutputStream output = CodedOutputStream.newInstance(body);
          output.writeEnum(3, PacketType.MESSAGES.getNumber());
  
          output.writeTag(5, WireFormat.WIRETYPE_LENGTH_DELIMITED);
          output.writeRawVarint32(messageSize);
          // message
          output.writeInt64(1, message.getId());
          for (ByteString rowEntry : rowEntries) {
              output.writeBytes(2, rowEntry);
          }
          output.checkNoSpaceLeft();
  
          return body;
      }
  }
  