Java 类io.netty.handler.codec.http.DefaultHttpResponse 实例源码

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable e)
        throws Exception
{
    // handle the case of to big requests.
    if (e.getCause() instanceof TooLongFrameException)
    {
        DefaultHttpResponse response = new DefaultHttpResponse(HTTP_1_1, REQUEST_ENTITY_TOO_LARGE);
        ctx.write(response).addListener(ChannelFutureListener.CLOSE);
    }
    else
    {
        if(ctx.channel().isActive()){ // 连接已断开就不打印了
            logger.warn("Exception caught by request handler", e);
        }
        ctx.close();
    }
}

private void onCreate(ChannelHandlerContext ctx)
  throws IOException, URISyntaxException {
  writeContinueHeader(ctx);

  final String nnId = params.namenodeId();
  final int bufferSize = params.bufferSize();
  final short replication = params.replication();
  final long blockSize = params.blockSize();
  final FsPermission permission = params.permission();

  EnumSet<CreateFlag> flags = params.overwrite() ?
    EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE)
      : EnumSet.of(CreateFlag.CREATE);

  final DFSClient dfsClient = newDfsClient(nnId, confForCreate);
  OutputStream out = dfsClient.createWrappedOutputStream(dfsClient.create(
    path, permission, flags, replication,
    blockSize, null, bufferSize, null), null);
  DefaultHttpResponse resp = new DefaultHttpResponse(HTTP_1_1, CREATED);

  final URI uri = new URI(HDFS_URI_SCHEME, nnId, path, null, null);
  resp.headers().set(LOCATION, uri.toString());
  resp.headers().set(CONTENT_LENGTH, 0);
  ctx.pipeline().replace(this, HdfsWriter.class.getSimpleName(),
    new HdfsWriter(dfsClient, out, resp));
}

private void onCreate(ChannelHandlerContext ctx)
  throws IOException, URISyntaxException {
  writeContinueHeader(ctx);

  final String nnId = params.namenodeId();
  final int bufferSize = params.bufferSize();
  final short replication = params.replication();
  final long blockSize = params.blockSize();
  final FsPermission permission = params.permission();

  EnumSet<CreateFlag> flags = params.overwrite() ?
    EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE)
      : EnumSet.of(CreateFlag.CREATE);

  final DFSClient dfsClient = newDfsClient(nnId, confForCreate);
  OutputStream out = dfsClient.createWrappedOutputStream(dfsClient.create(
    path, permission, flags, replication,
    blockSize, null, bufferSize, null), null);
  DefaultHttpResponse resp = new DefaultHttpResponse(HTTP_1_1, CREATED);

  final URI uri = new URI(HDFS_URI_SCHEME, nnId, path, null, null);
  resp.headers().set(LOCATION, uri.toString());
  resp.headers().set(CONTENT_LENGTH, 0);
  ctx.pipeline().replace(this, HdfsWriter.class.getSimpleName(),
    new HdfsWriter(dfsClient, out, resp));
}

@Override
public void writeResponseHead(Response restletResponse) throws IOException {
    setNettyResponse(new DefaultHttpResponse(HTTP_1_1, new HttpResponseStatus(getStatusCode(), getReasonPhrase())));
    HttpHeaders headers = getNettyResponse().headers();

    // this.response.clear();
    for (Header header : getResponseHeaders()) {
        headers.add(header.getName(), header.getValue());
    }

    // Decide whether to close the connection or not.
    if (isKeepAlive()) {
        headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
        getNettyChannel().write(getNettyResponse());
    } else {
        getNettyChannel().writeAndFlush(getNettyResponse()).addListener(ChannelFutureListener.CLOSE);
    }
}

private void startProxyServerWithFilterAnsweringStatusCode(int statusCode) {
    final HttpResponseStatus status = HttpResponseStatus.valueOf(statusCode);
    HttpFiltersSource filtersSource = new HttpFiltersSourceAdapter() {
        @Override
        public HttpFilters filterRequest(HttpRequest originalRequest) {
            return new HttpFiltersAdapter(originalRequest) {
                @Override
                public HttpResponse clientToProxyRequest(HttpObject httpObject) {
                    return new DefaultHttpResponse(HttpVersion.HTTP_1_1, status);
                }
            };
        }
    };

    proxyServer = DefaultHttpProxyServer.bootstrap()
            .withPort(0)
            .withFiltersSource(filtersSource)
            .start();
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    Attribute<HttpServerChannelHandler> attr = ctx.attr(SERVER_HANDLER_KEY);
    HttpServerChannelHandler handler = attr.get();
    if (handler != null) {
        handler.exceptionCaught(ctx, cause);
    } else {
        if (cause instanceof ClosedChannelException) {
            // The channel is closed so we do nothing here
            LOG.debug("Channel already closed. Ignoring this exception.");
            return;
        } else {
            // we cannot throw the exception here
            LOG.warn("HttpServerChannelHandler is not found as attachment to handle exception, send 404 back to the client.", cause);
            // Now we just send 404 back to the client
            HttpResponse response = new DefaultHttpResponse(HTTP_1_1, NOT_FOUND);
            response.headers().set(Exchange.CONTENT_TYPE, "text/plain");
            response.headers().set(Exchange.CONTENT_LENGTH, 0);
            ctx.writeAndFlush(response);
            ctx.close();
        }
    }
}

