@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    // ignore SSLHandshakeException when using a self-signed server certificate
    if (ignoreSslHandshakeErrors && cause.getCause() instanceof SSLHandshakeException) {
        return;
    }
    LOG.error("Unhandled exception in pipeline", cause);
    if (cause instanceof TimelyException) {
        this.sendHttpError(ctx, (TimelyException) cause);
    } else if (null != cause.getCause() && cause.getCause() instanceof TimelyException) {
        this.sendHttpError(ctx, (TimelyException) cause.getCause());
    } else {
        TimelyException e = new TimelyException(HttpResponseStatus.INTERNAL_SERVER_ERROR.code(), cause.getMessage(),
                "");
        this.sendHttpError(ctx, e);
    }
}
 

private void setDefaultModelVersion(
        ChannelHandlerContext ctx, String modelName, String newModelVersion)
        throws ModelNotFoundException, InternalServerException, RequestTimeoutException,
                ModelVersionNotFoundException {
    ModelManager modelManager = ModelManager.getInstance();
    HttpResponseStatus httpResponseStatus =
            modelManager.setDefaultVersion(modelName, newModelVersion);
    if (httpResponseStatus == HttpResponseStatus.NOT_FOUND) {
        throw new ModelNotFoundException("Model not found: " + modelName);
    } else if (httpResponseStatus == HttpResponseStatus.FORBIDDEN) {
        throw new ModelVersionNotFoundException(
                "Model version " + newModelVersion + " does not exist for model " + modelName);
    }
    String msg =
            "Default vesion succsesfully updated for model \""
                    + modelName
                    + "\" to \""
                    + newModelVersion
                    + "\"";
    SnapshotManager.getInstance().saveSnapshot();
    NettyUtils.sendJsonResponse(ctx, new StatusResponse(msg));
}
 

@Test
public void testGaugeEmptyMetricValue() throws IOException {

    String metricName = "gauge.a.b";
    BluefloodGauge gauge = new BluefloodGauge(metricName, null);
    FullHttpRequest request = createIngestRequest(createRequestBody(TENANT,
            new DefaultClockImpl().now().getMillis(), 0, new BluefloodGauge[]{gauge}, null, null, null));

    ArgumentCaptor<FullHttpResponse> argument = ArgumentCaptor.forClass(FullHttpResponse.class);
    handler.handle(context, request);
    verify(channel).write(argument.capture());

    String errorResponseBody = argument.getValue().content().toString(Charset.defaultCharset());
    ErrorResponse errorResponse = getErrorResponse(errorResponseBody);

    assertEquals("Number of errors invalid", 1, errorResponse.getErrors().size());
    assertEquals("Invalid error message", "may not be null", errorResponse.getErrors().get(0).getMessage());
    assertEquals("Invalid source", "gauges[0].value", errorResponse.getErrors().get(0).getSource());
    assertEquals("Invalid tenant", TENANT, errorResponse.getErrors().get(0).getTenantId());
    assertEquals("Invalid metric name", metricName, errorResponse.getErrors().get(0).getMetricName());
    assertEquals("Invalid status", HttpResponseStatus.BAD_REQUEST, argument.getValue().getStatus());
}
 

private void testModelRegisterWithDefaultWorkers(Channel mgmtChannel)
        throws NoSuchFieldException, IllegalAccessException, InterruptedException {
    setConfiguration("default_workers_per_model", "1");
    loadTests(mgmtChannel, "noop-v1.0", "noop_default_model_workers");

    result = null;
    latch = new CountDownLatch(1);
    HttpRequest req =
            new DefaultFullHttpRequest(
                    HttpVersion.HTTP_1_1, HttpMethod.GET, "/models/noop_default_model_workers");
    mgmtChannel.writeAndFlush(req);

    latch.await();
    DescribeModelResponse resp = JsonUtils.GSON.fromJson(result, DescribeModelResponse.class);
    Assert.assertEquals(httpStatus, HttpResponseStatus.OK);
    Assert.assertEquals(resp.getMinWorkers(), 1);
    unloadTests(mgmtChannel, "noop_default_model_workers");
    setConfiguration("default_workers_per_model", "0");
}
 

