@Override
protected void doWrite(SendHandler handler,ByteBuffer... data) {
    long timeout = getSendTimeout();
    if (timeout < 1) {
        timeout = Long.MAX_VALUE;

    }
    SendHandlerToCompletionHandler sh2ch =
            new SendHandlerToCompletionHandler(handler);
    try {
        channel.write(data,data.length,timeout,TimeUnit.MILLISECONDS,null,sh2ch);
    } catch (IllegalStateException ise) {
        sh2ch.Failed(ise,null);
    }
}

@Override
protected void doWrite(SendHandler handler,null);
    }
}

public void broadcast(@Observes @leaderDataQualifier String leaderboard) {
    for (final Session s : CLIENTS) {
        if (s != null && s.isopen()) {
            /**
             * Asynchronous push
             */
            s.getAsyncRemote().sendText(leaderboard,new SendHandler() {
                @Override
                public void onResult(SendResult result) {
                    if (result.isOK()) {
                        //Logger.getLogger(MeetupGroupsLiveleaderboardEndpoint.class.getName()).log(Level.INFO," sent to client {0}",s.getId());
                    } else {
                        Logger.getLogger(MeetupGroupsLiveleaderboardEndpoint.class.getName()).log(Level.SEVERE,"Could not send to client " + s.getId(),result.getException());
                    }
                }
            });
        }

    }

}

public void onopen(final Session session,EndpointConfig endpointConfig) {
    session.getAsyncRemote().sendText(
            "Client Success!Your id is: " + session.getId());
    session.addMessageHandler(new MessageHandler.Whole<String>() {
        @Override
        public void onMessage(String message) {
            session.getAsyncRemote().sendobject(message,new SendHandler() {
                @Override
                public void onResult(SendResult result) {
                    System.out.println(session.getId() + ":"
                            + result.isOK());
                }
            });
        }
    });
}

/** 
     * sendMessage is executed snychronously to avoid tomcat nativelib crashes.  
     * @param session
     * @param message
     * @param handler 
     */
    public synchronized static void sendMessage(final Session session,final String message,final SendHandler handler) {
//        synchronized (session) {
            try {
                session.getBasicRemote().sendText(message);
                handler.onResult(new SendResult());
            } catch (IOException ex) {
                Logger.getLogger(WebSocket.class.getName()).log(Level.SEVERE,ex);
                handler.onResult(new SendResult(ex));
                try {
                    //close broken session
                    session.close();
                } catch (IOException ex1) {
                    Logger.getLogger(WebSocket.class.getName()).log(Level.SEVERE,ex1);
                }
            }
//        }
//        }
    }

@OnMessage
public void onWebSocketText(String message)
{
    System.out.println("Received TEXT message: " + message);
    try {
        if ((session != null) && (session.isopen()))
        {
            System.out.println("Echoing back text message "+message);
            session.getAsyncRemote().sendText("Received: "+message,new SendHandler(){

    @Override
    public void onResult(SendResult arg0) {
        if (!arg0.isOK()){
            System.out.println("Error Sending Response: "+arg0.getException().getMessage());
        }
    }

            });
        }
    } catch (Exception e){
        System.out.println("Error: "+e.getMessage());
        e.printstacktrace();
    }
}

@Override
public void writeData(ByteBuffer data) {
    if (!this.isConnected()) {
        System.err.println("attempted to write before connection is open.");
        return;
    }
    new Thread(() -> 
    this.dataSession.getAsyncRemote().sendBinary(data,new SendHandler() {
        @Override
        public void onResult(SendResult arg0) {
            RemoteP2PConnection.this.executor.execute(new Runnable() {
                @Override
                public void run() {
                    if (RemoteP2PConnection.this.handler != null) RemoteP2PConnection.this.handler.onDataSent(RemoteP2PConnection.this);
                }
            });
        }
    })).start();
}

@Override
public void onopen(final Session session,EndpointConfig ec) {
    session.addMessageHandler(new MessageHandler.Whole<String>() {

        @Override
        public void onMessage(String data) {
            System.out.println("Received (MyEndpointHandler) : " + data);

            session.getAsyncRemote().sendText(data,new SendHandler() {

                @Override
                public void onResult(SendResult sr) {
                    if (sr.isOK()) {
                        System.out.println("Message written to the socket (handler)");
                    } else {
                        System.out.println("Message NOT written to the socket (handler)");
                        sr.getException().printstacktrace();
                    }

                }
            });
        }
    });
}

@Override
protected void doWrite(SendHandler handler,null);
    }
}

public void sendBytesByCompletion(ByteBuffer data,SendHandler handler) {
    if (data == null) {
        throw new IllegalArgumentException(sm.getString("wsRemoteEndpoint.nullData"));
    }
    if (handler == null) {
        throw new IllegalArgumentException(sm.getString("wsRemoteEndpoint.nullHandler"));
    }
    StateUpdateSendHandler sush = new StateUpdateSendHandler(handler);
    stateMachine.binaryStart();
    startMessage(Constants.OPCODE_BINARY,data,true,sush);
}

public void sendStringByCompletion(String text,SendHandler handler) {
    if (text == null) {
        throw new IllegalArgumentException(sm.getString("wsRemoteEndpoint.nullData"));
    }
    if (handler == null) {
        throw new IllegalArgumentException(sm.getString("wsRemoteEndpoint.nullHandler"));
    }
    stateMachine.textStart();
    TextMessageSendHandler tmsh = new TextMessageSendHandler(handler,CharBuffer.wrap(text),encoder,encoderBuffer,this);
    tmsh.write();
    // TextMessageSendHandler will update stateMachine when it completes
}

