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

/**
     * 适配
     */
    @Override
    protected ChannelHandler fixHandlerBeforeConnect(final ChannelHandler handler) {
        ChannelHandler result=new ShareableChannelInboundHandler() {
            @Override
            public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
                Channel ch=ctx.channel();
                ch.pipeline().addLast(new HttpClientCodec());
                ch.pipeline().addLast(new HttpObjectAggregator(64*1024));
                ch.pipeline().addLast(new WebSocketClientProtocolHandler(WebSocketClientHandshakerFactory.newHandshaker(uri, WebSocketVersion.V13, null, false, new DefaultHttpHeaders())));
                ch.pipeline().addLast(new WebSocketConnectedClientHandler(handler));
                ctx.pipeline().remove(this);//移除当前handler
                ctx.pipeline().fireChannelRegistered();//重新从第一个handler抛出事件
            }
        };
//      ChannelInitializer<SocketChannel> result=new ChannelInitializer<SocketChannel>() {
//            @Override
//            protected void initChannel(SocketChannel ch) {
//              ch.pipeline().addLast(new HttpClientCodec());
//              ch.pipeline().addLast(new HttpObjectAggregator(64*1024));
//              ch.pipeline().addLast(new WebSocketClientProtocolHandler(WebSocketClientHandshakerFactory.newHandshaker(uri, WebSocketVersion.V13, null, false, new DefaultHttpHeaders())));
//              ch.pipeline().addLast(new WebSocketConnectedClientHandler(handler));
//            }
//        };
        return result;
    }

/**
     * 适配
     */
    @Override
    protected ChannelHandler fixHandlerBeforeConnect(final ChannelHandler handler) {
        ChannelHandler result=new ShareableChannelInboundHandler() {
            @Override
            public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
                Channel ch=ctx.channel();
                ch.pipeline().addLast(new HttpClientCodec());
                ch.pipeline().addLast(new HttpObjectAggregator(64*1024));
                ch.pipeline().addLast(new WebSocketClientProtocolHandler(WebSocketClientHandshakerFactory.newHandshaker(uri, WebSocketVersion.V13, null, false, new DefaultHttpHeaders())));
                ch.pipeline().addLast(new WebSocketConnectedClientHandler(handler));
                ctx.pipeline().remove(this);//移除当前handler
                ctx.fireChannelRegistered();//重新从第一个handler抛出事件
            }
        };
//      ChannelInitializer<SocketChannel> result=new ChannelInitializer<SocketChannel>() {
//            @Override
//            protected void initChannel(SocketChannel ch) {
//              ch.pipeline().addLast(new HttpClientCodec());
//              ch.pipeline().addLast(new HttpObjectAggregator(64*1024));
//              ch.pipeline().addLast(new WebSocketClientProtocolHandler(WebSocketClientHandshakerFactory.newHandshaker(uri, WebSocketVersion.V13, null, false, new DefaultHttpHeaders())));
//              ch.pipeline().addLast(new WebSocketConnectedClientHandler(handler));
//            }
//        };
        return result;
    }

@Test
public void testConvert() throws IOException {
    Response responseMock = mock(Response.class);

    HttpHeaders headers = new DefaultHttpHeaders();
    headers.add(TEST_KEY_A, TEST_VALUE_B);
    headers.add(TEST_KEY_C, TEST_VALUE_D);

    when(responseMock.getHeaders()).thenReturn(headers);
    when(responseMock.getStatusCode()).thenReturn(STATUS_CODE);
    when(responseMock.getStatusText()).thenReturn(STATUS_TEXT);
    when(responseMock.getResponseBodyAsStream()).thenReturn(
            new ByteArrayInputStream(ENCODED_BODY.getBytes(StandardCharsets.UTF_8))
    );

    BiFunction<Response, Request, HttpResponse> converter = new AsyncResponseConverter();
    HttpResponse awsResponse = converter.apply(responseMock, null);

    assertThat(awsResponse.getHeaders().get(TEST_KEY_A)).isEqualTo(TEST_VALUE_B);
    assertThat(awsResponse.getHeaders().get(TEST_KEY_C)).isEqualTo(TEST_VALUE_D);
    assertThat(awsResponse.getStatusCode()).isEqualTo(STATUS_CODE);
    assertThat(awsResponse.getStatusText()).isEqualTo(STATUS_TEXT);
    assertThat(new BufferedReader(new InputStreamReader(awsResponse.getContent())).readLine())
            .isEqualTo(ENCODED_BODY);
}

@Override
public CompletableFuture<ResponseInfo<byte[]>> execute(RequestInfo<Void> request,
                                                       Executor longRunningTaskExecutor,
                                                       ChannelHandlerContext ctx) {
    /* Redirect requests from '/dashboard' to '/dashboard/'
         Without this redirect, assets would be requested with URI '/asset.ext' instead of '/dashboard/asset.ext'.
         This is important because the '/dashboard' prefix is needed in order to match this endpoint.
    */
    if (StringUtils.endsWith(request.getPath(), DASHBOARD_ENDPOINT_NO_TRAILING_SLASH)) {
        return CompletableFuture.completedFuture(
                ResponseInfo.<byte[]>newBuilder()
                        .withHttpStatusCode(HttpResponseStatus.MOVED_PERMANENTLY.code())
                        .withHeaders(new DefaultHttpHeaders().add(HttpHeaders.LOCATION, DASHBOARD_ENDPOINT))
                        .build());
    } else if (StringUtils.endsWith(request.getPath(), VERSION_FILE_NAME)) {
        String versionJson = String.format(VERSION_RESPONSE_FORMAT, cmsVersion);
        return CompletableFuture.completedFuture(
                ResponseInfo.<byte[]>newBuilder()
                        .withDesiredContentWriterMimeType("application/json")
                        .withContentForFullResponse(versionJson.getBytes(Charset.defaultCharset()))
                        .withHttpStatusCode(HttpResponseStatus.OK.code())
                        .build());
    } else {
        return CompletableFuture.completedFuture(getDashboardAsset(request));
    }
}