@Test
public void testLoginBadPassword() throws Exception {
   Login login = new Login();
   login.setUserId(UUID.randomUUID());
   login.setUsername("joe");
   login.setPassword("password");
   
   EasyMock
      .expect(authenticationDao.findLogin("joe"))
      .andReturn(login);
   
   replay();
   
   DefaultFullHttpRequest request = new DefaultFullHttpRequest(
         HttpVersion.HTTP_1_1, 
         HttpMethod.POST, 
         "http://localhost/client",
         Unpooled.wrappedBuffer("{username:\"joe\",password:\"wrong\"}".getBytes("UTF-8"))
   );
   
   FullHttpResponse response = authenticator.authenticateRequest(channel, request);
   assertEquals(HttpResponseStatus.UNAUTHORIZED, response.getStatus());
   assertCookieCleared(response);
   
   verify();
}
 

private List<HttpObject> handleChunkedResponse(int desiredResponseStatusCode, boolean responseShouldBeEmpty) {
    HttpResponse firstChunk = new DefaultHttpResponse(
        HttpVersion.HTTP_1_1,
        HttpResponseStatus.valueOf(desiredResponseStatusCode)
    );

    firstChunk.headers()
            .set(TRANSFER_ENCODING, CHUNKED)
            .set(CONNECTION, HttpHeaders.Values.KEEP_ALIVE)
            .set(SOME_EXPECTED_RESPONSE_HEADER.getKey(), SOME_EXPECTED_RESPONSE_HEADER.getValue());

    List<HttpObject> responseChunks = new ArrayList<>();
    
    responseChunks.add(firstChunk);

    if (!responseShouldBeEmpty) {
        RESPONSE_PAYLOAD_CHUNKS.forEach(chunkData -> responseChunks.add(
            new DefaultHttpContent(Unpooled.wrappedBuffer(chunkData.getBytes(CharsetUtil.UTF_8)))
        ));
    }

    responseChunks.add(LastHttpContent.EMPTY_LAST_CONTENT);

    return responseChunks;
}
 

private void sendResponse(final Task task, HttpResponseStatus status, String contentType, final byte[] response) {

    HttpServerResponse httpResponse = task.context.response().setStatusCode(status.code());

    CacheProfile cacheProfile = task.getCacheProfile();
    if (cacheProfile.browserTTL > 0) {
      httpResponse.putHeader(HttpHeaders.CACHE_CONTROL, "private, max-age=" + (cacheProfile.browserTTL / 1000));
    }

    if (response == null || response.length == 0) {
      httpResponse.end();
    } else if (response.length > getMaxResponseLength(task.context)) {
      sendErrorResponse(task.context, new HttpException(RESPONSE_PAYLOAD_TOO_LARGE, RESPONSE_PAYLOAD_TOO_LARGE_MESSAGE));
    } else {
      httpResponse.putHeader(CONTENT_TYPE, contentType);
      httpResponse.end(Buffer.buffer(response));
    }
  }
 

@Test
public void should_return_expected_response_when_chunked_request_not_exceeding_global_request_size() throws Exception {
    NettyHttpClientRequestBuilder request = request()
        .withMethod(HttpMethod.POST)
        .withUri(BasicEndpoint.MATCHING_PATH)
        .withPaylod(generatePayloadOfSizeInBytes(GLOBAL_MAX_REQUEST_SIZE))
        .withHeader(HttpHeaders.Names.TRANSFER_ENCODING, CHUNKED);

    // when
    NettyHttpClientResponse serverResponse = request.execute(serverConfig.endpointsPort(),
                                                            incompleteCallTimeoutMillis);

    // then
    assertThat(serverResponse.statusCode).isEqualTo(HttpResponseStatus.OK.code());
    assertThat(serverResponse.payload).isEqualTo(BasicEndpoint.RESPONSE_PAYLOAD);
}
 

@Test
public void should_return_expected_response_when_endpoint_disabled_ContentLength_header_above_global_size_validation() {
    ExtractableResponse response =
            given()
                    .baseUri(BASE_URI)
                    .port(serverConfig.endpointsPort())
                    .basePath(BasicEndpointWithRequestSizeValidationDisabled.MATCHING_PATH)
                    .log().all()
                    .body(generatePayloadOfSizeInBytes(GLOBAL_MAX_REQUEST_SIZE + 100))
                    .when()
                    .post()
                    .then()
                    .log().headers()
                    .extract();

    assertThat(response.statusCode()).isEqualTo(HttpResponseStatus.OK.code());
    assertThat(response.asString()).isEqualTo(BasicEndpointWithRequestSizeValidationDisabled.RESPONSE_PAYLOAD);
}
 