@Test
public void testMesosStreamIdIsSavedForSuccessfulSubscribeCall() throws Exception {
    final AtomicReference<String> mesosStreamId = new AtomicReference<>(null);

    final Func1<HttpClientResponse<ByteBuf>, Observable<ByteBuf>> f = MesosClient.verifyResponseOk(
        "Subscribe",
        mesosStreamId,
        StringMessageCodec.UTF8_STRING.mediaType()
    );

    final DefaultHttpResponse nettyResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
    nettyResponse.headers().add("Mesos-Stream-Id", "streamId");
    nettyResponse.headers().add("Content-Type", StringMessageCodec.UTF8_STRING.mediaType());
    final HttpClientResponse<ByteBuf> response = new HttpClientResponse<>(
        nettyResponse,
        UnicastContentSubject.create(1000, TimeUnit.MILLISECONDS)
    );

    f.call(response);


    assertThat(mesosStreamId.get()).isEqualTo("streamId");
}

@Test
public void testMesosStreamIdIsNotSavedForUnsuccessfulSubscribeCall() throws Exception {
    final AtomicReference<String> mesosStreamId = new AtomicReference<>(null);

    final Func1<HttpClientResponse<ByteBuf>, Observable<ByteBuf>> f = MesosClient.verifyResponseOk(
        "Subscribe",
        mesosStreamId,
        StringMessageCodec.UTF8_STRING.mediaType()
    );

    final DefaultHttpResponse nettyResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST);
    nettyResponse.headers().add("Mesos-Stream-Id", "streamId");
    nettyResponse.headers().add("Content-Type", StringMessageCodec.UTF8_STRING.mediaType());
    final HttpClientResponse<ByteBuf> response = new HttpClientResponse<>(
        nettyResponse,
        UnicastContentSubject.create(1000, TimeUnit.MILLISECONDS)
    );

    try {
        f.call(response);
    } catch (Mesos4xxException e) {
        // expected
    }

    assertThat(mesosStreamId.get()).isEqualTo(null);
}

@Test
public void testVerifyResponseOk_ensuresContentTypeOfResponseMatchesReceiveCodec() throws Exception {
    final Func1<HttpClientResponse<ByteBuf>, Observable<ByteBuf>> f = MesosClient.verifyResponseOk(
        "Subscribe",
        new AtomicReference<>(),
        StringMessageCodec.UTF8_STRING.mediaType()
    );

    final DefaultHttpResponse nettyResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
    nettyResponse.headers().add("Content-Type", "text/html");
    final HttpClientResponse<ByteBuf> response = new HttpClientResponse<>(
        nettyResponse,
        UnicastContentSubject.create(1000, TimeUnit.MILLISECONDS)
    );

    try {
        f.call(response);
    } catch (MesosException e) {
        final String expected = String.format(
            "Response had Content-Type \"%s\" expected \"%s\"",
            "text/html",
            StringMessageCodec.UTF8_STRING.mediaType()
        );
        assertThat(e.getContext().getMessage()).isEqualTo(expected);
    }
}

/**
 * Create a CarbonMessage for a specific status code.
 *
 * @param status HTTP status code
 * @param msg message text
 * @return CarbonMessage representing the status
 */
