@Override
public void onopen(Session session,EndpointConfig endpointConfig) {

    ServerEndpointConfig sec = (ServerEndpointConfig) endpointConfig;

    Object pojo;
    try {
        pojo = sec.getConfigurator().getEndpointInstance(sec.getEndpointClass());
    } catch (InstantiationException e) {
        throw new IllegalArgumentException(
                sm.getString("pojoEndpointServer.getPojoInstanceFail",sec.getEndpointClass().getName()),e);
    }
    setPojo(pojo);

    @SuppressWarnings("unchecked")
    Map<String,String> pathParameters = (Map<String,String>) sec.getUserProperties().get(POJO_PATH_ParaM_KEY);
    setPathParameters(pathParameters);

    PojoMethodMapping methodMapping = (PojoMethodMapping) sec.getUserProperties().get(POJO_METHOD_MAPPING_KEY);
    setMethodMapping(methodMapping);

    doOnopen(session,endpointConfig);
}

private static Object[] buildArgs(PojoPathParam[] pathParams,Map<String,String> pathParameters,Session session,EndpointConfig config,Throwable throwable,CloseReason closeReason) throws DecodeException {
    Object[] result = new Object[pathParams.length];
    for (int i = 0; i < pathParams.length; i++) {
        Class<?> type = pathParams[i].getType();
        if (type.equals(Session.class)) {
            result[i] = session;
        } else if (type.equals(EndpointConfig.class)) {
            result[i] = config;
        } else if (type.equals(Throwable.class)) {
            result[i] = throwable;
        } else if (type.equals(CloseReason.class)) {
            result[i] = closeReason;
        } else {
            String name = pathParams[i].getName();
            String value = pathParameters.get(name);
            try {
                result[i] = Util.coercetoType(type,value);
            } catch (Exception e) {
                throw new DecodeException(value,sm.getString("pojoMethodMapping.decodePathParamFail",value,type),e);
            }
        }
    }
    return result;
}

/**
 * 连接建立成功调用的方法-与前端JS代码对应
 * 
 * @param session 可选的参数。session为与某个客户端的连接会话，需要通过它来给客户端发送数据
 */
@Onopen
public void onopen(Session session,EndpointConfig config) {
    // 单个会话对象保存
    this.session = session;
    webSocketSet.add(this); // 加入set中
    this.httpSession = (HttpSession) config.getUserProperties().get(HttpSession.class.getName());
    String uId = (String) httpSession.getAttribute("userid"); // 获取当前用户
    String sessionId = httpSession.getId();
    this.userid = uId + "|" + sessionId;
    if (!OnlineUserlist.contains(this.userid)) {
        OnlineUserlist.add(userid); // 将用户名加入在线列表
    }
    routetabMap.put(userid,session); // 将用户名和session绑定到路由表
    System.out.println(userid + " -> 已上线");
    String message = getMessage(userid + " -> 已上线","notice",OnlineUserlist);
    broadcast(message); // 广播
}

private void initWebSocketSession(String url,int wsConnectionTimeout) throws Exception {
    CountDownLatch wsLatch = new CountDownLatch(1);
    ClientEndpointConfig cec = ClientEndpointConfig.Builder.create().build();
    ClientManager client = ClientManager.createClient();

    client.connectToServer(new Endpoint() {
        @Override
        public void onopen(Session session,EndpointConfig endpointConfig) {
            wsSession = session;
            wsLatch.countDown();
        }
    },cec,new URI(url));

    if (!wsLatch.await(wsConnectionTimeout,TimeUnit.SECONDS)) {
        throw new TimeoutException("Web socket connection timeout");
    }
}

/**连接建立成功调用的方法*/
@Onopen
public void onopen(Session session,EndpointConfig config){
    HttpSession httpSession = (HttpSession) config.getUserProperties().get(HttpSession.class.getName());
    if(StorageUtil.init(httpSession).getLoginMemberId()!=ReturnUtil.NOT_LOGIN_CODE){
        long userId = StorageUtil.init(httpSession).getLoginMemberId();
        mapUS.put(userId,session);
        mapSU.put(session,userId);
        //上线通知由客户端自主发起
        onlineCount++;           //在线数加1
        System.out.println("用户"+userId+"进入WebSocket！当前在线人数为" + onlineCount);
        getUserKey(userId);
    }else{
        try {
            session.close();
            System.out.println("未获取到用户信息，关闭WebSocket！");
        } catch (IOException e) {
            System.out.println("关闭WebSocket失败！");
        }
    }
}

@Override
public void onopen(Session session,EndpointConfig arg1) {
    final RemoteEndpoint.Basic remote = session.getBasicRemote();
    session.addMessageHandler(new MessageHandler.Whole<String>() {
        public void onMessage(String text) {
            try {
                remote.sendText(text.toupperCase());
            } catch (IOException ioe) {
                ioe.printstacktrace();
            }
        }
    });
}

@Override
public void onopen(Session session,EndpointConfig endpointConfig) {

    ServerEndpointConfig sec = (ServerEndpointConfig) endpointConfig;

    Object pojo;
    try {
        pojo = sec.getConfigurator().getEndpointInstance(
                sec.getEndpointClass());
    } catch (InstantiationException e) {
        throw new IllegalArgumentException(sm.getString(
                "pojoEndpointServer.getPojoInstanceFail",String> pathParameters =
            (Map<String,String>) sec.getUserProperties().get(
                    POJO_PATH_ParaM_KEY);
    setPathParameters(pathParameters);

    PojoMethodMapping methodMapping =
            (PojoMethodMapping) sec.getUserProperties().get(
                    POJO_METHOD_MAPPING_KEY);
    setMethodMapping(methodMapping);

    doOnopen(session,endpointConfig);
}

public Set<MessageHandler> getMessageHandlers(Object pojo,EndpointConfig config) {
    Set<MessageHandler> result = new HashSet<MessageHandler>();
    for (MessageHandlerInfo messageMethod : onMessage) {
        result.addAll(messageMethod.getMessageHandlers(pojo,pathParameters,session,config));
    }
    return result;
}

private static Object[] buildArgs(PojoPathParam[] pathParams,CloseReason closeReason)
        throws DecodeException {
    Object[] result = new Object[pathParams.length];
    for (int i = 0; i < pathParams.length; i++) {
        Class<?> type = pathParams[i].getType();
        if (type.equals(Session.class)) {
            result[i] = session;
        } else if (type.equals(EndpointConfig.class)) {
            result[i] = config;
        } else if (type.equals(Throwable.class)) {
            result[i] = throwable;
        } else if (type.equals(CloseReason.class)) {
            result[i] = closeReason;
        } else {
            String name = pathParams[i].getName();
            String value = pathParameters.get(name);
            try {
                result[i] = Util.coercetoType(type,sm.getString(
                        "pojoMethodMapping.decodePathParamFail",e);
            }
        }
    }
    return result;
}

private static List<Class<? extends Decoder>> matchDecoders(Class<?> target,EndpointConfig endpointConfig,boolean binary) {
    DecoderMatch decoderMatch = matchDecoders(target,endpointConfig);
    if (binary) {
        if (decoderMatch.getBinaryDecoders().size() > 0) {
            return decoderMatch.getBinaryDecoders();
        }
    } else if (decoderMatch.getTextDecoders().size() > 0) {
        return decoderMatch.getTextDecoders();
    }
    return null;
}

private static DecoderMatch matchDecoders(Class<?> target,EndpointConfig endpointConfig) {
    DecoderMatch decoderMatch;
    try {
        List<Class<? extends Decoder>> decoders =
                endpointConfig.getDecoders();
        List<DecoderEntry> decoderEntries = getDecoders(decoders);
        decoderMatch = new DecoderMatch(target,decoderEntries);
    } catch (DeploymentException e) {
        throw new IllegalArgumentException(e);
    }
    return decoderMatch;
}

@Onopen
public void onopen(@SuppressWarnings("unused") Session session,EndpointConfig config) {
    if (config == null) {
        throw new RuntimeException();
    }
}

@Override
public void onopen(Session sn,EndpointConfig ec) {
    try {
        sn.addMessageHandler(String.class,new MessageHandler.Whole<String>() {
            @Override
            public void onMessage(String m) {
                System.out.println("got message from server - " + m);
            }
        });
    } catch (Exception ex) {
        Logger.getLogger(WebSocketEndpointConcurrencyTest.class.getName()).log(Level.SEVERE,null,ex);
    }

}

@Onopen
public void onopen(final Session session,EndpointConfig ec) {
    currentSession = session;
    agent = getRandomSupportAgent(); 

    String greeting = getGreeting(agent); 
    currentSession.getAsyncRemote().sendText(greeting);
}

@Override
   public void onopen(Session session,EndpointConfig endpointConfig) {
this.session = session;
if (messageHandler != null) {
    session.addMessageHandler(messageHandler);
}
   }

@Onopen
public void opened(@PathParam("user") String user,EndpointConfig config) throws IOException{
    System.out.println("opened() Current thread "+ Thread.currentThread().getName());
    this.httpSession = (HttpSession) config.getUserProperties().get(user);
    System.out.println("User joined "+ user + " with http session id "+ httpSession.getId());
    String response = "User " + user + " | WebSocket session ID "+ session.getId() +" | HTTP session ID " + httpSession.getId();
    System.out.println(response);
    session.getBasicRemote().sendText(response);
}

private static List<Class<? extends Decoder>> matchDecoders(Class<?> target,endpointConfig);
    if (binary) {
        if (decoderMatch.getBinaryDecoders().size() > 0) {
            return decoderMatch.getBinaryDecoders();
        }
    } else if (decoderMatch.getTextDecoders().size() > 0) {
        return decoderMatch.getTextDecoders();
    }
    return null;
}

@Override
public void onopen(Session sn,EndpointConfig ec) {
    this.sn = sn;
    controlLatch.countDown();
    sn.addMessageHandler(String.class,s -> {
        response = s;
        controlLatch.countDown();
    });
}

private static DecoderMatch matchDecoders(Class<?> target,EndpointConfig endpointConfig) {
    DecoderMatch decoderMatch;
    try {
        List<Class<? extends Decoder>> decoders = endpointConfig.getDecoders();
        List<DecoderEntry> decoderEntries = getDecoders(decoders);
        decoderMatch = new DecoderMatch(target,decoderEntries);
    } catch (DeploymentException e) {
        throw new IllegalArgumentException(e);
    }
    return decoderMatch;
}

public void preInit(Endpoint ep,WsServerContainer wsc,WsHandshakeRequest handshakeRequest,List<Extension> negotiatedExtensionsPhase2,String subProtocol,Transformation transformation,boolean secure) {
    this.ep = ep;
    this.endpointConfig = endpointConfig;
    this.webSocketContainer = wsc;
    this.handshakeRequest = handshakeRequest;
    this.negotiatedExtensions = negotiatedExtensionsPhase2;
    this.subProtocol = subProtocol;
    this.transformation = transformation;
    this.pathParameters = pathParameters;
    this.secure = secure;
}

@Override
public void onopen(Session session,endpointConfig);
}

public Set<MessageHandler> getMessageHandlers(Object pojo,config));
    }
    return result;
}

private static Object[] buildArgs(PojoPathParam[] pathParams,e);
            }
        }
    }
    return result;
}

private static List<Class<? extends Decoder>> matchDecoders(Class<?> target,endpointConfig);
    if (binary) {
        if (decoderMatch.getBinaryDecoders().size() > 0) {
            return decoderMatch.getBinaryDecoders();
        }
    } else if (decoderMatch.getTextDecoders().size() > 0) {
        return decoderMatch.getTextDecoders();
    }
    return null;
}

private static DecoderMatch matchDecoders(Class<?> target,decoderEntries);
    } catch (DeploymentException e) {
        throw new IllegalArgumentException(e);
    }
    return decoderMatch;
}

public void preInit(Endpoint ep,boolean secure) {
    this.ep = ep;
    this.endpointConfig = endpointConfig;
    this.webSocketContainer = wsc;
    this.handshakeRequest = handshakeRequest;
    this.negotiatedExtensions = negotiatedExtensionsPhase2;
    this.subProtocol = subProtocol;
    this.transformation = transformation;
    this.pathParameters = pathParameters;
    this.secure = secure;
}

