Java 类io.netty.buffer.UnpooledByteBufAllocator 实例源码

public CachedSingleFileSystem(String path) throws IOException {
  this.path = path;
  File f = new File(path);
  long length = f.length();
  if (length > Integer.MAX_VALUE) {
    throw new UnsupportedOperationException("Cached file system only supports files of less than 2GB.");
  }
  System.out.println(length);
  try (InputStream is = new BufferedInputStream(new FileInputStream(path))) {
    byte[] buffer = new byte[64*1024];
    this.file = UnpooledByteBufAllocator.DEFAULT.directBuffer((int) length);
    int read;
    while ( (read = is.read(buffer)) > 0) {
      file.writeBytes(buffer, 0, read);
    }
  }
}

public CachedSingleFileSystem(String path) throws IOException {
  this.path = path;
  File f = new File(path);
  long length = f.length();
  if (length > Integer.MAX_VALUE) {
    throw new UnsupportedOperationException("Cached file system only supports files of less than 2GB.");
  }
  System.out.println(length);
  try (InputStream is = new BufferedInputStream(new FileInputStream(path))) {
    byte[] buffer = new byte[64*1024];
    this.file = UnpooledByteBufAllocator.DEFAULT.directBuffer((int) length);
    int read;
    while ( (read = is.read(buffer)) > 0) {
      file.writeBytes(buffer, 0, read);
    }
  }
}

public static byte[] createMessageWrittenToLedger(String msg) throws Exception {
    PulsarApi.MessageMetadata.Builder messageMetadataBuilder = PulsarApi.MessageMetadata.newBuilder();
    messageMetadataBuilder.setPublishTime(System.currentTimeMillis());
    messageMetadataBuilder.setProducerName("createMessageWrittenToLedger");
    messageMetadataBuilder.setSequenceId(1);
    PulsarApi.MessageMetadata messageMetadata = messageMetadataBuilder.build();
    ByteBuf data = UnpooledByteBufAllocator.DEFAULT.heapBuffer().writeBytes(msg.getBytes());

    int msgMetadataSize = messageMetadata.getSerializedSize();
    int payloadSize = data.readableBytes();
    int totalSize = 4 + msgMetadataSize + payloadSize;

    ByteBuf headers = PooledByteBufAllocator.DEFAULT.heapBuffer(totalSize, totalSize);
    ByteBufCodedOutputStream outStream = ByteBufCodedOutputStream.get(headers);
    headers.writeInt(msgMetadataSize);
    messageMetadata.writeTo(outStream);
    ByteBuf headersAndPayload = DoubleByteBuf.get(headers, data);
    byte[] byteMessage = headersAndPayload.nioBuffer().array();
    headersAndPayload.release();
    return byteMessage;
}

/**
 * Read a DeviceConnectInformation from a Base64 encoded String, which was read from a QR Code.
 */
public static DeviceConnectInformation fromDataString(String data) throws IOException {
    final ByteBuf base64 = UnpooledByteBufAllocator.DEFAULT.heapBuffer(data.length());
    ByteBufUtil.writeAscii(base64, data);
    final ByteBuf byteBuf = decode(base64);
    if (byteBuf.readableBytes() != DATA_LENGTH) {
        throw new IOException("too many bytes encoded");
    }

    final byte[] addressData = new byte[ADDRESS_LENGTH];
    byteBuf.readBytes(addressData);
    final InetAddress address = InetAddress.getByAddress(addressData);
    final int port = byteBuf.readUnsignedShort();
    final byte[] idData = new byte[DeviceID.ID_LENGTH];
    byteBuf.readBytes(idData);
    final DeviceID id = new DeviceID(idData);
    final byte[] encodedToken = new byte[TOKEN_BASE64_LENGTH];
    byteBuf.readBytes(encodedToken);
    final byte[] token = decodeToken(new String(encodedToken));

    return new DeviceConnectInformation(address, port, id, token);
}

public CachedSingleFileSystem(String path) throws IOException {
  this.path = path;
  File f = new File(path);
  long length = f.length();
  if (length > Integer.MAX_VALUE) {
    throw new UnsupportedOperationException("Cached file system only supports files of less than 2GB.");
  }
  System.out.println(length);
  try (InputStream is = new BufferedInputStream(new FileInputStream(path))) {
    byte[] buffer = new byte[64*1024];
    this.file = UnpooledByteBufAllocator.DEFAULT.directBuffer((int) length);
    int read;
    while ( (read = is.read(buffer)) > 0) {
      file.writeBytes(buffer, 0, read);
    }
  }
}

@Test
public void testParser() throws Exception {
  UnpooledByteBufAllocator allocator = new UnpooledByteBufAllocator(false);
  CollectdParser parser = new CollectdParser(getContext(), false, null, false, null, CHARSET);
  byte[] bytes = Files.readAllBytes(SINGLE_PACKET.toPath());
  ByteBuf buf = allocator.buffer(bytes.length);
  buf.writeBytes(bytes);
  List<Record> records = parser.parse(buf, null, null);

  Assert.assertEquals(23, records.size()); // 23 Value parts

  Record record0 = records.get(0);
  UDPTestUtil.verifyCollectdRecord(UDPTestUtil.expectedRecord0, record0);

  Record record2 = records.get(2);
  UDPTestUtil.verifyCollectdRecord(UDPTestUtil.expectedRecord2, record2);

}