public static HTTPCarbonMessage createTextResponse(int status, String msg) {
    HTTPCarbonMessage response = new HTTPCarbonMessage(
            new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.valueOf(status)));
    response.setProperty(Constants.HTTP_STATUS_CODE, status);
    if (msg != null) {
        response.setHeader(HttpHeaders.CONTENT_LENGTH, String.valueOf(msg.length()));
        byte[] msgArray = null;
        try {
            msgArray = msg.getBytes("UTF-8");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException("Failed to get the byte array from responseValue", e);
        }
        ByteBuffer byteBuffer = ByteBuffer.allocate(msgArray.length);
        byteBuffer.put(msgArray);
        byteBuffer.flip();
        response.addHttpContent(new DefaultLastHttpContent(Unpooled.wrappedBuffer(byteBuffer)));
    } else {
        response.setHeader(HttpHeaders.CONTENT_LENGTH, "0");
    }
    return response;
}

@Override
public void onMessage(HTTPCarbonMessage httpRequestMessage) {
    executor.execute(() -> {
        HTTPCarbonMessage cMsg = new HTTPCarbonMessage(new DefaultHttpResponse(HttpVersion.HTTP_1_1,
                HttpResponseStatus.OK));
        cMsg.setHeader(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.KEEP_ALIVE);
        cMsg.setHeader(HttpHeaders.Names.TRANSFER_ENCODING, HttpHeaders.Values.CHUNKED);
        cMsg.setHeader(HttpHeaders.Names.CONTENT_TYPE, Constants.TEXT_PLAIN);
        cMsg.setProperty(Constants.HTTP_STATUS_CODE, 200);
        try {
            httpRequestMessage.respond(cMsg);
        } catch (ServerConnectorException e) {
            logger.error("Error occurred during message notification: " + e.getMessage());
        }
        while (true) {
            HttpContent httpContent = httpRequestMessage.getHttpContent();
            cMsg.addHttpContent(httpContent);
            if (httpContent instanceof LastHttpContent) {
                cMsg.addHttpContent(new DefaultLastHttpContent());
                httpRequestMessage.release();
                break;
            }
        }
    });
}

public static void beginHTTPResponse(ChannelHandlerContext ctx, FullHttpRequest request, long lastModified,
        String path, long fileLength) {
    HttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK);
    HttpUtil.setContentLength(response, fileLength);
    setContentTypeHeader(response, path);

    response.headers().set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);

    setDateAndCacheHeaders(response, lastModified);
    if (HttpUtil.isKeepAlive(request)) {
        response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
    }

    // Write the initial line and the header.
    ctx.write(response);
}

/** Test that the handler correctly supports http error codes i.e. 404 (NOT FOUND). */
@Test
public void httpErrorsAreSupported() throws IOException {
  EmbeddedChannel ch = new EmbeddedChannel(new HttpDownloadHandler(null));
  ByteArrayOutputStream out = Mockito.spy(new ByteArrayOutputStream());
  DownloadCommand cmd = new DownloadCommand(CACHE_URI, true, "abcdef", out);
  ChannelPromise writePromise = ch.newPromise();
  ch.writeOneOutbound(cmd, writePromise);

  HttpResponse response =
      new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.NOT_FOUND);
  response.headers().set(HttpHeaders.CONNECTION, HttpHeaderValues.CLOSE);
  ch.writeInbound(response);
  assertThat(writePromise.isDone()).isTrue();
  assertThat(writePromise.cause()).isInstanceOf(HttpException.class);
  assertThat(((HttpException) writePromise.cause()).status())
      .isEqualTo(HttpResponseStatus.NOT_FOUND);
  // No data should have been written to the OutputStream and it should have been closed.
  assertThat(out.size()).isEqualTo(0);
  verify(out).close();
  assertThat(ch.isOpen()).isFalse();
}

@Test
public void shouldDecodeSuccessfulGetDesignDocumentResponse() throws Exception {
    String response = Resources.read("designdoc_success.json", this.getClass());
    HttpResponse responseHeader = new DefaultHttpResponse(HttpVersion.HTTP_1_1, new HttpResponseStatus(200, "OK"));
    HttpContent responseChunk = new DefaultLastHttpContent(Unpooled.copiedBuffer(response, CharsetUtil.UTF_8));

    GetDesignDocumentRequest requestMock = mock(GetDesignDocumentRequest.class);
    when(requestMock.name()).thenReturn("name");
    when(requestMock.development()).thenReturn(true);
    queue.add(requestMock);
    channel.writeInbound(responseHeader, responseChunk);
    latch.await(1, TimeUnit.SECONDS);
    assertEquals(1, firedEvents.size());
    GetDesignDocumentResponse inbound = (GetDesignDocumentResponse) firedEvents.get(0);

    assertTrue(inbound.status().isSuccess());
    assertEquals("name", inbound.name());
    assertEquals(true, inbound.development());
    assertEquals(response, inbound.content().toString(CharsetUtil.UTF_8));
    ReferenceCountUtil.releaseLater(inbound);
}