void endMessage(SendHandler handler,SendResult result) {
    boolean doWrite = false;
    MessagePart mpNext = null;
    synchronized (messagePartLock) {

        fragmented = nextFragmented;
        text = nextText;

        mpNext = messagePartQueue.poll();
        if (mpNext == null) {
            messagePartInProgress = false;
        } else if (!closed){
            // Session may have been closed unexpectedly in the middle of
            // sending a fragmented message closing the endpoint. If this
            // happens,clearly there is no point trying to send the rest of
            // the message.
            doWrite = true;
        }
    }
    if (doWrite) {
        // Actual write has to be outside sync block to avoid possible
        // deadlock between messagePartLock and writeLock in
        // o.a.coyote.http11.upgrade.AbstractServletoutputStream
        writeMessagePart(mpNext);
    }

    wsSession.updateLastActive();

    // Some handlers,such as the IntermediateMessageHandler,do not have a
    // nested handler so handler may be null.
    if (handler != null) {
        handler.onResult(result);
    }
}

public TextMessageSendHandler(SendHandler handler,CharBuffer message,boolean isLast,CharsetEncoder encoder,ByteBuffer encoderBuffer,WsRemoteEndpointImplBase endpoint) {
    this.handler = handler;
    this.message = message;
    this.isLast = isLast;
    this.encoder = encoder.reset();
    this.buffer = encoderBuffer;
    this.endpoint = endpoint;
}

public OutputBufferSendHandler(SendHandler completion,ByteBuffer headerBuffer,ByteBuffer payload,byte[] mask,ByteBuffer outputBuffer,boolean flushrequired,WsRemoteEndpointImplBase endpoint) {
    this.handler = completion;
    this.headerBuffer = headerBuffer;
    this.payload = payload;
    this.mask = mask;
    this.outputBuffer = outputBuffer;
    this.flushrequired = flushrequired;
    this.endpoint = endpoint;
}

@Override
protected void doWrite(SendHandler handler,ByteBuffer... buffers) {
    this.handler = handler;
    this.buffers = buffers;
    // This is definitely the same thread that triggered the write so a
    // dispatch will be required.
    onWritePossible(true);
}

/**
 *
 * @param t             The throwable associated with any error that
 *                      occurred
 * @param usedispatch   Should {@link SendHandler#onResult(SendResult)} be
 *                      called from a new thread,keeping in mind the
 *                      requirements of
 *                      {@link javax.websocket.RemoteEndpoint.Async}
 */
private void clearHandler(Throwable t,boolean usedispatch) {
    // Setting the result marks this (partial) message as
    // complete which means the next one may be sent which
    // Could update the value of the handler. Therefore,keep a
    // local copy before signalling the end of the (partial)
    // message.
    SendHandler sh = handler;
    handler = null;
    buffers = null;
    if (sh != null) {
        if (usedispatch) {
            OnResultRunnable r = onResultRunnables.poll();
            if (r == null) {
                r = new OnResultRunnable(onResultRunnables);
            }
            r.init(sh,t);
            if (executorService == null || executorService.isShutdown()) {
                // Can't use the executor so call the runnable directly.
                // This may not be strictly specification compliant in all
                // cases but during shutdown only close messages are going
                // to be sent so there should not be the issue of nested
                // calls leading to stack overflow as described in bug
                // 55715. The issues with nested calls was the reason for
                // the separate thread requirement in the specification.
                r.run();
            } else {
                executorService.execute(r);
            }
        } else {
            if (t == null) {
                sh.onResult(new SendResult());
            } else {
                sh.onResult(new SendResult(t));
            }
        }
    }
}

public MessagePart( boolean fin,int rsv,byte opCode,SendHandler intermediateHandler,SendHandler endHandler) {
    this.fin = fin;
    this.rsv = rsv;
    this.opCode = opCode;
    this.payload = payload;
    this.intermediateHandler = intermediateHandler;
    this.endHandler = endHandler;
}

public void sendBytesByCompletion(ByteBuffer data,sush);
}

public void sendStringByCompletion(String text,this);
    tmsh.write();
    // TextMessageSendHandler will update stateMachine when it completes
}

void endMessage(SendHandler handler,do not have a
    // nested handler so handler may be null.
    if (handler != null) {
        handler.onResult(result);
    }
}

public TextMessageSendHandler(SendHandler handler,WsRemoteEndpointImplBase endpoint) {
    this.handler = handler;
    this.message = message;
    this.isLast = isLast;
    this.encoder = encoder.reset();
    this.buffer = encoderBuffer;
    this.endpoint = endpoint;
}

public OutputBufferSendHandler(SendHandler completion,WsRemoteEndpointImplBase endpoint) {
    this.handler = completion;
    this.headerBuffer = headerBuffer;
    this.payload = payload;
    this.mask = mask;
    this.outputBuffer = outputBuffer;
    this.flushrequired = flushrequired;
    this.endpoint = endpoint;
}

@Override
protected void doWrite(SendHandler handler,ByteBuffer... buffers) {
    this.handler = handler;
    this.buffers = buffers;
    // This is definitely the same thread that triggered the write so a
    // dispatch will be required.
    onWritePossible(true);
}