private void writeResourceReport(Channel channel) {
  ByteBuf content = Unpooled.buffer();
  Writer writer = new OutputStreamWriter(new ByteBufOutputStream(content), CharsetUtil.UTF_8);
  try {
    reportAdapter.toJson(resourceReport.get(), writer);
    writer.close();
  } catch (IOException e) {
    LOG.error("error writing resource report", e);
    writeAndClose(channel, new DefaultFullHttpResponse(
      HttpVersion.HTTP_1_1, HttpResponseStatus.INTERNAL_SERVER_ERROR,
      Unpooled.copiedBuffer(e.getMessage(), StandardCharsets.UTF_8)));
    return;
  }

  FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, content);
  HttpUtil.setContentLength(response, content.readableBytes());
  response.headers().set(HttpHeaderNames.CONTENT_TYPE, "application/json; charset=UTF-8");
  channel.writeAndFlush(response);
}
 

@Test
public void handleEmptyRequestAndRedirect_get() throws Exception {
  try (final Response resp =
      client.newCall(new Request.Builder().get().url(service.url()).build()).execute()) {
    Assertions.assertThat(resp.code()).isEqualTo(HttpResponseStatus.PERMANENT_REDIRECT.code());
    final String location = resp.header("Location");
    Assertions.assertThat(location).isNotEmpty().isNotNull();
    final HttpUrl redirectUrl = resp.request().url().resolve(location);
    Assertions.assertThat(redirectUrl).isNotNull();
    final Request.Builder redirectBuilder = resp.request().newBuilder();
    redirectBuilder.get();
    // resp.body().close();
    try (final Response redirectResp =
        client.newCall(redirectBuilder.url(redirectUrl).build()).execute()) {
      Assertions.assertThat(redirectResp.code()).isEqualTo(HttpResponseStatus.BAD_REQUEST.code());
    }
  }
}
 

@Test
public void testCompressedFlag() {
  HelloRequest grpcRequest = HelloRequest.newBuilder().setName("myName").build();
  ByteBuf grpcRequestBuffer = bufferFor(grpcRequest, true);
  int streamId = 345;

  SegmentedRequestData segmentedRequest = fullGrpcRequest(grpcRequestBuffer, streamId, true);
  channel.writeInbound(segmentedRequest);

  Response response = channel.readOutbound();
  SegmentedData segmentedData = channel.readOutbound();

  assertEquals(HttpResponseStatus.OK, response.status());
  assertEquals(streamId, response.streamId());
  assertEquals("application/grpc+proto", response.headers().get(HttpHeaderNames.CONTENT_TYPE));

  assertEquals("12", Objects.requireNonNull(segmentedData.trailingHeaders()).get("grpc-status"));
  String actualMessage =
      grpcDecodedString(
          Objects.requireNonNull(
              Objects.requireNonNull(segmentedData.trailingHeaders()).get("grpc-message")));
  assertEquals("compression not supported", actualMessage);
  assertEquals(streamId, segmentedData.streamId());
  assertTrue(segmentedData.endOfMessage());
}
 

public static FullHttpResponse createResponse(HttpResponse homekitResponse) {

    FullHttpResponse response =
        new DefaultFullHttpResponse(
            homekitResponse.getVersion() == HttpResponse.HttpVersion.EVENT_1_0
                ? EVENT_VERSION
                : HttpVersion.HTTP_1_1,
            HttpResponseStatus.valueOf(homekitResponse.getStatusCode()),
            Unpooled.copiedBuffer(homekitResponse.getBody()));
    for (Entry<String, String> header : homekitResponse.getHeaders().entrySet()) {
      response.headers().add(header.getKey(), header.getValue());
    }
    response.headers().set(HttpHeaders.Names.CONTENT_LENGTH, response.content().readableBytes());
    response.headers().set(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.KEEP_ALIVE);
    return response;
  }
 