@Test
public void test_validateSecureRequestForEndpoint_adds_security_context_to_request() {
    final RequestInfo<Void> requestInfo = mock(RequestInfo.class);
    when(requestInfo.getUri()).thenReturn("https://localhost");
    final HttpHeaders httpHeaders = new DefaultHttpHeaders();
    httpHeaders.add(CmsRequestSecurityValidator.HEADER_X_VAULT_TOKEN, vaultToken);
    when(requestInfo.getHeaders()).thenReturn(httpHeaders);

    final Map<String, String> meta = Maps.newHashMap();
    meta.put(VaultAuthPrincipal.METADATA_KEY_IS_ADMIN, Boolean.TRUE.toString());
    meta.put(VaultAuthPrincipal.METADATA_KEY_USERNAME, "username");
    meta.put(VaultAuthPrincipal.METADATA_KEY_GROUPS, "group1,group2");
    final VaultClientTokenResponse clientTokenResponse = new VaultClientTokenResponse()
            .setId(vaultToken)
            .setMeta(meta);
    when(vaultAdminClient.lookupToken(vaultToken)).thenReturn(clientTokenResponse);

    subject.validateSecureRequestForEndpoint(requestInfo, securedEndpoint);

    verify(requestInfo).addRequestAttribute(eq(SECURITY_CONTEXT_ATTR_KEY), any(SecurityContext.class));
}

@Test
public void execute_throws_api_error_when_bad_auth_header() {
    final RequestInfo<Void> requestInfo = mock(RequestInfo.class);
    final HttpHeaders httpHeaders = new DefaultHttpHeaders();
    httpHeaders.add(HttpHeaders.Names.AUTHORIZATION, invalidAuthorizationHeader);
    when(requestInfo.getHeaders()).thenReturn(httpHeaders);

    try {
        final CompletableFuture<ResponseInfo<AuthResponse>> completableFuture =
                subject.execute(requestInfo, executor, null);
        completableFuture.join();
        fail("Expected exception not thrown.");
    } catch (CompletionException cex) {
        assertThat(cex.getCause()).isInstanceOf(ApiException.class);
    }
}

@Override
protected void initChannel(SocketChannel channel) throws SSLException {
    URI uri = config.getConnectionWebsocketUri();

    DefaultHttpHeaders headers = new DefaultHttpHeaders();
    headers.add(USER_ID_HEADER, config.getConnectionUserId().toString());
    headers.add(USER_PASSWORD_HEADER, config.getConnectionUserPassword());
    headers.add(SUPPLIER_ID_HEADER, config.getConnectionServerId());

    WebSocketClientHandshaker handshaker = WebSocketClientHandshakerFactory.newHandshaker(uri, WS_VERSION, null, false, headers);

    ChannelPipeline pipeline = channel.pipeline();
    if (config.isConnectionSecure()) {
        try {
            SslContext sslContext = SslContext.newClientContext(InsecureTrustManagerFactory.INSTANCE);
            pipeline.addLast(sslContext.newHandler(channel.alloc()));
        } catch (SSLException e) {
            logger.log(Level.SEVERE, "Shutting down client due to unexpected failure to create SSL context", e);
            throw e;
        }
    }
    pipeline.addLast(new HttpClientCodec());
    pipeline.addLast(new HttpObjectAggregator(8192));
    pipeline.addLast(new AudioConnectClientHandler(handshaker));
}

@Test
public void doChannelRead_creates_and_sets_RequestInfo_on_state_and_RequestInfo_is_marked_as_complete_with_all_chunks_if_msg_is_FullHttpRequest() {
    // given
    FullHttpRequest msgMock = mock(FullHttpRequest.class);
    String uri = "/some/url";
    HttpHeaders headers = new DefaultHttpHeaders();
    doReturn(uri).when(msgMock).getUri();
    doReturn(headers).when(msgMock).headers();
    doReturn(headers).when(msgMock).trailingHeaders();
    doReturn(byteBufMock).when(msgMock).content();
    doReturn(false).when(byteBufMock).isReadable();
    doReturn(HttpVersion.HTTP_1_1).when(msgMock).getProtocolVersion();

    // when
    PipelineContinuationBehavior result = handler.doChannelRead(ctxMock, msgMock);

    // then
    ArgumentCaptor<RequestInfo> requestInfoArgumentCaptor = ArgumentCaptor.forClass(RequestInfo.class);
    verify(stateMock).setRequestInfo(requestInfoArgumentCaptor.capture());
    RequestInfo requestInfo = requestInfoArgumentCaptor.getValue();
    assertThat(requestInfo.getUri()).isEqualTo(uri);
    assertThat(requestInfo.isCompleteRequestWithAllChunks()).isTrue();
    assertThat(result).isEqualTo(PipelineContinuationBehavior.CONTINUE);
}

@Before
public void beforeMethod() {
    stateMock = mock(HttpProcessingState.class);
    ctxMock = mock(ChannelHandlerContext.class);
    channelMock = mock(Channel.class);
    stateAttrMock = mock(Attribute.class);
    requestInfoMock = mock(RequestInfo.class);
    endpointMock = mock(StandardEndpoint.class);
    matcherMock = mock(Matcher.class);
    endpoints = new ArrayList<>(Collections.singleton(endpointMock));
    httpHeaders = new DefaultHttpHeaders();
    maxRequestSizeInBytes = 10;
    msg = mock(HttpRequest.class);

    doReturn(channelMock).when(ctxMock).channel();
    doReturn(stateAttrMock).when(channelMock).attr(ChannelAttributes.HTTP_PROCESSING_STATE_ATTRIBUTE_KEY);
    doReturn(stateMock).when(stateAttrMock).get();
    doReturn(endpointMock).when(stateMock).getEndpointForExecution();
    doReturn(matcherMock).when(endpointMock).requestMatcher();
    doReturn(Optional.of(defaultPath)).when(matcherMock).matchesPath(any(RequestInfo.class));
    doReturn(true).when(matcherMock).matchesMethod(any(RequestInfo.class));
    doReturn(requestInfoMock).when(stateMock).getRequestInfo();
    doReturn(httpHeaders).when(msg).headers();

    handlerSpy = spy(new RoutingHandler(endpoints, maxRequestSizeInBytes));
}