/**
 *
 * @param t             The throwable associated with any error that
 *                      occurred
 * @param usedispatch   Should {@link SendHandler#onResult(SendResult)} be
 *                      called from a new thread,t);
            if (executorService == null || executorService.isShutdown()) {
                // Can't use the executor so call the runnable directly.
                // This may not be strictly specification compliant in all
                // cases but during shutdown only close messages are going
                // to be sent so there should not be the issue of nested
                // calls leading to stack overflow as described in bug
                // 55715. The issues with nested calls was the reason for
                // the separate thread requirement in the specification.
                r.run();
            } else {
                executorService.execute(r);
            }
        } else {
            if (t == null) {
                sh.onResult(new SendResult());
            } else {
                sh.onResult(new SendResult(t));
            }
        }
    }
}

public MessagePart( boolean fin,SendHandler endHandler) {
    this.fin = fin;
    this.rsv = rsv;
    this.opCode = opCode;
    this.payload = payload;
    this.intermediateHandler = intermediateHandler;
    this.endHandler = endHandler;
}

public void sendBytesByCompletion(ByteBuffer data,sush);
}

public void sendStringByCompletion(String text,this);
    tmsh.write();
    // TextMessageSendHandler will update stateMachine when it completes
}

void endMessage(SendHandler handler,SendResult result) {
    boolean doWrite = false;
    MessagePart mpNext = null;
    synchronized (messagePartLock) {

        fragmented = nextFragmented;
        text = nextText;

        mpNext = messagePartQueue.poll();
        if (mpNext == null) {
            messagePartInProgress = false;
        } else if (!closed) {
            // Session may have been closed unexpectedly in the middle of
            // sending a fragmented message closing the endpoint. If this
            // happens,do not have a
    // nested handler so handler may be null.
    if (handler != null) {
        handler.onResult(result);
    }
}

public TextMessageSendHandler(SendHandler handler,WsRemoteEndpointImplBase endpoint) {
    this.handler = handler;
    this.message = message;
    this.isLast = isLast;
    this.encoder = encoder.reset();
    this.buffer = encoderBuffer;
    this.endpoint = endpoint;
}

public OutputBufferSendHandler(SendHandler completion,WsRemoteEndpointImplBase endpoint) {
    this.handler = completion;
    this.headerBuffer = headerBuffer;
    this.payload = payload;
    this.mask = mask;
    this.outputBuffer = outputBuffer;
    this.flushrequired = flushrequired;
    this.endpoint = endpoint;
}

@Override
protected void doWrite(SendHandler handler,ByteBuffer... data) {
    long timeout = getSendTimeout();
    if (timeout < 1) {
        timeout = Long.MAX_VALUE;

    }
    SendHandlerToCompletionHandler sh2ch = new SendHandlerToCompletionHandler(handler);
    try {
        channel.write(data,null);
    }
}

@Override
protected void doWrite(SendHandler handler,ByteBuffer... buffers) {
    this.handler = handler;
    this.buffers = buffers;
    // This is definitely the same thread that triggered the write so a
    // dispatch will be required.
    onWritePossible(true);
}

/**
 *
 * @param t
 *            The throwable associated with any error that occurred
 * @param usedispatch
 *            Should {@link SendHandler#onResult(SendResult)} be called from
 *            a new thread,keeping in mind the requirements of
 *            {@link javax.websocket.RemoteEndpoint.Async}
 */
private void clearHandler(Throwable t,t);
            if (executorService == null || executorService.isShutdown()) {
                // Can't use the executor so call the runnable directly.
                // This may not be strictly specification compliant in all
                // cases but during shutdown only close messages are going
                // to be sent so there should not be the issue of nested
                // calls leading to stack overflow as described in bug
                // 55715. The issues with nested calls was the reason for
                // the separate thread requirement in the specification.
                r.run();
            } else {
                executorService.execute(r);
            }
        } else {
            if (t == null) {
                sh.onResult(new SendResult());
            } else {
                sh.onResult(new SendResult(t));
            }
        }
    }
}

public MessagePart(boolean fin,SendHandler endHandler) {
    this.fin = fin;
    this.rsv = rsv;
    this.opCode = opCode;
    this.payload = payload;
    this.intermediateHandler = intermediateHandler;
    this.endHandler = endHandler;
}

public void sendStringByCompletion(String text,SendHandler handler) {
    stateMachine.textStart();
    TextMessageSendHandler tmsh = new TextMessageSendHandler(handler,this);
    tmsh.write();
    // TextMessageSendHandler will update stateMachine when it completes
}

void startMessage(byte opCode,boolean last,SendHandler handler) {

    wsSession.updateLastActive();

    MessagePart mp = new MessagePart(opCode,payload,last,handler,this);

    boolean doWrite = false;
    synchronized (messagePartLock) {
        if (Constants.OPCODE_CLOSE == mp.getopCode()) {
            try {
                setBatchingallowed(false);
            } catch (IOException e) {
                log.warn(sm.getString(
                        "wsRemoteEndpoint.flushOnCloseFailed"),e);
            }
        }
        if (messagePartInProgress) {
            // When a control message is sent while another message is being
            // sent,the control message is queued. Chances are the
            // subsequent data message part will end up queued while the
            // control message is sent. The logic in this class (state
            // machine,EndMessageHandler,TextMessageSendHandler) ensures
            // that there will only ever be one data message part in the
            // queue. There Could be multiple control messages in the queue.

            // Add it to the queue
            messagePartQueue.add(mp);
        } else {
            messagePartInProgress = true;
            doWrite = true;
        }
    }
    if (doWrite) {
        // Actual write has to be outside sync block to avoid possible
        // deadlock between messagePartLock and writeLock in
        // o.a.coyote.http11.upgrade.AbstractServletoutputStream
        writeMessagePart(mp);
    }
}

