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

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

protected ChannelHandler setupWSChannel(SslContext sslCtx, Configuration conf, DataStore datastore) {
    return new ChannelInitializer<SocketChannel>() {

        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline().addLast("ssl", sslCtx.newHandler(ch.alloc()));
            ch.pipeline().addLast("httpServer", new HttpServerCodec());
            ch.pipeline().addLast("aggregator", new HttpObjectAggregator(8192));
            ch.pipeline().addLast("sessionExtractor", new WebSocketHttpCookieHandler(config));
            ch.pipeline().addLast("idle-handler", new IdleStateHandler(conf.getWebsocket().getTimeout(), 0, 0));
            ch.pipeline().addLast("ws-protocol", new WebSocketServerProtocolHandler(WS_PATH, null, true));
            ch.pipeline().addLast("wsDecoder", new WebSocketRequestDecoder(datastore, config));
            ch.pipeline().addLast("error", new WSExceptionHandler());
        }
    };

}

@Override
protected void initChannel(Channel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    //编解码http请求
    pipeline.addLast(new HttpServerCodec());
    //聚合解码HttpRequest/HttpContent/LastHttpContent到FullHttpRequest
    //保证接收的Http请求的完整性
    pipeline.addLast(new HttpObjectAggregator(64 *1024));
    //写文件内容
    pipeline.addLast(new ChunkedWriteHandler());
    //处理FullHttpRequest
    pipeline.addLast(new HttpRequestHandler("/ws"));
    //处理其他的WebSocketFrame
    pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
    //处理TextWebSocketFrame
    pipeline.addLast(new TextWebSocketFrameHandler(group));
}

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    // 如果WebSocket握手完成
    if (evt == WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE) {
        // 删除ChannelPipeline中的HttpRequestHttpHandler
        ctx.pipeline().remove(HttpRequestHandler.class);
        String user = ChatUtils.addChannel(ctx.channel());
        Users us = new Users(user);
        ctx.channel().writeAndFlush(new TextWebSocketFrame(us.getCurrentUser()));
        // 写一个消息到ChannelGroup
        group.writeAndFlush(new TextWebSocketFrame(user + " 加入聊天室."));
        // 将channel添加到ChannelGroup
        group.add(ctx.channel());
        group.writeAndFlush(new TextWebSocketFrame(us.getAllUsers()));
    } else {
        super.userEventTriggered(ctx, evt);
    }
}

@Override
protected void initChannel(Channel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    //编解码http请求
    pipeline.addLast(new HttpServerCodec());
    //聚合解码HttpRequest/HttpContent/LastHttpContent到FullHttpRequest
    //保证接收的Http请求的完整性
    pipeline.addLast(new HttpObjectAggregator(64 *1024));
    //写文件内容
    pipeline.addLast(new ChunkedWriteHandler());
    //处理FullHttpRequest
    pipeline.addLast(new HttpRequestHandler("/ws"));
    //处理其他的WebSocketFrame
    pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
    //处理TextWebSocketFrame
    pipeline.addLast(new TextWebSocketFrameHandler(group));
}

/**
 * Processes an HTTP request
 * @param ctx The channel handler context
 * @param req The HTTP request
 */
public void handleRequest(final ChannelHandlerContext ctx, final FullHttpRequest req) {
    log.warn("HTTP Request: {}", req);
       if (req.method() != GET) {
           sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, FORBIDDEN));
           return;
       }
       String uri = req.uri();
       if(!"/ws".equals(uri)) {
        //channelRead(ctx, req);
        final WebSocketServerProtocolHandler wsProto = ctx.pipeline().get(WebSocketServerProtocolHandler.class);
        if(wsProto != null) {
            try {
                wsProto.acceptInboundMessage(req);
                return;
            } catch (Exception ex) {
                log.error("Failed to dispatch http request to WebSocketServerProtocolHandler on channel [{}]", ctx.channel(), ex);
            }
        }
       }
       log.error("Failed to handle HTTP Request [{}] on channel [{}]", req, ctx.channel());
       sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, HttpResponseStatus.MISDIRECTED_REQUEST));
}