public ConsumerService subscribeTopic(VertxTestContext context, String groupId, String name, JsonObject... partition) throws InterruptedException, ExecutionException, TimeoutException {
    CompletableFuture<Boolean> subscribe = new CompletableFuture<>();
    // subscribe to a topic
    JsonArray partitions = new JsonArray();
    for (JsonObject p : partition) {
        partitions.add(p);
    }

    JsonObject partitionsRoot = new JsonObject();
    partitionsRoot.put("partitions", partitions);

    postRequest(Urls.consumerInstanceAssignments(groupId, name))
            .putHeader(CONTENT_LENGTH.toString(), String.valueOf(partitionsRoot.toBuffer().length()))
            .putHeader(CONTENT_TYPE.toString(), BridgeContentType.KAFKA_JSON)
            .as(BodyCodec.jsonObject())
            .sendJsonObject(partitionsRoot, ar -> {
                context.verify(() -> {
                    assertThat(ar.succeeded(), is(true));
                    assertThat(ar.result().statusCode(), is(HttpResponseStatus.NO_CONTENT.code()));
                });
                subscribe.complete(true);
            });
    subscribe.get(HTTP_REQUEST_TIMEOUT, TimeUnit.SECONDS);
    return this;
}
 

public Observable<Void> simulateTimeout(HttpServerRequest<ByteBuf> httpRequest, HttpServerResponse<ByteBuf> response) {
    String uri = httpRequest.getUri();
    QueryStringDecoder decoder = new QueryStringDecoder(uri);
    List<String> timeout = decoder.parameters().get("timeout");
    byte[] contentBytes;
    HttpResponseStatus status = HttpResponseStatus.NO_CONTENT;
    if (null != timeout && !timeout.isEmpty()) {
        try {
            Thread.sleep(Integer.parseInt(timeout.get(0)));
            contentBytes = "".getBytes();
        } catch (Exception e) {
            contentBytes = e.getMessage().getBytes();
            status = HttpResponseStatus.INTERNAL_SERVER_ERROR;
        }
    } else {
        status = HttpResponseStatus.BAD_REQUEST;
        contentBytes = "Please provide a timeout parameter.".getBytes();
    }

    response.setStatus(status);
    return response.writeBytesAndFlush(contentBytes);
}
 

@Test
public void testDecodeFullResponseHeaders() throws Exception {
    EmbeddedChannel ch = new EmbeddedChannel(new Http2StreamFrameToHttpObjectCodec(false));
    Http2Headers headers = new DefaultHttp2Headers();
    headers.scheme(HttpScheme.HTTP.name());
    headers.status(HttpResponseStatus.OK.codeAsText());

    assertTrue(ch.writeInbound(new DefaultHttp2HeadersFrame(headers, true)));

    FullHttpResponse response = ch.readInbound();
    try {
        assertThat(response.status(), is(HttpResponseStatus.OK));
        assertThat(response.protocolVersion(), is(HttpVersion.HTTP_1_1));
        assertThat(response.content().readableBytes(), is(0));
        assertTrue(response.trailingHeaders().isEmpty());
        assertFalse(HttpUtil.isTransferEncodingChunked(response));
    } finally {
        response.release();
    }

    assertThat(ch.readInbound(), is(nullValue()));
    assertFalse(ch.finish());
}
 

/**
 * Test that the handler correctly supports http error codes i.e. 404 (NOT FOUND) with a
 * Content-Length header.
 */
@Test
public void httpErrorsWithContentAreSupported() {
  EmbeddedChannel ch = new EmbeddedChannel(new HttpUploadHandler(null, ImmutableList.of()));
  ByteArrayInputStream data = new ByteArrayInputStream(new byte[] {1, 2, 3, 4, 5});
  ChannelPromise writePromise = ch.newPromise();
  ch.writeOneOutbound(new UploadCommand(CACHE_URI, true, "abcdef", data, 5), writePromise);

  HttpRequest request = ch.readOutbound();
  assertThat(request).isInstanceOf(HttpRequest.class);
  HttpChunkedInput content = ch.readOutbound();
  assertThat(content).isInstanceOf(HttpChunkedInput.class);

  ByteBuf errorMsg = ByteBufUtil.writeAscii(ch.alloc(), "error message");
  FullHttpResponse response =
      new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.NOT_FOUND, errorMsg);
  response.headers().set(HttpHeaders.CONNECTION, HttpHeaderValues.KEEP_ALIVE);

  ch.writeInbound(response);

  assertThat(writePromise.isDone()).isTrue();
  assertThat(writePromise.cause()).isInstanceOf(HttpException.class);
  assertThat(((HttpException) writePromise.cause()).response().status())
      .isEqualTo(HttpResponseStatus.NOT_FOUND);
  assertThat(ch.isOpen()).isTrue();
}
 