void endMessage(SendHandler handler,clearly there is no point trying to send the rest of
            // the message.
            doWrite = true;
        }
    }
    if (doWrite) {
        // Actual write has to be outside sync block to avoid possible
        // deadlock between messagePartLock and writeLock in
        // o.a.coyote.http11.upgrade.AbstractServletoutputStream
        writeMessagePart(mpNext);
    }

    wsSession.updateLastActive();

    handler.onResult(result);
}

public MessagePart(byte opCode,SendHandler handler,WsRemoteEndpointImplBase endpoint) {
    this.opCode = opCode;
    this.payload = payload;
    this.last = last;
    this.handler = new EndMessageHandler(endpoint,handler);
}

public TextMessageSendHandler(SendHandler handler,WsRemoteEndpointImplBase endpoint) {
    this.handler = handler;
    this.message = message;
    this.isLast = isLast;
    this.encoder = encoder.reset();
    this.buffer = encoderBuffer;
    this.endpoint = endpoint;
}

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

    if (msg instanceof BinaryWebSocketFrame) {
        try {
            Object obj = protobufDecoder.decode(((BinaryWebSocketFrame) msg).content());
            resQueue.add((Response.HeshResMessage)obj);
        } catch (Exception e) {
            e.printstacktrace();
        }
    }

}

@Override
protected void encode(ChannelHandlerContext ctx,Proto proto,List<Object> list) throws Exception {
    ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer();
    if (proto.getBody() != null) {
        byteBuf.writeInt(Proto.HEADER_LENGTH + proto.getBody().length);
        byteBuf.writeShort(Proto.HEADER_LENGTH);
        byteBuf.writeShort(Proto.VERSION);
        byteBuf.writeInt(proto.getoperation());
        byteBuf.writeInt(proto.getSeqId());
        byteBuf.writeBytes(proto.getBody());
    } else {
        byteBuf.writeInt(Proto.HEADER_LENGTH);
        byteBuf.writeShort(Proto.HEADER_LENGTH);
        byteBuf.writeShort(Proto.VERSION);
        byteBuf.writeInt(proto.getoperation());
        byteBuf.writeInt(proto.getSeqId());
    }

    list.add(new BinaryWebSocketFrame(byteBuf));

    logger.debug("encode: {}",proto);
}

@Test
public void testVersion() throws Exception {
    try {
        String uuid = UUID.randomUUID().toString();
        VersionRequest request = new VersionRequest();
        request.setRequestId(uuid);
        ch.writeAndFlush(new BinaryWebSocketFrame(Unpooled.wrappedBuffer(JsonSerializer.getobjectMapper()
                .writeValueAsBytes(request))));
        // Confirm receipt of all data sent to this point
        List<byte[]> response = handler.getResponses();
        while (response.size() == 0 && handler.isConnected()) {
            LOG.info("Waiting for web socket response");
            sleepUninterruptibly(500,TimeUnit.MILLISECONDS);
            response = handler.getResponses();
        }
        assertEquals(1,response.size());
        VersionResponse version = JsonSerializer.getobjectMapper()
                .readValue(response.get(0),VersionResponse.class);
        assertEquals(VersionResponse.VERSION,version.getVersion());
        assertEquals(uuid,version.getRequestId());
    } finally {
        ch.close().sync();
        s.shutdown();
        group.shutdownGracefully();
    }
}

@SuppressWarnings({ "unchecked","rawtypes" })
@Override
   protected void encode(ChannelHandlerContext ctx,Packet packet,List out) throws Exception {
       ByteBuf buf = ctx.alloc().buffer().order(ByteOrder.LITTLE_ENDIAN);
       int packetId = PacketRegistry.SERVER2CLIENT.getPacketId(packet.getClass());
       if (packetId == -1) {
           throw new IllegalArgumentException("Provided packet is not registered as a clientbound packet!");
       }

       buf.writeByte(packetId);
       packet.writeData(buf);
       new BinaryWebSocketFrame(buf);
       out.add(new BinaryWebSocketFrame(buf));

       Log.logDebug("Sent packet ID " + packetId + " (" + packet.getClass().getSimpleName() + ") to " + ctx.channel().remoteAddress());
   }

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

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;
        }
    }

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

    if(msg instanceof WebSocketFrame){
        WebSocketFrame frame = (WebSocketFrame)msg;
        if(frame instanceof BinaryWebSocketFrame){
            handleWebSocketFrame(ctx,frame);
        }
        return;
    }
    return;
}

@Override
public void channelRead0(ChannelHandlerContext ctx,frame);
        }
        return;
    }
    sendRealTimeMessageTest(ctx);
    return;
}

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

private ReactiveSocketWebSocketClient(WebSocketConnection wsConn) {
    this.reactiveSocket = ReactiveSocket.createRequestor();
    connect = this.reactiveSocket.connect(
            new DuplexConnection() {
                @Override
                public Publisher<Frame> getinput() {
                    return toPublisher(wsConn.getinput().map(frame -> {
                        return Frame.from(frame.content().nioBuffer());
                    }));
                }

                @Override
                public Publisher<Void> addOutput(Publisher<Frame> o) {
                    // had to use writeAndFlushOnEach instead of write for frames to get through
                    // Todo determine if that's expected or not
                    Publisher<Void> p = toPublisher(wsConn.writeAndFlushOnEach(toObservable(o)
                            .map(frame -> new BinaryWebSocketFrame(Unpooled.wrappedBuffer(frame.getByteBuffer())))
                    ));
                    return p;
                }
            });
}