@Onopen
public void onopen(@SuppressWarnings("unused") Session session,EndpointConfig config) {
    if (config == null) {
        throw new RuntimeException();
    }
}

public Set<MessageHandler> getMessageHandlers(Object pojo,config));
    }
    return result;
}

@Override
    public void onopen(Session session,EndpointConfig config) {
        System.out.println("Server session established");
        //conn to redis
        jedis = new Jedis("192.168.99.100",6379,10000);
        session.addMessageHandler(new MessageHandler.Whole<MeetupRSVP>() {
            @Override
            public void onMessage(MeetupRSVP message) {
                List<GroupTopic> groupTopics = message.getGroup().getGroupTopics();
                for (GroupTopic groupTopic : groupTopics) {
                    try {

                        if(GROUPS_IN_REdis.contains(groupTopic.getTopicName())){
                            jedis.zincrby(leaderBOARD_REdis_KEY,1,groupTopic.getTopicName());
                        }else{
                            //zscore = jedis.zscore(leaderBOARD_REdis_KEY,groupTopic.getTopicName());
                            jedis.zadd(leaderBOARD_REdis_KEY,groupTopic.getTopicName());
                            GROUPS_IN_REdis.add(groupTopic.getTopicName());
                        }

//                        Double zscore = jedis.zscore(leaderBOARD_REdis_KEY,groupTopic.getTopicName());;
//                        if(zscore == null){
//                            jedis.zadd(leaderBOARD_REdis_KEY,groupTopic.getTopicName());
//                        }else{
//                            jedis.zincrby(leaderBOARD_REdis_KEY,groupTopic.getTopicName());
//                        }
                    } catch (Exception e) {
                        e.printstacktrace();
                    }

                }
            }
        });

    }

public void onClose(Session arg0,EndpointConfig arg1) {
    System.out.println("close...");
}

public void onError(Session arg0,EndpointConfig arg1) {
    System.out.println("error...");
}

@Override
public void onopen(Session session,EndpointConfig config) {
    sessionClient = session;
    session.getAsyncRemote().sendText("hi");
}

public void init(EndpointConfig arg0) {
    tmp = new JSONObject();
}

public void init(EndpointConfig arg0) {
    tmp = new JSONObject();
}

@Override
public void onopen(Session session,EndpointConfig config) {
    session.addMessageHandler(new LoggregatorMessageHandler(listener));
}

@Override
public void onopen(Session session,EndpointConfig config) {
    doOnopen(session,config);
}

public Object[] getonopenArgs(Map<String,EndpointConfig config) throws DecodeException {
    return buildArgs(onopenParams,config,null);
}

private static PojoPathParam[] getPathParams(Method m,MethodType methodType) throws DeploymentException {
    if (m == null) {
        return new PojoPathParam[0];
    }
    boolean foundThrowable = false;
    Class<?>[] types = m.getParameterTypes();
    Annotation[][] paramsAnnotations = m.getParameterannotations();
    PojoPathParam[] result = new PojoPathParam[types.length];
    for (int i = 0; i < types.length; i++) {
        Class<?> type = types[i];
        if (type.equals(Session.class)) {
            result[i] = new PojoPathParam(type,null);
        } else if (methodType == MethodType.ON_OPEN &&
                type.equals(EndpointConfig.class)) {
            result[i] = new PojoPathParam(type,null);
        } else if (methodType == MethodType.ON_ERROR
                && type.equals(Throwable.class)) {
            foundThrowable = true;
            result[i] = new PojoPathParam(type,null);
        } else if (methodType == MethodType.ON_CLOSE &&
                type.equals(CloseReason.class)) {
            result[i] = new PojoPathParam(type,null);
        } else {
            Annotation[] paramAnnotations = paramsAnnotations[i];
            for (Annotation paramAnnotation : paramAnnotations) {
                if (paramAnnotation.annotationType().equals(
                        PathParam.class)) {
                    // Check that the type is valid. "0" coerces to every
                    // valid type
                    try {
                        Util.coercetoType(type,"0");
                    } catch (IllegalArgumentException iae) {
                        throw new DeploymentException(sm.getString(
                                "pojoMethodMapping.invalidpathParamType"),iae);
                    }
                    result[i] = new PojoPathParam(type,((PathParam) paramAnnotation).value());
                    break;
                }
            }
            // Parameters without annotations are not permitted
            if (result[i] == null) {
                throw new DeploymentException(sm.getString(
                        "pojoMethodMapping.paramWithoutAnnotation",type,m.getName(),m.getClass().getName()));
            }
        }
    }
    if (methodType == MethodType.ON_ERROR && !foundThrowable) {
        throw new DeploymentException(sm.getString(
                "pojoMethodMapping.onErrorNoThrowable",m.getDeclaringClass().getName()));
    }
    return result;
}

@Onopen
public void processOpen(@SuppressWarnings("unused") Session session,EndpointConfig  epc) {
    subprotocols = ((ServerEndpointConfig)epc).getSubprotocols();
}

@Test(expected = RuntimeException.class,timeout = 120000)
public void testFailingSocketInterceptor() {
    Config config = new Config();
    config.setProperty(GroupProperties.PROP_MAX_JOIN_SECONDS,"3");
    SocketInterceptorConfig sic = new SocketInterceptorConfig();
    MySocketInterceptor mySocketInterceptor = new MySocketInterceptor(false);
    sic.setImplementation(mySocketInterceptor);
    config.getNetworkConfig().setSocketInterceptorConfig(sic);
    HazelcastInstance h1 = Hazelcast.newHazelcastInstance(config);
    HazelcastInstance h2 = Hazelcast.newHazelcastInstance(config);
}

@Test(expected = RuntimeException.class,timeout = 120000)
public void testFailingClientSocketInterceptor() {
    Config config = new Config();
    SocketInterceptorConfig sic = new SocketInterceptorConfig();
    MySocketInterceptor mySocketInterceptor = new MySocketInterceptor(true);
    sic.setImplementation(mySocketInterceptor);
    config.getNetworkConfig().setSocketInterceptorConfig(sic);
    HazelcastInstance h1 = Hazelcast.newHazelcastInstance(config);
    HazelcastInstance h2 = Hazelcast.newHazelcastInstance(config);
    int count = 1000;
    for (int i = 0; i < count; i++) {
        h1.getMap("default").put(i,"value" + i);
        h2.getMap("default").put(i,"value" + i);
    }
    assertEquals(2,h2.getCluster().getMembers().size());
    assertTrue(mySocketInterceptor.getAcceptCallCount() >= 1);
    assertTrue(mySocketInterceptor.getConnectCallCount() >= 1);
    assertEquals(2,mySocketInterceptor.getinitCallCount());
    assertEquals(0,mySocketInterceptor.getAcceptFailureCount());
    assertEquals(0,mySocketInterceptor.getConnectFailureCount());
    ClientConfig clientConfig = new ClientConfig();
    clientConfig.setGroupConfig(new GroupConfig("dev","dev-pass")).addAddress("localhost");
    MySocketInterceptor myClientSocketInterceptor = new MySocketInterceptor(false);
    clientConfig.setSocketInterceptor(myClientSocketInterceptor);
    HazelcastInstance client = HazelcastClient.newHazelcastClient(clientConfig);
    for (int i = 0; i < count; i++) {
        client.getMap("default").put(i,"value" + i);
    }
    assertTrue(mySocketInterceptor.getAcceptCallCount() >= 2);
    assertTrue(mySocketInterceptor.getConnectCallCount() >= 1);
    assertEquals(1,myClientSocketInterceptor.getConnectCallCount());
    assertEquals(0,myClientSocketInterceptor.getAcceptCallCount());
    assertEquals(1,myClientSocketInterceptor.getAcceptFailureCount());
    assertEquals(1,myClientSocketInterceptor.getConnectFailureCount());
}

@Test(expected = RuntimeException.class,"3");
    SocketInterceptorConfig sic = new SocketInterceptorConfig();
    MySocketInterceptor mySocketInterceptor = new MySocketInterceptor(false);
    sic.setImplementation(mySocketInterceptor).setEnabled(true);
    config.getNetworkConfig().setSocketInterceptorConfig(sic);
    HazelcastInstance h1 = Hazelcast.newHazelcastInstance(config);
    HazelcastInstance h2 = Hazelcast.newHazelcastInstance(config);
}

@Test(expected = RuntimeException.class,timeout = 120000)
public void testFailingClientSocketInterceptor() {
    Config config = new Config();
    SocketInterceptorConfig sic = new SocketInterceptorConfig();
    MySocketInterceptor mySocketInterceptor = new MySocketInterceptor(true);
    sic.setImplementation(mySocketInterceptor).setEnabled(true);
    config.getNetworkConfig().setSocketInterceptorConfig(sic);
    HazelcastInstance h1 = Hazelcast.newHazelcastInstance(config);
    HazelcastInstance h2 = Hazelcast.newHazelcastInstance(config);
    int count = 1000;
    for (int i = 0; i < count; i++) {
        h1.getMap("default").put(i,myClientSocketInterceptor.getConnectFailureCount());
}

public void init(final SocketInterceptorConfig socketInterceptorConfig) {
}

public SocketInterceptorConfig getSocketInterceptorConfig() {
    return node.getConfig().getNetworkConfig().getSocketInterceptorConfig();
}

public ConnectionManager(IOService ioService,ServerSocketChannel serverSocketChannel) {
    this.ioService = ioService;
    this.ipV6ScopeId = ioService.getThisAddress().getScopeId();
    this.serverSocketChannel = serverSocketChannel;
    this.logger = ioService.getLogger(ConnectionManager.class.getName());
    this.soCKET_RECEIVE_BUFFER_SIZE = ioService.getSocketReceiveBufferSize() * KILO_BYTE;
    this.soCKET_SEND_BUFFER_SIZE = ioService.getSocketSendBufferSize() * KILO_BYTE;
    this.soCKET_LINGER_SECONDS = ioService.getSocketLingerSeconds();
    this.soCKET_KEEP_ALIVE = ioService.getSocketKeepAlive();
    this.soCKET_NO_DELAY = ioService.getSocketNoDelay();
    int selectorCount = ioService.getSelectorThreadCount();
    selectors = new InOutSelector[selectorCount];
    SSLConfig sslConfig = ioService.getSSLConfig();
    if (sslConfig != null && sslConfig.isEnabled()) {
        socketChannelWrapperFactory = new SSLSocketChannelWrapperFactory(sslConfig);
        logger.log(Level.INFO,"SSL is enabled");
    } else {
        socketChannelWrapperFactory = new DefaultSocketChannelWrapperFactory();
    }
    SocketInterceptorConfig sic = ioService.getSocketInterceptorConfig();
    if (sic != null && sic.isEnabled()) {
        SocketInterceptor implementation = (SocketInterceptor) sic.getImplementation();
        if (implementation == null && sic.getClassName() != null) {
            try {
                implementation = (SocketInterceptor) Class.forName(sic.getClassName()).newInstance();
            } catch (Throwable e) {
                logger.log(Level.SEVERE,"SocketInterceptor class cannot be instantiated!" + sic.getClassName(),e);
            }
        }
        if (implementation != null) {
            if (!(implementation instanceof MemberSocketInterceptor)) {
                logger.log(Level.SEVERE,"SocketInterceptor must be instance of " + MemberSocketInterceptor.class.getName());
                implementation = null;
            } else {
                logger.log(Level.INFO,"SocketInterceptor is enabled");
            }
        }
        if (implementation != null) {
            memberSocketInterceptor = (MemberSocketInterceptor) implementation;
            memberSocketInterceptor.init(sic);
        } else {
            memberSocketInterceptor = null;
        }
    } else {
        memberSocketInterceptor = null;
    }
}

@Test(timeout = 120000)
public void testSuccessfulSocketInterceptor() {
    Config config = new Config();
    SocketInterceptorConfig sic = new SocketInterceptorConfig();
    MySocketInterceptor mySocketInterceptor = new MySocketInterceptor(true);
    sic.setImplementation(mySocketInterceptor);
    config.getNetworkConfig().setSocketInterceptorConfig(sic);
    HazelcastInstance h1 = Hazelcast.newHazelcastInstance(config);
    HazelcastInstance h2 = Hazelcast.newHazelcastInstance(config);
    HazelcastInstance h3 = Hazelcast.newHazelcastInstance(config);
    HazelcastInstance h4 = Hazelcast.newHazelcastInstance(config);
    int count = 1000;
    for (int i = 0; i < count; i++) {
        h1.getMap("default").put(i,"value" + i);
        h3.getMap("default").put(i,"value" + i);
        h4.getMap("default").put(i,"value" + i);
    }
    assertEquals(4,h4.getCluster().getMembers().size());
    assertTrue(mySocketInterceptor.getAcceptCallCount() >= 6);
    assertTrue(mySocketInterceptor.getConnectCallCount() >= 6);
    assertEquals(4,"dev-pass")).addAddress("localhost");
    MySocketInterceptor myClientSocketInterceptor = new MySocketInterceptor(true);
    clientConfig.setSocketInterceptor(myClientSocketInterceptor);
    HazelcastInstance client = HazelcastClient.newHazelcastClient(clientConfig);
    for (int i = 0; i < count; i++) {
        client.getMap("default").put(i,"value" + i);
    }
    assertTrue(mySocketInterceptor.getAcceptCallCount() >= 7);
    assertTrue(mySocketInterceptor.getConnectCallCount() >= 6);
    assertEquals(1,myClientSocketInterceptor.getAcceptCallCount());
    assertEquals(0,mySocketInterceptor.getConnectFailureCount());
    assertEquals(0,myClientSocketInterceptor.getAcceptFailureCount());
    assertEquals(0,myClientSocketInterceptor.getConnectFailureCount());
}