@Test
public void testParserExcludeInterval() throws Exception {
  UnpooledByteBufAllocator allocator = new UnpooledByteBufAllocator(false);
  CollectdParser parser = new CollectdParser(getContext(), false, null, true, null, CHARSET);
  byte[] bytes = Files.readAllBytes(SINGLE_PACKET.toPath());
  ByteBuf buf = allocator.buffer(bytes.length);
  buf.writeBytes(bytes);
  List<Record> records = parser.parse(buf, null, null);

  Assert.assertEquals(23, records.size()); // 23 Value parts

  Record record0 = records.get(0);
  UDPTestUtil.verifyCollectdRecord(UDPTestUtil.expectedRecordNoInterval0, record0);

  Record record2 = records.get(2);
  UDPTestUtil.verifyCollectdRecord(UDPTestUtil.expectedRecordNoInterval2, record2);

}

@Test
public void testEncryptedRecord() throws Exception {
  // If unlimited strength encryption is not available, we cant run this test.
  Assume.assumeFalse(Cipher.getMaxAllowedKeyLength("AES") < 256);

  UnpooledByteBufAllocator allocator = new UnpooledByteBufAllocator(false);
  CollectdParser parser = new CollectdParser(getContext(), false, null, false, AUTH_FILE_PATH, CHARSET);
  byte[] bytes = Files.readAllBytes(SINGLE_ENCRYPTED_PACKET.toPath());
  ByteBuf buf = allocator.buffer(bytes.length);
  buf.writeBytes(bytes);
  List<Record> records = parser.parse(buf, null, null);

  Assert.assertEquals(24, records.size()); // 24 value parts
  Record record14 = records.get(14);
  UDPTestUtil.verifyCollectdRecord(UDPTestUtil.encryptedRecord14, record14);
  LOG.info("Num records: {}", records.size());
}

@Test
public void testParseFailure() throws Exception {
  SyslogParser parser = new SyslogParser(getContext(), StandardCharsets.UTF_8);
  String msg = "<123>                    ";
  byte[] bytes = msg.getBytes(StandardCharsets.UTF_8);
  UnpooledByteBufAllocator allocator = new UnpooledByteBufAllocator(false);
  ByteBuf buffer = allocator.buffer(bytes.length);
  buffer.writeBytes(bytes);
  try {
    parser.parse(
        buffer,
        InetSocketAddress.createUnresolved("localhost", 5000),
        InetSocketAddress.createUnresolved("localhost", 50000)
    );
    Assert.fail("Expected OnRecordErrorException");
  } catch (OnRecordErrorException ex) {
    Record record = ex.getRecord();
    Assert.assertEquals(msg, record.get().getValueAsString());
  }
}

@Test
public void testFullRequestWithBody() throws Exception {
  outputReceived = new CountDownLatch(1);
  String payload = "body";
  ByteBuf body = ByteBufUtil.writeUtf8(UnpooledByteBufAllocator.DEFAULT, payload);
  FullHttpRequest requestIn = new DefaultFullHttpRequest(HTTP_1_1, GET, "/", body);

  channel.writeInbound(requestIn);

  channel.runPendingTasks(); // blocks

  Uninterruptibles.awaitUninterruptibly(outputReceived);

  Request requestOut = requests.remove(0);

  assertTrue(requestOut != null);
  assertTrue(requestOut instanceof FullRequest);
  assertEquals("HTTP/1.1", requestOut.version());
  assertEquals(HttpMethod.GET, requestOut.method());
  assertEquals("/", requestOut.path());
  assertTrue(requestOut.hasBody());
  assertFalse(requestOut.body() == null);
  assertEquals(body, requestOut.body());
}

@Test
public void testFullResponse() throws Exception {
  outputReceived = new CountDownLatch(2);
  ByteBuf body = ByteBufUtil.writeUtf8(UnpooledByteBufAllocator.DEFAULT, "response");

  FullHttpRequest requestIn = new DefaultFullHttpRequest(HTTP_1_1, GET, "/");
  FullResponse responseIn = ResponseBuilders.newOk().body(body).build();

  channel.writeInbound(requestIn);
  channel.runPendingTasks(); // blocks
  channel.writeOutbound(responseIn);

  channel.runPendingTasks(); // blocks

  Uninterruptibles.awaitUninterruptibly(outputReceived);

  HttpResponse responseOut = (HttpResponse) responses.remove(0);

  assertTrue(responseOut != null);
  assertTrue(responseOut instanceof FullHttpResponse);
  assertEquals(HTTP_1_1, responseOut.protocolVersion());
  assertEquals(OK, responseOut.status());
  assertFalse(((FullHttpResponse) responseOut).content() == null);
  assertEquals(body, ((FullHttpResponse) responseOut).content());
}

@Test
public void testFullRequestWithBody() throws Exception {
  outputReceived = new CountDownLatch(1);
  ByteBuf body = ByteBufUtil.writeUtf8(UnpooledByteBufAllocator.DEFAULT, "body");
  FullRequest requestIn = RequestBuilders.newPost("/").body(body).build();

  channel.writeOutbound(requestIn);

  channel.runPendingTasks(); // blocks

  Uninterruptibles.awaitUninterruptibly(outputReceived);

  FullHttpRequest requestOut = (FullHttpRequest) requests.remove(0);

  assertTrue(requestOut != null);
  assertEquals(HTTP_1_1, requestOut.protocolVersion());
  assertEquals(HttpMethod.POST, requestOut.method());
  assertEquals("/", requestOut.uri());
  assertFalse(requestOut.content() == null);
  assertEquals(body, requestOut.content());
}