/**
   * Use this method as the RxNetty HttpServer WebSocket handler.
   * 
   * @param ws
   * @return
   */
  public Observable<Void> acceptWebsocket(WebSocketConnection ws) {
    return toObservable(reactiveSocket.connect(new DuplexConnection() {
    @Override
    public Publisher<Frame> getinput() {
        return toPublisher(ws.getinput().map(frame -> {
            // Todo is this copying bytes?
            try {
                return Frame.from(frame.content().nioBuffer());
            } catch (Exception e) {
                e.printstacktrace();
                throw new RuntimeException(e);
            }
        }));
    }

    @Override
    public Publisher<Void> addOutput(Publisher<Frame> o) {
        // had to use writeAndFlushOnEach instead of write for frames to reliably get through
        // Todo determine if that's expected or not
        return toPublisher(ws.writeAndFlushOnEach(toObservable(o).map(frame -> {
            return new BinaryWebSocketFrame(Unpooled.wrappedBuffer(frame.getByteBuffer()));
        })));
    }
}));
  }

@Override
public void channelRead0(ChannelHandlerContext ctx,Object msg) throws Exception {
  Channel ch = ctx.channel();

  if (!handshaker.isHandshakeComplete()) {
    handshaker.finishHandshake(ch,(FullHttpResponse) msg);
    handshakeFuture.setSuccess();
    return;
  }

  if (!(msg instanceof BinaryWebSocketFrame)) {
    ch.close();
    log.warn("Received {},closing",msg);
    return;
  }

  byte[] b = extractBytes(msg);

  ctx.fireChannelRead(b);
}

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

@Override
protected void channelRead0(ChannelHandlerContext ctx,WebSocketFrame frame) throws Exception {
    if (frame instanceof TextWebSocketFrame) {
        // Echos the same text
        String text = ((TextWebSocketFrame) frame).text();
        if (PING.equals(text)) {
            ctx.writeAndFlush(new PingWebSocketFrame(Unpooled.wrappedBuffer(new byte[]{1,2,3,4})));
            return;
        }
        ctx.channel().writeAndFlush(new TextWebSocketFrame(text));
    } else if (frame instanceof BinaryWebSocketFrame) {
        ctx.channel().writeAndFlush(frame.retain());
    } else if (frame instanceof CloseWebSocketFrame) {
        ctx.close();
    } else {
        String message = "unsupported frame type: " + frame.getClass().getName();
        throw new UnsupportedOperationException(message);
    }
}

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

private void performSend(byte[] raw) throws IOException
{
    if (this.outBuf != null)
    {
        this.outBuf.write(raw);
        raw = this.outBuf.toByteArray();
        this.outBuf = null;
    }
    //char[] encoded = Base64.encode(raw);

    if (this.binary)
    {
        this.ctx.channel().write(new BinaryWebSocketFrame(Unpooled.wrappedBuffer(raw)));
    }
    else
    {
        this.ctx.channel().write(new TextWebSocketFrame(StringUtil.toUtfString(raw)));
    }

}

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

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

public void handle(final Object msg) {
  ready();
  if (msg instanceof TextWebSocketFrame) {
    onTextCallback.accept(((TextWebSocketFrame) msg).text());
  } else if (msg instanceof BinaryWebSocketFrame) {
    onBinaryCallback.accept(((BinaryWebSocketFrame) msg).content().nioBuffer());
  } else if (msg instanceof CloseWebSocketFrame) {
    CloseWebSocketFrame closeFrame = ((CloseWebSocketFrame) msg).retain();
    int statusCode = closeFrame.statusCode();
    onCloseCallback.accept(statusCode == -1 ? WebSocket.norMAL.code() : statusCode,Optional.ofNullable(closeFrame.reasonText()));
    handshaker.close(ctx.channel(),closeFrame).addListener(CLOSE);
  } else if (msg instanceof Throwable) {
    onErrorCallback.accept((Throwable) msg);
  }
}

private void handleWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame 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 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 decode(final ChannelHandlerContext ctx,final WebSocketFrame msg,final List<Object> out)
    throws Exception {
    if (msg instanceof BinaryWebSocketFrame) {
        ByteBuf content = msg.content();
        // the content is passed to other handlers so they need to be retained.
        content.retain();
        fragments.add(content);
        if (msg.isFinalFragment()) {
            if (fragments.size() == 1) {
                out.add(fragments.get(0));
            } else {
                ByteBuf[] array = fragments.toArray(BYTE_BUF_TYPE);
                out.add(Unpooled.wrappedBuffer(array));
            }
            fragments.clear();
        }
    } else if (msg instanceof TextWebSocketFrame) {
        LOG.warn("Recieved a Websocket text frame. This was not expected. Ignoring it.");
    }
}

/**
 * 向当前客户端发送数据
 * @param message
 */
public void send(Message message) {
    byte[] bytes = message.toByteArray();

    ByteBuf b = Unpooled.buffer(bytes.length);
    b.writeBytes(bytes);

    WebSocketFrame frame = new BinaryWebSocketFrame(b);

    channel.writeAndFlush(frame);
}

public ChannelFuture write(Communication.HeshReqMessage message) {
    byte[] bytes = message.toByteArray();

    ByteBuf b = Unpooled.buffer(bytes.length);
    b.writeBytes(bytes);

    WebSocketFrame frame = new BinaryWebSocketFrame(b);
    return channel.writeAndFlush(frame);
}