@Test
public void getHeaderMapDelegatesToRequestInfoAndCachesResult() {
    Map<String, List<String>> expectedHeaderMap = new TreeMap<>(MapBuilder.<String, List<String>>builder()
                                                                          .put("header1", Arrays.asList("h1val1"))
                                                                          .put("header2", Arrays.asList("h2val1", "h2val2"))
                                                                          .build());

    HttpHeaders nettyHeaders = new DefaultHttpHeaders();
    for (Map.Entry<String, List<String>> headerEntry : expectedHeaderMap.entrySet()) {
        nettyHeaders.add(headerEntry.getKey(), headerEntry.getValue());
    }
    setFieldOnRequestInfo("headers", nettyHeaders);
    Map<String, List<String>> actualHeaderMap = adapter.getHeadersMap();
    assertThat(actualHeaderMap, is(expectedHeaderMap));
    assertThat(adapter.getHeadersMap(), sameInstance(actualHeaderMap));
}

@Test
public void uber_constructor_for_full_response_sets_fields_as_expected() {
    // given
    int httpStatusCode = 200;
    HttpHeaders headers = new DefaultHttpHeaders();
    String mimeType = "text/text";
    Charset contentCharset = CharsetUtil.UTF_8;
    Set<Cookie> cookies = Sets.newHashSet(new DefaultCookie("key1", "val1"), new DefaultCookie("key2", "val2"));
    boolean preventCompressedResponse = true;

    // when
    BaseResponseInfo<?> responseInfo = createNewBaseResponseInfoForTesting(httpStatusCode, headers, mimeType, contentCharset, cookies, preventCompressedResponse);

    // then
    assertThat(responseInfo.getHttpStatusCode(), is(httpStatusCode));
    assertThat(responseInfo.getHeaders(), is(headers));
    assertThat(responseInfo.getDesiredContentWriterMimeType(), is(mimeType));
    assertThat(responseInfo.getDesiredContentWriterEncoding(), is(contentCharset));
    assertThat(responseInfo.getCookies(), is(cookies));
    assertThat(responseInfo.getUncompressedRawContentLength(), nullValue());
    assertThat(responseInfo.isPreventCompressedOutput(), is(preventCompressedResponse));
    assertThat(responseInfo.isResponseSendingStarted(), is(false));
    assertThat(responseInfo.isResponseSendingLastChunkSent(), is(false));
}

@Test
@DataProvider(value = {
        "text/text charset=US-ASCII |   UTF-8   | US-ASCII",
        "text/text charset=us-ascii |   UTF-8   | US-ASCII",
        "text/text                  |   UTF-8   | UTF-8",
        "                           |   UTF-8   | UTF-8",
        "null                       |   UTF-8   | UTF-8",
}, splitBy = "\\|")
public void determineCharsetFromContentType_works(String contentTypeHeader, String defaultCharsetString, String expectedCharsetString) {
    // given
    Charset defaultCharset = Charset.forName(defaultCharsetString);
    Charset expectedCharset = Charset.forName(expectedCharsetString);
    HttpHeaders headers = new DefaultHttpHeaders().add(HttpHeaders.Names.CONTENT_TYPE, String.valueOf(contentTypeHeader));

    // when
    Charset actualCharset = HttpUtils.determineCharsetFromContentType(headers, defaultCharset);

    // then
    assertThat(actualCharset, is(expectedCharset));
}

@Test
public void extractCookies_works_if_cookies_defined_in_headers() {
    // given
    Cookie cookie1 = new DefaultCookie(UUID.randomUUID().toString(), UUID.randomUUID().toString());
    Cookie cookie2 = new DefaultCookie(UUID.randomUUID().toString(), UUID.randomUUID().toString());
    HttpHeaders headers = new DefaultHttpHeaders().add(HttpHeaders.Names.COOKIE, ClientCookieEncoder.LAX.encode(cookie1, cookie2));

    HttpRequest nettyRequestMock = mock(HttpRequest.class);
    doReturn(headers).when(nettyRequestMock).headers();

    // when
    Set<Cookie> extractedCookies = HttpUtils.extractCookies(nettyRequestMock);

    // then
    assertThat(extractedCookies.contains(cookie1), is(true));
    assertThat(extractedCookies.contains(cookie2), is(true));
}

@Test
public void extractCookies_works_if_cookies_defined_in_trailing_headers() {
    // given
    Cookie cookie1 = new DefaultCookie(UUID.randomUUID().toString(), UUID.randomUUID().toString());
    Cookie cookie2 = new DefaultCookie(UUID.randomUUID().toString(), UUID.randomUUID().toString());
    HttpHeaders trailingHeaders = new DefaultHttpHeaders().add(HttpHeaders.Names.COOKIE, ClientCookieEncoder.LAX.encode(cookie1, cookie2));

    FullHttpRequest nettyRequestMock = mock(FullHttpRequest.class);
    doReturn(new DefaultHttpHeaders()).when(nettyRequestMock).headers();
    doReturn(trailingHeaders).when(nettyRequestMock).trailingHeaders();

    // when
    Set<Cookie> extractedCookies = HttpUtils.extractCookies(nettyRequestMock);

    // then
    assertThat(extractedCookies.contains(cookie1), is(true));
    assertThat(extractedCookies.contains(cookie2), is(true));
}

@Test
public void extractCookies_handles_cookie_values_leniently() {
    // given
    //these are cookie values seen in the wild...
    Cookie cookie1 = new DefaultCookie(UUID.randomUUID().toString(), "2094%3Az%7C2021%3Ab");
    Cookie cookie2 = new DefaultCookie(UUID.randomUUID().toString(), "geoloc=cc=US,rc=OR,tp=vhigh,tz=PST,la=45.4978,lo=-122.6937,bw=5000");
    Cookie cookie3 = new DefaultCookie(UUID.randomUUID().toString(), "\"dm=n.com&si=27431295-a282-4745-8cd5-542e7fce" +
            "429e&ss=1477551008358&sl=76&tt=437632&obo=12&sh=1477552753923%3D76%3A12%3A437632%2C1477552698670%3D75%3" +
            "A12%3A429879%2C1477552677137%3D74%3A12%3A426596%2C1477552672564%3D73%3A12%3A425585%2C1477552669893%3D72" +
            "%3A12%3A423456&bcn=%2F%2F3408178b.mpstat.us%2F&ld=1477552753923&r=http%3A%2F%2Fwww.nike.com%2Fbe%2Fde_de%" +
            "2F&ul=1477552756811\"");
    HttpHeaders headers = new DefaultHttpHeaders().add(HttpHeaders.Names.COOKIE, ClientCookieEncoder.LAX.encode(cookie1, cookie2, cookie3));

    HttpRequest nettyRequestMock = mock(HttpRequest.class);
    doReturn(headers).when(nettyRequestMock).headers();

    // when
    Set<Cookie> extractedCookies = HttpUtils.extractCookies(nettyRequestMock);

    // then
    assertThat(extractedCookies.contains(cookie1), is(true));
    assertThat(extractedCookies.contains(cookie2), is(true));
    assertThat(extractedCookies.contains(cookie3), is(true));
}

