NettyMessageProcessorTest.java 文件源码

/**
 * Does the post test by sending the request and content to {@link NettyMessageProcessor} through an
 * {@link EmbeddedChannel} and returns the data stored in the {@link InMemoryRouter} as a result of the post.
 * @param postRequest the POST request as a {@link HttpRequest}.
 * @param contentToSend the content to be sent as a part of the POST.
 * @return the data stored in the {@link InMemoryRouter} as a result of the POST.
 * @throws InterruptedException
 */
private ByteBuffer doPostTest(HttpRequest postRequest, List<ByteBuffer> contentToSend) throws InterruptedException {
  EmbeddedChannel channel = createChannel();

  // POST
  notificationSystem.reset();
  postRequest.headers().set(RestUtils.Headers.AMBRY_CONTENT_TYPE, "application/octet-stream");
  HttpUtil.setKeepAlive(postRequest, false);
  channel.writeInbound(postRequest);
  if (contentToSend != null) {
    for (ByteBuffer content : contentToSend) {
      channel.writeInbound(new DefaultHttpContent(Unpooled.wrappedBuffer(content)));
    }
    channel.writeInbound(LastHttpContent.EMPTY_LAST_CONTENT);
  }
  if (!notificationSystem.operationCompleted.await(100, TimeUnit.MILLISECONDS)) {
    fail("Post did not succeed after 100ms. There is an error or timeout needs to increase");
  }
  assertNotNull("Blob id operated on cannot be null", notificationSystem.blobIdOperatedOn);
  return router.getActiveBlobs().get(notificationSystem.blobIdOperatedOn).getBlob();
}