@Override
protected void decode(ChannelHandlerContext ctx,BinaryWebSocketFrame wsFrame,List<Object> out) throws Exception {
    ByteBuf buf = wsFrame.content();

    this.messageNewDecoder.decode(ctx,buf,out);
}

@Override
protected void encode(ChannelHandlerContext ctx,Message msg,List<Object> out) throws Exception {
    if (msg == null || !(msg instanceof Message))
        return;

    byte[] data = ((Message) msg).toBytes();

    out.add(new BinaryWebSocketFrame(Unpooled.wrappedBuffer(data)));
}

/**
 * 将webSocket消息转换为bytebuf类型,以适配后面的解码器
 */
@Override
protected void decode(ChannelHandlerContext paramChannelHandlerContext,WebSocketFrame paramINBOUND_IN,List<Object> paramList)
        throws Exception {
    if(paramINBOUND_IN instanceof BinaryWebSocketFrame)
    {
        BinaryWebSocketFrame msg=(BinaryWebSocketFrame)paramINBOUND_IN;
        ByteBuf data = msg.content();
        paramList.add(data);
        data.retain();
    }
}

/**
 * 对于业务层直接发送的bytebuf实例将其转换为websocket消息
 */
@Override
protected void encode(ChannelHandlerContext paramChannelHandlerContext,ByteBuf paramOUTBOUND_IN,List<Object> paramList) throws Exception {
    paramList.add(new BinaryWebSocketFrame(paramOUTBOUND_IN));
    paramOUTBOUND_IN.retain();
}

@Override
public void run() {
    try {
        VersionResponse response = new VersionResponse();
        response.setRequestId(this.request.getRequestId());
        ctx.writeAndFlush(new BinaryWebSocketFrame(Unpooled.wrappedBuffer(om.writeValueAsBytes(response))));
    } catch (JsonProcessingException e) {
        LOG.error("Error serializing version response",e);
    }
}

@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
public void channelRead0(final ChannelHandlerContext ctx,WebSocketFrame frame) throws Exception {
    webSocketServerThread.log(Level.FInesT,"channel read,frame="+frame);
    // Todo: log at INFO level if this the first data we received from a client (new first connection),to
    // help detect clients connecting but not sending authentication commands (in newPlayer)

    if (this.checkIPBans) {
        String ip = webSocketServerThread.getRemoteIP(ctx.channel());
        if (this.ipBans.contains(ip)) {
            webSocketServerThread.sendLine(ctx.channel(),"T,Banned from server"); // Todo: show reason,getBanList
            return;
        }
    }

    if (frame instanceof BinaryWebSocketFrame) {
        ByteBuf content = frame.content();

        byte[] bytes = new byte[content.capacity()];
        content.getBytes(0,bytes);

        final String string = new String(bytes);
        webSocketServerThread.log(Level.FInesT,"received "+content.capacity()+" bytes: "+string);

        this.webSocketServerThread.scheduleSyncTask(new Runnable() {
            @Override
            public void run() {
                webSocketServerThread.handle(string,ctx);
            }
        });
    } else {
        String message = "unsupported frame type: " + frame.getClass().getName();
        throw new UnsupportedOperationException(message);
    }
}

public void broadcastLineExcept(ChannelId excludeChannelId,String message) {
    for (Channel channel: allUsersGroup) {
        if (channel.id().equals(excludeChannelId)) {
            continue;
        }

        channel.writeAndFlush(new BinaryWebSocketFrame(Unpooled.copiedBuffer((message + "\n").getBytes())));
    }
}

@SuppressWarnings({ "unchecked","rawtypes" })
@Override
protected void encode(ChannelHandlerContext ctx,List out) throws Exception {
    ByteBuf buf = ctx.alloc().buffer().order(ByteOrder.LITTLE_ENDIAN);
    int packetId = reg.CLIENTBOUND.getPacketId(packet.getClass());
    if (packetId == -1) {
        throw new IllegalArgumentException("Provided packet is not registered as a clientbound packet!");
    }

    buf.writeByte(packetId);
    packet.writeData(buf);
    out.add(new BinaryWebSocketFrame(buf));

    Server.log.finest("Sent packet ID " + packetId + " (" + packet.getClass().getSimpleName() + ") to " + ctx.channel().remoteAddress());
}

@Override
protected void decode(ChannelHandlerContext chc,BinaryWebSocketFrame frame,List<Object> out) throws Exception
{
    //convert the frame to a ByteBuf
    ByteBuf bb = frame.content();
    bb.retain();
    out.add(bb);
}

@Override
protected void encode(ChannelHandlerContext chc,ByteBuf bb,List<Object> out) throws Exception
{
    //convert the ByteBuf to a WebSocketFrame
    BinaryWebSocketFrame result = new BinaryWebSocketFrame();
    result.content().writeBytes(bb);
    out.add(result);
}