/**
 * <p>
 * Process server response:
 * </p>
 *
 * <pre>
 * HTTP/1.1 101 Switching Protocols
 * Upgrade: websocket
 * Connection: Upgrade
 * Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
 * Sec-WebSocket-Protocol: chat
 * </pre>
 *
 * @param response
 *            HTTP response returned from the server for the request sent by beginOpeningHandshake00().
 * @throws WebSocketHandshakeException
 */
@Override
protected void verify(FullHttpResponse response) {
    final HttpResponseStatus status = HttpResponseStatus.SWITCHING_PROTOCOLS;
    final HttpHeaders headers = response.headers();

    if (!response.status().equals(status)) {
        throw new WebSocketHandshakeException("Invalid handshake response getStatus: " + response.status());
    }

    CharSequence upgrade = headers.get(HttpHeaderNames.UPGRADE);
    if (!HttpHeaderValues.WEBSOCKET.contentEqualsIgnoreCase(upgrade)) {
        throw new WebSocketHandshakeException("Invalid handshake response upgrade: " + upgrade);
    }

    if (!headers.containsValue(HttpHeaderNames.CONNECTION, HttpHeaderValues.UPGRADE, true)) {
        throw new WebSocketHandshakeException("Invalid handshake response connection: "
                + headers.get(HttpHeaderNames.CONNECTION));
    }

    CharSequence accept = headers.get(HttpHeaderNames.SEC_WEBSOCKET_ACCEPT);
    if (accept == null || !accept.equals(expectedChallengeResponseString)) {
        throw new WebSocketHandshakeException(String.format(
                "Invalid challenge. Actual: %s. Expected: %s", accept, expectedChallengeResponseString));
    }
}
 

@Test(
        alwaysRun = true,
        dependsOnMethods = {"testPredictionsModelNotFound"})
public void testInvalidManagementUri() throws InterruptedException {
    Channel channel = TestUtils.connect(true, configManager);
    Assert.assertNotNull(channel);

    HttpRequest req =
            new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/InvalidUrl");
    channel.writeAndFlush(req).sync();
    channel.closeFuture().sync();

    ErrorResponse resp = JsonUtils.GSON.fromJson(TestUtils.getResult(), ErrorResponse.class);

    Assert.assertEquals(resp.getCode(), HttpResponseStatus.NOT_FOUND.code());
    Assert.assertEquals(resp.getMessage(), ERROR_NOT_FOUND);
}
 

/**
 * Quickreply, send message over event-bus and pass the result directly to rest response
 *
 * @param targetId the target id to send to
 * @param message the message to send
 * @param options the event-bus delivery serverOptions
 */
public void sendAndRespondRequest(String targetId, Object message, DeliveryOptions options) {
  final Vertx vertx = vxmsShared.getVertx();
  vertx
      .eventBus()
      .send(
          targetId,
          message,
          options != null ? options : new DeliveryOptions(),
          event -> {
            final HttpServerResponse response = context.response();
            if (event.failed()) {
              response.setStatusCode(HttpResponseStatus.SERVICE_UNAVAILABLE.code()).end();
            }
            Optional.ofNullable(event.result())
                .ifPresent(
                    result ->
                        Optional.ofNullable(result.body())
                            .ifPresent(resp -> respond(response, resp)));
          });
}
 