@Override
protected void decode(ChannelHandlerContext ctx, HttpObject msg, List<Object> out)
{
    if(!added && msg instanceof HttpRequest)
    {
        String path = ((HttpRequest)msg).getUri();
        WsServerHandler handler = findHandler(path);
        if(handler != null)
        {
            ctx.pipeline().addAfter("switch", "aggregator", new HttpObjectAggregator(65536));
            ctx.pipeline().addAfter("aggregator", "wsprotocol", new WebSocketServerProtocolHandler(path, null, true));
            ctx.pipeline().addAfter("wsprotocol", "wshandler", new WsFrameHandler(handler));
            added = true;
        }
    }
    ReferenceCountUtil.retain(msg);
    out.add(msg);
}

protected ChannelInitializer<Channel> createInitializer() {

    return new ChannelInitializer<Channel>() {

        @Override
        protected void initChannel(Channel ch) throws Exception {
            ChannelPipeline p = ch.pipeline();
            p.addLast(new HttpServerCodec() );
            p.addLast(new ChunkedWriteHandler());
            p.addLast(new HttpObjectAggregator(64 * 1024));
            p.addLast(new EchoServerHttpRequestHandler("/ws"));
            p.addLast(new WebSocketServerProtocolHandler("/ws"));
            p.addLast(new EchoServerWSHandler());
        }
    };
}

private void initWebSocketPipeline(ChannelHandlerContext ctx, String websocketPath) {
    ChannelPipeline pipeline = ctx.pipeline();

    //websockets specific handlers
    pipeline.addLast("WSWebSocketServerProtocolHandler", new WebSocketServerProtocolHandler(websocketPath, true));
    pipeline.addLast("WSWebSocket", new WebSocketHandler(stats));
    pipeline.addLast("WSMessageDecoder", new MessageDecoder(stats));
    pipeline.addLast("WSSocketWrapper", new WebSocketWrapperEncoder());
    pipeline.addLast("WSMessageEncoder", new MessageEncoder(stats));
    pipeline.addLast("WSWebSocketGenericLoginHandler", genericLoginHandler);
    pipeline.remove(this);
    pipeline.remove(ChunkedWriteHandler.class);
    pipeline.remove(UrlReWriterHandler.class);
    pipeline.remove(StaticFileHandler.class);
    pipeline.remove(LetsEncryptHandler.class);
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline p = ch.pipeline();
    if (sslCtx != null) {
        p.addLast(sslCtx.newHandler(ch.alloc()));
    }
    if (enableGzip) {
        p.addLast(new HttpContentCompressor());
    }
    p.addLast(new HttpServerCodec(36192 * 2, 36192 * 8, 36192 * 16, false));
    p.addLast(new HttpServerExpectContinueHandler());
    p.addLast(new HttpObjectAggregator(Integer.MAX_VALUE));
    p.addLast(new ChunkedWriteHandler());
    if (enableCors) {
        CorsConfig corsConfig = CorsConfigBuilder.forAnyOrigin().allowNullOrigin().allowCredentials().build();
        p.addLast(new CorsHandler(corsConfig));
    }
    if (null != blade.webSocketPath()) {
        p.addLast(new WebSocketServerProtocolHandler(blade.webSocketPath(), null, true));
        p.addLast(new WebSockerHandler(blade));
    }
    service.scheduleWithFixedDelay(() -> date = new AsciiString(DateKit.gmtDate(LocalDateTime.now())), 1000, 1000, TimeUnit.MILLISECONDS);
    p.addLast(new HttpServerHandler());
}

public ChannelFuture run() {

    final ServerBootstrap httpServerBootstrap = new ServerBootstrap();
    httpServerBootstrap.group(bossGroup, workerGroup)
      .channel(NioServerSocketChannel.class)
      .localAddress(new InetSocketAddress(port))
      .childHandler(new ChannelInitializer<SocketChannel>() {

        public void initChannel(final SocketChannel ch) throws Exception {
          ch.pipeline().addLast(
            new HttpResponseEncoder(),
            new HttpRequestDecoder(),
            new HttpObjectAggregator(65536),
            new WebSocketServerProtocolHandler("/debug-session"),
            new DebugProtocolHandler(debugWebsocketConfiguration));
        }

    });

    LOGG.log(Level.INFO, "starting camunda BPM debug HTTP websocket interface on port "+port+".");

    return httpServerBootstrap.bind(port);


  }