@SuppressWarnings({ "deprecation","unchecked",List out) throws Exception {
       ByteBuf buf = ctx.alloc().buffer().order(ByteOrder.BIG_ENDIAN);
       int packetId = PacketRegistry.CLIENTBOUND.getPacketId(packet.getClass());
       if (packetId == -1) {
           throw new IllegalArgumentException("Provided packet is not registered as a clientbound packet!");
       }

       buf.writeByte(packetId);
       packet.writeData(buf);
       out.add(new BinaryWebSocketFrame(buf));

       ClitherServer.log.finest("Sent packet " + " (" + packet.getClass().getSimpleName() + ") to " + ctx.channel().remoteAddress());
   }

@Override
public void channelRead0(ChannelHandlerContext context,Object message) throws Exception {
  Channel channel = context.channel();
  if (message instanceof FullHttpResponse) {
    checkState(!handshaker.isHandshakeComplete());
    try {
      handshaker.finishHandshake(channel,(FullHttpResponse) message);
      delegate.onopen();
    } catch (WebSocketHandshakeException e) {
      delegate.onError(e);
    }
  } else if (message instanceof TextWebSocketFrame) {
    delegate.onMessage(((TextWebSocketFrame) message).text());
  } else {
    checkState(message instanceof CloseWebSocketFrame);
    delegate.onClose();
  }
}

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

@Test
public void testCreateSubscriptionWithMissingSessionId() throws Exception {
    decoder = new WebSocketRequestDecoder(config);
    // @formatter:off
    String request = "{ "+ 
      "\"operation\" : \"create\"," +
      "\"subscriptionId\" : \"1234\"" + 
    " }";
    // @formatter:on
    TextWebSocketFrame frame = new TextWebSocketFrame();
    frame.content().writeBytes(request.getBytes(StandardCharsets.UTF_8));
    decoder.decode(ctx,frame,results);
    Assert.assertNotNull(ctx.msg);
    Assert.assertEquals(CloseWebSocketFrame.class,ctx.msg.getClass());
    Assert.assertEquals(1008,((CloseWebSocketFrame) ctx.msg).statusCode());
    Assert.assertEquals("User must log in",((CloseWebSocketFrame) ctx.msg).reasonText());
}

@Test
public void testCreateSubscriptionWithInvalidSessionIdAndNonAnonymousAccess() throws Exception {
    ctx.channel().attr(SubscriptionRegistry.SESSION_ID_ATTR)
            .set(URLEncoder.encode(UUID.randomUUID().toString(),StandardCharsets.UTF_8.name()));
    decoder = new WebSocketRequestDecoder(config);
    // @formatter:off
    String request = "{ "+ 
      "\"operation\" : \"create\",((CloseWebSocketFrame) ctx.msg).reasonText());
}

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

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 channelRead0(ChannelHandlerContext ctx,Object msg) throws Exception {
  Channel channel = ctx.channel();
  if (!handshaker.isHandshakeComplete()) {
    handshaker.finishHandshake(channel,(FullHttpResponse) msg);
    handshakeFuture.setSuccess();
    eventBus.post(new Connected());
    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;
    eventBus.post(new Response(textFrame.text()));
  } else if (frame instanceof CloseWebSocketFrame) {
    channel.close();
    eventBus.post(new disconnected());
  }
}

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

@Override
protected void channelRead0(ChannelHandlerContext ctx,WebSocketFrame frame) throws Exception {
    if (frame instanceof TextWebSocketFrame) {
        // Echos the same text
        String text = ((TextWebSocketFrame) frame).text();
        if (PING.equals(text)) {
            ctx.writeAndFlush(new PingWebSocketFrame(Unpooled.wrappedBuffer(new byte[]{1,2,3,4})));
            return;
        }
        ctx.channel().writeAndFlush(new TextWebSocketFrame(text));
    } else if (frame instanceof BinaryWebSocketFrame) {
        ctx.channel().writeAndFlush(frame.retain());
    } else if (frame instanceof CloseWebSocketFrame) {
        ctx.close();
    } else {
        String message = "unsupported frame type: " + frame.getClass().getName();
        throw new UnsupportedOperationException(message);
    }
}

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

public void handle(final Object msg) {
  ready();
  if (msg instanceof TextWebSocketFrame) {
    onTextCallback.accept(((TextWebSocketFrame) msg).text());
  } else if (msg instanceof BinaryWebSocketFrame) {
    onBinaryCallback.accept(((BinaryWebSocketFrame) msg).content().nioBuffer());
  } else if (msg instanceof CloseWebSocketFrame) {
    CloseWebSocketFrame closeFrame = ((CloseWebSocketFrame) msg).retain();
    int statusCode = closeFrame.statusCode();
    onCloseCallback.accept(statusCode == -1 ? WebSocket.norMAL.code() : statusCode,Optional.ofNullable(closeFrame.reasonText()));
    handshaker.close(ctx.channel(),closeFrame).addListener(CLOSE);
  } else if (msg instanceof Throwable) {
    onErrorCallback.accept((Throwable) msg);
  }
}

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

/**
 * disconnect from the AudioConnect server and reset.<br>
 * If a connection is not established or being established,this will do nothing.
 * @return a Future for when the connection has been fully disconnected and closed
 */
public Future<?> disconnect() {
    Connection connection;
    synchronized (connectionLock) {
        connection = this.connection;
        this.connection = null;
    }

    if (connection != null) {
        playerScheduler.clear();
        connection.playerConnections.clear();

        // Remove channelCloseListener to not reconnect
        connection.channel.closeFuture().removeListener(channelCloseListener);

        if (connection.channel.isActive()) {
            final Promise<Object> disconnectPromise = bootstrap.group().next().newPromise();

            Object closeFrame = new CloseWebSocketFrame(WEBSOCKET_CLOSE_CODE_GOING_AWAY,"Going offline");
            connection.channel.writeAndFlush(closeFrame).addListener(new ChannelFutureListener() {

                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    future.channel().close().addListener(new PromiseNotifier<>(disconnectPromise));
                }
            });
            return disconnectPromise;
        }
    }
    return bootstrap.group().next().newSucceededFuture(null);
}