@Test
public void testFullResponse() throws Exception {
  outputReceived = new CountDownLatch(1);
  ByteBuf body = ByteBufUtil.writeUtf8(UnpooledByteBufAllocator.DEFAULT, "response");

  FullHttpResponse responseIn = new DefaultFullHttpResponse(HTTP_1_1, OK, body);

  channel.writeInbound(responseIn);

  channel.runPendingTasks(); // blocks

  Uninterruptibles.awaitUninterruptibly(outputReceived);

  Response responseOut = responses.remove(0);

  assertTrue(responseOut != null);
  assertTrue(responseOut instanceof FullResponse);
  assertEquals("HTTP/1.1", responseOut.version());
  assertEquals(OK, responseOut.status());
  assertTrue(responseOut.hasBody());
  assertFalse(responseOut.body() == null);
  assertEquals(body, responseOut.body());
}

@Override
public String toString() {

    final StringBuilder sb = new StringBuilder();
    sb.append(getClass().getSimpleName());

    ByteBuf buffer = UnpooledByteBufAllocator.DEFAULT.buffer(singularArguments.size() * 10);
    encode(buffer);
    buffer.resetReaderIndex();

    byte[] bytes = new byte[buffer.readableBytes()];
    buffer.readBytes(bytes);
    sb.append(" [buffer=").append(new String(bytes));
    sb.append(']');
    buffer.release();

    return sb.toString();
}

public void sendOpenFlowMessageToCore(ZeroMQBaseConnector coreConnector, DataObject msg, short ofVersion, long xId,
        long datapathId, int moduleId) {

    SerializationFactory factory = createSerializationFactory();
    SerializerRegistry registry = new SerializerRegistryImpl();
    registry.init();
    ByteBuf output = UnpooledByteBufAllocator.DEFAULT.buffer();
    factory.setSerializerTable(registry);
    factory.messageToBuffer(ofVersion, output, msg);
    byte[] bytes = new byte[output.readableBytes()];
    output.readBytes(bytes);
    Message message = new Message(NetIPUtils.StubHeaderFromPayload(bytes), bytes);
    message.getHeader().setMessageType(MessageType.OPENFLOW);
    message.getHeader().setDatapathId(datapathId);
    message.getHeader().setModuleId(moduleId);
    message.getHeader().setTransactionId((int) xId);
    coreConnector.SendData(message.toByteRepresentation());
}

private void serializeFlowBody(MultipartReplyBody body, ByteBuf outBuffer, MultipartReplyMessage message) {
    MultipartReplyFlowCase flowCase = (MultipartReplyFlowCase) body;
    MultipartReplyFlow flow = flowCase.getMultipartReplyFlow();
    for (FlowStats flowStats : flow.getFlowStats()) {
        ByteBuf flowStatsBuff = UnpooledByteBufAllocator.DEFAULT.buffer();
        flowStatsBuff.writeShort(EncodeConstants.EMPTY_LENGTH);
        flowStatsBuff.writeByte(new Long(flowStats.getTableId()).byteValue());
        flowStatsBuff.writeZero(FLOW_STATS_PADDING_1);
        OFSerializer<MatchV10> matchSerializer = registry
                .getSerializer(new MessageTypeKey<>(message.getVersion(), MatchV10.class));
        matchSerializer.serialize(flowStats.getMatchV10(), flowStatsBuff);
        flowStatsBuff.writeInt(flowStats.getDurationSec().intValue());
        flowStatsBuff.writeInt(flowStats.getDurationNsec().intValue());
        flowStatsBuff.writeShort(flowStats.getPriority());
        flowStatsBuff.writeShort(flowStats.getIdleTimeout());
        flowStatsBuff.writeShort(flowStats.getHardTimeout());
        flowStatsBuff.writeZero(FLOW_STATS_PADDING_2);
        flowStatsBuff.writeLong(flowStats.getCookie().longValue());
        flowStatsBuff.writeLong(flowStats.getPacketCount().longValue());
        flowStatsBuff.writeLong(flowStats.getByteCount().longValue());
        ListSerializer.serializeList(flowStats.getAction(), ACTION_KEY_MAKER, registry, flowStatsBuff);
        flowStatsBuff.setShort(FLOW_STATS_LENGTH_INDEX, flowStatsBuff.readableBytes());
        outBuffer.writeBytes(flowStatsBuff);
    }
}

private void serializeMeterBody(MultipartReplyBody body, ByteBuf outBuffer) {
    MultipartReplyMeterCase meterCase = (MultipartReplyMeterCase) body;
    MultipartReplyMeter meter = meterCase.getMultipartReplyMeter();
    for (MeterStats meterStats : meter.getMeterStats()) {
        ByteBuf meterStatsBuff = UnpooledByteBufAllocator.DEFAULT.buffer();
        meterStatsBuff.writeInt(meterStats.getMeterId().getValue().intValue());
        meterStatsBuff.writeInt(EncodeConstants.EMPTY_LENGTH);
        meterStatsBuff.writeZero(METER_PADDING);
        meterStatsBuff.writeInt(meterStats.getFlowCount().intValue());
        meterStatsBuff.writeLong(meterStats.getPacketInCount().longValue());
        meterStatsBuff.writeLong(meterStats.getByteInCount().longValue());
        meterStatsBuff.writeInt(meterStats.getDurationSec().intValue());
        meterStatsBuff.writeInt(meterStats.getDurationNsec().intValue());
        for (MeterBandStats meterBandStats : meterStats.getMeterBandStats()) {
            meterStatsBuff.writeLong(meterBandStats.getPacketBandCount().longValue());
            meterStatsBuff.writeLong(meterBandStats.getByteBandCount().longValue());
        }
        meterStatsBuff.setInt(METER_LENGTH_INDEX, meterStatsBuff.readableBytes());
        outBuffer.writeBytes(meterStatsBuff);
    }
}