@Override
protected void initChannel(Channel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    //Decode  bytes  to  HTTP  requests  /  encode  HTTP requests to bytes.
    pipeline.addLast(new HttpServerCodec());
    //Allows to write a file content.
    pipeline.addLast(new ChunkedWriteHandler());
    //Aggregate decoded HttpRequest / HttpContent / LastHttpContent to FullHttpRequest. This way you will always receive only full Http requests
    pipeline.addLast(new HttpObjectAggregator(64 * 1024));
    //Handle FullHttpRequest which are not send to /ws URI and so serve the index.html page
    pipeline.addLast(new HttpRequestHandler("/ws"));
    //Handle the WebSocket upgrade and Ping/Pong/Close WebSocket frames to be RFC compliant
    pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
    //Handles Text frames and handshake completion events
    pipeline.addLast(new TextWebSocketFrameHandler(group));
}

@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
    ChannelPipeline pipeline = socketChannel.pipeline();
    pipeline.addLast("timeout", new ReadTimeoutHandler(15));
    pipeline.addLast("codec-http", new HttpServerCodec());
    pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
    pipeline.addLast("handler", new HTTPHandler(plugin));
    pipeline.addLast("websocket", new WebSocketServerProtocolHandler("/server"));
    pipeline.addLast("packet-decoder", new PacketDecoder());
    pipeline.addLast("packet-encoder", new PacketEncoder());
    pipeline.addLast("packet-handler", new ClientHandler(socketChannel, plugin));

    socketChannel.config().setAllocator(PooledByteBufAllocator.DEFAULT);

    plugin.getWebHandler().getChannelGroup().add(socketChannel);
}

@Override
public void initChannel(SocketChannel ch) {
  ChannelPipeline p = ch.pipeline();
  p.addLast(new ReadTimeoutHandler(60, TimeUnit.SECONDS));
  if (sslContext != null) {
    p.addLast(sslContext.newHandler(ch.alloc()));
  }
  p.addLast(new HttpContentCompressor(5));
  p.addLast(new HttpServerCodec());
  p.addLast(new HttpObjectAggregator(1048576));
  p.addLast(new ChunkedWriteHandler());
  if (null != corsConfig) {
    p.addLast(new CorsHandler(corsConfig));
  }
  p.addLast(new WebSocketServerCompressionHandler());
  p.addLast(new WebSocketServerProtocolHandler(webSocketPath, null, true));
  p.addLast(new LaputaServerHandler(null != sslContext, requestProcessor));
}

protected void initChannel(NioSocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    // 编解码 http 请求
    pipeline.addLast(new HttpServerCodec());
    // 写文件内容
    pipeline.addLast(new ChunkedWriteHandler());
    // 聚合解码 HttpRequest/HttpContent/LastHttpContent 到 FullHttpRequest
    // 保证接收的 Http 请求的完整性
    pipeline.addLast(new HttpObjectAggregator(64 * 1024));
    // 处理其他的 WebSocketFrame
    pipeline.addLast(new WebSocketServerProtocolHandler("/chat"));
    // 处理 TextWebSocketFrame
    pipeline.addLast(protoCodec);
    pipeline.addLast(serverHandler);
}

private void switchToHttp(ChannelHandlerContext ctx) {
    ChannelPipeline p = ctx.pipeline();
    p.addLast(new HttpServerCodec());
    p.addLast(new HttpObjectAggregator(65536));
    p.addLast(new WebSocketServerCompressionHandler());
    p.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH, "ws", true));
    p.addLast(new NetoJsonStringToMapWebSocketDecoder());
    p.addLast(new NetoMessageToWebsocketFrameEncoder());
    p.remove(this);

    // 핸들러를 다시 등록 했으므로 이벤트를 전파
    ctx.fireChannelActive();
}

@Override
public void initChannel(final SocketChannel ch) throws Exception {
    ch.pipeline().addLast(new HttpServerCodec(),
            new MqttMessageWebSocketFrameEncoder(),
            new HttpObjectAggregator(65536), httpRequestHandler,
            new WebSocketServerProtocolHandler(websocketUri),
            new MqttMessageWebSocketFrameDecoder(),
            new MqttMessageHandler());
}