@Override
public CompletableFuture<ResponseInfo<String>> execute(
    RequestInfo<String> request, Executor longRunningTaskExecutor, ChannelHandlerContext ctx
) {
    if (!EXPECTED_REQUEST_PAYLOAD.equals(request.getContent()))
        throw new ApiException(MISSING_EXPECTED_REQ_PAYLOAD);

    if (!EXPECTED_HEADER_VAL.equals(request.getHeaders().get(EXPECTED_HEADER_KEY)))
        throw new ApiException(MISSING_EXPECTED_HEADER);

    return CompletableFuture.completedFuture(
        ResponseInfo.newBuilder(RESPONSE_PAYLOAD)
                    .withHeaders(
                        new DefaultHttpHeaders()
                            .set(TraceHeaders.TRACE_ID, request.getHeaders().get(TraceHeaders.TRACE_ID))
                            .set(TraceHeaders.PARENT_SPAN_ID,
                                 request.getHeaders().get(TraceHeaders.PARENT_SPAN_ID))
                    )
                    .build()
    );
}

@Test(groups = "standalone")
public void testCookieParseExpires() {
    // e.g. "Tue, 27 Oct 2015 12:54:24 GMT";
    SimpleDateFormat sdf = new SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss z", Locale.US);
    sdf.setTimeZone(TimeZone.getTimeZone("GMT"));

    Date date = new Date(System.currentTimeMillis() + 60000);
    final String cookieDef = String.format("efmembercheck=true; expires=%s; path=/; domain=.eclipse.org", sdf.format(date));

    HttpResponseHeaders responseHeaders = new HttpResponseHeaders(new DefaultHttpHeaders().add(HttpHeaders.Names.SET_COOKIE, cookieDef));
    NettyResponse response = new NettyResponse(new NettyResponseStatus(null, null, null, null), responseHeaders, null);

    List<Cookie> cookies = response.getCookies();
    assertEquals(cookies.size(), 1);

    Cookie cookie = cookies.get(0);
    assertTrue(cookie.getMaxAge() >= 58 && cookie.getMaxAge() <= 60);
}

@Test
public void getWithHeaders() throws Throwable {
    withClient().run(client -> {
        withServer(server).run(server -> {
            HttpHeaders h = new DefaultHttpHeaders();
            for (int i = 1; i < 5; i++) {
                h.add("Test" + i, "Test" + i);
            }

            server.enqueueEcho();

            client.executeRequest(get(getTargetUrl()).setHeaders(h), new AsyncCompletionHandlerAdapter() {

                @Override
                public Response onCompleted(Response response) throws Exception {
                    assertEquals(response.getStatusCode(), 200);
                    for (int i = 1; i < 5; i++) {
                        assertEquals(response.getHeader("X-Test" + i), "Test" + i);
                    }
                    return response;
                }
            }).get(TIMEOUT, SECONDS);
        });
    });
}

public Object[] paramsForFromHttpHeaders() {
  return new Object[][] {
    {new DefaultHttpHeaders(), new Headers(Collections.emptyMap())},
    {
      new DefaultHttpHeaders()
          .set("Rpc-Header-Foo", "ABC")
          .set("Rpc-Header-Bar", "Def")
          .set("Rpc-Procedure", "foo"),
      new Headers(
          ImmutableMap.<String, String>builder().put("Foo", "ABC").put("BAR", "Def").build())
    },
    {
      new DefaultHttpHeaders().set("Rpc-Header-Rpc-Procedure", "foo").set("Rpc-Procedure", "bar"),
      new Headers(ImmutableMap.<String, String>builder().put("Rpc-Procedure", "foo").build())
    }
  };
}

public WebSocketClient(WebSocketConfig config) throws URISyntaxException, SSLException, InterruptedException {
    final int port = config.serverUri.getPort();
    final String scheme = config.serverUri.getScheme().endsWith("s") ? "wss" : "ws";
    final boolean isWss = "wss".equalsIgnoreCase(scheme);

    this.headers = new DefaultHttpHeaders();
    if (config.apiKey != null) headers.add("X-Samebug-ApiKey", config.apiKey);
    if (config.workspaceId != null) headers.add("X-Samebug-WorkspaceId", config.workspaceId);

    this.port = port == -1 ? (isWss ? 443 : 80) : port;
    this.host = config.serverUri.getHost();
    this.wsEndpoint = new URI(scheme, null, host, port, "/sockets/websocket", null, null);
    this.eventHandler = config.eventHandler;
    this.group = config.group;
    this.sslContext = isWss ? SslContextBuilder.forClient().trustManager(InsecureTrustManagerFactory.INSTANCE).build() : null;
    // IMPROVE the constructor blocks the thread with networking!
    this.channel = connect();
}

public WebSocketClient(String host, int port, String path, boolean isSSL) throws Exception {
    super(host, port, new Random());

    String scheme = isSSL ? "wss://" : "ws://";
    URI uri = new URI(scheme + host + ":" + port + path);

    if (isSSL) {
        sslCtx = SslContextBuilder.forClient().sslProvider(SslProvider.JDK).trustManager(InsecureTrustManagerFactory.INSTANCE).build();
    } else {
        sslCtx = null;
    }

    this.handler = new WebSocketClientHandler(
                    WebSocketClientHandshakerFactory.newHandshaker(
                            uri, WebSocketVersion.V13, null, false, new DefaultHttpHeaders()));
}