private void serializeGroupDescBody(MultipartReplyBody body, ByteBuf outBuffer, MultipartReplyMessage message) {
    MultipartReplyGroupDescCase groupDescCase = (MultipartReplyGroupDescCase) body;
    MultipartReplyGroupDesc group = groupDescCase.getMultipartReplyGroupDesc();
    for (GroupDesc groupDesc : group.getGroupDesc()) {
        ByteBuf groupDescBuff = UnpooledByteBufAllocator.DEFAULT.buffer();
        groupDescBuff.writeShort(EncodeConstants.EMPTY_LENGTH);
        groupDescBuff.writeByte(groupDesc.getType().getIntValue());
        groupDescBuff.writeZero(GROUP_DESC_PADDING);
        groupDescBuff.writeInt(groupDesc.getGroupId().getValue().intValue());
        for (BucketsList bucket : groupDesc.getBucketsList()) {
            ByteBuf bucketBuff = UnpooledByteBufAllocator.DEFAULT.buffer();
            bucketBuff.writeShort(EncodeConstants.EMPTY_LENGTH);
            bucketBuff.writeShort(bucket.getWeight());
            bucketBuff.writeInt(bucket.getWatchPort().getValue().intValue());
            bucketBuff.writeInt(bucket.getWatchGroup().intValue());
            bucketBuff.writeZero(BUCKET_PADDING);
            ListSerializer.serializeList(bucket.getAction(),
                    TypeKeyMakerFactory.createActionKeyMaker(message.getVersion()), registry, bucketBuff);
            bucketBuff.setShort(BUCKET_LENGTH_INDEX, bucketBuff.readableBytes());
            groupDescBuff.writeBytes(bucketBuff);
        }
        groupDescBuff.setShort(GROUP_DESC_LENGTH_INDEX, groupDescBuff.readableBytes());
        outBuffer.writeBytes(groupDescBuff);
    }
}

private void serializeGroupBody(MultipartReplyBody body, ByteBuf outBuffer) {
    MultipartReplyGroupCase groupCase = (MultipartReplyGroupCase) body;
    MultipartReplyGroup group = groupCase.getMultipartReplyGroup();
    for (GroupStats groupStats : group.getGroupStats()) {
        ByteBuf groupStatsBuff = UnpooledByteBufAllocator.DEFAULT.buffer();
        groupStatsBuff.writeShort(EncodeConstants.EMPTY_LENGTH);
        groupStatsBuff.writeZero(GROUP_STATS_PADDING_1);
        groupStatsBuff.writeInt(groupStats.getGroupId().getValue().intValue());
        groupStatsBuff.writeInt(groupStats.getRefCount().intValue());
        groupStatsBuff.writeZero(GROUP_STATS_PADDING_2);
        groupStatsBuff.writeLong(groupStats.getPacketCount().longValue());
        groupStatsBuff.writeLong(groupStats.getByteCount().longValue());
        groupStatsBuff.writeInt(groupStats.getDurationSec().intValue());
        groupStatsBuff.writeInt(groupStats.getDurationNsec().intValue());
        for (BucketStats bucketStats : groupStats.getBucketStats()) {
            groupStatsBuff.writeLong(bucketStats.getPacketCount().longValue());
            groupStatsBuff.writeLong(bucketStats.getByteCount().longValue());
        }
        groupStatsBuff.setShort(GROUP_STATS_LENGTH_INDEX, groupStatsBuff.readableBytes());
        outBuffer.writeBytes(groupStatsBuff);
    }
}

@Test
public void testSetAsyncInputWithNullMasks() throws Exception {
    GetAsyncOutputBuilder builder = new GetAsyncOutputBuilder();
    BufferHelper.setupHeader(builder, EncodeConstants.OF13_VERSION_ID);
    builder.setPacketInMask(null);
    builder.setPortStatusMask(null);
    builder.setFlowRemovedMask(null);
    GetAsyncOutput message = builder.build();
    GetAsyncReplyMessageFactory serializer = new GetAsyncReplyMessageFactory();
    SerializerRegistry registry = new NetIdeSerializerRegistryImpl();
    registry.init();
    ByteBuf serializedBuffer = UnpooledByteBufAllocator.DEFAULT.buffer();
    serializer.serialize(message, serializedBuffer);
    BufferHelper.checkHeaderV13(serializedBuffer, MESSAGE_TYPE, 8);
    Assert.assertTrue("Unexpected data", serializedBuffer.readableBytes() == 0);
}

@Test
public void testDescBodySerialize() {
    builder.setType(MultipartType.forValue(0));
    MultipartReplyDescCaseBuilder descCase = new MultipartReplyDescCaseBuilder();
    MultipartReplyDescBuilder desc = new MultipartReplyDescBuilder();
    desc.setMfrDesc("Test");
    desc.setHwDesc("Test");
    desc.setSwDesc("Test");
    desc.setSerialNum("12345");
    desc.setDpDesc("Test");
    descCase.setMultipartReplyDesc(desc.build());
    builder.setMultipartReplyBody(descCase.build());
    message = builder.build();
    OF10StatsReplyMessageFactory serializer = new OF10StatsReplyMessageFactory();
    SerializerRegistry registry = new NetIdeSerializerRegistryImpl();
    registry.init();
    serializer.injectSerializerRegistry(registry);
    ByteBuf serializedBuffer = UnpooledByteBufAllocator.DEFAULT.buffer();
    serializer.serialize(message, serializedBuffer);
    BufferHelper.checkHeaderV10(serializedBuffer, MESSAGE_TYPE, 1068);
    Assert.assertEquals("Wrong type", MultipartType.OFPMPDESC.getIntValue(), serializedBuffer.readShort());
    Assert.assertEquals("Wrong flags", message.getFlags(),
            createMultipartRequestFlags(serializedBuffer.readShort()));
    Assert.assertEquals("Wrong desc body", message.getMultipartReplyBody(), decodeDescBody(serializedBuffer));
}