@Override
protected final void initChannel(Channel ch) throws Exception {
    ChannelPipeline pipeline=ch.pipeline();
    pipeline.addLast(new HttpServerCodec());
       pipeline.addLast(new ChunkedWriteHandler());
       pipeline.addLast(new HttpObjectAggregator(64*1024));
       pipeline.addLast(new WebSocketServerProtocolHandler(uri));
       pipeline.addLast(new WebSocketConnectedServerHandler());//连接成功监听handler
}

@Override
public void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    if (sslCtx != null) {
        pipeline.addLast(sslCtx.newHandler(ch.alloc()));
    }
    pipeline.addLast(new HttpServerCodec());
    pipeline.addLast(new HttpObjectAggregator(65536));
    pipeline.addLast(new WebSocketServerCompressionHandler());
    pipeline.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH, "binary", true));
    pipeline.addLast(new WebSocketIndexPageHandler(pluginDataFolder));
    pipeline.addLast(new WebSocketFrameHandler(webSocketServerThread, checkIPBans));
}

@Override
public void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();

    pipeline.addLast(new HttpServerCodec());
    pipeline.addLast(new HttpObjectAggregator(64*1024));
    pipeline.addLast(new ChunkedWriteHandler());
    pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
    pipeline.addLast(new WebSocketFrameHandler()); // 使用了 WebSocketFrameHandler 来对 Channel 进行处理
}

@Override
protected void initPipeline(ChannelPipeline pipeline) {
    pipeline.addLast(new HttpServerCodec());
    pipeline.addLast(new HttpObjectAggregator(65536));
    pipeline.addLast(new WebSocketServerCompressionHandler());
    pipeline.addLast(new WebSocketServerProtocolHandler(CC.mp.net.ws_path, null, true));
    pipeline.addLast(new WebSocketIndexPageHandler());
    pipeline.addLast(getChannelHandler());
}

protected void configurePipeline(final ChannelPipeline pipeline) 
{
    pipeline.addLast("httpEncoder", new HttpResponseEncoder());
    pipeline.addLast("httpDecoder", new HttpRequestDecoder());
    pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
    pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", "mqtt, mqttv3.1, mqttv3.1.1"));
    pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
    pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
    pipeline.addLast("filter", new AuthenticationHandler(m_mqttListerner));
    pipeline.addLast("forward", new ForwardToMQTTBrokerHandler(m_mqttBrokerHost, m_mqttBrokerPort));
}

@Override
protected void initChannel(Channel ch) throws Exception {
    //添加websocket相关内容
    ChannelPipeline pip = ch.pipeline();
    pip.addLast(new HttpServerCodec());
    pip.addLast(new HttpObjectAggregator(65536));
    pip.addLast(new WebSocketServerProtocolHandler("/"));
    pip.addLast(new WebSocketDecoder());
    pip.addLast(new WebSocketEncoder());
    pip.addLast(new MessageDecoder(builder.getImessageandhandler()));
    pip.addLast(new MessageExecutor(builder.getConsumer(), builder.getListener()));
    for (ChannelHandler handler : builder.getExtraHandlers()) {
        pip.addLast(handler);
    }
}

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    //如果WebSocket握手完成
    if(evt == WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE) {
        //删除ChannelPipeline中的HttpRequestHttpHandler
        ctx.pipeline().remove(HttpRequestHandler.class);
        //写一个消息到ChannelGroup
        group.writeAndFlush(new TextWebSocketFrame("Client " + ctx.channel() + " joined."));
        //将channel添加到ChannelGroup
        group.add(ctx.channel());
    } else {
        super.userEventTriggered(ctx, evt);
    }
}