@Override
public void push(final String method, final String path, final Map<String, Object> headers) {
  ctx.channel().eventLoop().execute(() -> {
    AsciiString streamIdHeader = HttpConversionUtil.ExtensionHeaderNames.STREAM_ID.text();
    Http2Connection connection = encoder.connection();
    int nextStreamId = connection.local().incrementAndGetNextStreamId();
    Http2Headers h2headers = new DefaultHttp2Headers()
        .path(path)
        .method(method)
        .authority(authority)
        .scheme(scheme);
    headers.forEach((n, v) -> h2headers.add(n, v.toString()));
    encoder.writePushPromise(ctx, streamId, nextStreamId, h2headers, 0, ctx.newPromise());

    // TODO: Is there another way of handling a push promise?
    DefaultFullHttpRequest pushRequest = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1,
        HttpMethod.valueOf(method.toUpperCase()), path, Unpooled.EMPTY_BUFFER,
        new DefaultHttpHeaders(false).set(streamIdHeader, nextStreamId),
        EmptyHttpHeaders.INSTANCE);
    ctx.pipeline().fireChannelRead(pushRequest);
    ctx.pipeline().fireChannelReadComplete();
  });
}

@Test
public void noHeaders() throws Exception {
  int bufferSize = 8192;
  boolean keepAlive = true;
  List<String> v = Collections.emptyList();
  new MockUnit(ChannelHandlerContext.class)
      .expect(unit -> {
        DefaultHttpHeaders headers = unit.mockConstructor(DefaultHttpHeaders.class);
        expect(headers.getAll("h")).andReturn(null);
      })
      .run(unit -> {
        assertEquals(v, new NettyResponse(unit.get(ChannelHandlerContext.class), bufferSize,
            keepAlive)
                .headers("h"));
      });
}

@BeforeMethod
public void setup() {
    applicationMapping = mock(ApplicationMapping.class);
    request = mock(FullHttpRequest.class);
    response = mock(FullHttpResponse.class);
    when(request.getMethod()).thenReturn(HttpMethod.GET);
    when(request.getUri()).thenReturn("/test");
    when(request.headers()).thenReturn(new DefaultHttpHeaders());
    when(request.content()).thenReturn(mock(ByteBuf.class));
    when(request.getProtocolVersion()).thenReturn(HttpVersion.HTTP_1_1);
    when(response.getStatus()).thenReturn(HttpResponseStatus.PROCESSING);
    application = mock(Application.class);
    context = mock(MessageContext.class);
    router = spy(new Router(applicationMapping));
    doReturn(new ApplicationHandler()).when(router).getApplicationHandler(application);
    when(application.getPath()).thenReturn(URI.create("/app"));
    when(context.getRequest()).thenReturn(request);
    when(context.getResponse()).thenReturn(response);
    when(context.getApplication()).thenReturn(application);
    when(context.getBaseUri()).thenReturn(URI.create("http://localhost:8080"));
    when(applicationMapping.resolve(request)).thenReturn(application);
}

/**
 * Tests multipart POST and verifies it via GET operations.
 * @throws Exception
 */
@Test
public void multipartPostGetHeadTest() throws Exception {
  Account refAccount = ACCOUNT_SERVICE.createAndAddRandomAccount();
  Container refContainer = refAccount.getContainerById(Container.DEFAULT_PUBLIC_CONTAINER_ID);
  doPostGetHeadDeleteTest(0, refAccount, refContainer, refAccount.getName(), !refContainer.isCacheable(),
      refAccount.getName(), refContainer.getName(), true);
  doPostGetHeadDeleteTest(FRONTEND_CONFIG.frontendChunkedGetResponseThresholdInBytes * 3, refAccount, refContainer,
      refAccount.getName(), !refContainer.isCacheable(), refAccount.getName(), refContainer.getName(), true);

  // failure case
  // size of content being POSTed is higher than what is allowed via multipart/form-data
  long maxAllowedSizeBytes = new NettyConfig(FRONTEND_VERIFIABLE_PROPS).nettyMultipartPostMaxSizeBytes;
  ByteBuffer content = ByteBuffer.wrap(TestUtils.getRandomBytes((int) maxAllowedSizeBytes + 1));
  HttpHeaders headers = new DefaultHttpHeaders();
  setAmbryHeadersForPut(headers, 7200, !refContainer.isCacheable(), refAccount.getName(), "application/octet-stream",
      null, refAccount.getName(), refContainer.getName());
  HttpRequest httpRequest = RestTestUtils.createRequest(HttpMethod.POST, "/", headers);
  HttpPostRequestEncoder encoder = createEncoder(httpRequest, content, ByteBuffer.allocate(0));
  ResponseParts responseParts = nettyClient.sendRequest(encoder.finalizeRequest(), encoder, null).get();
  HttpResponse response = getHttpResponse(responseParts);
  assertEquals("Unexpected response status", HttpResponseStatus.REQUEST_ENTITY_TOO_LARGE, response.status());
  assertTrue("No Date header", response.headers().getTimeMillis(HttpHeaderNames.DATE, -1) != -1);
  assertFalse("Channel should not be active", HttpUtil.isKeepAlive(response));
}

/**
 * Gets the blob with blob ID {@code blobId} and verifies that the blob is not returned as blob is not modified
 * @param blobId the blob ID of the blob to GET.
 * @param getOption the options to use while getting the blob.
 * @param isPrivate {@code true} if the blob is private, {@code false} if not.
 * @throws Exception
 */