@Override
protected void channelRead0(ChannelHandlerContext ctx,AddSubscription add) throws Exception {
    Subscription s = SubscriptionRegistry.get().get(add.getSubscriptionId());
    if (null != s) {
        String metric = add.getMetric();
        if (null == metric) {
            LOG.error("Metric name cannot be null in add subscription");
            ctx.writeAndFlush(new CloseWebSocketFrame(1008,"Metric name cannot be null in add subscription"));
        }
        Map<String,String> tags = null;
        Long startTime = 0L;
        Long endTime = 0L;
        Long delayTime = 5000L;
        if (add.getTags().isPresent()) {
            tags = add.getTags().get();
        }
        if (add.getStartTime().isPresent()) {
            startTime = add.getStartTime().get();
        }
        if (add.getEndTime().isPresent()) {
            endTime = add.getEndTime().get();
        }
        if (add.getDelayTime().isPresent()) {
            delayTime = add.getDelayTime().get();
        }
        s.addMetric(metric,tags,startTime,endTime,delayTime);
    } else {
        LOG.error("UnkNown subscription id,create subscription first");
        ctx.writeAndFlush(new CloseWebSocketFrame(1003,"UnkNown subscription id,create subscription first"));
    }
}

@Override
protected void channelRead0(ChannelHandlerContext ctx,CloseSubscription close) throws Exception {
    Subscription s = SubscriptionRegistry.get().remove(close.getSubscriptionId());
    if (null != s) {
        s.close();
    }
    ctx.writeAndFlush(new CloseWebSocketFrame(1000,"Client requested close."));
}

@Override
protected void channelRead0(ChannelHandlerContext ctx,QueryRequest msg) throws Exception {
    try {
        String response = JsonUtil.getobjectMapper().writeValueAsstring(dataStore.query(msg));
        ctx.writeAndFlush(new TextWebSocketFrame(response));
    } catch (TimelyException e) {
        if (e.getMessage().contains("No matching tags")) {
            LOG.trace(e.getMessage());
        } else {
            LOG.error(e.getMessage(),e);
        }
        ctx.writeAndFlush(new CloseWebSocketFrame(1008,e.getMessage()));
    }
}

@Override
protected void channelRead0(ChannelHandlerContext ctx,SuggestRequest msg) throws Exception {
    try {
        String response = JsonUtil.getobjectMapper().writeValueAsstring(dataStore.suggest(msg));
        ctx.writeAndFlush(new TextWebSocketFrame(response));
    } catch (TimelyException e) {
        LOG.error(e.getMessage(),e);
        ctx.writeAndFlush(new CloseWebSocketFrame(1008,e.getMessage()));
    }
}

@Override
protected void channelRead0(ChannelHandlerContext ctx,SearchLookupRequest msg) throws Exception {
    try {
        String response = JsonUtil.getobjectMapper().writeValueAsstring(dataStore.lookup(msg));
        ctx.writeAndFlush(new TextWebSocketFrame(response));
    } catch (TimelyException e) {
        LOG.error(e.getMessage(),e.getMessage()));
    }
}

@Test
public void testCreateSubscriptionWithoutSubscriptionId() throws Exception {
    decoder = new WebSocketRequestDecoder(anonConfig);
    String request = "{ \"operation\" : \"create\" }";
    TextWebSocketFrame frame = new TextWebSocketFrame();
    frame.content().writeBytes(request.getBytes(StandardCharsets.UTF_8));
    decoder.decode(ctx,((CloseWebSocketFrame) ctx.msg).statusCode());
    Assert.assertEquals("Subscription ID is required.",((CloseWebSocketFrame) ctx.msg).reasonText());
}

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 {

}

@Override
public void onopen(Session session,EndpointConfig config) {
    String id = session.getId();
    log.debug("{}: open ws proxy ",id);
    try {
        ChannelFuture cf = backend.connect().sync();
        Channel channel = cf.channel();
        WebSocketClientProtocolHandler wscph = makeWsProtocolHandler(session);
        WebSocketClientHandshaker handshaker = wscph.handshaker();
        WsHandler handler = new WsHandler(handshaker,channel,session);
        channel.pipeline().addLast(new HttpObjectAggregator(1024 * 4),WebSocketClientCompressionHandler.INSTANCE,wscph,handler);
        handshaker.handshake(channel);
        log.debug("{}: wait messages",id);
        session.addMessageHandler(String.class,handler::onFrontString);
        session.addMessageHandler(ByteBuffer.class,handler::onFrontBytes);
    } catch (Exception e) {
        log.error("{}: can not establish ws connect with backed",id,e);
    }

}

@Override
protected void initChannel(final Channel channel) throws Exception
{
    final ChannelPipeline pipeline = channel.pipeline();
    pipeline.addLast("Bridge|SSLContext",this.webSocketConnection.getSslContext().newHandler(channel.alloc(),this.webSocketConnection.getIp(),this.webSocketConnection.getPort()));
    pipeline.addLast("Bridge|HttpClientCodec",new HttpClientCodec());
    pipeline.addLast("Bridge|HttpObjectAggregator",new HttpObjectAggregator(8192));
    pipeline.addLast("Bridge|WebSocketClientCompressionHandler",WebSocketClientCompressionHandler.INSTANCE);
    pipeline.addLast(new WebSocketHandler(this.webSocketConnection));
}

/**
 * @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,WebSocketVersion.V13,subProtocol,true,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(),host,port));
            }
            p.addLast(new HttpClientCodec(),new HttpObjectAggregator(8192),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,null,new DefaultHttpHeaders(),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;
    }
}

/**
 * @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());
    }
}