@Test
public void testAggregateBodySerialize() {
    builder.setType(MultipartType.forValue(2));
    MultipartReplyAggregateCaseBuilder aggregateCase = new MultipartReplyAggregateCaseBuilder();
    MultipartReplyAggregateBuilder aggregate = new MultipartReplyAggregateBuilder();
    aggregate.setPacketCount(BigInteger.valueOf(1234L));
    aggregate.setByteCount(BigInteger.valueOf(1234L));
    aggregate.setFlowCount(1L);
    aggregateCase.setMultipartReplyAggregate(aggregate.build());
    builder.setMultipartReplyBody(aggregateCase.build());
    message = builder.build();
    OF10StatsReplyMessageFactory serializer = new OF10StatsReplyMessageFactory();
    SerializerRegistry registry = new NetIdeSerializerRegistryImpl();
    registry.init();
    serializer.injectSerializerRegistry(registry);
    ByteBuf serializedBuffer = UnpooledByteBufAllocator.DEFAULT.buffer();
    serializer.serialize(message, serializedBuffer);
    BufferHelper.checkHeaderV10(serializedBuffer, MESSAGE_TYPE, 36);
    Assert.assertEquals("Wrong type", MultipartType.OFPMPAGGREGATE.getIntValue(), serializedBuffer.readShort());
    Assert.assertEquals("Wrong flags", message.getFlags(),
            createMultipartRequestFlags(serializedBuffer.readShort()));
    Assert.assertEquals("Wrong Packet count", 1234L, serializedBuffer.readLong());
    Assert.assertEquals("Wrong Byte count", 1234L, serializedBuffer.readLong());
    Assert.assertEquals("Wrong flow count", 1L, serializedBuffer.readInt());
    serializedBuffer.skipBytes(4);
}

@Test
public void testSerialize() {
    GetFeaturesOutputFactory serializer = new GetFeaturesOutputFactory();
    SerializerRegistry registry = new NetIdeSerializerRegistryImpl();
    registry.init();
    serializer.injectSerializerRegistry(registry);
    ByteBuf serializedBuffer = UnpooledByteBufAllocator.DEFAULT.buffer();
    serializer.serialize(message, serializedBuffer);
    BufferHelper.checkHeaderV13(serializedBuffer, MESSAGE_TYPE, 32);
    Assert.assertEquals("Wrong DatapathId", message.getDatapathId().longValue(), serializedBuffer.readLong());
    Assert.assertEquals("Wrong Buffer ID", message.getBuffers().longValue(), serializedBuffer.readInt());
    Assert.assertEquals("Wrong tables", message.getTables().shortValue(), serializedBuffer.readUnsignedByte());
    Assert.assertEquals("Wrong auxiliary ID", message.getAuxiliaryId().shortValue(),
            serializedBuffer.readUnsignedByte());
    serializedBuffer.skipBytes(PADDING);
    Assert.assertEquals("Wrong Capabilities", message.getCapabilities(),
            createCapabilities(serializedBuffer.readInt()));
    Assert.assertEquals("Wrong reserved", message.getReserved().longValue(), serializedBuffer.readInt());
}

@Test
public void testSerialize() {
    OF10PortStatusMessageFactory serializer = new OF10PortStatusMessageFactory();
    ByteBuf serializedBuffer = UnpooledByteBufAllocator.DEFAULT.buffer();
    serializer.serialize(message, serializedBuffer);
    BufferHelper.checkHeaderV10(serializedBuffer, MESSAGE_TYPE, 64);
    Assert.assertEquals("Wrong reason", message.getReason().getIntValue(), serializedBuffer.readUnsignedByte());
    serializedBuffer.skipBytes(7);
    Assert.assertEquals("Wrong port No", message.getPortNo().intValue(), serializedBuffer.readShort());
    byte[] address = new byte[6];
    serializedBuffer.readBytes(address);
    Assert.assertEquals("Wrong MacAddress", message.getHwAddr().getValue().toLowerCase(),
            new MacAddress(ByteBufUtils.macAddressToString(address)).getValue().toLowerCase());
    byte[] name = new byte[16];
    serializedBuffer.readBytes(name);
    Assert.assertEquals("Wrong name", message.getName(), new String(name).trim());
    Assert.assertEquals("Wrong config", message.getConfigV10(), createPortConfig(serializedBuffer.readInt()));
    Assert.assertEquals("Wrong state", message.getStateV10(), createPortState(serializedBuffer.readInt()));
    Assert.assertEquals("Wrong current", message.getCurrentFeaturesV10(),
            createPortFeatures(serializedBuffer.readInt()));
    Assert.assertEquals("Wrong advertised", message.getAdvertisedFeaturesV10(),
            createPortFeatures(serializedBuffer.readInt()));
    Assert.assertEquals("Wrong supported", message.getSupportedFeaturesV10(),
            createPortFeatures(serializedBuffer.readInt()));
    Assert.assertEquals("Wrong peer", message.getPeerFeaturesV10(), createPortFeatures(serializedBuffer.readInt()));
}