private void getNotModifiedBlobAndVerify(String blobId, GetOption getOption, boolean isPrivate) throws Exception {
  HttpHeaders headers = new DefaultHttpHeaders();
  if (getOption != null) {
    headers.add(RestUtils.Headers.GET_OPTION, getOption.toString());
  }
  headers.add(RestUtils.Headers.IF_MODIFIED_SINCE, new Date());
  FullHttpRequest httpRequest = buildRequest(HttpMethod.GET, blobId, headers, null);
  ResponseParts responseParts = nettyClient.sendRequest(httpRequest, null, null).get();
  HttpResponse response = getHttpResponse(responseParts);
  assertEquals("Unexpected response status", HttpResponseStatus.NOT_MODIFIED, response.status());
  assertNotNull("Date header should be set", response.headers().get(RestUtils.Headers.DATE));
  assertNotNull("Last-Modified header should be set", response.headers().get("Last-Modified"));
  assertNull("Content-Length should not be set", response.headers().get(RestUtils.Headers.CONTENT_LENGTH));
  assertNull("Accept-Ranges should not be set", response.headers().get(RestUtils.Headers.ACCEPT_RANGES));
  assertNull("Content-Range header should not be set", response.headers().get(RestUtils.Headers.CONTENT_RANGE));
  assertNull(RestUtils.Headers.BLOB_SIZE + " should have been null ",
      response.headers().get(RestUtils.Headers.BLOB_SIZE));
  assertNull("Content-Type should have been null", response.headers().get(RestUtils.Headers.CONTENT_TYPE));
  verifyCacheHeaders(isPrivate, response);
  assertNoContent(responseParts.queue);
}

/**
 * Gets the user metadata of the blob with blob ID {@code blobId} and verifies them against what is expected.
 * @param blobId the blob ID of the blob to HEAD.
 * @param getOption the options to use while getting the blob.
 * @param expectedHeaders the expected headers in the response.
 * @param usermetadata if non-null, this is expected to come as the body.
 * @throws ExecutionException
 * @throws InterruptedException
 */
private void getUserMetadataAndVerify(String blobId, GetOption getOption, HttpHeaders expectedHeaders,
    byte[] usermetadata) throws ExecutionException, InterruptedException {
  HttpHeaders headers = new DefaultHttpHeaders();
  if (getOption != null) {
    headers.add(RestUtils.Headers.GET_OPTION, getOption.toString());
  }
  FullHttpRequest httpRequest =
      buildRequest(HttpMethod.GET, blobId + "/" + RestUtils.SubResource.UserMetadata, headers, null);
  ResponseParts responseParts = nettyClient.sendRequest(httpRequest, null, null).get();
  HttpResponse response = getHttpResponse(responseParts);
  assertEquals("Unexpected response status", HttpResponseStatus.OK, response.status());
  checkCommonGetHeadHeaders(response.headers());
  verifyUserMetadata(expectedHeaders, response, usermetadata, responseParts.queue);
  assertTrue("Channel should be active", HttpUtil.isKeepAlive(response));
}

/**
 * Gets the blob info of the blob with blob ID {@code blobId} and verifies them against what is expected.
 * @param blobId the blob ID of the blob to HEAD.
 * @param getOption the options to use while getting the blob.
 * @param expectedHeaders the expected headers in the response.
 * @param isPrivate {@code true} if the blob is expected to be private
 * @param accountName the expected account name in the response.
 * @param containerName the expected container name in response.
 * @param usermetadata if non-null, this is expected to come as the body.
 * @throws ExecutionException
 * @throws InterruptedException
 */
private void getBlobInfoAndVerify(String blobId, GetOption getOption, HttpHeaders expectedHeaders, boolean isPrivate,
    String accountName, String containerName, byte[] usermetadata) throws ExecutionException, InterruptedException {
  HttpHeaders headers = new DefaultHttpHeaders();
  if (getOption != null) {
    headers.add(RestUtils.Headers.GET_OPTION, getOption.toString());
  }
  FullHttpRequest httpRequest =
      buildRequest(HttpMethod.GET, blobId + "/" + RestUtils.SubResource.BlobInfo, headers, null);
  ResponseParts responseParts = nettyClient.sendRequest(httpRequest, null, null).get();
  HttpResponse response = getHttpResponse(responseParts);
  assertEquals("Unexpected response status", HttpResponseStatus.OK, response.status());
  checkCommonGetHeadHeaders(response.headers());
  verifyBlobProperties(expectedHeaders, isPrivate, response);
  verifyAccountAndContainerHeaders(accountName, containerName, response);
  verifyUserMetadata(expectedHeaders, response, usermetadata, responseParts.queue);
  assertTrue("Channel should be active", HttpUtil.isKeepAlive(response));
}

/**
 * Verifies that the right response code is returned for GET, HEAD and DELETE once a blob is deleted.
 * @param blobId the ID of the blob that was deleted.
 * @param expectedHeaders the expected headers in the response if the right options are provided.
 * @param isPrivate {@code true} if the blob is expected to be private
 * @param accountName the expected account name in {@code response}.
 * @param containerName the expected container name in {@code response}.
 * @param expectedContent the expected content of the blob if the right options are provided.
 * @param usermetadata if non-null, this is expected to come as the body.
 * @throws Exception
 */
private void verifyOperationsAfterDelete(String blobId, HttpHeaders expectedHeaders, boolean isPrivate,
    String accountName, String containerName, ByteBuffer expectedContent, byte[] usermetadata) throws Exception {
  HttpHeaders headers = new DefaultHttpHeaders().add(RestUtils.Headers.GET_OPTION, GetOption.None.toString());
  FullHttpRequest httpRequest = buildRequest(HttpMethod.GET, blobId, null, null);
  verifyDeleted(httpRequest, HttpResponseStatus.GONE);
  httpRequest = buildRequest(HttpMethod.GET, blobId, headers, null);
  verifyDeleted(httpRequest, HttpResponseStatus.GONE);

  httpRequest = buildRequest(HttpMethod.HEAD, blobId, null, null);
  verifyDeleted(httpRequest, HttpResponseStatus.GONE);
  httpRequest = buildRequest(HttpMethod.HEAD, blobId, headers, null);
  verifyDeleted(httpRequest, HttpResponseStatus.GONE);

  httpRequest = buildRequest(HttpMethod.DELETE, blobId, null, null);
  verifyDeleted(httpRequest, HttpResponseStatus.ACCEPTED);

  GetOption[] options = {GetOption.Include_Deleted_Blobs, GetOption.Include_All};
  for (GetOption option : options) {
    getBlobAndVerify(blobId, null, option, expectedHeaders, isPrivate, expectedContent);
    getNotModifiedBlobAndVerify(blobId, option, isPrivate);
    getUserMetadataAndVerify(blobId, option, expectedHeaders, usermetadata);
    getBlobInfoAndVerify(blobId, option, expectedHeaders, isPrivate, accountName, containerName, usermetadata);
    getHeadAndVerify(blobId, null, option, expectedHeaders, isPrivate, accountName, containerName);
  }
}