@Test
@SuppressWarnings("unchecked")
public void shouldParseErrorWithEmptyRows() throws Exception {
    String response = Resources.read("error_empty_rows.json", this.getClass());
    HttpResponse responseHeader = new DefaultHttpResponse(HttpVersion.HTTP_1_1, new HttpResponseStatus(200, "OK"));
    HttpContent responseChunk1 = new DefaultLastHttpContent(Unpooled.copiedBuffer(response, CharsetUtil.UTF_8));

    ViewQueryRequest requestMock = mock(ViewQueryRequest.class);
    queue.add(requestMock);
    channel.writeInbound(responseHeader, responseChunk1);
    latch.await(1, TimeUnit.SECONDS);
    assertEquals(1, firedEvents.size());
    ViewQueryResponse inbound = (ViewQueryResponse) firedEvents.get(0);
    assertTrue(inbound.status().isSuccess());

    assertEquals(0, countAndRelease(inbound.rows()));

    String error = inbound.error().toBlocking().single();
    Map<String, Object> parsed = mapper.readValue(error, Map.class);
    assertEquals(1, parsed.size());
    assertNotNull(parsed.get("errors"));
}

@Test
@SuppressWarnings("unchecked")
public void shouldParseErrorAfterRows() throws Exception {
    String response = Resources.read("error_rows.json", this.getClass());
    HttpResponse responseHeader = new DefaultHttpResponse(HttpVersion.HTTP_1_1, new HttpResponseStatus(200, "OK"));
    HttpContent responseChunk1 = new DefaultLastHttpContent(Unpooled.copiedBuffer(response, CharsetUtil.UTF_8));

    ViewQueryRequest requestMock = mock(ViewQueryRequest.class);
    queue.add(requestMock);
    channel.writeInbound(responseHeader, responseChunk1);
    latch.await(1, TimeUnit.SECONDS);
    assertEquals(1, firedEvents.size());
    ViewQueryResponse inbound = (ViewQueryResponse) firedEvents.get(0);
    assertTrue(inbound.status().isSuccess());

    assertEquals(10, countAndRelease(inbound.rows()));

    String error = inbound.error().toBlocking().single();
    Map<String, Object> parsed = mapper.readValue(error, Map.class);
    assertEquals(1, parsed.size());
    assertNotNull(parsed.get("errors"));
}

@Test
public void shouldParseErrorWithDesignNotFound() throws Exception {
    String response = Resources.read("designdoc_notfound.json", this.getClass());
    HttpResponse responseHeader = new DefaultHttpResponse(HttpVersion.HTTP_1_1, new HttpResponseStatus(404, "Object Not Found"));
    HttpContent responseChunk1 = new DefaultLastHttpContent(Unpooled.copiedBuffer(response, CharsetUtil.UTF_8));

    ViewQueryRequest requestMock = mock(ViewQueryRequest.class);
    queue.add(requestMock);
    channel.writeInbound(responseHeader, responseChunk1);
    latch.await(1, TimeUnit.SECONDS);
    assertEquals(1, firedEvents.size());
    ViewQueryResponse inbound = (ViewQueryResponse) firedEvents.get(0);
    assertFalse(inbound.status().isSuccess());
    assertEquals(ResponseStatus.NOT_EXISTS, inbound.status());

    assertEquals(0, countAndRelease(inbound.rows()));

    String error = inbound.error().toBlocking().single();
    assertEquals("{\"errors\":[{\"error\":\"not_found\",\"reason\":\"Design document _design/designdoc not found\"}]}", error);

}

@Test
public void shouldDecodeSuccessBucketConfigResponse() throws Exception {
    HttpResponse responseHeader = new DefaultHttpResponse(HttpVersion.HTTP_1_1, new HttpResponseStatus(200, "OK"));
    HttpContent responseChunk1 = new DefaultHttpContent(Unpooled.copiedBuffer("foo", CHARSET));
    HttpContent responseChunk2 = new DefaultLastHttpContent(Unpooled.copiedBuffer("bar", CHARSET));

    BucketConfigRequest requestMock = mock(BucketConfigRequest.class);
    requestQueue.add(requestMock);
    channel.writeInbound(responseHeader, responseChunk1, responseChunk2);
    channel.readInbound();

    assertEquals(1, eventSink.responseEvents().size());
    BucketConfigResponse event = (BucketConfigResponse) eventSink.responseEvents().get(0).getMessage();

    assertEquals(ResponseStatus.SUCCESS, event.status());
    assertEquals("foobar", event.config());
    assertTrue(requestQueue.isEmpty());
}