@Test
public void testAdvanced_redirect() {
  // Sets up EventLoopGroup with 1 thread to be blocking.
  EventLoopGroup group = new NioEventLoopGroup(1);

  // Sets up embedded channel.
  setup("", makeBootstrap(group), false);
  setupChannel(initialProtocol);

  // Initializes LocalAddress with unique String.
  LocalAddress address = new LocalAddress(TARGET_HOST);

  // stores future
  ChannelFuture future = actionHandler.getFinishedFuture();
  channel.writeOutbound(msg);

  // Path that we test WebWhoisActionHandler uses.
  String path = "/test";

  // Sets up the local server that the handler will be redirected to.
  TestServer.webWhoisServer(group, address, "", TARGET_HOST, path);

  FullHttpResponse response =
      new HttpResponseMessage(
          makeRedirectResponse(
              HttpResponseStatus.MOVED_PERMANENTLY, HTTP_REDIRECT + TARGET_HOST + path, true));

  // checks that future has not been set to successful or a failure
  assertThat(future.isDone()).isFalse();

  channel.writeInbound(response);

  // makes sure old channel is shut down when attempting redirection
  assertThat(channel.isActive()).isFalse();

  // assesses that we successfully received good response and protocol is unchanged
  assertThat(future.syncUninterruptibly().isSuccess()).isTrue();
}
 

@Override
public void handle(final RoutingContext context, final JsonRpcRequest rpcRequest) {
  LOG.debug("Internally responding to {}, id={}", rpcRequest.getMethod(), rpcRequest.getId());
  final JsonRpcBody providedBody = responseBodyProvider.getBody(rpcRequest);

  if (providedBody.hasError()) {
    context.fail(new JsonRpcException(providedBody.error()));
  } else {
    final JsonRpcSuccessResponse result =
        jsonDecoder.decodeValue(providedBody.body(), JsonRpcSuccessResponse.class);
    responder.create(context.request(), HttpResponseStatus.OK.code(), result);
  }
}
 

@Test
public void testOptionsWithRegex() throws Exception {
  router.optionsWithRegex("\\/somepath\\/.*").handler(rc -> rc.response().setStatusMessage("foo").end());
  testRequest(HttpMethod.OPTIONS, "/somepath/whatever", 200, "foo");
  testRequest(HttpMethod.OPTIONS, "/otherpath/whatever", 404, "Not Found");
  testRequest(HttpMethod.GET, "/somepath/whatever", HttpResponseStatus.METHOD_NOT_ALLOWED);
  testRequest(HttpMethod.POST, "/somepath/whatever", HttpResponseStatus.METHOD_NOT_ALLOWED);
  testRequest(HttpMethod.PUT, "/somepath/whatever", HttpResponseStatus.METHOD_NOT_ALLOWED);
  testRequest(HttpMethod.DELETE, "/somepath/whatever", HttpResponseStatus.METHOD_NOT_ALLOWED);
  testRequest(HttpMethod.HEAD, "/somepath/whatever", HttpResponseStatus.METHOD_NOT_ALLOWED);
}
 

public FilteredResponse(HttpRequest originalRequest,
                        String blackList, int matchType,
                        HttpResponseStatus httpResponseStatus){
    super(originalRequest, null);
    this.blackList = blackList;
    this.blackListArr = blackList.split("["+SqliteDBHelper.SPLIT_CHAR+"]");
    this.matchType = matchType;
    this.httpResponseStatus = httpResponseStatus;
}
 

/**
 * response输出
 * @param text
 * @param contentType
 */
public static FullHttpResponse render(String text, String contentType){
	if(text == null){
		text = "";
	}
	ByteBuf byteBuf = Unpooled.wrappedBuffer(text.getBytes());
	FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, byteBuf);
	response.headers().add(CONTENT_TYPE, contentType);
	response.headers().add(CONTENT_LENGTH, String.valueOf(byteBuf.readableBytes()));
	return response;
}
 

@Override
public FullHttpResponse respond(FullHttpRequest req, ChannelHandlerContext ctx) throws Exception {
	// FIXME this should be highly cache-able
	String contents = service.render("apple-app-site-association", templateContext);
	FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
	response.headers().set(HttpHeaders.Names.CONTENT_TYPE, BridgeHeaders.CONTENT_TYPE_JSON_UTF8);
	response.content().writeBytes(contents.getBytes(Charsets.UTF_8));
	return response;
}
 