/**
 * Asks the server to write more data than the set Content-Length and checks behavior.
 * @param chunkCount the number of chunks of {@link MockNettyMessageProcessor#CHUNK} to use to set Content-Length.
 * @throws Exception
 */
private void doWriteMoreThanContentLengthTest(int chunkCount) throws Exception {
  EmbeddedChannel channel = createEmbeddedChannel();
  MockNettyMessageProcessor processor = channel.pipeline().get(MockNettyMessageProcessor.class);
  HttpHeaders httpHeaders = new DefaultHttpHeaders();
  httpHeaders.set(MockNettyMessageProcessor.CHUNK_COUNT_HEADER_NAME, chunkCount);
  HttpRequest httpRequest =
      RestTestUtils.createRequest(HttpMethod.POST, TestingUri.WriteMoreThanContentLength.toString(), httpHeaders);
  HttpUtil.setKeepAlive(httpRequest, true);
  channel.writeInbound(httpRequest);

  try {
    verifyCallbacks(processor);
    fail("One of the callbacks should have failed because the data written was more than Content-Length");
  } catch (IllegalStateException e) {
    // expected. Nothing to do.
  }

  // It doesn't matter what the response is - because it may either fail or succeed depending on certain race
  // conditions. What matters is that the programming error is caught appropriately by NettyResponseChannel and it
  // makes a callback with the right exception.
  while (channel.readOutbound() != null) {
  }
  channel.close();
}

private void upgradeChannel(final ChannelHandlerContext ctx, FullHttpRequest httpRequest) {
    handshaker = new WebSocketServerHandshakerFactory(
        "ws://" + httpRequest.headers().get(HOST) + UPGRADE_CHANNEL_FOR_UI_WEB_SOCKET_URI,
        null,
        true,
        Integer.MAX_VALUE
    ).newHandshaker(httpRequest);
    if (handshaker == null) {
        WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
    } else {
        handshaker.handshake(
            ctx.channel(),
            httpRequest,
            new DefaultHttpHeaders(),
            ctx.channel().newPromise()
        ).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                clientRegistry.put(ctx, request());
            }
        });
    }
}

private void upgradeChannel(final ChannelHandlerContext ctx, FullHttpRequest httpRequest) {
    handshaker = new WebSocketServerHandshakerFactory(
        "ws://" + httpRequest.headers().get(HOST) + UPGRADE_CHANNEL_FOR_CALLBACK_WEB_SOCKET_URI,
        null,
        true,
        Integer.MAX_VALUE
    ).newHandshaker(httpRequest);
    if (handshaker == null) {
        WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
    } else {
        final String clientId = UUID.randomUUID().toString();
        handshaker.handshake(
            ctx.channel(),
            httpRequest,
            new DefaultHttpHeaders().add("X-CLIENT-REGISTRATION-ID", clientId),
            ctx.channel().newPromise()
        ).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                webSocketClientRegistry.registerClient(clientId, ctx);
                // TODO(jamesdbloom) remove mockserver codec and handler
            }
        });
    }
}

WebSocketClientHandler(
    URI uri, String userAgent, WebsocketConnection.WSClientEventHandler delegate) {
  this.delegate = checkNotNull(delegate, "delegate must not be null");
  checkArgument(!Strings.isNullOrEmpty(userAgent), "user agent must not be null or empty");
  this.handshaker = WebSocketClientHandshakerFactory.newHandshaker(
      uri, WebSocketVersion.V13, null, true,
      new DefaultHttpHeaders().add("User-Agent", userAgent));
}

/**
 * 通道注册的时候配置websocket解码handler
 */
@Override
protected final void initChannel(Channel ch) throws Exception {
    ChannelPipeline pipeline=ch.pipeline();
    pipeline.addLast(new HttpClientCodec());
    pipeline.addLast(new ChunkedWriteHandler());
    pipeline.addLast(new HttpObjectAggregator(64*1024));
    pipeline.addLast(new WebSocketClientProtocolHandler(WebSocketClientHandshakerFactory.newHandshaker(new URI(url), WebSocketVersion.V13, null, false, new DefaultHttpHeaders())));
       pipeline.addLast(new WebSocketConnectedClientHandler());//连接成功监听handler
}

@Before
public void setup() throws Exception {
    s = new Server(conf);
    s.run();

    Connector con = mac.getConnector("root", "secret");
    con.securityOperations().changeUserAuthorizations("root", new Authorizations("A", "B", "C", "D", "E", "F"));

    this.sessionId = UUID.randomUUID().toString();
    AuthCache.getCache().put(sessionId, token);
    group = new NioEventLoopGroup();
    SslContext ssl = SslContextBuilder.forClient().trustManager(InsecureTrustManagerFactory.INSTANCE).build();

    String cookieVal = ClientCookieEncoder.STRICT.encode(Constants.COOKIE_NAME, sessionId);
    HttpHeaders headers = new DefaultHttpHeaders();
    headers.add(Names.COOKIE, cookieVal);

    WebSocketClientHandshaker handshaker = WebSocketClientHandshakerFactory.newHandshaker(LOCATION,
            WebSocketVersion.V13, (String) null, false, headers);
    handler = new ClientHandler(handshaker);
    Bootstrap boot = new Bootstrap();
    boot.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {

        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline().addLast("ssl", ssl.newHandler(ch.alloc(), "127.0.0.1", WS_PORT));
            ch.pipeline().addLast(new HttpClientCodec());
            ch.pipeline().addLast(new HttpObjectAggregator(8192));
            ch.pipeline().addLast(handler);
        }
    });
    ch = boot.connect("127.0.0.1", WS_PORT).sync().channel();
    // Wait until handshake is complete
    while (!handshaker.isHandshakeComplete()) {
        sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
        LOG.debug("Waiting for Handshake to complete");
    }
}

public HttpRequest adapt(SdkHttpRequest sdkRequest) {
    String uri = sdkRequest.getUri().toString();
    HttpMethod method = toNettyHttpMethod(sdkRequest.method());
    HttpHeaders headers = new DefaultHttpHeaders();
    DefaultHttpRequest request = new DefaultHttpRequest(HttpVersion.HTTP_1_1, method, uri, headers);
    sdkRequest.headers().forEach(request.headers()::add);
    return request;
}