@Test
public void shouldDecodeAuthFailureBucketConfigResponse() throws Exception {
    HttpResponse responseHeader = new DefaultHttpResponse(HttpVersion.HTTP_1_1,
        new HttpResponseStatus(401, "Unauthorized"));
    HttpContent responseChunk = LastHttpContent.EMPTY_LAST_CONTENT;

    BucketConfigRequest requestMock = mock(BucketConfigRequest.class);
    requestQueue.add(requestMock);
    channel.writeInbound(responseHeader, responseChunk);

    assertEquals(1, eventSink.responseEvents().size());
    BucketConfigResponse event = (BucketConfigResponse) eventSink.responseEvents().get(0).getMessage();

    assertEquals(ResponseStatus.ACCESS_ERROR, event.status());
    assertEquals("Unauthorized", event.config());
    assertTrue(requestQueue.isEmpty());
}

@Test
public void shouldDecodeNotFoundBucketConfigResponse() throws Exception {
    HttpResponse responseHeader = new DefaultHttpResponse(HttpVersion.HTTP_1_1,
        new HttpResponseStatus(404, "Object Not Found"));
    HttpContent responseChunk = new DefaultLastHttpContent(Unpooled.copiedBuffer("Not found.", CharsetUtil.UTF_8));

    BucketConfigRequest requestMock = mock(BucketConfigRequest.class);
    requestQueue.add(requestMock);
    channel.writeInbound(responseHeader, responseChunk);

    assertEquals(1, eventSink.responseEvents().size());
    BucketConfigResponse event = (BucketConfigResponse) eventSink.responseEvents().get(0).getMessage();

    assertEquals(ResponseStatus.NOT_EXISTS, event.status());
    assertEquals("Not found.", event.config());
    assertTrue(requestQueue.isEmpty());
}

@Test
public void shouldDecodeSuccessFlushResponse() throws Exception {
    HttpResponse responseHeader = new DefaultHttpResponse(HttpVersion.HTTP_1_1,
        new HttpResponseStatus(200, "OK"));
    HttpContent responseChunk = LastHttpContent.EMPTY_LAST_CONTENT;

    FlushRequest requestMock = mock(FlushRequest.class);
    requestQueue.add(requestMock);
    channel.writeInbound(responseHeader, responseChunk);

    assertEquals(1, eventSink.responseEvents().size());
    FlushResponse event = (FlushResponse) eventSink.responseEvents().get(0).getMessage();

    assertEquals(ResponseStatus.SUCCESS, event.status());
    assertEquals("OK", event.content());
    assertTrue(requestQueue.isEmpty());
}

@Test
public void shouldDecodeFlushNotEnabledResponse() throws Exception {
    String content = "{\"_\":\"Flush is disabled for the bucket\"}";
    HttpResponse responseHeader = new DefaultHttpResponse(HttpVersion.HTTP_1_1,
        new HttpResponseStatus(400, "Bad Request"));
    HttpContent responseChunk = new DefaultLastHttpContent(Unpooled.copiedBuffer(content, CharsetUtil.UTF_8));

    FlushRequest requestMock = mock(FlushRequest.class);
    requestQueue.add(requestMock);
    channel.writeInbound(responseHeader, responseChunk);

    assertEquals(1, eventSink.responseEvents().size());
    FlushResponse event = (FlushResponse) eventSink.responseEvents().get(0).getMessage();

    assertEquals(ResponseStatus.INVALID_ARGUMENTS, event.status());
    assertEquals("{\"_\":\"Flush is disabled for the bucket\"}", event.content());
    assertTrue(requestQueue.isEmpty());
}

@Test
public void shouldDecodeListDesignDocumentsResponse() throws Exception {
    HttpResponse responseHeader = new DefaultHttpResponse(HttpVersion.HTTP_1_1, new HttpResponseStatus(200, "OK"));
    HttpContent responseChunk1 = new DefaultHttpContent(Unpooled.copiedBuffer("foo", CharsetUtil.UTF_8));
    HttpContent responseChunk2 = new DefaultLastHttpContent(Unpooled.copiedBuffer("bar", CharsetUtil.UTF_8));

    GetDesignDocumentsRequest requestMock = mock(GetDesignDocumentsRequest.class);
    requestQueue.add(requestMock);
    channel.writeInbound(responseHeader, responseChunk1, responseChunk2);

    assertEquals(1, eventSink.responseEvents().size());
    GetDesignDocumentsResponse event = (GetDesignDocumentsResponse) eventSink.responseEvents().get(0).getMessage();

    assertEquals(ResponseStatus.SUCCESS, event.status());
    assertEquals("foobar", event.content());
    assertTrue(requestQueue.isEmpty());
}