protected ChannelHandler setupWSChannel(SslContext sslCtx, Configuration conf) {
    return new ChannelInitializer<SocketChannel>() {

        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline().addLast("ssl", sslCtx.newHandler(ch.alloc()));
            ch.pipeline().addLast("httpServer", new HttpServerCodec());
            ch.pipeline().addLast("aggregator", new HttpObjectAggregator(8192));
            ch.pipeline().addLast("sessionExtractor", new WebSocketHttpCookieHandler(config));
            ch.pipeline().addLast("idle-handler", new IdleStateHandler(conf.getWebsocket().getTimeout(), 0, 0));
            ch.pipeline().addLast("ws-protocol", new WebSocketServerProtocolHandler(WS_PATH, null, true));
            ch.pipeline().addLast("wsDecoder", new WebSocketRequestDecoder(config));
            ch.pipeline().addLast("aggregators", new WSAggregatorsRequestHandler());
            ch.pipeline().addLast("metrics", new WSMetricsRequestHandler(config));
            ch.pipeline().addLast("query", new WSQueryRequestHandler(dataStore));
            ch.pipeline().addLast("lookup", new WSSearchLookupRequestHandler(dataStore));
            ch.pipeline().addLast("suggest", new WSSuggestRequestHandler(dataStore));
            ch.pipeline().addLast("version", new WSVersionRequestHandler());
            ch.pipeline().addLast("put", new WSMetricPutHandler(dataStore));
            ch.pipeline().addLast("create", new WSCreateSubscriptionRequestHandler(dataStore, config));
            ch.pipeline().addLast("add", new WSAddSubscriptionRequestHandler());
            ch.pipeline().addLast("remove", new WSRemoveSubscriptionRequestHandler());
            ch.pipeline().addLast("close", new WSCloseSubscriptionRequestHandler());
            ch.pipeline().addLast("error", new WSTimelyExceptionHandler());
        }
    };

}

/**
 * 채널 파이프라인 설정.
 * Netty.Server.Configuration.NettyServerConfiguration 에서 등록한 Bean 을 이용해 사용자의 통신을 처리할 Handler 도 등록.
 * Netty.Server.Handler.JsonHandler 에서 실제 사용자 요청 처리.
 *
 * @param channel
 * @throws Exception
 */
@Override
protected void initChannel(Channel channel) throws Exception {

    ChannelPipeline channelPipeline = channel.pipeline();

    switch (transferType) {

        case "websocket":

            channelPipeline
                    .addLast(new HttpServerCodec())
                    .addLast(new HttpObjectAggregator(65536))
                    .addLast(new WebSocketServerCompressionHandler())
                    .addLast(new WebSocketServerProtocolHandler(transferWebsocketPath, transferWebsocketSubProtocol, transferWebsocketAllowExtensions))
                    .addLast(new LoggingHandler(LogLevel.valueOf(logLevelPipeline)))
                    .addLast(websocketHandler);

        case "tcp":
        default:

            channelPipeline
                    .addLast(new LineBasedFrameDecoder(Integer.MAX_VALUE))
                    .addLast(STRING_DECODER)
                    .addLast(STRING_ENCODER)
                    .addLast(new LoggingHandler(LogLevel.valueOf(logLevelPipeline)))
                    .addLast(jsonHandler);

    }

}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    if ("true".equalsIgnoreCase(Settings.INSTANCE.getProperty("webserver.logging.writelogOfNettyLogger"))) {
        ch.pipeline().addLast("log", new LoggingHandler("lannister.web/server", LogLevel.DEBUG));
    }

    if (useSsl) {
        SslContext sslCtx = SslContextBuilder
                .forServer(Settings.INSTANCE.certChainFile(), Settings.INSTANCE.privateKeyFile()).build();

        logger.debug("SSL Provider : {}", SslContext.defaultServerProvider());

        ch.pipeline().addLast(sslCtx.newHandler(ch.alloc()));
    }

    ch.pipeline().addLast(HttpServerCodec.class.getName(), new HttpServerCodec());
    ch.pipeline().addLast(HttpObjectAggregator.class.getName(), new HttpObjectAggregator(1048576));
    ch.pipeline().addLast(HttpContentCompressor.class.getName(), new HttpContentCompressor());
    ch.pipeline().addLast(HttpRequestRouter.class.getName(), new HttpRequestRouter());

    if (websocketFrameHandlerClass != null) {
        WebsocketFrameHandler wsfh = websocketFrameHandlerClass.newInstance();

        ch.pipeline().addLast(WebSocketServerProtocolHandler.class.getName(), new WebSocketServerProtocolHandler(
                wsfh.websocketPath(), wsfh.subprotocols(), wsfh.allowExtensions(), wsfh.maxFrameSize()));

        ch.pipeline().addLast(wsfh);
    }
}