public void init(SocketInterceptorConfig socketInterceptorConfig) {
    initCallCount.incrementAndGet();
}

@Override
public SocketInterceptorConfig getSocketInterceptorConfig() {
    return null;
}

public void init(final SocketInterceptorConfig socketInterceptorConfig) {
}

public ConnectionManager(IOService ioService,ServerSocketChannel serverSocketChannel) {
    this.ioService = ioService;
    this.serverSocketChannel = serverSocketChannel;
    this.logger = ioService.getLogger(ConnectionManager.class.getName());
    this.soCKET_RECEIVE_BUFFER_SIZE = ioService.getSocketReceiveBufferSize() * KILO_BYTE;
    this.soCKET_SEND_BUFFER_SIZE = ioService.getSocketSendBufferSize() * KILO_BYTE;
    this.soCKET_LINGER_SECONDS = ioService.getSocketLingerSeconds();
    this.soCKET_KEEP_ALIVE = ioService.getSocketKeepAlive();
    this.soCKET_NO_DELAY = ioService.getSocketNoDelay();
    int selectorCount = ioService.getSelectorThreadCount();
    selectors = new InOutSelector[selectorCount];
    final Collection<Integer> ports = ioService.getoutboundPorts();
    outboundPortCount = ports == null ? 0 : ports.size();
    if (ports != null) {
        outboundPorts.addAll(ports);
    }
    SSLConfig sslConfig = ioService.getSSLConfig();
    if (sslConfig != null && sslConfig.isEnabled()) {
        socketChannelWrapperFactory = new SSLSocketChannelWrapperFactory(sslConfig);
        logger.log(Level.INFO,"SocketInterceptor is enabled");
            }
        }
        if (implementation != null) {
            memberSocketInterceptor = (MemberSocketInterceptor) implementation;
            memberSocketInterceptor.init(sic);
        } else {
            memberSocketInterceptor = null;
        }
    } else {
        memberSocketInterceptor = null;
    }
}

@Test(timeout = 120000)
public void testSuccessfulSocketInterceptor() {
    Config config = new Config();
    SocketInterceptorConfig sic = new SocketInterceptorConfig();
    MySocketInterceptor mySocketInterceptor = new MySocketInterceptor(true);
    sic.setImplementation(mySocketInterceptor).setEnabled(true);
    config.getNetworkConfig().setSocketInterceptorConfig(sic);
    HazelcastInstance h1 = Hazelcast.newHazelcastInstance(config);
    HazelcastInstance h2 = Hazelcast.newHazelcastInstance(config);
    HazelcastInstance h3 = Hazelcast.newHazelcastInstance(config);
    HazelcastInstance h4 = Hazelcast.newHazelcastInstance(config);
    int count = 1000;
    for (int i = 0; i < count; i++) {
        h1.getMap("default").put(i,myClientSocketInterceptor.getConnectFailureCount());
}

public void init(SocketInterceptorConfig socketInterceptorConfig) {
    initCallCount.incrementAndGet();
}

SocketInterceptorConfig getSocketInterceptorConfig();

void init(SocketInterceptorConfig socketInterceptorConfig);

SocketInterceptorConfig getSocketInterceptorConfig();

void init(SocketInterceptorConfig socketInterceptorConfig);

private void handleWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame frame) {
    if (logger.isLoggable(Level.FINE)) {
        logger.fine(String.format(
                "Channel %s received %s",ctx.channel().hashCode(),StringUtil.simpleClassName(frame)));
    }

    if (frame instanceof CloseWebSocketFrame) {
        handshaker.close(ctx.channel(),(CloseWebSocketFrame) frame);
    } else if (frame instanceof PingWebSocketFrame) {
        ctx.write(new PongWebSocketFrame(frame.isFinalFragment(),frame.rsv(),frame.content()),ctx.voidPromise());
    } else if (frame instanceof TextWebSocketFrame) {
        ctx.write(frame,ctx.voidPromise());
    } else if (frame instanceof BinaryWebSocketFrame) {
        ctx.write(frame,ctx.voidPromise());
    } else if (frame instanceof ContinuationWebSocketFrame) {
        ctx.write(frame,ctx.voidPromise());
    } else if (frame instanceof PongWebSocketFrame) {
        frame.release();
        // Ignore
    } else {
        throw new UnsupportedOperationException(String.format("%s frame types not supported",frame.getClass()
                .getName()));
    }
}

private void handleWebSocketFrame(ChannelHandlerContext ctx,frame.getClass()
                .getName()));
    }
}

@Override
public void write(ChannelHandlerContext ctx,Object msg,ChannelPromise promise) throws Exception {
    LOG.trace("NettyServerHandler: Channel write: {}",msg);
    if (isWebSocketServer() && msg instanceof ByteBuf) {
        if(isFragmentWrites()) {
            ByteBuf orig = (ByteBuf) msg;
            int origIndex = orig.readerIndex();
            int split = orig.readableBytes()/2;

            ByteBuf part1 = orig.copy(origIndex,split);
            LOG.trace("NettyServerHandler: Part1: {}",part1);
            orig.readerIndex(origIndex + split);
            LOG.trace("NettyServerHandler: Part2: {}",orig);

            BinaryWebSocketFrame frame1 = new BinaryWebSocketFrame(false,part1);
            ctx.writeAndFlush(frame1);
            ContinuationWebSocketFrame frame2 = new ContinuationWebSocketFrame(true,orig);
            ctx.write(frame2,promise);
        } else {
            BinaryWebSocketFrame frame = new BinaryWebSocketFrame((ByteBuf) msg);
            ctx.write(frame,promise);
        }
    } else {
        ctx.write(msg,promise);
    }
}

private void handleWebSocketFrame(ChannelHandlerContext ctx,frame.content()));
    } else if (frame instanceof TextWebSocketFrame) {
        ctx.write(frame);
    } else if (frame instanceof BinaryWebSocketFrame) {
        ctx.write(frame);
    } else if (frame instanceof ContinuationWebSocketFrame) {
        ctx.write(frame);
    } else if (frame instanceof PongWebSocketFrame) {
        frame.release();
        // Ignore
    } else {
        throw new UnsupportedOperationException(String.format("%s frame types not supported",frame.getClass()
                .getName()));
    }
}

protected void handleWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame frame) {
   logger.debug("Received incoming frame [{}]",frame.getClass().getName());
   // Check for closing frame
   if (frame instanceof CloseWebSocketFrame) {
      if (frameBuffer != null) {
          handleMessageCompleted(ctx,frameBuffer.toString());
      }
      handshaker.close(ctx.channel(),(CloseWebSocketFrame) frame.retain());
      return;
   }

   if (frame instanceof PingWebSocketFrame) {
      ctx.channel().writeAndFlush(new PongWebSocketFrame(frame.content().retain()));
      return;
   }

   if (frame instanceof PongWebSocketFrame) {
      logger.info("Pong frame received");
      return;
   }

   if (frame instanceof TextWebSocketFrame) {
      frameBuffer = new StringBuilder();
      frameBuffer.append(((TextWebSocketFrame)frame).text());
   } else if (frame instanceof ContinuationWebSocketFrame) {
      if (frameBuffer != null) {
         frameBuffer.append(((ContinuationWebSocketFrame)frame).text());
      } else {
         logger.warn("Continuation frame received without initial frame.");
      }
   } else {
      throw new UnsupportedOperationException(String.format("%s frame types not supported",frame.getClass().getName()));
   }

   // Check if Text or Continuation Frame is final fragment and handle if needed.
   if (frame.isFinalFragment()) {
      handleMessageCompleted(ctx,frameBuffer.toString());
      frameBuffer = null;
   }
}

protected void handleWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame frame)
{
    logger.debug("Received incoming frame [{}]",frame.getClass().getName());
    // Check for closing frame
    if ( frame instanceof CloseWebSocketFrame) {
        if ( frameBuffer != null) {
            handleMessageCompleted( ctx,frameBuffer.toString());
        }
        handshaker.close(ctx.channel(),(CloseWebSocketFrame) frame.retain());
        return;
    }

    if (frame instanceof PingWebSocketFrame) {
        ctx.channel().writeAndFlush(new PongWebSocketFrame(frame.content().retain()));
        return;
    }

    if (frame instanceof PongWebSocketFrame) {
        logger.info("Pong frame received");
        return;
    }

    if (frame instanceof TextWebSocketFrame) {
        frameBuffer = new StringBuilder();
        frameBuffer.append(((TextWebSocketFrame)frame).text());
    }
    else if (frame instanceof ContinuationWebSocketFrame) {
        if (frameBuffer != null) {
            frameBuffer.append(((ContinuationWebSocketFrame)frame).text());
        }
        else {
            logger.warn("Continuation frame received without initial frame.");
        }
    }
    else {
        throw new UnsupportedOperationException(String.format("%s frame types not supported",frame.getClass().getName()));
    }

    // Check if Text or Continuation Frame is final fragment and handle if needed.
    if (frame.isFinalFragment()) {
        handleMessageCompleted(ctx,frameBuffer.toString());
        frameBuffer = null;
    }
}

/**
 * Handle web socket frame.
 *
 * @param ctx the ctx
 * @param frame the frame
 */
private void handleWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame frame) {

    try{

        // Check for closing frame
        if (frame instanceof CloseWebSocketFrame) {
            dominoServer.onClose(this.newWrapper(ctx));
            handshaker.close(ctx.channel(),(CloseWebSocketFrame) frame.retain());
            return;

        }
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }


        if(frame instanceof PongWebSocketFrame){
            return;//do nothing.

        }


        if(frame instanceof TextWebSocketFrame){
            String message = ((TextWebSocketFrame) frame).text();
            textBuffer.append(message);
        }else if(frame instanceof ContinuationWebSocketFrame){
            textBuffer.append(((ContinuationWebSocketFrame) frame).text());
        }


        if(frame.isFinalFragment()){
            dominoServer.onMessage(this.newWrapper(ctx),textBuffer.toString());
            textBuffer = new StringBuilder();
        }


    }catch(Exception e){
        e.printstacktrace();
    }
}