@Test
public void shouldDecodeInitialBucketStreamingResponse() throws Exception {
    HttpResponse responseHeader = new DefaultHttpResponse(HttpVersion.HTTP_1_1, new HttpResponseStatus(200, "OK"));

    BucketStreamingRequest requestMock = mock(BucketStreamingRequest.class);
    requestQueue.add(requestMock);
    channel.writeInbound(responseHeader);

    assertEquals(1, eventSink.responseEvents().size());
    BucketStreamingResponse event = (BucketStreamingResponse) eventSink.responseEvents().get(0).getMessage();

    assertEquals(ResponseStatus.SUCCESS, event.status());
    assertNotNull(event.configs());
    assertNotNull(event.host());
    assertEquals(0, requestQueue.size());
    ReferenceCountUtil.releaseLater(event);
    ReferenceCountUtil.releaseLater(responseHeader);
}

@Test
public void shouldDecodeFailingInitialBucketStreamingResponse() throws Exception {
    HttpResponse responseHeader = new DefaultHttpResponse(HttpVersion.HTTP_1_1, new HttpResponseStatus(404, "Object Not Found"));

    BucketStreamingRequest requestMock = mock(BucketStreamingRequest.class);
    requestQueue.add(requestMock);
    channel.writeInbound(responseHeader);

    assertEquals(1, eventSink.responseEvents().size());
    BucketStreamingResponse event = (BucketStreamingResponse) eventSink.responseEvents().get(0).getMessage();

    assertEquals(ResponseStatus.NOT_EXISTS, event.status());
    assertNull(event.configs());
    assertNotNull(event.host());
    assertEquals(0, requestQueue.size());
    ReferenceCountUtil.releaseLater(responseHeader);
    ReferenceCountUtil.releaseLater(event);

}

@Override
public void end() {
  if (ctx != null) {
    Attribute<NettyWebSocket> ws = ctx.channel().attr(NettyWebSocket.KEY);
    if (ws != null && ws.get() != null) {
      status = HttpResponseStatus.SWITCHING_PROTOCOLS;
      ws.get().hankshake();
      ctx = null;
      committed = true;
      return;
    }
    if (!committed) {
      DefaultHttpResponse rsp = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, status);
      headers.set(HttpHeaderNames.CONTENT_LENGTH, 0);
      // dump headers
      rsp.headers().set(headers);
      if (keepAlive) {
        ctx.write(rsp, ctx.voidPromise());
      } else {
        ctx.write(rsp).addListener(CLOSE);
      }
      committed = true;
    }
    ctx = null;
  }
}

@Test
public void testGetCookie() throws Exception {
    DefaultHttpResponse nettyResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.NOT_FOUND);
    String cookie1Name = "PREF";
    String cookie1Value = "ID=a95756377b78e75e:FF=0:TM=1392709628:LM=1392709628:S=a5mOVvTB7DBkexgi";
    String cookie1Domain = ".google.com";
    String cookie1Path = "/";
    String cookie1Header = cookie1Name + '=' + cookie1Value
                           + "; expires=Thu, 18-Feb-2016 07:47:08 GMT; path=" + cookie1Path + "; domain=" + cookie1Domain;
    nettyResponse.headers().add(HttpHeaders.Names.SET_COOKIE, cookie1Header);
    HttpClientResponse<ByteBuf> response = new HttpClientResponse<ByteBuf>(nettyResponse, PublishSubject.<ByteBuf>create());
    Map<String,Set<Cookie>> cookies = response.getCookies();
    Assert.assertNotNull("Cookies are null.", cookies);
    Assert.assertEquals("Cookies are empty.", 1, cookies.size());
    Set<Cookie> cookies1 = cookies.get(cookie1Name);
    Assert.assertNotNull("No cookies found with name: " + cookie1Name, cookies1);
    Assert.assertEquals("Unexpected number of cookies found.", 1, cookies1.size());
    Cookie cookieFound = cookies1.iterator().next();
    Assert.assertEquals("unexpected cookie name.", cookie1Name, cookieFound.getName());
    Assert.assertEquals("unexpected cookie value.", cookie1Value, cookieFound.getValue());
    Assert.assertEquals("unexpected cookie path.", cookie1Path, cookieFound.getPath());
    Assert.assertEquals("unexpected cookie domain.", cookie1Domain, cookieFound.getDomain());
}