private void initializeWebSocketTransport(final NettyMQTTHandler handler, IConfig props) throws IOException {
    String webSocketPortProp = props.getProperty(BrokerConstants.WEB_SOCKET_PORT_PROPERTY_NAME);
    if (webSocketPortProp == null) {
        //Do nothing no WebSocket configured
        LOG.info("WebSocket is disabled");
        return;
    }
    int port = Integer.parseInt(webSocketPortProp);

    final SmartConnectorIdleTimeoutHandler timeoutHandler = new SmartConnectorIdleTimeoutHandler();

    String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
    initFactory(host, port, new PipelineInitializer() {
        @Override
        void init(ChannelPipeline pipeline) {
            pipeline.addLast("httpEncoder", new HttpResponseEncoder());
            pipeline.addLast("httpDecoder", new HttpRequestDecoder());
            pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
            pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", "mqtt, mqttv3.1, mqttv3.1.1"));
            pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
            pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
            pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
            pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
            pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
            pipeline.addLast("decoder", new MQTTDecoder());
            pipeline.addLast("encoder", new MQTTEncoder());
            pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
            pipeline.addLast("handler", handler);
        }
    });
}

private void initializeWSSTransport(final NettyMQTTHandler handler, IConfig props, final SSLContext sslContext) throws IOException {
    String sslPortProp = props.getProperty(BrokerConstants.WSS_PORT_PROPERTY_NAME);
    if (sslPortProp == null) {
        //Do nothing no SSL configured
        LOG.info("SSL is disabled");
        return;
    }
    int sslPort = Integer.parseInt(sslPortProp);
    final SmartConnectorIdleTimeoutHandler timeoutHandler = new SmartConnectorIdleTimeoutHandler();
    String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
    initFactory(host, sslPort, new PipelineInitializer() {
        @Override
        void init(ChannelPipeline pipeline) throws Exception {
            pipeline.addLast("ssl", createSslHandler(sslContext));
            pipeline.addLast("httpEncoder", new HttpResponseEncoder());
            pipeline.addLast("httpDecoder", new HttpRequestDecoder());
            pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
            pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", "mqtt mqttv3.1, mqttv3.1.1"));
            pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
            pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
            pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
            pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
            pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
            pipeline.addLast("decoder", new MQTTDecoder());
            pipeline.addLast("encoder", new MQTTEncoder());
            pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
            pipeline.addLast("handler", handler);
        }
    });
}

@Override
public void configure(final ChannelPipeline pipeline) {
    if (logger.isDebugEnabled())
        pipeline.addLast(new LoggingHandler("log-io", LogLevel.DEBUG));

    logger.debug("HttpRequestDecoder settings - maxInitialLineLength={}, maxHeaderSize={}, maxChunkSize={}",
            settings.maxInitialLineLength, settings.maxHeaderSize, settings.maxChunkSize);
    pipeline.addLast("http-request-decoder", new HttpRequestDecoder(settings.maxInitialLineLength, settings.maxHeaderSize, settings.maxChunkSize));

    if (logger.isDebugEnabled())
        pipeline.addLast(new LoggingHandler("log-decoder-aggregator", LogLevel.DEBUG));

    logger.debug("HttpObjectAggregator settings - maxContentLength={}, maxAccumulationBufferComponents={}",
            settings.maxContentLength, settings.maxAccumulationBufferComponents);
    final HttpObjectAggregator aggregator = new HttpObjectAggregator(settings.maxContentLength);
    aggregator.setMaxCumulationBufferComponents(settings.maxAccumulationBufferComponents);
    pipeline.addLast("aggregator", aggregator);

    if (logger.isDebugEnabled())
        pipeline.addLast(new LoggingHandler("log-aggregator-encoder", LogLevel.DEBUG));

    pipeline.addLast("http-response-encoder", new HttpResponseEncoder());
    pipeline.addLast("request-handler", new WebSocketServerProtocolHandler("/gremlin", null, false, settings.maxContentLength));

    if (logger.isDebugEnabled())
        pipeline.addLast(new LoggingHandler("log-aggregator-encoder", LogLevel.DEBUG));

    pipeline.addLast("ws-frame-encoder", wsGremlinResponseFrameEncoder);
    pipeline.addLast("response-frame-encoder", gremlinResponseFrameEncoder);
    pipeline.addLast("request-text-decoder", wsGremlinTextRequestDecoder);
    pipeline.addLast("request-binary-decoder", wsGremlinBinaryRequestDecoder);
    pipeline.addLast("request-close-decoder", wsGremlinCloseRequestDecoder);

    if (logger.isDebugEnabled())
        pipeline.addLast(new LoggingHandler("log-aggregator-encoder", LogLevel.DEBUG));

    if (authenticationHandler != null)
        pipeline.addLast(PIPELINE_AUTHENTICATOR, authenticationHandler);
}