@Override
protected void channelRead0(ChannelHandlerContext ctx,Object message) throws Exception {
   LOG.trace("New data read: incoming: {}",message);

   Channel ch = ctx.channel();
   if (!handshaker.isHandshakeComplete()) {
      handshaker.finishHandshake(ch,(FullHttpResponse) message);
      LOG.trace("WebSocket Client connected! {}",ctx.channel());
      // Now trigger super processing as we are really connected.
      NettyWSTransport.super.handleConnected(ch);
      return;
   }

   // We shouldn't get this since we handle the handshake prevIoUsly.
   if (message instanceof FullHttpResponse) {
      FullHttpResponse response = (FullHttpResponse) message;
      throw new IllegalStateException(
         "Unexpected FullHttpResponse (getStatus=" + response.status() + ",content=" + response.content().toString(StandardCharsets.UTF_8) + ')');
   }

   WebSocketFrame frame = (WebSocketFrame) message;
   if (frame instanceof TextWebSocketFrame) {
      TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
      LOG.warn("WebSocket Client received message: " + textFrame.text());
      ctx.fireExceptionCaught(new IOException("Received invalid frame over WebSocket."));
   } else if (frame instanceof BinaryWebSocketFrame) {
      BinaryWebSocketFrame binaryFrame = (BinaryWebSocketFrame) frame;
      LOG.trace("WebSocket Client received data: {} bytes",binaryFrame.content().readableBytes());
      listener.onData(binaryFrame.content());
   } else if (frame instanceof ContinuationWebSocketFrame) {
      ContinuationWebSocketFrame continuationFrame = (ContinuationWebSocketFrame) frame;
      LOG.trace("WebSocket Client received data continuation: {} bytes",continuationFrame.content().readableBytes());
      listener.onData(continuationFrame.content());
   } else if (frame instanceof PingWebSocketFrame) {
      LOG.trace("WebSocket Client received ping,response with pong");
      ch.write(new PongWebSocketFrame(frame.content()));
   } else if (frame instanceof CloseWebSocketFrame) {
      LOG.trace("WebSocket Client received closing");
      ch.close();
   }
}

@Override
public void channelRead0(ChannelHandlerContext ctx,Object msg) throws Exception {
  Channel ch = ctx.channel();
  if (!handshaker.isHandshakeComplete()) {
    handshaker.finishHandshake(ch,(FullHttpResponse) msg);
    log.debug("{} WebSocket Client connected!",label);
    handshakeFuture.setSuccess();
    return;
  }

  if (msg instanceof FullHttpResponse) {
    FullHttpResponse response = (FullHttpResponse) msg;
    throw new IllegalStateException(
        "Unexpected FullHttpResponse (getStatus=" + response.status() + ",content="
            + response.content().toString(CharsetUtil.UTF_8) + ')');
  }

  WebSocketFrame frame = (WebSocketFrame) msg;
  if (frame instanceof TextWebSocketFrame) {
    TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
    if (textFrame.isFinalFragment()) {
      receivedTextMessage(textFrame.text());
    } else {
      partialText.append(textFrame.text());
    }
  } else if (frame instanceof ContinuationWebSocketFrame) {
    ContinuationWebSocketFrame continuationFrame = (ContinuationWebSocketFrame) frame;
    partialText.append(continuationFrame.text());
    if (continuationFrame.isFinalFragment()) {
      receivedTextMessage(partialText.toString());
      partialText.setLength(0);
    }
  } else if (frame instanceof CloseWebSocketFrame) {
    CloseWebSocketFrame closeFrame = (CloseWebSocketFrame) frame;
    log.info("{} Received close frame from server. Will close client! Reason: {}",label,closeFrame.reasonText());
  } else {
    log.warn("{} Received frame of type {}. Will be ignored",frame.getClass().getSimpleName());
  }

}

@Override
protected void channelRead0(ChannelHandlerContext ctx,Object message) throws Exception {
    LOG.trace("New data read: incoming: {}",message);

    Channel ch = ctx.channel();
    if (!handshaker.isHandshakeComplete()) {
        handshaker.finishHandshake(ch,(FullHttpResponse) message);
        LOG.trace("WebSocket Client connected! {}",ctx.channel());
        // Now trigger super processing as we are really connected.
        NettyWsTransport.super.handleConnected(ch);
        return;
    }

    // We shouldn't get this since we handle the handshake prevIoUsly.
    if (message instanceof FullHttpResponse) {
        FullHttpResponse response = (FullHttpResponse) message;
        throw new IllegalStateException(
            "Unexpected FullHttpResponse (getStatus=" + response.status() +
            ",content=" + response.content().toString(StandardCharsets.UTF_8) + ')');
    }

    WebSocketFrame frame = (WebSocketFrame) message;
    if (frame instanceof TextWebSocketFrame) {
        TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
        LOG.warn("WebSocket Client received message: " + textFrame.text());
        ctx.fireExceptionCaught(new IOException("Received invalid frame over WebSocket."));
    } else if (frame instanceof BinaryWebSocketFrame) {
        BinaryWebSocketFrame binaryFrame = (BinaryWebSocketFrame) frame;
        LOG.trace("WebSocket Client received data: {} bytes",binaryFrame.content().readableBytes());
        listener.onData(binaryFrame.content());
    } else if (frame instanceof ContinuationWebSocketFrame) {
        ContinuationWebSocketFrame continuationFrame = (ContinuationWebSocketFrame) frame;
        LOG.trace("WebSocket Client received data continuation: {} bytes",continuationFrame.content().readableBytes());
        listener.onData(continuationFrame.content());
    } else if (frame instanceof PingWebSocketFrame) {
        LOG.trace("WebSocket Client received ping,response with pong");
        ch.write(new PongWebSocketFrame(frame.content()));
    } else if (frame instanceof CloseWebSocketFrame) {
        LOG.trace("WebSocket Client received closing");
        ch.close();
    }
}

@Override
protected void messageReceived(ChannelHandlerContext ctx,ContinuationWebSocketFrame msg) throws Exception {

}

private void handlerWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame frame) {
    // 判断是否关闭链路的指令
    if (frame instanceof CloseWebSocketFrame) {
        socketServerHandshaker.close(ctx.channel(),(CloseWebSocketFrame) frame.retain());
    }
    // 判断是否ping消息
    if (frame instanceof PingWebSocketFrame) {
        ctx.channel().write(
                new PongWebSocketFrame(frame.content().retain()));
        return;
    }
    // 本例程仅支持文本消息，不支持二进制消息
    if (!(frame instanceof TextWebSocketFrame)) {
        throw new UnsupportedOperationException(String.format(
                "%s frame types not supported",frame.getClass().getName()));
    }
    // 返回应答消息
    String request = ((TextWebSocketFrame) frame).text();
    System.out.println("服务端收到：" + request);
    TextWebSocketFrame tws = new TextWebSocketFrame(new Date().toString()
            + ctx.channel().id() + "：" + request);
    // 群发
    group.writeAndFlush(tws);
}

private void handleFrame(Channel channel,WebSocketFrame frame,WebSocketUpgradeHandler handler,NettyWebSocket webSocket) throws Exception {
    if (frame instanceof CloseWebSocketFrame) {
        Channels.setdiscard(channel);
        CloseWebSocketFrame closeFrame = (CloseWebSocketFrame) frame;
        webSocket.onClose(closeFrame.statusCode(),closeFrame.reasonText());
    } else {
        ByteBuf buf = frame.content();
        if (buf != null && buf.readableBytes() > 0) {
            HttpResponseBodyPart part = config.getResponseBodyPartFactory().newResponseBodyPart(buf,frame.isFinalFragment());
            handler.onBodyPartReceived(part);

            if (frame instanceof BinaryWebSocketFrame) {
                webSocket.onBinaryFragment(part);
            } else if (frame instanceof TextWebSocketFrame) {
                webSocket.onTextFragment(part);
            } else if (frame instanceof PingWebSocketFrame) {
                webSocket.onPing(part);
            } else if (frame instanceof PongWebSocketFrame) {
                webSocket.onPong(part);
            }
        }
    }
}

@Override
public void onInboundNext(ChannelHandlerContext ctx,Object frame) {
    if (frame instanceof CloseWebSocketFrame && ((CloseWebSocketFrame) frame).isFinalFragment()) {
        if (log.isDebugEnabled()) {
            log.debug("CloseWebSocketFrame detected. Closing Websocket");
        }
        CloseWebSocketFrame close = (CloseWebSocketFrame) frame;
        sendClose(new CloseWebSocketFrame(true,close.rsv(),close.content()),f -> onHandlerTerminate());
        return;
    }
    if (frame instanceof PingWebSocketFrame) {
        ctx.writeAndFlush(new PongWebSocketFrame(((PingWebSocketFrame) frame).content()));
        ctx.read();
        return;
    }
    super.onInboundNext(ctx,frame);
}

private Message decodeWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame frame) {
    // Check for closing frame
    if (frame instanceof CloseWebSocketFrame) {
        handshaker.close(ctx.channel(),(CloseWebSocketFrame) frame.retain());
        return null;
    }

    if (frame instanceof PingWebSocketFrame) {
        ctx.write(new PongWebSocketFrame(frame.content().retain()));
        return null;
    }

    if (frame instanceof TextWebSocketFrame) {
        TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
        return parseMessage(textFrame.content());
    }

    if (frame instanceof BinaryWebSocketFrame) {
        BinaryWebSocketFrame binFrame = (BinaryWebSocketFrame) frame;
        return parseMessage(binFrame.content());
    }

    log.warn("Message format error: " + frame); 
    return null;
}

private void handleWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame frame) {

        // Check for closing frame
        if (frame instanceof CloseWebSocketFrame) {
            handshaker.close(ctx.channel(),(CloseWebSocketFrame) frame.retain());
            return;
        }
        if (frame instanceof PingWebSocketFrame) {
            ctx.write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        if (frame instanceof TextWebSocketFrame) {
            // Echo the frame
            ctx.write(frame.retain());
            return;
        }
        if (frame instanceof BinaryWebSocketFrame) {
            // Echo the frame
            ctx.write(frame.retain());
            return;
        }
    }

private void handleWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame frame) {
    // Check for closing frame
    if (frame instanceof CloseWebSocketFrame) {
        addTraceForFrame(frame,"close");
        handshaker.close(ctx.channel(),(CloseWebSocketFrame) frame.retain());
        return;
    }
    if (frame instanceof PingWebSocketFrame) {
        addTraceForFrame(frame,"ping");
        ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
        return;
    }
    if (!(frame instanceof TextWebSocketFrame)) {
        throw new UnsupportedOperationException(String.format("%s frame types not supported",frame.getClass()
            .getName()));
    }

    // todo [om] think about BinaryWebsocketFrame

    handleTextWebSocketFrameInternal((TextWebSocketFrame) frame,ctx);
}

@Override
protected void channelRead0(ChannelHandlerContext context,Object message) throws Exception {
    final Channel channel = context.channel();
    if (!handshaker.isHandshakeComplete()) {
        handshaker.finishHandshake(channel,(FullHttpResponse) message);
        channel.pipeline().addBefore(HANDLER_NAME,"websocket-frame-aggregator",new WebSocketFrameAggregator(64 * 1024));
        subscriber.onStart();
        return;
    }

    if (message instanceof FullHttpResponse) {
        final FullHttpResponse response = (FullHttpResponse) message;
        throw new IllegalStateException(
                "Unexpected FullHttpResponse (getStatus=" + response.getStatus() +
                        ",content=" + response.content().toString(CharsetUtil.UTF_8) + ')');
    }

    final WebSocketFrame frame = (WebSocketFrame) message;
    if (frame instanceof PingWebSocketFrame) {
        context.writeAndFlush(new PongWebSocketFrame(((PingWebSocketFrame)frame).retain().content()));
    } else if (frame instanceof BinaryWebSocketFrame) {
        final ByteBufInputStream input = new ByteBufInputStream(((BinaryWebSocketFrame)message).content());
        final Envelope envelope = Envelope.ADAPTER.decode(input);
        subscriber.onNext(envelope);
    }
}

private void handleWebSocketFrame(ChannelHandlerContext ctx,(CloseWebSocketFrame) frame.retain());
            return;
        }
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        if (!(frame instanceof TextWebSocketFrame)) {
            throw new UnsupportedOperationException(String.format("%s frame types not supported",frame.getClass()
                    .getName()));
        }

        // Send the uppercase string back.
        String request = ((TextWebSocketFrame) frame).text();
        System.err.printf("%s received %s%n",ctx.channel(),request);
        ctx.channel().write(new TextWebSocketFrame(request.toupperCase()));
    }