@Test
public void testMeterConfigSerialize() {
    builder.setType(MultipartType.forValue(10));
    MultipartReplyMeterConfigCaseBuilder meterConfigCase = new MultipartReplyMeterConfigCaseBuilder();
    MultipartReplyMeterConfigBuilder meterConfigBuilder = new MultipartReplyMeterConfigBuilder();
    meterConfigBuilder.setMeterConfig(createMeterConfig());
    meterConfigCase.setMultipartReplyMeterConfig(meterConfigBuilder.build());
    builder.setMultipartReplyBody(meterConfigCase.build());
    message = builder.build();
    ByteBuf serializedBuffer = UnpooledByteBufAllocator.DEFAULT.buffer();
    serializer.serialize(message, serializedBuffer);
    BufferHelper.checkHeaderV13(serializedBuffer, MESSAGE_TYPE, 48);
    Assert.assertEquals("Wrong type", MultipartType.OFPMPMETERCONFIG.getIntValue(), serializedBuffer.readShort());
    Assert.assertEquals("Wrong flags", message.getFlags(),
            createMultipartRequestFlags(serializedBuffer.readShort()));
    serializedBuffer.skipBytes(PADDING);
    MultipartReplyMeterConfigCase body = (MultipartReplyMeterConfigCase) message.getMultipartReplyBody();
    MultipartReplyMeterConfig messageOutput = body.getMultipartReplyMeterConfig();
    MeterConfig meterConfig = messageOutput.getMeterConfig().get(0);
    Assert.assertEquals("Wrong len", 32, serializedBuffer.readShort());
    Assert.assertEquals("Wrong flags", meterConfig.getFlags(), createMeterFlags(serializedBuffer.readShort()));
    Assert.assertEquals("Wrong meterId", meterConfig.getMeterId().getValue().intValue(),
            serializedBuffer.readInt());
    Assert.assertEquals("Wrong bands", meterConfig.getBands(), decodeBandsList(serializedBuffer));
}

@Test
public void testTableSerialize() {
    builder.setType(MultipartType.forValue(3));
    MultipartReplyTableCaseBuilder tableCase = new MultipartReplyTableCaseBuilder();
    MultipartReplyTableBuilder table = new MultipartReplyTableBuilder();
    table.setTableStats(createTableStats());
    tableCase.setMultipartReplyTable(table.build());
    builder.setMultipartReplyBody(tableCase.build());
    message = builder.build();
    ByteBuf serializedBuffer = UnpooledByteBufAllocator.DEFAULT.buffer();
    serializer.serialize(message, serializedBuffer);
    BufferHelper.checkHeaderV13(serializedBuffer, MESSAGE_TYPE, 40);
    Assert.assertEquals("Wrong type", MultipartType.OFPMPTABLE.getIntValue(), serializedBuffer.readShort());
    Assert.assertEquals("Wrong flags", message.getFlags(),
            createMultipartRequestFlags(serializedBuffer.readShort()));
    serializedBuffer.skipBytes(PADDING);
    MultipartReplyTableCase body = (MultipartReplyTableCase) message.getMultipartReplyBody();
    MultipartReplyTable messageOutput = body.getMultipartReplyTable();
    TableStats tableStats = messageOutput.getTableStats().get(0);
    Assert.assertEquals("Wrong tableId", tableStats.getTableId().shortValue(), serializedBuffer.readUnsignedByte());
    serializedBuffer.skipBytes(3);
    Assert.assertEquals("Wrong active count", tableStats.getActiveCount().longValue(), serializedBuffer.readInt());
    Assert.assertEquals("Wrong lookup count", tableStats.getLookupCount().longValue(), serializedBuffer.readLong());
    Assert.assertEquals("Wrong matched count", tableStats.getMatchedCount().longValue(),
            serializedBuffer.readLong());
}

@Test
public void testAggregateSerialize() {
    builder.setType(MultipartType.forValue(2));
    MultipartReplyAggregateCaseBuilder aggregateCase = new MultipartReplyAggregateCaseBuilder();
    MultipartReplyAggregateBuilder aggregate = new MultipartReplyAggregateBuilder();
    aggregate.setPacketCount(BigInteger.valueOf(1L));
    aggregate.setByteCount(BigInteger.valueOf(1L));
    aggregate.setFlowCount(1L);
    aggregateCase.setMultipartReplyAggregate(aggregate.build());
    builder.setMultipartReplyBody(aggregateCase.build());
    message = builder.build();
    ByteBuf serializedBuffer = UnpooledByteBufAllocator.DEFAULT.buffer();
    serializer.serialize(message, serializedBuffer);
    BufferHelper.checkHeaderV13(serializedBuffer, MESSAGE_TYPE, 40);
    Assert.assertEquals("Wrong type", MultipartType.OFPMPAGGREGATE.getIntValue(), serializedBuffer.readShort());
    Assert.assertEquals("Wrong flags", message.getFlags(),
            createMultipartRequestFlags(serializedBuffer.readShort()));
    serializedBuffer.skipBytes(PADDING);
    MultipartReplyAggregateCase body = (MultipartReplyAggregateCase) message.getMultipartReplyBody();
    MultipartReplyAggregate messageOutput = body.getMultipartReplyAggregate();
    Assert.assertEquals("Wrong Packet count", messageOutput.getPacketCount().longValue(),
            serializedBuffer.readLong());
    Assert.assertEquals("Wrong Byte count", messageOutput.getByteCount().longValue(), serializedBuffer.readLong());
    Assert.assertEquals("Wrong Flow count", messageOutput.getFlowCount().longValue(), serializedBuffer.readInt());
    serializedBuffer.skipBytes(4);
}