@Override
protected void initChannel(final SocketChannel ch) throws Exception {
    final ChannelPipeline pipeline = ch.pipeline();
       pipeline.addLast(new HttpServerCodec());
       pipeline.addLast(new HttpObjectAggregator(65536));
       pipeline.addLast(new WebSocketServerCompressionHandler());
       pipeline.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH, null, true));
       pipeline.addLast(webSockServiceHandler);

}

private void initializeWebSocketTransport(final NettyMQTTHandler handler, IConfig props) throws IOException {
    String webSocketPortProp = props.getProperty(BrokerConstants.WEB_SOCKET_PORT_PROPERTY_NAME);
    if (webSocketPortProp == null) {
        //Do nothing no WebSocket configured
        LOG.info("WebSocket is disabled");
        return;
    }
    int port = Integer.parseInt(webSocketPortProp);

    final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();

    String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
    initFactory(host, port, new PipelineInitializer() {
        @Override
        void init(ChannelPipeline pipeline) {
            pipeline.addLast("httpEncoder", new HttpResponseEncoder());
            pipeline.addLast("httpDecoder", new HttpRequestDecoder());
            pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
            pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", "mqtt, mqttv3.1, mqttv3.1.1"));
            pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
            pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
            pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
            pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
            pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
            pipeline.addLast("decoder", new MQTTDecoder());
            pipeline.addLast("encoder", new MQTTEncoder());
            pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
            pipeline.addLast("handler", handler);
        }
    });
}

private void initializeWSSTransport(final NettyMQTTHandler handler, IConfig props, final SSLContext sslContext) throws IOException {
    String sslPortProp = props.getProperty(BrokerConstants.WSS_PORT_PROPERTY_NAME);
    if (sslPortProp == null) {
        //Do nothing no SSL configured
        LOG.info("SSL is disabled");
        return;
    }
    int sslPort = Integer.parseInt(sslPortProp);
    final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
    String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
    String sNeedsClientAuth = props.getProperty(BrokerConstants.NEED_CLIENT_AUTH, "false");
    final boolean needsClientAuth =  Boolean.valueOf(sNeedsClientAuth);
    initFactory(host, sslPort, new PipelineInitializer() {
        @Override
        void init(ChannelPipeline pipeline) throws Exception {
            pipeline.addLast("ssl", createSslHandler(sslContext, needsClientAuth));
            pipeline.addLast("httpEncoder", new HttpResponseEncoder());
            pipeline.addLast("httpDecoder", new HttpRequestDecoder());
            pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
            pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", "mqtt mqttv3.1, mqttv3.1.1"));
            pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
            pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
            pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
            pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
            pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
            pipeline.addLast("decoder", new MQTTDecoder());
            pipeline.addLast("encoder", new MQTTEncoder());
            pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
            pipeline.addLast("handler", handler);
        }
    });
}

@Override
public void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    if (sslCtx != null) {
        pipeline.addLast(sslCtx.newHandler(ch.alloc()));
    }
    pipeline.addLast(new HttpServerCodec());
    pipeline.addLast(new HttpObjectAggregator(65536));
    pipeline.addLast(new WebSocketServerCompressionHandler());
    pipeline.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH, subProtocols, true));
    pipeline.addLast(new WebSocketRemoteServerFrameHandler());
}

@Override
public void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    if (sslCtx != null) {
        pipeline.addLast(sslCtx.newHandler(ch.alloc()));
    }
    pipeline.addLast(new HttpServerCodec());
    pipeline.addLast(new HttpObjectAggregator(65536));
    pipeline.addLast(new WebSocketServerCompressionHandler());
    pipeline.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH, null, true));
    pipeline.addLast(new WebSocketIndexPageHandler(WEBSOCKET_PATH));
    pipeline.addLast(new WebSocketFrameHandler());
}