@Test
public void testSetCookie() throws Exception {
    DefaultHttpResponse nettyResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.NOT_FOUND);
    HttpServerResponse<ByteBuf> response = new HttpServerResponse<ByteBuf>(new NoOpChannelHandlerContext(),
                                                               nettyResponse);
    String cookieName = "name";
    String cookieValue = "value";
    response.addCookie(new DefaultCookie(cookieName, cookieValue));
    String cookieHeader = nettyResponse.headers().get(HttpHeaders.Names.SET_COOKIE);
    Assert.assertNotNull("Cookie header not found.", cookieHeader);
    Set<Cookie> decode = CookieDecoder.decode(cookieHeader);
    Assert.assertNotNull("Decoded cookie not found.", decode);
    Assert.assertEquals("Unexpected number of decoded cookie not found.", 1, decode.size());
    Cookie cookie = decode.iterator().next();
    Assert.assertEquals("Unexpected cookie name.", cookieName, cookie.getName());
    Assert.assertEquals("Unexpected cookie value.", cookieValue, cookie.getValue());

}

/**
 * Create new HTTP carbon messge.
 *
 * @param isRequest
 * @return
 */
private static HTTPCarbonMessage createHttpCarbonMessage(boolean isRequest) {
    HTTPCarbonMessage httpCarbonMessage;
    if (isRequest) {
        httpCarbonMessage = new HTTPCarbonMessage(
                new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, ""));
        httpCarbonMessage.setEndOfMsgAdded(true);
    } else {
        httpCarbonMessage = new HTTPCarbonMessage(
                new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK));
        httpCarbonMessage.setEndOfMsgAdded(true);
    }
    return httpCarbonMessage;
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
  releaseDfsResources();
  DefaultHttpResponse resp = ExceptionHandler.exceptionCaught(cause);
  resp.headers().set(CONNECTION, CLOSE);
  ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
  LOG.debug("Error ", cause);
  DefaultHttpResponse resp = ExceptionHandler.exceptionCaught(cause);
  resp.headers().set(CONNECTION, CLOSE);
  ctx.writeAndFlush(resp).addListener(ChannelFutureListener.CLOSE);
}

private void onAppend(ChannelHandlerContext ctx) throws IOException {
  writeContinueHeader(ctx);
  final String nnId = params.namenodeId();
  final int bufferSize = params.bufferSize();

  DFSClient dfsClient = newDfsClient(nnId, conf);
  OutputStream out = dfsClient.append(path, bufferSize,
      EnumSet.of(CreateFlag.APPEND), null, null);
  DefaultHttpResponse resp = new DefaultHttpResponse(HTTP_1_1, OK);
  resp.headers().set(CONTENT_LENGTH, 0);
  ctx.pipeline().replace(this, HdfsWriter.class.getSimpleName(),
    new HdfsWriter(dfsClient, out, resp));
}

@Override
public void channelRead0
  (final ChannelHandlerContext ctx, final HttpRequest req) {
  uri = req.getUri();
  final Channel client = ctx.channel();
  Bootstrap proxiedServer = new Bootstrap()
    .group(client.eventLoop())
    .channel(NioSocketChannel.class)
    .handler(new ChannelInitializer<SocketChannel>() {
      @Override
      protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline p = ch.pipeline();
        p.addLast(new HttpRequestEncoder(), new Forwarder(uri, client));
      }
    });
  ChannelFuture f = proxiedServer.connect(host);
  proxiedChannel = f.channel();
  f.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
      if (future.isSuccess()) {
        ctx.channel().pipeline().remove(HttpResponseEncoder.class);
        HttpRequest newReq = new DefaultFullHttpRequest(HTTP_1_1,
          req.getMethod(), req.getUri());
        newReq.headers().add(req.headers());
        newReq.headers().set(CONNECTION, Values.CLOSE);
        future.channel().writeAndFlush(newReq);
      } else {
        DefaultHttpResponse resp = new DefaultHttpResponse(HTTP_1_1,
          INTERNAL_SERVER_ERROR);
        resp.headers().set(CONNECTION, Values.CLOSE);
        LOG.info("Proxy " + uri + " failed. Cause: ", future.cause());
        ctx.writeAndFlush(resp).addListener(ChannelFutureListener.CLOSE);
        client.close();
      }
    }
  });
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
  releaseDfsResources();
  DefaultHttpResponse resp = ExceptionHandler.exceptionCaught(cause);
  resp.headers().set(CONNECTION, CLOSE);
  ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
  LOG.debug("Error ", cause);
  DefaultHttpResponse resp = ExceptionHandler.exceptionCaught(cause);
  resp.headers().set(CONNECTION, CLOSE);
  ctx.writeAndFlush(resp).addListener(ChannelFutureListener.CLOSE);
}