private void handleWebSocketFrame(ChannelHandlerContext ctx,(CloseWebSocketFrame) frame.retain());
            return;
        }
        if (frame instanceof PingWebSocketFrame) {
            ctx.write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        if (frame instanceof TextWebSocketFrame) {
            // Echo the frame
            ctx.write(frame.retain());
            return;
        }
        if (frame instanceof BinaryWebSocketFrame) {
            // Echo the frame
            ctx.write(frame.retain());
            return;
        }
    }

private void handleWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame frame) {
    if (logger.isLoggable(Level.FINE)) {
        logger.fine(String.format(
                "Channel %s received %s",ctx.channel().hashCode(),StringUtil.simpleClassName(frame)));
    }

    if (frame instanceof CloseWebSocketFrame) {
        handshaker.close(ctx.channel(),(CloseWebSocketFrame) frame);
    } else if (frame instanceof PingWebSocketFrame) {
        ctx.write(new PongWebSocketFrame(frame.isFinalFragment(),frame.rsv(),frame.content()),ctx.voidPromise());
    } else if (frame instanceof TextWebSocketFrame) {
        ctx.write(frame,ctx.voidPromise());
    } else if (frame instanceof BinaryWebSocketFrame) {
        ctx.write(frame,ctx.voidPromise());
    } else if (frame instanceof ContinuationWebSocketFrame) {
        ctx.write(frame,ctx.voidPromise());
    } else if (frame instanceof PongWebSocketFrame) {
        frame.release();
        // Ignore
    } else {
        throw new UnsupportedOperationException(String.format("%s frame types not supported",frame.getClass()
                .getName()));
    }
}

@Override
public void accept(ChannelHandlerContext ctx,WebSocketFrame frame) {
    if (frame instanceof CloseWebSocketFrame) {
        handshaker.close(ctx.channel(),(CloseWebSocketFrame) frame.retain());
        endpoint.releaseReferences();
        endpoint.onClose();
        return;
    }

    if (frame instanceof PingWebSocketFrame) {
        ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
        return;
    }

    if (frame instanceof TextWebSocketFrame) {
        endpoint.onMessage(((TextWebSocketFrame) frame).text());
        return;
    }

    throw new UnsupportedOperationException(String.format("Unsupported websocket frame of type %s",frame.getClass().getName()));
}

private void handlerWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame frame) {
    // 判断是否关闭链路的指令
    if (frame instanceof CloseWebSocketFrame) {
        handshaker.close(ctx.channel(),(CloseWebSocketFrame) frame.retain());
        return;
    }
    // 判断是否ping消息
    if (frame instanceof PingWebSocketFrame) {
        ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
        return;
    }
    // 仅支持文本消息，不支持二进制消息
    if (!(frame instanceof TextWebSocketFrame)) {
        ctx.close();//(String.format("%s frame types not supported",frame.getClass().getName()));
        return;
    }

}

public void handle(ChannelHandlerContext ctx,(CloseWebSocketFrame) frame);
        onClose(ctx);
        return;
    }
    if (frame instanceof PingWebSocketFrame) {
        ctx.channel().write(new PongWebSocketFrame(frame.content()));
        return;
    }
    if (!(frame instanceof TextWebSocketFrame)) {
        throw new UnsupportedOperationException(String.format("%s frame types not supported",frame.getClass()
                .getName()));
    }

    String msg = ((TextWebSocketFrame) frame).text();
    onMessage(ctx,msg);
}

public void handleWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame frame) {

        // 判断是否是关闭链路的指令
        if (frame instanceof CloseWebSocketFrame) {
            handshaker.close(ctx.channel(),(CloseWebSocketFrame) frame.retain());
            return;
        }
        // 判断是否是Ping消息
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }

        if (!(frame instanceof TextWebSocketFrame)) {
            throw new UnsupportedOperationException(String.format("%s frame types not supported",frame.getClass().getName()));
        }
        //返回应答消息
        String request= ((TextWebSocketFrame)frame).text();
        System.out.println(String.format("%s received %s",request));

        ctx.channel().write(new TextWebSocketFrame(request+",现在时刻："+new Date()));

    }

@Override
public void channelRead(ChannelHandlerContext ctx,Object msg)
        throws UnkNownWebSocketFrameTypeException,ServerConnectorException {
    if (!(msg instanceof WebSocketFrame)) {
        logger.error("Expecting WebSocketFrame. UnkNown type.");
        throw new UnkNownWebSocketFrameTypeException("Expecting WebSocketFrame. UnkNown type.");
    }
    if (msg instanceof TextWebSocketFrame) {
        notifyTextMessage((TextWebSocketFrame) msg);
    } else if (msg instanceof BinaryWebSocketFrame) {
        notifyBinaryMessage((BinaryWebSocketFrame) msg);
    } else if (msg instanceof CloseWebSocketFrame) {
        notifyCloseMessage((CloseWebSocketFrame) msg);
    } else if (msg instanceof PingWebSocketFrame) {
        notifyPingMessage((PingWebSocketFrame) msg);
    } else if (msg instanceof PongWebSocketFrame) {
        notifyPongMessage((PongWebSocketFrame) msg);
    }
}

private void handleWebSocketFrame(ChannelHandlerContext ctx,frame.getClass()
                .getName()));
    }
}

private void handleWebSocketFrame(ChannelHandlerContext ctx,frame.getClass()
                    .getName()));
        }

        // Send the uppercase string back.
        String request = ((TextWebSocketFrame) frame).text();
        if (logger.isLoggable(Level.FINE)) {
            logger.fine(String.format("%s received %s",request));
        }
        ctx.channel().write(new TextWebSocketFrame(request.toupperCase()));
    }

private void handleWebSocketFrame(ChannelHandlerContext ctx,request));
        }
        ctx.channel().write(new TextWebSocketFrame(request.toupperCase()));
    }

@Override
protected void channelRead0(ChannelHandlerContext ctx,WebSocketFrame frame) throws Exception
{
    this.last = ctx;
    if (frame instanceof CloseWebSocketFrame)
    {
        this.log.debug("recevied close frame");
        this.server.unsubscribe(this);
        this.handshaker.close(ctx.channel(),(CloseWebSocketFrame)frame);
    }
    else if (frame instanceof PingWebSocketFrame)
    {
        this.log.debug("recevied ping frame");
        ctx.write(new PongWebSocketFrame(frame.content()));
    }
    else if (frame instanceof TextWebSocketFrame)
    {
        this.log.debug("recevied text frame");
        this.handleTextWebSocketFrame(ctx,(TextWebSocketFrame)frame);
    }
    else
    {
        this.log.info("recevied unkNown incompatible frame");
        ctx.close();
    }
}

private void handleWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame frame) {
    _logger.debug("Handling websocket frame");
    // Check for closing frame
    if (frame instanceof CloseWebSocketFrame) {
        _handshaker.close(ctx.channel(),(CloseWebSocketFrame) frame.retain());
        return;
    }
    if (frame instanceof PingWebSocketFrame) {
        ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
        return;
    }
    if (!(frame instanceof TextWebSocketFrame)) {
        throw new UnsupportedOperationException(String.format("%s frame types not supported",frame.getClass()
                .getName()));
    }

    String request = ((TextWebSocketFrame) frame).text();
    _logger.debug("{} received {}",request);
    _messageQueue.add(frame.content().retain());
    //ctx.channel().write(new TextWebSocketFrame(request.toupperCase()));
}

private void handleWebSocketFrame(ChannelHandlerContext ctx,request));
        }
        ctx.channel().write(new TextWebSocketFrame(request.toupperCase()));
    }

private void handleWebSocketFrame(ChannelHandlerContext ctx,request));
        }
        ctx.channel().write(new TextWebSocketFrame(request.toupperCase()));
    }

private void handleWebSocketFrame(ChannelHandlerContext ctx,frame.content()));
    } else if (frame instanceof TextWebSocketFrame) {
        ctx.write(frame);
    } else if (frame instanceof BinaryWebSocketFrame) {
        ctx.write(frame);
    } else if (frame instanceof ContinuationWebSocketFrame) {
        ctx.write(frame);
    } else if (frame instanceof PongWebSocketFrame) {
        frame.release();
        // Ignore
    } else {
        throw new UnsupportedOperationException(String.format("%s frame types not supported",frame.getClass()
                .getName()));
    }
}

private void handleWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame frame) {
  // Check for closing frame
  if (frame instanceof CloseWebSocketFrame) {
    handshaker.close(ctx.channel(),(CloseWebSocketFrame) frame.retain());
    return;
  }
  if (frame instanceof PingWebSocketFrame) {
    ctx.write(new PongWebSocketFrame(frame.content().retain()));
    return;
  }
  if (frame instanceof TextWebSocketFrame) {
    // Echo the frame
    ctx.write(frame.retain());
    return;
  }
  if (frame instanceof BinaryWebSocketFrame) {
    // Echo the frame
    ctx.write(frame.retain());
  }
}

private void handleWebSocketFrame(ChannelHandlerContext ctx,(CloseWebSocketFrame) frame.retain());
        return;
    }
    if (frame instanceof PingWebSocketFrame) {
        ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
        return;
    }

    if (frame instanceof BinaryWebSocketFrame)
        try {
            this.connection.onMessage(((BinaryWebSocketFrame) frame).content().retain());
        } catch (Exception e) {
            logger.error("onMessage error",e);
            handshaker.close(ctx.channel(),new CloseWebSocketFrame(true,frame.content().clear()
                                    .writeShort(1000)
                                    .writeBytes(e.getMessage().getBytes(CharsetUtil.UTF_8))
                                    .retain()));
        }
}

private void handleWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame msg) throws Exception {
  if (log.isDebugEnabled())
    log.debug("Received {} WebSocketFrame: {} from channel: {}",getTransportType().getName(),msg,ctx.channel());

  if (msg instanceof CloseWebSocketFrame) {
    sessionIdByChannel.remove(ctx.channel());
    ChannelFuture f = ctx.writeAndFlush(msg);
    f.addListener(ChannelFutureListener.CLOSE);
  } else if (msg instanceof PingWebSocketFrame) {
    ctx.writeAndFlush(new PongWebSocketFrame(msg.content()));
  } else if (msg instanceof TextWebSocketFrame || msg instanceof BinaryWebSocketFrame){
    Packet packet = PacketDecoder.decodePacket(msg.content());
    packet.setTransportType(getTransportType());
    String sessionId = sessionIdByChannel.get(ctx.channel());
    packet.setSessionId(sessionId);
    msg.release();
    ctx.fireChannelRead(packet);
  } else {
    msg.release();
    log.warn("{} frame type is not supported",msg.getClass().getName());
  }
}

@Override
protected void channelRead0(ChannelHandlerContext ctx,Object msg) throws Exception {
    LOG.info("Received msg: {}",msg);
    if (!this.handshaker.isHandshakeComplete()) {
        this.handshaker.finishHandshake(ctx.channel(),(FullHttpResponse) msg);
        LOG.info("Client connected.");
        this.connected = true;
        this.handshakeFuture.setSuccess();
        return;
    }
    if (msg instanceof FullHttpResponse) {
        throw new IllegalStateException("Unexpected response: " + msg.toString());
    }
    WebSocketFrame frame = (WebSocketFrame) msg;
    if (frame instanceof TextWebSocketFrame) {
        synchronized (responses) {
            responses.add(((TextWebSocketFrame) frame).text().getBytes(StandardCharsets.UTF_8));
        }
    } else if (frame instanceof BinaryWebSocketFrame) {
        ByteBuf buf = frame.content();
        byte[] b = new byte[buf.readableBytes()];
        buf.readBytes(b);
        synchronized (responses) {
            responses.add(b);
        }
    } else if (frame instanceof PingWebSocketFrame) {
        LOG.info("Returning pong message");
        ctx.writeAndFlush(new PongWebSocketFrame());
    } else if (frame instanceof CloseWebSocketFrame) {
        LOG.info("Received message from server to close the channel.");
        ctx.close();
    } else {
        LOG.warn("Unhandled frame type received: " + frame.getClass());
    }
}