@Test
public void testFlowSerialize() {
    builder.setType(MultipartType.forValue(1));
    MultipartReplyFlowCaseBuilder flowCase = new MultipartReplyFlowCaseBuilder();
    MultipartReplyFlowBuilder flow = new MultipartReplyFlowBuilder();
    flow.setFlowStats(createFlowStats());
    flowCase.setMultipartReplyFlow(flow.build());
    builder.setMultipartReplyBody(flowCase.build());
    message = builder.build();
    ByteBuf serializedBuffer = UnpooledByteBufAllocator.DEFAULT.buffer();
    serializer.serialize(message, serializedBuffer);
    BufferHelper.checkHeaderV13(serializedBuffer, MESSAGE_TYPE, 192);
    Assert.assertEquals("Wrong type", MultipartType.OFPMPFLOW.getIntValue(), serializedBuffer.readShort());
    Assert.assertEquals("Wrong flags", message.getFlags(),
            createMultipartRequestFlags(serializedBuffer.readShort()));
    serializedBuffer.skipBytes(PADDING);
    testFlowBody(message.getMultipartReplyBody(), serializedBuffer);
}

@Test
public void testDescSerialize() {
    builder.setType(MultipartType.forValue(0));
    MultipartReplyDescCaseBuilder descCase = new MultipartReplyDescCaseBuilder();
    MultipartReplyDescBuilder desc = new MultipartReplyDescBuilder();
    desc.setMfrDesc("Test");
    desc.setHwDesc("Test");
    desc.setSwDesc("Test");
    desc.setSerialNum("12345");
    desc.setDpDesc("Test");
    descCase.setMultipartReplyDesc(desc.build());
    builder.setMultipartReplyBody(descCase.build());
    message = builder.build();
    ByteBuf serializedBuffer = UnpooledByteBufAllocator.DEFAULT.buffer();
    serializer.serialize(message, serializedBuffer);
    BufferHelper.checkHeaderV13(serializedBuffer, MESSAGE_TYPE, 1072);
    Assert.assertEquals("Wrong type", MultipartType.OFPMPDESC.getIntValue(), serializedBuffer.readShort());
    Assert.assertEquals("Wrong flags", message.getFlags(),
            createMultipartRequestFlags(serializedBuffer.readShort()));
    serializedBuffer.skipBytes(PADDING);
    Assert.assertEquals("Wrong desc body", message.getMultipartReplyBody(), decodeDescBody(serializedBuffer));
}

@Test
public void testSerialize() {
    OF10QueueGetConfigReplyMessageFactory serializer = new OF10QueueGetConfigReplyMessageFactory();
    ByteBuf serializedBuffer = UnpooledByteBufAllocator.DEFAULT.buffer();
    serializer.serialize(message, serializedBuffer);
    BufferHelper.checkHeaderV10(serializedBuffer, MESSAGE_TYPE, 40);
    Assert.assertEquals("Wrong port", message.getPort().getValue().longValue(), serializedBuffer.readShort());
    serializedBuffer.skipBytes(6);
    Assert.assertEquals("Wrong queue Id", message.getQueues().get(0).getQueueId().getValue().longValue(),
            serializedBuffer.readInt());
    Assert.assertEquals("Wrong length", 24, serializedBuffer.readShort());
    serializedBuffer.skipBytes(2);
    List<QueueProperty> properties = message.getQueues().get(0).getQueueProperty();
    Assert.assertEquals("Wrong property", properties.get(0).getProperty().getIntValue(),
            serializedBuffer.readShort());
    Assert.assertEquals("Wrong property length", 16, serializedBuffer.readShort());
    serializedBuffer.skipBytes(4);
    RateQueueProperty rateQueueProperty = properties.get(0).getAugmentation(RateQueueProperty.class);
    Assert.assertEquals("Wrong rate", rateQueueProperty.getRate().intValue(), serializedBuffer.readShort());
    serializedBuffer.skipBytes(6);
}

@Test
public void test() throws Exception {
    GetFeaturesInput expectedMessage = createMessage();
    SerializerRegistry registry = new SerializerRegistryImpl();
    registry.init();
    OFSerializer<GetFeaturesInput> serializer = registry
            .getSerializer(new MessageTypeKey<>(EncodeConstants.OF13_VERSION_ID, GetFeaturesInput.class));
    ByteBuf originalBuffer = UnpooledByteBufAllocator.DEFAULT.buffer();
    serializer.serialize(expectedMessage, originalBuffer);

    // TODO: Skipping first 4 bytes due to the way deserializer is
    // implemented
    // Skipping version, type and length from OF header
    originalBuffer.skipBytes(4);
    GetFeaturesInput deserializedMessage = BufferHelper.deserialize(factory, originalBuffer);
    Assert.assertEquals("Wrong version", expectedMessage.getVersion(), deserializedMessage.getVersion());
    Assert.assertEquals("Wrong XId", expectedMessage.getXid(), deserializedMessage.getXid());
}

private ChannelFuture createTcpBootstrap(final TcpServerPipelineFactory tcpServerPipelineFactory, final int tcpPort) throws InterruptedException {
  log.info("Initializing TCP...");
  ServerBootstrap tcpBootstrap = new ServerBootstrap();
  tcpBootstrap.group(eventLoopGroup);
  tcpBootstrap.channel(NioServerSocketChannel.class);
  tcpBootstrap.childHandler(tcpServerPipelineFactory);
  tcpBootstrap.option(ChannelOption.ALLOCATOR, UnpooledByteBufAllocator.DEFAULT);

  final ChannelFuture channelFuture = tcpBootstrap.bind(tcpPort).addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(final ChannelFuture future) throws Exception {
      throttler.setServerChannel(future.channel());
    }
  });
  log.info("Binding to TCP port {}", tcpPort);
  return channelFuture;
}