private void onCreate(ChannelHandlerContext ctx)
  throws IOException, URISyntaxException {
  writeContinueHeader(ctx);

  final String nnId = params.namenodeId();
  final int bufferSize = params.bufferSize();
  final short replication = params.replication();
  final long blockSize = params.blockSize();
  final FsPermission permission = params.permission();
  final boolean createParent = params.createParent();

  EnumSet<CreateFlag> flags = params.createFlag();
  if (flags.equals(EMPTY_CREATE_FLAG)) {
    flags = params.overwrite() ?
        EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE)
        : EnumSet.of(CreateFlag.CREATE);
  } else {
    if(params.overwrite()) {
      flags.add(CreateFlag.OVERWRITE);
    }
  }

  final DFSClient dfsClient = newDfsClient(nnId, confForCreate);
  OutputStream out = dfsClient.createWrappedOutputStream(dfsClient.create(
      path, permission, flags, createParent, replication, blockSize, null,
      bufferSize, null), null);

  DefaultHttpResponse resp = new DefaultHttpResponse(HTTP_1_1, CREATED);

  final URI uri = new URI(HDFS_URI_SCHEME, nnId, path, null, null);
  resp.headers().set(LOCATION, uri.toString());
  resp.headers().set(CONTENT_LENGTH, 0);
  ctx.pipeline().replace(this, HdfsWriter.class.getSimpleName(),
    new HdfsWriter(dfsClient, out, resp));
}

private void onAppend(ChannelHandlerContext ctx) throws IOException {
  writeContinueHeader(ctx);
  final String nnId = params.namenodeId();
  final int bufferSize = params.bufferSize();

  DFSClient dfsClient = newDfsClient(nnId, conf);
  OutputStream out = dfsClient.append(path, bufferSize,
      EnumSet.of(CreateFlag.APPEND), null, null);
  DefaultHttpResponse resp = new DefaultHttpResponse(HTTP_1_1, OK);
  resp.headers().set(CONTENT_LENGTH, 0);
  ctx.pipeline().replace(this, HdfsWriter.class.getSimpleName(),
    new HdfsWriter(dfsClient, out, resp));
}

@Override
public void channelRead0
  (final ChannelHandlerContext ctx, final HttpRequest req) {
  uri = req.uri();
  final Channel client = ctx.channel();
  Bootstrap proxiedServer = new Bootstrap()
    .group(client.eventLoop())
    .channel(NioSocketChannel.class)
    .handler(new ChannelInitializer<SocketChannel>() {
      @Override
      protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline p = ch.pipeline();
        p.addLast(new HttpRequestEncoder(), new Forwarder(uri, client));
      }
    });
  ChannelFuture f = proxiedServer.connect(host);
  proxiedChannel = f.channel();
  f.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
      if (future.isSuccess()) {
        ctx.channel().pipeline().remove(HttpResponseEncoder.class);
        HttpRequest newReq = new DefaultFullHttpRequest(HTTP_1_1,
          req.method(), req.uri());
        newReq.headers().add(req.headers());
        newReq.headers().set(CONNECTION, HttpHeaderValues.CLOSE);
        future.channel().writeAndFlush(newReq);
      } else {
        DefaultHttpResponse resp = new DefaultHttpResponse(HTTP_1_1,
          INTERNAL_SERVER_ERROR);
        resp.headers().set(CONNECTION, HttpHeaderValues.CLOSE);
        LOG.info("Proxy " + uri + " failed. Cause: ", future.cause());
        ctx.writeAndFlush(resp).addListener(ChannelFutureListener.CLOSE);
        client.close();
      }
    }
  });
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
  releaseDfsResources();
  DefaultHttpResponse resp = ExceptionHandler.exceptionCaught(cause);
  resp.headers().set(CONNECTION, CLOSE);
  ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
  LOG.debug("Error ", cause);
  DefaultHttpResponse resp = ExceptionHandler.exceptionCaught(cause);
  resp.headers().set(CONNECTION, CLOSE);
  ctx.writeAndFlush(resp).addListener(ChannelFutureListener.CLOSE);
}