@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    if (evt instanceof HandshakeComplete) {
        HandshakeComplete handshake = (HandshakeComplete) evt;
        // http request uri
        String uri = handshake.requestUri();
        logger.info("websocket handshake complete, uri: {}", uri);

        MultiValueMap<String, String> parameters = UriComponentsBuilder.fromUriString(uri).build().getQueryParams();
        String method = parameters.getFirst("method");

        if ("connectArthas".equals(method)) { // form browser
            connectArthas(ctx, parameters);
        } else if ("agentRegister".equals(method)) { // form arthas agent, register
            agentRegister(ctx, uri);
        }
        if ("openTunnel".equals(method)) { // from arthas agent open tunnel
            String clientConnectionId = parameters.getFirst("clientConnectionId");
            openTunnel(ctx, clientConnectionId);
        }
    } else {
        ctx.fireUserEventTriggered(evt);
    }
}
 

@SuppressWarnings("deprecation")
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt)throws Exception {
	if (evt == WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE)
	{//旧版本
		log.debug("excute webSocketHandComplete……");
		webSocketHandComplete(ctx);
		ctx.pipeline().remove(this);
		log.debug("excuted webSocketHandComplete:"+ctx.pipeline().toMap().toString());
		return;
	}
	if(evt instanceof HandshakeComplete)
	{//新版本
		HandshakeComplete hc=(HandshakeComplete)evt;
		log.debug("excute webSocketHandComplete……,HandshakeComplete="+hc);
		webSocketHandComplete(ctx);
		ctx.pipeline().remove(this);
		log.debug("excuted webSocketHandComplete:"+ctx.pipeline().toMap().toString());
		return;
	}
	super.userEventTriggered(ctx, evt);
}
 

@Override
public void userEventTriggered(ChannelHandlerContext context, Object payload) {
    if (payload instanceof HandshakeComplete) {
        handshakeComplete = (HandshakeComplete) payload;
        handshakeCompletion.countDown();
    }
}
 

public HandshakeComplete getHandshakeComplete() {
    return handshakeComplete;
}
 

@Test(timeout = 60 * 1000)
public void testConfiguredHttpHeadersArriveAtServer() throws Exception {
    try (NettyEchoServer server = createEchoServer(createServerOptions())) {
        server.start();

        int port = server.getServerPort();
        URI serverLocation = new URI("tcp://localhost:" + port);

        TransportOptions clientOptions = createClientOptions();
        clientOptions.getHttpHeaders().put("test-header1", "FOO");
        clientOptions.getHttpHeaders().put("test-header2", "BAR");

        Transport transport = createTransport(serverLocation, testListener, clientOptions);
        try {
            transport.connect(null, null);
            LOG.info("Connected to server:{} as expected.", serverLocation);
        } catch (Exception e) {
            fail("Should have connected to the server at " + serverLocation + " but got exception: " + e);
        }

        assertTrue(transport.isConnected());
        assertEquals(serverLocation, transport.getRemoteLocation());

        assertTrue("HandshakeCompletion not set within given time", server.awaitHandshakeCompletion(2000));
        HandshakeComplete handshake = server.getHandshakeComplete();
        assertNotNull("completion should not be null", handshake);
        HttpHeaders requestHeaders = handshake.requestHeaders();

        assertTrue(requestHeaders.contains("test-header1"));
        assertTrue(requestHeaders.contains("test-header2"));

        assertEquals("FOO", requestHeaders.get("test-header1"));
        assertEquals("BAR", requestHeaders.get("test-header2"));

        transport.close();
    }

    assertTrue(!transportClosed);  // Normal shutdown does not trigger the event.
    assertTrue(exceptions.isEmpty());
    assertTrue(data.isEmpty());
}