public byte[] encode() {
    ByteBuf out = UnpooledByteBufAllocator.DEFAULT.buffer();
    out.writeInt(destInstanceID);// 来源instance
    out.writeInt(srcInstanceID);// 目的instance
    out.writeLong(seqNum);// 相应次数
    out.writeInt(flag);// 标志
    out.writeInt(protocolID);// 协议ID
    userInfo.encode(out);// 用户标识
    return out.readBytes(out.readableBytes()).array();
}

@Test
public void testURIMatchOnly() throws IOException {
    AppConfiguration appConfig = new AppConfiguration(new ConfigLoader(),
            "./src/test/resources/testuserconfig.json");
    appConfig.init();
    RequestKeyGenerator keyGen = new RequestKeyGenerator(appConfig);

    ByteBuf buffer = UnpooledByteBufAllocator.DEFAULT.buffer();
    buffer.writeBytes("{\"fromDate\":1464251112185,\"toDate\":1464337512185}"
            .getBytes());
    DefaultFullHttpRequest req1 = new DefaultFullHttpRequest(
            HttpVersion.HTTP_1_0, HttpMethod.GET,
            "http://test.ebay.com/v1/s1", buffer);

    String key1 = keyGen.generateKey(req1);

    buffer = UnpooledByteBufAllocator.DEFAULT.buffer();
    buffer.writeBytes("{\"fromDate\":1464251113750,\"toDate\":1464337513750}"
            .getBytes());
    DefaultFullHttpRequest req2 = new DefaultFullHttpRequest(
            HttpVersion.HTTP_1_0, HttpMethod.GET,
            "http://test.ebay.com/v1/s1", buffer);

    String key2 = keyGen.generateKey(req2);
    assertEquals(key1, key2);

    HttpHeaders.setContentLength(req2, 758);
    key2 = keyGen.generateKey(req2);
    assertEquals(key1, key2);

    appConfig.put("uriMatchOnly", null);
    keyGen = new RequestKeyGenerator(appConfig);
    key1 = keyGen.generateKey(req1);
    key2 = keyGen.generateKey(req2);

    assertNotEquals(key1, key2);
}

private Bootstrap newBootstrap(Class<? extends Channel> socketChannelClass, EventLoopGroup eventLoopGroup, AsyncHttpClientConfig config) {
    @SuppressWarnings("deprecation")
    Bootstrap bootstrap = new Bootstrap().channel(socketChannelClass).group(eventLoopGroup)//
            // default to PooledByteBufAllocator
            .option(ChannelOption.ALLOCATOR, config.isUsePooledMemory() ? PooledByteBufAllocator.DEFAULT : UnpooledByteBufAllocator.DEFAULT)//
            .option(ChannelOption.TCP_NODELAY, config.isTcpNoDelay())//
            .option(ChannelOption.SO_REUSEADDR, config.isSoReuseAddress())//
            .option(ChannelOption.AUTO_CLOSE, false);

    if (config.getConnectTimeout() > 0) {
        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, config.getConnectTimeout());
    }

    if (config.getSoLinger() >= 0) {
        bootstrap.option(ChannelOption.SO_LINGER, config.getSoLinger());
    }

    if (config.getSoSndBuf() >= 0) {
        bootstrap.option(ChannelOption.SO_SNDBUF, config.getSoSndBuf());
    }

    if (config.getSoRcvBuf() >= 0) {
        bootstrap.option(ChannelOption.SO_RCVBUF, config.getSoRcvBuf());
    }

    for (Entry<ChannelOption<Object>, Object> entry : config.getChannelOptions().entrySet()) {
        bootstrap.option(entry.getKey(), entry.getValue());
    }

    return bootstrap;
}

/**
 * Serialize this Object to a String which can be converted to a QR Code
 */
public String toDataString() {
    final ByteBuf byteBuf = UnpooledByteBufAllocator.DEFAULT.heapBuffer(DATA_LENGTH, DATA_LENGTH);
    byteBuf.writeBytes(address.getAddress());
    byteBuf.writeShort(port);
    byteBuf.writeBytes(id.getIDBytes());
    byteBuf.writeBytes(encodeToken(token).getBytes());
    return encode(byteBuf).toString(Charsets.US_ASCII);
}

public TestCustomTunnel() {
  buf1 = UnpooledByteBufAllocator.DEFAULT.buffer(1024);
  Random r = new Random();
  this.expected = new byte[1024];
  r.nextBytes(expected);
  buf1.writeBytes(expected);
}

@Override
protected ByteBuf getWriteByteBuf() {

    int length = payload.length;
    CompositeByteBuf result = new CompositeByteBuf(UnpooledByteBufAllocator.DEFAULT, false, payload.length + 1);
    String prefix = String.format("%c%d\r\n", ASTERISK_BYTE, length);
    result.addComponent(Unpooled.wrappedBuffer(prefix.getBytes()));
    for(Object o : payload){
        ByteBuf buff = ParserManager.parse(o);
        result.addComponent(buff);
    }
    result.setIndex(0, result.capacity());
    return result;
}

@Override protected HttpClient newClient(int port) {
    return Exceptions.uncheck(() -> new ZipkinHttpClientImpl( HttpClient.of(s -> s
        .poolSize(0)
        .byteBufAllocator(UnpooledByteBufAllocator.DEFAULT)
        .maxContentLength(ServerConfig.DEFAULT_MAX_CONTENT_LENGTH)
    ), httpTracing));
}

public static ByteBuf allocateUnpooledBuffer() {
    if (direct) {
        return UnpooledByteBufAllocator.DEFAULT.directBuffer();
    } else {
        return UnpooledByteBufAllocator.DEFAULT.heapBuffer();
    }
}