@Override
public void sendResponse(FullHttpRequest req, ChannelHandlerContext ctx) throws Exception {
   long startTime = System.nanoTime();

   HttpPostRequestDecoder decoder = new HttpPostRequestDecoder(req);
   try {
      String place = null;
      int num = 0;
      while(decoder.hasNext()) {
         num++;

         InterfaceHttpData httpData = decoder.next();
         if(httpData.getHttpDataType() == HttpDataType.Attribute && httpData.getName().equalsIgnoreCase("place")) {
            place = ((Attribute) httpData).getValue();
         } else if(httpData.getHttpDataType() == HttpDataType.FileUpload) {
            String camProtAddr = URLDecoder.decode(httpData.getName(), "UTF-8");
            Device d = findCamera(camProtAddr);
            if(d == null) {
               UPLOAD_UNKNOWN.inc();
               logger.warn("ignoring preview upload for non-existent camera {}", camProtAddr);
               continue;
            }
            write(place, d, (FileUpload) httpData);
         }
      }
      HttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
      ChannelFuture future = ctx.writeAndFlush(response);
      if(!HttpHeaders.isKeepAlive(req)) {
         future.addListener(ChannelFutureListener.CLOSE);
      }

      UPLOAD_NUM.update(num);
      UPLOAD_SUCCESS.update(System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
   } catch (Exception ex) {
      UPLOAD_FAIL.update(System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
   } finally {
      decoder.cleanFiles();
   }
}
 

private HttpResponse createOfflineResponse() {
    ByteBuf buffer = Unpooled.wrappedBuffer("Offline response".getBytes());
    HttpResponse response = new DefaultFullHttpResponse(
            HttpVersion.HTTP_1_1, HttpResponseStatus.OK, buffer);
    HttpHeaders.setContentLength(response, buffer.readableBytes());
    HttpHeaders.setHeader(response, HttpHeaders.Names.CONTENT_TYPE,
            "text/html");
    return response;
}
 

public static FullHttpResponse buildContent(String content, String contentType) {
  FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
  response.headers().set(HttpHeaderNames.CONTENT_TYPE, AsciiString.cached(contentType));
  if (content != null) {
    response.content().writeBytes(content.getBytes(Charset.forName("utf-8")));
  }
  response.headers().set(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
  return response;
}
 

/**
 * HTTP upgrade response will be decode by Netty as 2 objects:
 * - 1 HttpObject corresponding to the 101 SWITCHING PROTOCOL headers
 * - 1 EMPTY_LAST_CONTENT
 *
 * The HTTP upgrade is successful whne the 101 SWITCHING PROTOCOL has been received (handshakeComplete = true)
 * but the latch is count down only when the following EMPTY_LAST_CONTENT is also received.
 * Otherwise this ChannelHandler would be removed too soon and the ActiveMQChannelHandler would handle the
 * EMPTY_LAST_CONTENT (while it is expecitng only ByteBuf).
 */
@Override
public void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
   if (logger.isDebugEnabled()) {
      logger.debug("Received msg=" + msg);
   }
   if (msg instanceof HttpResponse) {
      HttpResponse response = (HttpResponse) msg;
      if (response.status().code() == HttpResponseStatus.SWITCHING_PROTOCOLS.code() && response.headers().get(HttpHeaderNames.UPGRADE).equals(ACTIVEMQ_REMOTING)) {
         String accept = response.headers().get(SEC_ACTIVEMQ_REMOTING_ACCEPT);
         String expectedResponse = createExpectedResponse(MAGIC_NUMBER, ctx.channel().attr(REMOTING_KEY).get());

         if (expectedResponse.equals(accept)) {
            // HTTP upgrade is successful but let's wait to receive the EMPTY_LAST_CONTENT to count down the latch
            handshakeComplete = true;
         } else {
            // HTTP upgrade failed
            ActiveMQClientLogger.LOGGER.httpHandshakeFailed(msg);
            ctx.close();
            latch.countDown();
         }
         return;
      }
   } else if (msg == LastHttpContent.EMPTY_LAST_CONTENT && handshakeComplete) {
      // remove the http handlers and flag the activemq channel handler as active
      pipeline.remove(httpClientCodec);
      pipeline.remove(this);
      ActiveMQChannelHandler channelHandler = pipeline.get(ActiveMQChannelHandler.class);
      channelHandler.active = true;
   }
   if (!handshakeComplete) {
      ActiveMQClientLogger.LOGGER.httpHandshakeFailed(msg);
      ctx.close();
   }
   latch.countDown();
}