private ResponseInfo<Void> deleteSafeDepositBox(final RequestInfo<Void> request) {
    final Optional<SecurityContext> securityContext =
            CmsRequestSecurityValidator.getSecurityContextForRequest(request);

    if (securityContext.isPresent()) {
        final VaultAuthPrincipal vaultAuthPrincipal = (VaultAuthPrincipal) securityContext.get().getUserPrincipal();

        String sdbId = request.getPathParam("id");
        Optional<String> sdbNameOptional = safeDepositBoxService.getSafeDepositBoxNameById(sdbId);
        String sdbName = sdbNameOptional.orElse(String.format("(Failed to lookup name from id: %s)", sdbId));

        log.info("{}: {}, Delete SDB Event: the principal: {} from ip: {} is attempting to delete sdb name: '{}' and id: '{}'",
                HEADER_X_CERBERUS_CLIENT,
                getClientVersion(request),
                vaultAuthPrincipal.getName(),
                getXForwardedClientIp(request),
                sdbName,
                sdbId);

        safeDepositBoxService.deleteSafeDepositBox(vaultAuthPrincipal, sdbId);
        return ResponseInfo.<Void>newBuilder().withHttpStatusCode(HttpResponseStatus.OK.code())
                .withHeaders(new DefaultHttpHeaders().set(HEADER_X_REFRESH_TOKEN, Boolean.TRUE.toString()))
                .build();
    }

    throw ApiException.newBuilder().withApiErrors(DefaultApiError.AUTH_BAD_CREDENTIALS).build();
}

private ResponseInfo<Void> updateSafeDepositBox(final RequestInfo<SafeDepositBoxV1> request) {
    final Optional<SecurityContext> securityContext =
            CmsRequestSecurityValidator.getSecurityContextForRequest(request);

    if (securityContext.isPresent()) {
        final VaultAuthPrincipal vaultAuthPrincipal = (VaultAuthPrincipal) securityContext.get().getUserPrincipal();

        String sdbId = request.getPathParam("id");
        Optional<String> sdbNameOptional = safeDepositBoxService.getSafeDepositBoxNameById(sdbId);
        String sdbName = sdbNameOptional.orElseGet(() -> String.format("(Failed to lookup name from id: %s)", sdbId));

        log.info("{}: {}, Update SDB Event: the principal: {} from ip: {} is attempting to update sdb name: '{}' and id: '{}'",
                HEADER_X_CERBERUS_CLIENT,
                getClientVersion(request),
                vaultAuthPrincipal.getName(),
                getXForwardedClientIp(request),
                sdbName,
                sdbId);

        safeDepositBoxService.updateSafeDepositBoxV1(request.getContent(),
                vaultAuthPrincipal,
                sdbId);
        return ResponseInfo.<Void>newBuilder().withHttpStatusCode(HttpResponseStatus.NO_CONTENT.code())
                .withHeaders(new DefaultHttpHeaders().set(HEADER_X_REFRESH_TOKEN, Boolean.TRUE.toString()))
                .build();
    }

    throw ApiException.newBuilder().withApiErrors(DefaultApiError.AUTH_BAD_CREDENTIALS).build();
}

private ResponseInfo<SafeDepositBoxV2> createSafeDepositBox(final RequestInfo<SafeDepositBoxV2> request,
                                                          final String basePath) {
    final Optional<SecurityContext> securityContext =
            CmsRequestSecurityValidator.getSecurityContextForRequest(request);

    if (securityContext.isPresent()) {
        final VaultAuthPrincipal vaultAuthPrincipal = (VaultAuthPrincipal) securityContext.get().getUserPrincipal();

        log.info("{}: {}, Create SDB Event: the principal: {} from ip: {} is attempting to create sdb name: '{}'",
                HEADER_X_CERBERUS_CLIENT,
                getClientVersion(request),
                vaultAuthPrincipal.getName(),
                getXForwardedClientIp(request),
                request.getContent().getName());

        final SafeDepositBoxV2 safeDepositBox =
                safeDepositBoxService.createSafeDepositBoxV2(request.getContent(), vaultAuthPrincipal.getName());
        final String location = basePath + "/" + safeDepositBox.getId();
        return ResponseInfo.newBuilder(safeDepositBox)
                .withHeaders(new DefaultHttpHeaders()
                        .set(LOCATION, location)
                        .set(HEADER_X_REFRESH_TOKEN, Boolean.TRUE.toString()))
                .withHttpStatusCode(HttpResponseStatus.CREATED.code())
                .build();
    }

    throw ApiException.newBuilder().withApiErrors(DefaultApiError.AUTH_BAD_CREDENTIALS).build();
}

private ResponseInfo<Map<String, String>> createSafeDepositBox(final RequestInfo<SafeDepositBoxV1> request,
                                                               final String basePath) {
    final Optional<SecurityContext> securityContext =
            CmsRequestSecurityValidator.getSecurityContextForRequest(request);

    if (securityContext.isPresent()) {
        final VaultAuthPrincipal vaultAuthPrincipal = (VaultAuthPrincipal) securityContext.get().getUserPrincipal();

        log.info("{}: {}, Create SDB Event: the principal: {} from ip; {} is attempting to create sdb name: '{}'",
                HEADER_X_CERBERUS_CLIENT,
                getClientVersion(request),
                vaultAuthPrincipal.getName(),
                getXForwardedClientIp(request),
                request.getContent().getName());

        final String id =
                safeDepositBoxService.createSafeDepositBoxV1(request.getContent(), vaultAuthPrincipal.getName());

        final String location = basePath + "/" + id;
        final Map<String, String> map = Maps.newHashMap();
        map.put("id", id);
        return ResponseInfo.newBuilder(map)
                .withHeaders(new DefaultHttpHeaders()
                        .set(LOCATION, location)
                        .set(HEADER_X_REFRESH_TOKEN, Boolean.TRUE.toString()))
                .withHttpStatusCode(HttpResponseStatus.CREATED.code())
                .build();
    }

    throw ApiException.newBuilder().withApiErrors(DefaultApiError.AUTH_BAD_CREDENTIALS).build();
}