@Override
protected void channelRead0(ChannelHandlerContext ctx,Object msg)
        throws Exception {
    Channel ch = ctx.channel();
    if (!handshaker.isHandshakeComplete()) {
        handshaker.finishHandshake(ch,(FullHttpResponse) msg);
        System.out.println("WebSocket Client connected!");
        handshakeFuture.setSuccess();
        return;
    }
    if (msg instanceof FullHttpResponse) {
        FullHttpResponse response = (FullHttpResponse) msg;
        throw new IllegalStateException(
                "Unexpected FullHttpResponse (getStatus=" + response.getStatus() +
                    ",content=" + response.content().toString(CharsetUtil.UTF_8) + ')');
    } else if (msg instanceof WebSocketFrame) {
        WebSocketFrame frame = (WebSocketFrame) msg;
        if (msg instanceof TextWebSocketFrame) {
            TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
            System.out.println("WebSocket Client received message: " + textFrame.text());
        } else if (msg instanceof PongWebSocketFrame) {
            System.out.println("WebSocket Client received pong");
        } else if (msg instanceof CloseWebSocketFrame) {
            System.out.println("WebSocket Client received closing");
            ch.close();
        }
    }
}

@Override
protected void channelRead0(ChannelHandlerContext ctx,(FullHttpResponse) msg);
        LOG.info("Client connected.");
        this.connected = true;
        this.handshakeFuture.setSuccess();
        return;
    }
    if (msg instanceof FullHttpResponse) {
        throw new IllegalStateException("Unexpected response: " + msg.toString());
    }
    WebSocketFrame frame = (WebSocketFrame) msg;
    if (frame instanceof TextWebSocketFrame) {
        synchronized (responses) {
            responses.add(((TextWebSocketFrame) frame).text());
        }
    } else if (frame instanceof PingWebSocketFrame) {
        LOG.info("Returning pong message");
        ctx.writeAndFlush(new PongWebSocketFrame());
    } else if (frame instanceof CloseWebSocketFrame) {
        LOG.info("Received message from server to close the channel.");
        ctx.close();
    } else {
        LOG.warn("Unhandled frame type received: " + frame.getClass());
    }
}

@Override
protected void channelRead0(final ChannelHandlerContext ctx,final Object msg) throws Exception {
    final Channel ch = ctx.channel();
    if (!handshaker.isHandshakeComplete()) {
        // web socket client connected
        handshaker.finishHandshake(ch,(FullHttpResponse) msg);
        handshakeFuture.setSuccess();
        return;
    }

    if (msg instanceof FullHttpResponse) {
        final FullHttpResponse response = (FullHttpResponse) msg;
        throw new Exception("Unexpected FullHttpResponse (getStatus=" + response.getStatus() + ",content="
                + response.content().toString(CharsetUtil.UTF_8) + ')');
    }

    // a close frame doesn't mean much here.  errors raised from closed channels will mark the host as dead
    final WebSocketFrame frame = (WebSocketFrame) msg;
    if (frame instanceof TextWebSocketFrame) {
        ctx.fireChannelRead(frame.retain(2));
    } else if (frame instanceof PongWebSocketFrame) {
    } else if (frame instanceof BinaryWebSocketFrame) {
        ctx.fireChannelRead(frame.retain(2));
    } else if (frame instanceof CloseWebSocketFrame)
        ch.close();

}

@Override
public void channelRead0(ChannelHandlerContext ctx,Object msg) throws Exception {

    Channel ch = ctx.channel();
    if (!handshaker.isHandshakeComplete()) {
        handshaker.finishHandshake(ch,(FullHttpResponse) msg);
        handshakeFuture.setSuccess();

        //connection is opened.
        client.onopen(handshaker);

        return;
    }

    if (msg instanceof FullHttpResponse) {
        FullHttpResponse response = (FullHttpResponse) msg;
        throw new IllegalStateException(
                "Unexpected FullHttpResponse (getStatus=" + response.status() +
                ",content=" + response.content().toString(CharsetUtil.UTF_8) + ')');
    }

    WebSocketFrame frame = (WebSocketFrame) msg;
    if (frame instanceof TextWebSocketFrame) {
        TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
        client.onMessage(textFrame.text());

    } else if (frame instanceof PongWebSocketFrame) {
        /*
         * placeholder.  maybe add onPong method to the RhinoClient
         */
    } else if (frame instanceof CloseWebSocketFrame) {
        client.onClose();
        ch.close();
    }
}

@Override
public void channelRead0(ChannelHandlerContext ctx,Object msg) throws Exception {
    Channel ch = ctx.channel();
    if (!handshaker.isHandshakeComplete()) {
        handshaker.finishHandshake(ch,(FullHttpResponse) msg);
        System.out.println("WebSocket Client connected!");
        handshakeFuture.setSuccess();
        return;
    }

    if (msg instanceof FullHttpResponse) {
        FullHttpResponse response = (FullHttpResponse) msg;
        throw new IllegalStateException(
                "Unexpected FullHttpResponse (getStatus=" + response.status() +
                        ",content=" + response.content().toString(CharsetUtil.UTF_8) + ')');
    }

    WebSocketFrame frame = (WebSocketFrame) msg;
    if (frame instanceof TextWebSocketFrame) {
        TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
        System.out.println("WebSocket Client received message: " + textFrame.text());
    } else if (frame instanceof PongWebSocketFrame) {
        System.out.println("WebSocket Client received pong");
    } else if (frame instanceof CloseWebSocketFrame) {
        System.out.println("WebSocket Client received closing");
        ch.close();
    }
}

protected void handleWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame frame) {
   logger.debug("Received incoming frame [{}]",frame.getClass().getName());
   // Check for closing frame
   if (frame instanceof CloseWebSocketFrame) {
      if (frameBuffer != null) {
          handleMessageCompleted(ctx,frameBuffer.toString());
      }
      handshaker.close(ctx.channel(),(CloseWebSocketFrame) frame.retain());
      return;
   }

   if (frame instanceof PingWebSocketFrame) {
      ctx.channel().writeAndFlush(new PongWebSocketFrame(frame.content().retain()));
      return;
   }

   if (frame instanceof PongWebSocketFrame) {
      logger.info("Pong frame received");
      return;
   }

   if (frame instanceof TextWebSocketFrame) {
      frameBuffer = new StringBuilder();
      frameBuffer.append(((TextWebSocketFrame)frame).text());
   } else if (frame instanceof ContinuationWebSocketFrame) {
      if (frameBuffer != null) {
         frameBuffer.append(((ContinuationWebSocketFrame)frame).text());
      } else {
         logger.warn("Continuation frame received without initial frame.");
      }
   } else {
      throw new UnsupportedOperationException(String.format("%s frame types not supported",frame.getClass().getName()));
   }

   // Check if Text or Continuation Frame is final fragment and handle if needed.
   if (frame.isFinalFragment()) {
      handleMessageCompleted(ctx,frameBuffer.toString());
      frameBuffer = null;
   }
}

private void handleWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame frame) {

// 判断是否是关闭链路的指令
if (frame instanceof CloseWebSocketFrame) {
    handshaker.close(ctx.channel(),(CloseWebSocketFrame) frame.retain());
    return;
}
// 判断是否是Ping消息
if (frame instanceof PingWebSocketFrame) {
    ctx.channel().write(
        new PongWebSocketFrame(frame.content().retain()));
    return;
}
// 本例程仅支持文本消息，不支持二进制消息
if (!(frame instanceof TextWebSocketFrame)) {
    throw new UnsupportedOperationException(String.format(
        "%s frame types not supported",frame.getClass().getName()));
}

// 返回应答消息
String request = ((TextWebSocketFrame) frame).text();
if (logger.isLoggable(Level.FINE)) {
    logger.fine(String.format("%s received %s",request));
}
ctx.channel().write(
    new TextWebSocketFrame(request
        + ",欢迎使用Netty WebSocket服务，现在时刻："
        + new java.util.Date().toString()));
   }

@Override
public void channelRead0(ChannelHandlerContext ctx,(FullHttpResponse) msg);
        System.out.println("WebSocket Client connected!");
        handshakeFuture.setSuccess();
        return;
    }

    if (msg instanceof FullHttpResponse) {
        FullHttpResponse response = (FullHttpResponse) msg;
        throw new IllegalStateException(
                "Unexpected FullHttpResponse (getStatus=" + response.getStatus() +
                        ",content=" + response.content().toString(CharsetUtil.UTF_8) + ')');
    }

    WebSocketFrame frame = (WebSocketFrame) msg;
    if (frame instanceof TextWebSocketFrame) {
        TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
        System.out.println("WebSocket Client received message: " + textFrame.text());
    } else if (frame instanceof PongWebSocketFrame) {
        System.out.println("WebSocket Client received pong");
    } else if (frame instanceof CloseWebSocketFrame) {
        System.out.println("WebSocket Client received closing");
        ch.close();
    }
}

@Override
public void channelRead0(ChannelHandlerContext ctx,(FullHttpResponse) msg);
        logger.debug("WebSocket Client connected!");
        handshakeFuture.setSuccess();
        return;
    }

    if (msg instanceof FullHttpResponse) {
        FullHttpResponse response = (FullHttpResponse) msg;
        throw new IllegalStateException(
                "Unexpected FullHttpResponse (getStatus=" + response.status() +
                ",content=" + response.content().toString(CharsetUtil.UTF_8) + ')');
    }

    WebSocketFrame frame = (WebSocketFrame) msg;
    if (frame instanceof TextWebSocketFrame) {
        TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
        logger.debug("WebSocket Client received text message: " + textFrame.text());
        textReceived = textFrame.text();
    } else if (frame instanceof BinaryWebSocketFrame) {
        BinaryWebSocketFrame binaryFrame = (BinaryWebSocketFrame) frame;
        bufferReceived = binaryFrame.content().nioBuffer();
        logger.debug("WebSocket Client received  binary message: " + bufferReceived.toString());
    } else if (frame instanceof PongWebSocketFrame) {
        logger.debug("WebSocket Client received pong");
        PongWebSocketFrame pongFrame = (PongWebSocketFrame) frame;
        bufferReceived = pongFrame.content().nioBuffer();
    } else if (frame instanceof CloseWebSocketFrame) {
        logger.debug("WebSocket Client received closing");
        ch.close();
    }
}

private void notifyPongMessage(PongWebSocketFrame pongWebSocketFrame) throws ServerConnectorException {
    //Control message for WebSocket is Pong Message
    ByteBuf byteBuf = pongWebSocketFrame.content();
    ByteBuffer byteBuffer = byteBuf.nioBuffer();
    WebSocketMessageImpl webSocketControlMessage =
            new WebSocketControlMessageImpl(WebSocketControlSignal.PONG,byteBuffer);
    webSocketControlMessage = setupCommonProperties(webSocketControlMessage);
    connectorFuture.notifyWSListener((WebSocketControlMessage) webSocketControlMessage);
}

@Override
public void channelRead0(ChannelHandlerContext ctx,URISyntaxException,ServerConnectorException {
    Channel ch = ctx.channel();
    if (!handshaker.isHandshakeComplete()) {
        handshaker.finishHandshake(ch,(FullHttpResponse) msg);
        log.debug("WebSocket Client connected!");
        handshakeFuture.setSuccess();
        channelSession = WebSocketUtil.getSession(ctx,isSecure,requestedUri);
        return;
    }

    if (msg instanceof FullHttpResponse) {
        FullHttpResponse response = (FullHttpResponse) msg;
        throw new IllegalStateException(
                "Unexpected FullHttpResponse (getStatus=" + response.status() +
                        ",content=" + response.content().toString(CharsetUtil.UTF_8) + ')');
    }
    WebSocketFrame frame = (WebSocketFrame) msg;
    if (frame instanceof TextWebSocketFrame) {
        notifyTextMessage((TextWebSocketFrame) frame,ctx);
    } else if (frame instanceof BinaryWebSocketFrame) {
        notifyBinaryMessage((BinaryWebSocketFrame) frame,ctx);
    } else if (frame instanceof PongWebSocketFrame) {
        notifyPongMessage((PongWebSocketFrame) frame,ctx);
    } else if (frame instanceof PingWebSocketFrame) {
        notifyPingMessage((PingWebSocketFrame) frame,ctx);
    } else if (frame instanceof CloseWebSocketFrame) {
        if (channelSession != null) {
            channelSession.setIsOpen(false);
        }
        notifyCloseMessage((CloseWebSocketFrame) frame,ctx);
        ch.close();
    } else {
        throw new UnkNownWebSocketFrameTypeException("Cannot identify the WebSocket frame type");
    }
}

private void notifyPongMessage(PongWebSocketFrame pongWebSocketFrame,ChannelHandlerContext ctx)
        throws ServerConnectorException {
    //Control message for WebSocket is Pong Message
    ByteBuf byteBuf = pongWebSocketFrame.content();
    ByteBuffer byteBuffer = byteBuf.nioBuffer();
    WebSocketMessageImpl webSocketControlMessage =
            new WebSocketControlMessageImpl(WebSocketControlSignal.PONG,byteBuffer);
    webSocketControlMessage = setupCommonProperties(webSocketControlMessage,ctx);
    connectorListener.onMessage((WebSocketControlMessage) webSocketControlMessage);
}

private boolean handleWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame frame) {

      // Check for closing frame
      if (frame instanceof CloseWebSocketFrame) {
         this.handshaker.close(ctx.channel(),((CloseWebSocketFrame) frame).retain());
         return false;
      } else if (frame instanceof PingWebSocketFrame) {
         ctx.writeAndFlush(new PongWebSocketFrame(frame.content().retain()));
         return false;
      } else if (!(frame instanceof TextWebSocketFrame) && !(frame instanceof BinaryWebSocketFrame)) {
         throw new UnsupportedOperationException(String.format("%s frame types not supported",frame.getClass().getName()));
      }
      return true;
   }