@Override
public void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    if (sslCtx != null) {
        pipeline.addLast(sslCtx.newHandler(ch.alloc()));
    }
    pipeline.addLast(new HttpServerCodec());
    pipeline.addLast(new HttpObjectAggregator(65536));
    pipeline.addLast(new WebSocketServerCompressionHandler());
    pipeline.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH, null, true));
    pipeline.addLast(new WebSocketIndexPageHandler(WEBSOCKET_PATH));
    pipeline.addLast(new TextWebSocketFrameHandler());
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast(new HttpServerCodec());
    pipeline.addLast(new HttpObjectAggregator(65536));
    pipeline.addLast(new WebSocketServerCompressionHandler());
    pipeline.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH, null, true));
    pipeline.addLast(new WatcherServerIndexPageHandler(WEBSOCKET_PATH));
    pipeline.addLast(new WatcherServerHandler());
}

private void initializeWebSocketTransport(IMessaging messaging, Properties props)
    throws IOException {
  String webSocketPortProp = props.getProperty(Constants.WEB_SOCKET_PORT_PROPERTY_NAME);
  if (webSocketPortProp == null) {
    //Do nothing no WebSocket configured
    LOG.info("WebSocket is disabled");
    return;
  }
  int port = Integer.parseInt(webSocketPortProp);

  final NettyMQTTHandler mqttHandler = new NettyMQTTHandler();
  final PubsubHandler handler = new PubsubHandler(pubsub, mqttHandler);
  handler.setMessaging(messaging);

  String host = props.getProperty(Constants.HOST_PROPERTY_NAME);
  initFactory(host, port, new PipelineInitializer() {
    @Override
    void init(ChannelPipeline pipeline) {
      pipeline.addLast("httpEncoder", new HttpResponseEncoder());
      pipeline.addLast("httpDecoder", new HttpRequestDecoder());
      pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
      pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt"/*"/mqtt"*/,
          "mqttv3.1, mqttv3.1.1"));
      //pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler(null, "mqtt"));
      pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
      pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
      pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0,
          Constants.DEFAULT_CONNECT_TIMEOUT));
      pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimeoutHandler());
      pipeline.addFirst("bytemetrics", new BytesMetricsHandler(bytesMetricsCollector));
      pipeline.addLast("decoder", new MQTTDecoder());
      pipeline.addLast("encoder", new MQTTEncoder());
      pipeline.addLast("metrics", new MessageMetricsHandler(metricsCollector));
      pipeline.addLast("handler", handler);
    }
  });
}

private void initializeWssTransport(IMessaging messaging, Properties props,
    final SslHandler sslHandler) throws IOException {
  String sslPortProp = props.getProperty(Constants.WSS_PORT_PROPERTY_NAME);
  if (sslPortProp == null) {
    //Do nothing no SSL configured
    LOG.info("SSL is disabled");
    return;
  }
  int sslPort = Integer.parseInt(sslPortProp);
  final NettyMQTTHandler mqttHandler = new NettyMQTTHandler();
  final PubsubHandler handler = new PubsubHandler(pubsub, mqttHandler);
  handler.setMessaging(messaging);
  String host = props.getProperty(Constants.HOST_PROPERTY_NAME);
  initFactory(host, sslPort, new PipelineInitializer() {
    @Override
    void init(ChannelPipeline pipeline) throws Exception {
      pipeline.addLast("ssl", sslHandler);
      pipeline.addLast("httpEncoder", new HttpResponseEncoder());
      pipeline.addLast("httpDecoder", new HttpRequestDecoder());
      pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
      pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt",
          "mqttv3.1, mqttv3.1.1"));
      pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
      pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
      pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0,
          Constants.DEFAULT_CONNECT_TIMEOUT));
      pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimeoutHandler());
      pipeline.addFirst("bytemetrics", new BytesMetricsHandler(bytesMetricsCollector));
      pipeline.addLast("decoder", new MQTTDecoder());
      pipeline.addLast("encoder", new MQTTEncoder());
      pipeline.addLast("metrics", new MessageMetricsHandler(metricsCollector));
      pipeline.addLast("handler", handler);
    }
  });
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
  ChannelPipeline pipeline = ch.pipeline();
  pipeline.addLast(new HttpServerCodec());
  pipeline.addLast(new ChunkedWriteHandler());
  pipeline.addLast(new HttpObjectAggregator(64 * 1024));
  pipeline.addLast(new HttpRequestHandler("/ws"));
  pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
  pipeline.addLast(new TtyWebSocketFrameHandler(group, handler));
}