/**
     * 适配
     */
    @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,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,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,new DefaultHttpHeaders())));
//              ch.pipeline().addLast(new WebSocketConnectedClientHandler(handler));
//            }
//        };
        return result;
    }

@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,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));
}

@Override
public void configure(final ChannelPipeline pipeline) {
    final String scheme = connection.getUri().getScheme();
    if (!"ws".equalsIgnoreCase(scheme) && !"wss".equalsIgnoreCase(scheme))
        throw new IllegalStateException("Unsupported scheme (only ws: or wss: supported): " + scheme);

    if (!supportsSsl() && "wss".equalsIgnoreCase(scheme))
        throw new IllegalStateException("To use wss scheme ensure that enableSsl is set to true in configuration");

    final int maxContentLength = cluster.connectionPoolSettings().maxContentLength;
    handler = new WebSocketClientHandler(
            WebSocketClientHandshakerFactory.newHandshaker(
                    connection.getUri(),HttpHeaders.EMPTY_HEADERS,maxContentLength));

    pipeline.addLast("http-codec",new HttpClientCodec());
    pipeline.addLast("aggregator",new HttpObjectAggregator(maxContentLength));
    pipeline.addLast("ws-handler",handler);
    pipeline.addLast("gremlin-encoder",webSocketGremlinRequestEncoder);
    pipeline.addLast("gremlin-decoder",webSocketGremlinResponseDecoder);
}

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,new DefaultHttpHeaders()));
}

@Override
public void configure(final ChannelPipeline pipeline) {
    final String scheme = connection.getUri().getScheme();
    if (!"ws".equalsIgnoreCase(scheme) && !"wss".equalsIgnoreCase(scheme))
        throw new IllegalStateException("Unsupported scheme (only ws: or wss: supported): " + scheme);

    if (!supportsSsl() && "wss".equalsIgnoreCase(scheme))
        throw new IllegalStateException("To use wss scheme ensure that enableSsl is set to true in configuration");

    final int maxContentLength = cluster.connectionPoolSettings().maxContentLength;
    handler = new WebSocketClientHandler(
            WebSocketClientHandshakerFactory.newHandshaker(
                    connection.getUri(),webSocketGremlinResponseDecoder);
}

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,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),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,(String) null,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");
    }
}

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

    Connector con = mac.getConnector("root",TimeUnit.MILLISECONDS);
        LOG.debug("Waiting for Handshake to complete");
    }
}

public WebSocketClient(final URI uri) {
    super("ws-client-%d");
    final Bootstrap b = new Bootstrap().group(group);
    b.option(ChannelOption.ALLOCATOR,PooledByteBufAllocator.DEFAULT);

    final String protocol = uri.getScheme();
    if (!"ws".equals(protocol))
        throw new IllegalArgumentException("Unsupported protocol: " + protocol);

    try {
        final WebSocketClientHandler wsHandler =
                new WebSocketClientHandler(
                        WebSocketClientHandshakerFactory.newHandshaker(
                                uri,new DefaultHttpHeaders()));
        final MessageSerializer serializer = new GryoMessageSerializerV1d0();
        b.channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(final SocketChannel ch) {
                        final ChannelPipeline p = ch.pipeline();
                        p.addLast(
                                new HttpClientCodec(),new HttpObjectAggregator(8192),wsHandler,new WebSocketGremlinRequestEncoder(true,serializer),new WebSocketGremlinResponseDecoder(serializer),callbackResponseHandler);
                    }
                });

        channel = b.connect(uri.getHost(),uri.getPort()).sync().channel();
        wsHandler.handshakeFuture().sync();
    } catch (Exception ex) {
        throw new RuntimeException(ex);
    }
}

public void openConnection() throws InterruptedException{
    Bootstrap bootstrap = new Bootstrap();

    final WebSocketClientHandler handler =
            new WebSocketClientHandler(
                    WebSocketClientHandshakerFactory.newHandshaker(
                            mUri,WebSocketVersion.V08,new DefaultHttpHeaders()));

    bootstrap.group(mGroup)
    .channel(NioSocketChannel.class)
    .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel socketChannel){
                    ChannelPipeline channelPipeline =
                            socketChannel.pipeline();
                    channelPipeline.addLast(mSslContext.newHandler(
                            socketChannel.alloc(),mUri.getHost(),PORT));
                    channelPipeline.addLast(new HttpClientCodec(),handler);
                }
            });

    mChannel = bootstrap.connect(mUri.getHost(),PORT).sync().channel();
    handler.handshakeFuture().sync();
    setConnected(Boolean.TRUE);
}

/**
 * Inits the.
 *
 * @throws URISyntaxException
 *           the URI Syntax exception
 */
public void init() throws URISyntaxException {
  handshaker = WebSocketClientHandshakerFactory.newHandshaker(createUri(),customHeaders);
}

public WebSocketClient(final URI uri) {
    super("ws-client-%d");
    final Bootstrap b = new Bootstrap().group(group);
    b.option(ChannelOption.ALLOCATOR,65536));
        final MessageSerializer serializer = new GryoMessageSerializerV3d0();
        b.channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(final SocketChannel ch) {
                        final ChannelPipeline p = ch.pipeline();
                        p.addLast(
                                new HttpClientCodec(),new HttpObjectAggregator(65536),uri.getPort()).sync().channel();
        wsHandler.handshakeFuture().get(10000,TimeUnit.MILLISECONDS);
    } catch (Exception ex) {
        throw new RuntimeException(ex);
    }
}

/**
 * Creates a new WebSocketClientHandler that manages the BlaubotWebsocketConnection
 * @param uri                  The uri to connect with
 * @param remoteUniquedeviceid the unique device id of the device we are connecting to
 * @param listenerReference    a reference Object that handles the connection listener
 */
public WebsocketClientHandler(URI uri,String remoteUniquedeviceid,atomicreference<IBlaubotIncomingConnectionListener> listenerReference) {
    // Connect with V13 (RFC 6455 aka HyBi-17).
    // other options are V08 or V00.
    // If V00 is used,ping is not supported and remember to change
    // HttpResponseDecoder to WebSocketHttpResponseDecoder in the pipeline.
    this.handshaker = WebSocketClientHandshakerFactory.newHandshaker(uri,new DefaultHttpHeaders(),BlaubotWebsocketAdapter.MAX_WEBSOCKET_FRAME_SIZE);
    this.remoteDeviceUniquedeviceid = remoteUniquedeviceid;
    this.incomingConnectionListenerReference = listenerReference;
}

public ClientHandler(SocketInfo info) {
    super(info,false);

    HttpHeaders customHeaders = new DefaultHttpHeaders();
    customHeaders.add("x-DivConq-Layer","dcPrivate");

    this.handshaker = WebSocketClientHandshakerFactory.newHandshaker(info.getUri(),customHeaders);
}

@Override
protected void preparePipeline(ChannelPipeline pipeline) {
    pipeline.addLast(
            new HttpClientCodec(),new HttpObjectAggregator(8192));
    pipeline.addLast("handler",new WebSocketClientHandler(this,WebSocketClientHandshakerFactory.newHandshaker(
                            this.uri,new DefaultHttpHeaders())));
}

public void run() throws Exception {
        URI uri = new URI(url);

        String scheme = uri.getScheme() == null? "ws" : uri.getScheme();
        final String host = uri.getHost() == null? "127.0.0.1" : uri.getHost();
        final int port;
        if (uri.getPort() == -1) {
            if ("ws".equalsIgnoreCase(scheme)) {
                port = 80;
            } else if ("wss".equalsIgnoreCase(scheme)) {
                port = 443;
            } else {
                port = -1;
            }
        } else {
            port = uri.getPort();
        }

        if (!"ws".equalsIgnoreCase(scheme) && !"wss".equalsIgnoreCase(scheme)) {
            System.err.println("Only WS(S) is supported.");
            return;
        }

        final boolean ssl = "wss".equalsIgnoreCase(scheme);
        final SslContext sslCtx;
        if (ssl) {
            sslCtx = SslContextBuilder.forClient()
                .trustManager(InsecureTrustManagerFactory.INSTANCE).build();
        } else {
            sslCtx = null;
        }

        EventLoopGroup group = new NioEventLoopGroup();
        try {

            // Connect with V13 (RFC 6455 aka HyBi-17). You can change it to V08 or V00.
            // If you change it to V00,ping is not supported and remember to change
            // HttpResponseDecoder to WebSocketHttpResponseDecoder in the pipeline.
            this.handler = new WebSocketClientHandler(uid,WebSocketClientHandshakerFactory.newHandshaker(
                            uri,new DefaultHttpHeaders()));

            Bootstrap b = new Bootstrap();
            b.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline p = ch.pipeline();
                            if (sslCtx != null) {
                                p.addLast(sslCtx.newHandler(ch.alloc(),host,port));
                            }

                            p.addLast(
                                    new HttpClientCodec(),new HttpObjectAggregator(8192)
                            );
//
//                            p.addLast("frameDecoder",new ProtobufVarint32FrameDecoder());
//                            p.addLast("protobufDecoder",new ProtobufDecoder(Response.HeshResMessage.getDefaultInstance()));
//
//                            p.addLast("frameEncoder",new ProtobufVarint32LengthFieldPrepender());
//                            p.addLast("protobufEncoder",new ProtobufEncoder());

                            p.addLast(handler);

                        }
                    });

            this.channel = b.connect(uri.getHost(),port).sync().channel();
//            handler.handshakeFuture().sync();

        } catch (Exception e) {
            e.printstacktrace();
        } finally {
//            group.shutdownGracefully();
        }
    }

public WebSocketClientHandler(URI uri) {
    handshaker = WebSocketClientHandshakerFactory.newHandshaker(uri,new DefaultHttpHeaders());
}

public WebSocketHandler(final AbstractWebSocketConnection webSocketConnection)
{
    this.webSocketConnection = webSocketConnection;
    this.handshaker = WebSocketClientHandshakerFactory.newHandshaker(webSocketConnection.getUri(),new DefaultHttpHeaders());
}

/**
 * @return true if the handshake is done properly.
 * @throws URISyntaxException   throws if there is an error in the URI Syntax.
 * @throws InterruptedException throws if the connecting the server is interrupted.
 */
public boolean handhshake() throws InterruptedException,URISyntaxException,SSLException,ProtocolException {
    boolean isSuccess;
    URI uri = new URI(url);
    String scheme = uri.getScheme() == null ? "ws" : uri.getScheme();
    final String host = uri.getHost() == null ? "127.0.0.1" : uri.getHost();
    final int port;
    if (uri.getPort() == -1) {
        if ("ws".equalsIgnoreCase(scheme)) {
            port = 80;
        } else if ("wss".equalsIgnoreCase(scheme)) {
            port = 443;
        } else {
            port = -1;
        }
    } else {
        port = uri.getPort();
    }

    if (!"ws".equalsIgnoreCase(scheme) && !"wss".equalsIgnoreCase(scheme)) {
        logger.error("Only WS(S) is supported.");
        return false;
    }

    final boolean ssl = "wss".equalsIgnoreCase(scheme);
    final SslContext sslCtx;
    if (ssl) {
        sslCtx = SslContextBuilder.forClient().trustManager(InsecureTrustManagerFactory.INSTANCE).build();
    } else {
        sslCtx = null;
    }

    group = new NioEventLoopGroup();

    HttpHeaders headers = new DefaultHttpHeaders();
    for (Map.Entry<String,String> entry : customHeaders.entrySet()) {
        headers.add(entry.getKey(),entry.getValue());
    }
    // Connect with V13 (RFC 6455 aka HyBi-17). You can change it to V08 or V00.
    // If you change it to V00,ping is not supported and remember to change
    // HttpResponseDecoder to WebSocketHttpResponseDecoder in the pipeline.
    handler = new WebSocketClientHandler(
            WebSocketClientHandshakerFactory.newHandshaker(uri,subProtocol,headers),latch);

    Bootstrap bootstrap = new Bootstrap();
    bootstrap.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) {
            ChannelPipeline p = ch.pipeline();
            if (sslCtx != null) {
                p.addLast(sslCtx.newHandler(ch.alloc(),port));
            }
            p.addLast(new HttpClientCodec(),WebSocketClientCompressionHandler.INSTANCE,handler);
        }
    });

    channel = bootstrap.connect(uri.getHost(),port).sync().channel();
    isSuccess = handler.handshakeFuture().sync().isSuccess();
    logger.info("WebSocket Handshake successful : " + isSuccess);
    return isSuccess;
}

@Override
public void connect() throws InterruptedException{
    // Connect with V13 (RFC 6455 aka HyBi-17). You can change it to V08 or V00.
    // If you change it to V00,ping is not supported and remember to change
    // HttpResponseDecoder to WebSocketHttpResponseDecoder in the pipeline.
    handler =
            new WebSocketClientHandler(
                    WebSocketClientHandshakerFactory.newHandshaker(uri,this.getMaxPayload()));


    //make sure the handler has a refernce to this object.
    handler.setClient(this);

    Bootstrap clientBoot = new Bootstrap();
    clientBoot.group(group)
    .channel(NioSocketChannel.class)
    .handler(new ChannelInitializer<SocketChannel>() {

        @Override
        protected void initChannel(SocketChannel ch) {
            ChannelPipeline p = ch.pipeline();
            SSLEngine sslEngine=null;
            if(AbstractClient.this.isEncrypted()){
                if(sslContext == null){
                    sslEngine = new SSLFactory().createClientSslCtx(Config.getInstance()).newEngine(ch.alloc(),uri.getHost(),uri.getPort());
                }else{
                    sslEngine = sslContext.newEngine(ch.alloc(),uri.getPort());
                }

                sslEngine.setEnabledProtocols(Const.TLS_PROTOCOLS);
                sslEngine.setUseClientMode(true);
                p.addLast(new SslHandler(sslEngine));
            }

            p.addLast( new HttpClientCodec());
            p.addLast(new HttpObjectAggregator(8192));
            if(AbstractClient.this.isCompress()){
                p.addLast(WebSocketClientCompressionHandler.INSTANCE);
            }
            p.addLast(handler);


        }
    });


    this.ch = clientBoot.connect(uri.getHost(),uri.getPort()).sync().channel();
    handler.handshakeFuture().sync();   

}

/**
 * @return true if the handshake is done properly.
 * @throws URISyntaxException throws if there is an error in the URI Syntax.
 * @throws InterruptedException throws if the connecting the server is interrupted.
 */
public boolean handhshake() throws InterruptedException,SSLException {
    boolean isDone;
    URI uri = new URI(url);
    String scheme = uri.getScheme() == null ? "ws" : uri.getScheme();
    final String host = uri.getHost() == null ? "127.0.0.1" : uri.getHost();
    final int port;
    if (uri.getPort() == -1) {
        if ("ws".equalsIgnoreCase(scheme)) {
            port = 80;
        } else if ("wss".equalsIgnoreCase(scheme)) {
            port = 443;
        } else {
            port = -1;
        }
    } else {
        port = uri.getPort();
    }

    if (!"ws".equalsIgnoreCase(scheme) && !"wss".equalsIgnoreCase(scheme)) {
        logger.error("Only WS(S) is supported.");
        return false;
    }

    final boolean ssl = "wss".equalsIgnoreCase(scheme);
    final SslContext sslCtx;
    if (ssl) {
        sslCtx = SslContextBuilder.forClient()
                                  .trustManager(InsecureTrustManagerFactory.INSTANCE).build();
    } else {
        sslCtx = null;
    }

    group = new NioEventLoopGroup();

    HttpHeaders headers = new DefaultHttpHeaders();
    customHeaders.entrySet().forEach(
            header -> headers.add(header.getKey(),header.getValue())
    );
    try {
        // Connect with V13 (RFC 6455 aka HyBi-17). You can change it to V08 or V00.
        // If you change it to V00,ping is not supported and remember to change
        // HttpResponseDecoder to WebSocketHttpResponseDecoder in the pipeline.
        handler =
                new WebSocketClientHandler(
                        WebSocketClientHandshakerFactory.newHandshaker(
                                uri,headers));

        Bootstrap b = new Bootstrap();
        b.group(group)
         .channel(NioSocketChannel.class)
         .handler(new ChannelInitializer<SocketChannel>() {
             @Override
             protected void initChannel(SocketChannel ch) {
                 ChannelPipeline p = ch.pipeline();
                 if (sslCtx != null) {
                     p.addLast(sslCtx.newHandler(ch.alloc(),port));
                 }
                 p.addLast(
                         new HttpClientCodec(),handler);
             }
         });

        channel = b.connect(uri.getHost(),port).sync().channel();
        isDone = handler.handshakeFuture().sync().isSuccess();
        logger.debug("WebSocket Handshake successful : " + isDone);
        return isDone;
    } catch (Exception e) {
        logger.error("Handshake unsuccessful : " + e.getMessage(),e);
        return false;
    }
}

@Override
public void connect(WsClient client) {
    if (client == null) {
        throw new NullPointerException("client");
    }
    final URLInfo url = client.getUrl();
    String full = url.protocol + "://" + url.host
            + ":" + url.port + url.path;
    URI uri;
    try {
        uri = new URI(full);
    } catch (URISyntaxException e) {
        throw new RuntimeException(e);
    }
    WebSocketVersion v = WebSocketVersion.V13;
    HttpHeaders h = new DefaultHttpHeaders();
    final WebSocketClientHandshaker wsch = WebSocketClientHandshakerFactory
            .newHandshaker(uri,v,h,Integer.MAX_VALUE);
    final WebSocketHandler handler = new WebSocketHandler(wsch,client);

    Bootstrap b = new Bootstrap();
    b.group(Sharedobjects.getLoop());
    b.channel(NioSocketChannel.class);
    b.handler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline p = ch.pipeline();
            if (url.secure) {
                TrustManagerFactory man = InsecureTrustManagerFactory.INSTANCE;
                SslContext con = SslContext.newClientContext(man);
                p.addLast(con.newHandler(ch.alloc()));
            }

            p.addLast(new HttpClientCodec());
            p.addLast(new HttpObjectAggregator(8192));
            p.addLast(handler);
        }
    });

    ChannelFuture fut = b.connect(url.host,url.port);
    fut.syncUninterruptibly();
    handler.handshakeFuture().syncUninterruptibly();
}

@Override
public void connect(String url) throws Exception {
  URI uri = new URI(url);
  setConnected(false);

  String scheme = uri.getScheme() == null ? "http" : uri.getScheme();
  final String host = uri.getHost() == null ? "127.0.0.1" : uri.getHost();
  final int port;
  if (uri.getPort() == -1) {
    if ("http".equalsIgnoreCase(scheme)) {
      port = 80;
    } else if ("https".equalsIgnoreCase(scheme)) {
      port = 443;
    } else {
      port = -1;
    }
  } else {
    port = uri.getPort();
  }

  if (!"ws".equalsIgnoreCase(scheme) && !"wss".equalsIgnoreCase(scheme)) {
    Notifications.Bus.notify(
        new Notification(
            "Websocket Client","Unable to connect","Only WS(S) is supported.",NotificationType.ERROR)
    );
    return;
  }

  EventLoopGroup group = new NioEventLoopGroup();
  try {
    WebSocketClientHandshaker handshaker = WebSocketClientHandshakerFactory.newHandshaker(
        uri,new DefaultHttpHeaders());
    WebSocketClientHandler webSocketClientHandler = new WebSocketClientHandler(handshaker);

    Bootstrap bootstrap = new Bootstrap();
    bootstrap.group(group)
        .channel(NioSocketChannel.class)
        .handler(new ChannelInitializer<SocketChannel>() {
          @Override
          protected void initChannel(SocketChannel ch) {
            ChannelPipeline p = ch.pipeline();

            p.addLast(
                new HttpClientCodec(),webSocketClientHandler);
          }
        });

    channel = bootstrap.connect(uri.getHost(),port).sync().channel();
    webSocketClientHandler.handshakeFuture().sync();
    setConnected(true);

    for (; ; );
  } finally {
    group.shutdownGracefully();
    setConnected(false);
  }
}

/**
 * @return true if the handshake is done properly.
 * @throws URISyntaxException throws if there is an error in the URI Syntax.
 * @throws InterruptedException throws if the connecting the server is interrupted.
 */
public boolean handhshake() throws InterruptedException,ProtocolException {
    boolean isSuccess;
    URI uri = new URI(url);
    String scheme = uri.getScheme() == null ? "ws" : uri.getScheme();
    final String host = uri.getHost() == null ? "127.0.0.1" : uri.getHost();
    final int port;
    if (uri.getPort() == -1) {
        if ("ws".equalsIgnoreCase(scheme)) {
            port = 80;
        } else if ("wss".equalsIgnoreCase(scheme)) {
            port = 443;
        } else {
            port = -1;
        }
    } else {
        port = uri.getPort();
    }

    if (!"ws".equalsIgnoreCase(scheme) && !"wss".equalsIgnoreCase(scheme)) {
        logger.error("Only WS(S) is supported.");
        return false;
    }

    final boolean ssl = "wss".equalsIgnoreCase(scheme);
    final SslContext sslCtx;
    if (ssl) {
        sslCtx = SslContextBuilder.forClient()
                .trustManager(InsecureTrustManagerFactory.INSTANCE).build();
    } else {
        sslCtx = null;
    }

    group = new NioEventLoopGroup();

    HttpHeaders headers = new DefaultHttpHeaders();
    customHeaders.entrySet().forEach(
            header -> headers.add(header.getKey(),ping is not supported and remember to change
        // HttpResponseDecoder to WebSocketHttpResponseDecoder in the pipeline.
        handler =
                new WebSocketClientHandler(
                        WebSocketClientHandshakerFactory.newHandshaker(uri,latch);

        Bootstrap b = new Bootstrap();
        b.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ChannelPipeline p = ch.pipeline();
                        if (sslCtx != null) {
                            p.addLast(sslCtx.newHandler(ch.alloc(),port));
                        }
                        p.addLast(
                                new HttpClientCodec(),handler);
                    }
                });

        channel = b.connect(uri.getHost(),port).sync().channel();
        isSuccess = handler.handshakeFuture().sync().isSuccess();
        logger.debug("WebSocket Handshake successful : " + isSuccess);
        return isSuccess;
    } catch (Exception e) {
        logger.error("Handshake unsuccessful : " + e.getMessage());
        throw new ProtocolException("Protocol exception: " + e.getMessage());
    }
}

NettyWebSocketTransportHandler() {
   handshaker = WebSocketClientHandshakerFactory.newHandshaker(
      getRemoteLocation(),options.getWsSubProtocol(),getMaxFrameSize());
}

public NettyWebSocketTransportHandler() {
    handshaker = WebSocketClientHandshakerFactory.newHandshaker(
        getRemoteLocation(),AMQP_SUB_PROTOCOL,getMaxFrameSize());
}