Java 类io.netty.handler.timeout.IdleStateHandler 实例源码

项目：fresco_floodlight 文件：RPCChannelInitializer.java

@Override
protected void initChannel(Channel ch) throws Exception {
    RPCChannelHandler channelHandler = 
            new RPCChannelHandler(syncManager, rpcService);

    IdleStateHandler idleHandler = 
            new IdleStateHandler(5, 10, 0);
    ReadTimeoutHandler readTimeoutHandler = 
            new ReadTimeoutHandler(30);

    ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast("idle", idleHandler);
    pipeline.addLast("timeout", readTimeoutHandler);
    pipeline.addLast("handshaketimeout",
                     new HandshakeTimeoutHandler(channelHandler, timer, 10));

    pipeline.addLast("syncMessageDecoder",
                     new SyncMessageDecoder(maxFrameSize));
    pipeline.addLast("syncMessageEncoder",
                     new SyncMessageEncoder());

    pipeline.addLast("handler", channelHandler);
}

项目：star-map 文件：StarClientProtocol.java

@Override
    public void open() {
        EventLoopGroup eventLoop = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(eventLoop);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3 * 1000);
        bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline()
//                        .addLast("logging",new LoggingHandler(LogLevel.INFO))
                        .addLast("decoder", new ObjectDecoder(ClassResolvers.cacheDisabled(getClass().getClassLoader()))) // in 1
                        .addLast("handler", new ClientReadHandler()) // in 2
                        .addLast("encoder", new ObjectEncoder())// out 3
                        .addLast("idleStateHandler", new IdleStateHandler(0, 1, 0))
                        .addLast(new ClientIdleHandler());

            }
        });
    }

项目：hekate 文件：NettyServerClient.java

private Optional<IdleStateHandler> mayBeCreateIdleStateHandler() {
    if (hbInterval > 0 && hbLossThreshold > 0) {
        int interval = hbInterval;
        int readTimeout = hbInterval * hbLossThreshold;

        if (hbDisabled) {
            interval = 0;

            if (debug) {
                log.debug("Registering heartbeatless timeout handler [from={}, read-timeout={}]", address(), readTimeout);
            }
        } else {
            if (debug) {
                log.debug("Registering heartbeat handler [from={}, interval={}, loss-threshold={}, read-timeout={}]",
                    address(), interval, hbLossThreshold, readTimeout);
            }
        }

        return Optional.of(new IdleStateHandler(readTimeout, interval, 0, TimeUnit.MILLISECONDS));
    }

    return Optional.empty();
}

项目：qonduit 文件：Server.java

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

}

项目：graphiak 文件：GraphiakInitializer.java

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

    // removes idle connections after READER_IDLE_SECONDS seconds
    p.addLast("idleStateHandler",
            new IdleStateHandler(READER_IDLE_SECONDS, 0, 0));

    // handle new connections and idle timeouts
    p.addLast("auth", authHandler);

    // break each data chunk by newlines and split out metrics
    p.addLast("line", new GraphiteMetricDecoder(maxLength));

    // batch up metrics and store
    p.addLast("metrics", new MetricHandler(store));
}

项目：FFS-PubSub 文件：NetworkManager.java

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    try {
        ch.config().setOption(ChannelOption.TCP_NODELAY, true);
        ch.config().setOption(ChannelOption.IP_TOS, 0x18);
    } catch (ChannelException ex) {
        // IP_TOS not supported by platform, ignore
    }
    ch.config().setAllocator(PooledByteBufAllocator.DEFAULT);

    PacketRegistry r = new PacketRegistry();

    ch.pipeline().addLast("idleStateHandler", new IdleStateHandler(0, 0, 30));
    ch.pipeline().addLast(new HttpServerCodec());
    ch.pipeline().addLast(new HttpObjectAggregator(65536));
    ch.pipeline().addLast(new WebSocketHandler());
    ch.pipeline().addLast(new PacketDecoder(r));
    ch.pipeline().addLast(new PacketEncoder(r));
    ch.pipeline().addLast(mExecutorGroup, "serverHandler", new ClientHandler(mServer));
}

项目：SDN-Multicast 文件：RPCChannelInitializer.java

@Override
protected void initChannel(Channel ch) throws Exception {
    RPCChannelHandler channelHandler = 
            new RPCChannelHandler(syncManager, rpcService);

    IdleStateHandler idleHandler = 
            new IdleStateHandler(5, 10, 0);
    ReadTimeoutHandler readTimeoutHandler = 
            new ReadTimeoutHandler(30);

    ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast("idle", idleHandler);
    pipeline.addLast("timeout", readTimeoutHandler);
    pipeline.addLast("handshaketimeout",
                     new HandshakeTimeoutHandler(channelHandler, timer, 10));

    pipeline.addLast("syncMessageDecoder",
                     new SyncMessageDecoder(maxFrameSize));
    pipeline.addLast("syncMessageEncoder",
                     new SyncMessageEncoder());

    pipeline.addLast("handler", channelHandler);
}

项目：arscheduler 文件：RPCChannelInitializer.java

@Override
protected void initChannel(Channel ch) throws Exception {
    RPCChannelHandler channelHandler = 
            new RPCChannelHandler(syncManager, rpcService);

    IdleStateHandler idleHandler = 
            new IdleStateHandler(5, 10, 0);
    ReadTimeoutHandler readTimeoutHandler = 
            new ReadTimeoutHandler(30);

    ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast("idle", idleHandler);
    pipeline.addLast("timeout", readTimeoutHandler);
    pipeline.addLast("handshaketimeout",
                     new HandshakeTimeoutHandler(channelHandler, timer, 10));

    pipeline.addLast("syncMessageDecoder",
                     new SyncMessageDecoder(maxFrameSize));
    pipeline.addLast("syncMessageEncoder",
                     new SyncMessageEncoder());

    pipeline.addLast("handler", channelHandler);
}

项目：sailfish 文件：AbstractConfigurableExchangeChannelGroup.java

private ChannelInitializer<SocketChannel> newChannelInitializer(final NegotiateConfig config,
        final ExchangeChannelGroup channelGroup, final EventExecutorGroup executorGroup) {
    return new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            ch.attr(ChannelAttrKeys.maxIdleTimeout).set(config.maxIdleTimeout());
            ch.attr(ChannelAttrKeys.channelGroup).set(channelGroup);
            ch.attr(ChannelAttrKeys.clientSide).set(true);
            ch.attr(OneTime.awaitNegotiate).set(new CountDownLatch(1));
            ch.attr(OneTime.channelConfig).set(config);
            // TODO should increase ioRatio when every ChannelHandler bind to executorGroup?
            pipeline.addLast(executorGroup, 
                    RemotingEncoder.INSTANCE, 
                    new RemotingDecoder(), 
                    new IdleStateHandler(config.idleTimeout(), 0, 0), 
                    HeartbeatChannelHandler.INSTANCE,
                    NegotiateChannelHandler.INSTANCE,
                    ConcreteRequestHandler.INSTANCE);
        }
    };
}

项目：raft-java 文件：StartServer.java

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    ch.config().setAllowHalfClosure(true);
    ChannelPipeline pipeline = ch.pipeline();

    //IdleStateHandler 与客户端链接后，根据超出配置的时间自动触发userEventTriggered
    //readerIdleTime服务端长时间没有读到数据，则为读空闲，触发读空闲监听，并自动关闭链路连接，周期性按readerIdleTime的超时间触发空闲监听方法
    //writerIdleTime服务端长时间没有发送写请求，则为空闲，触发写空闲监听,空闲期间，周期性按writerIdleTime的超时间触发空闲监听方法
    //allIdleTime 服务端在allIdleTime时间内未接收到客户端消息，或者，也未去向客户端发送消息，则触发周期性操作
    pipeline.addLast("ping", new IdleStateHandler(10, 20, 35, TimeUnit.SECONDS));
    // 以("\n")为结尾分割的 解码器
    pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
    // 字符串解码 和 编码
    pipeline.addLast("decoder", new StringDecoder());
    pipeline.addLast("encoder", new StringEncoder());
    // 自己的逻辑Handler
    pipeline.addLast("handler", new DataServerHandler(nodeInfo));
}

项目：raft-java 文件：StartServer.java

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    ch.config().setAllowHalfClosure(true);
    ChannelPipeline pipeline = ch.pipeline();

    //IdleStateHandler 与客户端链接后，根据超出配置的时间自动触发userEventTriggered
    //readerIdleTime服务端长时间没有读到数据，则为读空闲，触发读空闲监听，并自动关闭链路连接，周期性按readerIdleTime的超时间触发空闲监听方法
    //writerIdleTime服务端长时间没有发送写请求，则为空闲，触发写空闲监听,空闲期间，周期性按writerIdleTime的超时间触发空闲监听方法
    //allIdleTime 服务端在allIdleTime时间内未接收到客户端消息，或者，也未去向客户端发送消息，则触发周期性操作
    pipeline.addLast("ping", new IdleStateHandler(10, 20, 35, TimeUnit.SECONDS));
    // 以("\n")为结尾分割的 解码器
    pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
    // 字符串解码 和 编码
    pipeline.addLast("decoder", new StringDecoder());
    pipeline.addLast("encoder", new StringEncoder());
    // 自己的逻辑Handler
    pipeline.addLast("handler", new ElectionServerHandler(nodeInfo));
}

项目：shortcircuit-proxy 文件：NettyProxyFrontendInitializer.java

@Override
public void initChannel(SocketChannel ch) {
    /* Netty default: {@code maxInitialLineLength (4096)} */
    int maxInitialLineLength = 4096 * 2;
    /* Netty default: {@code maxHeaderSize (8192)} */
    int maxHeaderSize = 8192 * 2;
    /* Netty default: {@code maxChunkSize (8192)} */
    int maxChunkSize = 8192 * 2;
    int readerIdleTimeSeconds = 0;
    int writerIdleTimeSeconds = 0;
    int allIdleTimeSeconds = 10;
    ch.pipeline().addLast(new LoggingHandler(NettyProxyFrontendHandler.class), //
            new HttpRequestDecoder(maxInitialLineLength, maxHeaderSize, maxChunkSize), //
            new IdleStateHandler(readerIdleTimeSeconds, writerIdleTimeSeconds, allIdleTimeSeconds), //
            new NettyProxyFrontendHandler());
}

项目：floodlight1.2-delay 文件：RPCChannelInitializer.java

@Override
protected void initChannel(Channel ch) throws Exception {
    RPCChannelHandler channelHandler = 
            new RPCChannelHandler(syncManager, rpcService);

    IdleStateHandler idleHandler = 
            new IdleStateHandler(5, 10, 0);
    ReadTimeoutHandler readTimeoutHandler = 
            new ReadTimeoutHandler(30);

    ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast("idle", idleHandler);
    pipeline.addLast("timeout", readTimeoutHandler);
    pipeline.addLast("handshaketimeout",
                     new HandshakeTimeoutHandler(channelHandler, timer, 10));

    pipeline.addLast("syncMessageDecoder",
                     new SyncMessageDecoder(maxFrameSize));
    pipeline.addLast("syncMessageEncoder",
                     new SyncMessageEncoder());

    pipeline.addLast("handler", channelHandler);
}

项目：netty 文件：NettyClientBootstrap.java

private void start() throws InterruptedException {
    EventLoopGroup eventLoopGroup=new NioEventLoopGroup();
    Bootstrap bootstrap=new Bootstrap();
    bootstrap.channel(NioSocketChannel.class);
    bootstrap.option(ChannelOption.SO_KEEPALIVE,true);
    bootstrap.group(eventLoopGroup);
    bootstrap.remoteAddress(host,port);
    bootstrap.handler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel socketChannel) throws Exception {
            socketChannel.pipeline().addLast(new IdleStateHandler(20,10,0));
            socketChannel.pipeline().addLast(new ObjectEncoder());
            socketChannel.pipeline().addLast(new ObjectDecoder(ClassResolvers.cacheDisabled(null)));
            socketChannel.pipeline().addLast(new NettyClientHandler());
        }
    });
    ChannelFuture future =bootstrap.connect(host,port).sync();
    if (future.isSuccess()) {
        socketChannel = (SocketChannel)future.channel();
        System.out.println("connect server  成功---------");
    }
}

项目：EasyMessage 文件：NettyAcceptor.java

private void initializePlainTCPTransport(final NettyMQTTHandler handler, IConfig props) throws IOException {
    final SmartConnectorIdleTimeoutHandler timeoutHandler = new SmartConnectorIdleTimeoutHandler();
    String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
    int port = Integer.parseInt(props.getProperty(BrokerConstants.PORT_PROPERTY_NAME));
    initFactory(host, port, new PipelineInitializer() {
        @Override
        void init(ChannelPipeline pipeline) {
            pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
            pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
            //pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
            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);
        }
    });
}

项目：jrpc 文件：NettyConnector.java

public NettyConnector(InetSocketAddress isa, final TransportConfig transportConfig) {
    workerGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("N4C-Work"));
    clientBoot = new Bootstrap().group(workerGroup).channel(NioSocketChannel.class);
    clientBoot.option(ChannelOption.TCP_NODELAY, true);
    clientBoot.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, transportConfig.getConnectTimeout());
    clientBoot.option(ChannelOption.SO_RCVBUF, 8 * 1024).option(ChannelOption.SO_SNDBUF, 8 * 1024);
    clientBoot.handler(new ChannelInitializer<SocketChannel>() {

        @Override
        public void initChannel(SocketChannel ch) throws Exception {
            TransportProtocolDecoder decoder = new TransportProtocolDecoder();
            decoder.setMaxObjectSize(transportConfig.getMaxSize());
            TransportProtocolEncoder encoder = new TransportProtocolEncoder();
            encoder.setMaxObjectSize(transportConfig.getMaxSize());
            ch.pipeline().addLast("TransportProtocolDecoder", decoder);
            ch.pipeline().addLast("TransportProtocolEncoder", encoder);

            int intervalSeconds = transportConfig.getHeartbeatIntervalSeconds();
            ch.pipeline().addLast("IdleStateHandler", new IdleStateHandler(0, intervalSeconds, 0));
            ch.pipeline().addLast("NettyClientHandler", new NettyClientHandler());
        }
    });

    clientBoot.remoteAddress(isa);
}

项目：jrpc 文件：NettyConnector.java

public NettyConnector(InetSocketAddress isa, final TransportConfig transportConfig) {
    workerGroup = new NioEventLoopGroup(1, new DefaultExecutorServiceFactory("N5C-Work"));
    clientBoot = new Bootstrap().group(workerGroup).channel(NioSocketChannel.class);
    clientBoot.option(ChannelOption.TCP_NODELAY, true);
    clientBoot.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, transportConfig.getConnectTimeout());
    clientBoot.option(ChannelOption.SO_RCVBUF, 8 * 1024).option(ChannelOption.SO_SNDBUF, 8 * 1024);
    clientBoot.handler(new ChannelInitializer<SocketChannel>() {

        @Override
        public void initChannel(SocketChannel ch) throws Exception {
            TransportProtocolDecoder decoder = new TransportProtocolDecoder();
            decoder.setMaxObjectSize(transportConfig.getMaxSize());
            TransportProtocolEncoder encoder = new TransportProtocolEncoder();
            encoder.setMaxObjectSize(transportConfig.getMaxSize());
            ch.pipeline().addLast("TransportProtocolDecoder", decoder);
            ch.pipeline().addLast("TransportProtocolEncoder", encoder);

            int intervalSeconds = transportConfig.getHeartbeatIntervalSeconds();
            ch.pipeline().addLast("IdleStateHandler", new IdleStateHandler(0, intervalSeconds, 0));
            ch.pipeline().addLast("NettyClientHandler", new NettyClientHandler());
        }
    });

    clientBoot.remoteAddress(isa);
}

项目：jim 文件：NettyAcceptor.java

private void initializePlainTCPTransport(final NettyMQTTHandler handler, IConfig props) throws IOException {
    final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();
    String host = props.getProperty(BrokerConstants.HOST_PROPERTY_NAME);
    int port = Integer.parseInt(props.getProperty(BrokerConstants.PORT_PROPERTY_NAME));
    initFactory(host, port, new PipelineInitializer() {
        @Override
        void init(ChannelPipeline pipeline) {
            pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
            pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
            //pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
            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);
        }
    });
}

项目：floodlight-hardware 文件：RPCChannelInitializer.java

@Override
protected void initChannel(Channel ch) throws Exception {
    RPCChannelHandler channelHandler =
            new RPCChannelHandler(syncManager, rpcService);

    IdleStateHandler idleHandler =
            new IdleStateHandler(5, 10, 0);
    ReadTimeoutHandler readTimeoutHandler =
            new ReadTimeoutHandler(30);

    ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast("idle", idleHandler);
    pipeline.addLast("timeout", readTimeoutHandler);
    pipeline.addLast("handshaketimeout",
                     new HandshakeTimeoutHandler(channelHandler, timer, 10));

    pipeline.addLast("syncMessageDecoder",
                     new SyncMessageDecoder(maxFrameSize));
    pipeline.addLast("syncMessageEncoder",
                     new SyncMessageEncoder());

    pipeline.addLast("handler", channelHandler);
}

项目：ACAMPController 文件：RPCChannelInitializer.java

@Override
protected void initChannel(Channel ch) throws Exception {
    RPCChannelHandler channelHandler = 
            new RPCChannelHandler(syncManager, rpcService);

    IdleStateHandler idleHandler = 
            new IdleStateHandler(5, 10, 0);
    ReadTimeoutHandler readTimeoutHandler = 
            new ReadTimeoutHandler(30);

    ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast("idle", idleHandler);
    pipeline.addLast("timeout", readTimeoutHandler);
    pipeline.addLast("handshaketimeout",
                     new HandshakeTimeoutHandler(channelHandler, timer, 10));

    pipeline.addLast("syncMessageDecoder",
                     new SyncMessageDecoder(maxFrameSize));
    pipeline.addLast("syncMessageEncoder",
                     new SyncMessageEncoder());

    pipeline.addLast("handler", channelHandler);
}

项目：zbus 文件：TcpServer.java

@Override
protected void initChannel(SocketChannel ch) throws Exception {  
    ChannelPipeline p = ch.pipeline(); 
    int timeout = loop.getIdleTimeInSeconds();
    p.addLast(new IdleStateHandler(0, 0, timeout));
    SslContext sslCtx = loop.getSslContext();
    if(sslCtx != null){
        p.addLast(sslCtx.newHandler(ch.alloc()));
    }
    CodecInitializer initializer = getCodecInitializer();
    if(initializer != null){
        List<ChannelHandler> handlers = new ArrayList<ChannelHandler>();
        initializer.initPipeline(handlers);
        for(ChannelHandler handler : handlers){
             p.addLast((ChannelHandler)handler); 
        }
    }    
    p.addLast(this.nettyToIoAdaptor);
}

项目：SecureSmartHome 文件：ClientHandshakeHandler.java

/**
 * Called once the TCP connection is established.
 * Configures the per-connection pipeline that is responsible for handling incoming and outgoing data.
 * After an incoming packet is decrypted, decoded and verified,
 * it will be sent to its target {@link de.unipassau.isl.evs.ssh.core.handler.MessageHandler}
 * by the {@link IncomingDispatcher}.
 */
@Override
public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
    Log.v(TAG, "channelRegistered " + ctx);
    ctx.attr(ATTR_HANDSHAKE_FINISHED).set(false);

    // Add (de-)serialization Handlers before this Handler
    ctx.pipeline().addBefore(ctx.name(), ObjectEncoder.class.getSimpleName(), new ObjectEncoder());
    ctx.pipeline().addBefore(ctx.name(), ObjectDecoder.class.getSimpleName(), new ObjectDecoder(
            ClassResolvers.weakCachingConcurrentResolver(getClass().getClassLoader())));
    ctx.pipeline().addBefore(ctx.name(), LoggingHandler.class.getSimpleName(), new LoggingHandler(LogLevel.TRACE));

    // Timeout Handler
    ctx.pipeline().addBefore(ctx.name(), IdleStateHandler.class.getSimpleName(),
            new IdleStateHandler(READER_IDLE_TIME, WRITER_IDLE_TIME, ALL_IDLE_TIME));
    ctx.pipeline().addBefore(ctx.name(), TimeoutHandler.class.getSimpleName(), new TimeoutHandler());

    // Add exception handler
    ctx.pipeline().addAfter(ctx.name(), PipelinePlug.class.getSimpleName(), new PipelinePlug());

    super.channelRegistered(ctx);
    Log.v(TAG, "Pipeline after register: " + ctx.pipeline());
}

项目：NPush 文件：ServerInitializer.java

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

    /*
     * 使用ObjectDecoder和ObjectEncoder
     * 因为双向都有写数据和读数据，所以这里需要两个都设置
     * 如果只读，那么只需要ObjectDecoder即可
     */
    pipeline.addLast("decoder", new ObjectDecoder(ClassResolvers.cacheDisabled(this.getClass().getClassLoader())));
    pipeline.addLast("encoder", new ObjectEncoder());

    /*
     * 这里只监听读操作
     * 可以根据需求，监听写操作和总得操作
     */
    pipeline.addLast("idleStateHandler", new IdleStateHandler(0, 0, Constants.ALL_IDLE_TIME, TimeUnit.SECONDS));

    pipeline.addLast("handler", new ServerHandler());
}

项目：jlogstash-input-plugin 文件：Server.java

/**
 * 加入打日志、ssl、idleState、BeatsHandler和BeatsParser这几个handler。
 */
public void initChannel(SocketChannel socket) throws SSLException {
    ChannelPipeline pipeline = socket.pipeline();

    pipeline.addLast(LOGGER_HANDLER, loggingHandler);

    if(server.isSslEnable()) {
        SslHandler sslHandler = sslBuilder.build(socket.alloc());
        pipeline.addLast(SSL_HANDLER, sslHandler);
    }

    // We have set a specific executor for the idle check, because the `beatsHandler` can be
    // blocked on the queue, this the idleStateHandler manage the `KeepAlive` signal.
    pipeline.addLast(idleExecutorGroup, KEEP_ALIVE_HANDLER, new IdleStateHandler(60*15, 5, 0));
    pipeline.addLast(BEATS_PARSER, new BeatsParser());
    pipeline.addLast(BEATS_HANDLER, this.beatsHandler);
}

项目：flashback 文件：ProxyInitializer.java

@Override
public void initChannel(SocketChannel socketChannel) {
  ChannelPipeline channelPipeline = socketChannel.pipeline();
  channelPipeline.addLast("decoder", new HttpRequestDecoder());
  channelPipeline.addLast("encoder", new HttpResponseEncoder());
  channelPipeline.addLast("idle", new IdleStateHandler(0, 0, _proxyServer.getClientConnectionIdleTimeout()));
  ChannelMediator channelMediator = new ChannelMediator(socketChannel,
      _proxyServer.getProxyModeControllerFactory(),
      _proxyServer.getDownstreamWorkerGroup(),
      _proxyServer.getServerConnectionIdleTimeout(),
      _proxyServer.getAllChannels());
  ClientChannelHandler clientChannelHandler =
      new ClientChannelHandler(channelMediator, _proxyServer.getConnectionFlowRegistry());

  channelPipeline.addLast("handler", clientChannelHandler);
}

项目：cloud-pubsub-mqtt-proxy 文件：NettyAcceptor.java

private void initializePlainTcpTransport(IMessaging messaging, Properties props)
    throws IOException {
  final NettyMQTTHandler mqttHandler = new NettyMQTTHandler();
  final PubsubHandler handler = new PubsubHandler(pubsub, mqttHandler);
  handler.setMessaging(messaging);
  String host = props.getProperty(Constants.HOST_PROPERTY_NAME);
  int port = Integer.parseInt(props.getProperty(Constants.PORT_PROPERTY_NAME));
  initFactory(host, port, new PipelineInitializer() {
    @Override
    void init(ChannelPipeline pipeline) {
      pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0,
          Constants.DEFAULT_CONNECT_TIMEOUT));
      pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimeoutHandler());
      //pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
      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);
    }
  });
}

项目：Thor 文件：JSyncServer.java

protected void doOpen() throws Exception {
    bossGroup = new NioEventLoopGroup(1, (new NamedThreadFactory("NettyServerBoss", true)));
    workerGroup = new NioEventLoopGroup(DEFAULT_IO_THREADS, new NamedThreadFactory("NettyServerWorker", true));

    ServerBootstrap bootstrap = new ServerBootstrap();
    bootstrap.group(bossGroup, workerGroup)//
    .channel(NioServerSocketChannel.class)//
    .childOption(ChannelOption.TCP_NODELAY, false)//
    .childHandler(new ChannelInitializer<Channel>() {

        @Override
        public void initChannel(Channel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            pipeline.addLast(new NettyDecoder());
            pipeline.addLast(new NettyEncoder());
            pipeline.addLast(new IdleStateHandler(0, 0, serverChannelMaxIdleTimeSeconds));
            pipeline.addLast(new NettyServerHandler());
        }
    });

    ChannelFuture channelFuture = bootstrap.bind(getBindAddress()).sync();
    channelFuture.awaitUninterruptibly();
    channel = channelFuture.channel();
    // channel.closeFuture().sync();
}

项目：Thor 文件：JSyncClient.java

protected void doOpen() throws Exception {
    bootstrap = new Bootstrap();
    bootstrap.channel(NioSocketChannel.class);
    bootstrap.group(WORKER_GROUP);
    bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
    bootstrap.option(ChannelOption.TCP_NODELAY, true);
    bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1000);
    bootstrap.handler(new ChannelInitializer<SocketChannel>() {

        public void initChannel(SocketChannel ch) {
            ChannelPipeline pipeline = ch.pipeline();
            pipeline.addLast(new NettyDecoder());
            pipeline.addLast(new NettyEncoder());
            pipeline.addLast(new IdleStateHandler(0, 0, clientChannelMaxIdleTimeSeconds));
            pipeline.addLast(new NettyClientHandler());
        }
    });

    ChannelFuture channelFuture = bootstrap.connect(getBindAddress()).sync();
    channelFuture.awaitUninterruptibly();
    channel = channelFuture.channel();
    channel.closeFuture().sync();
}

项目：ircd4j 文件：IRCChannelInitializer.java

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    LineBasedFrameDecoder lineDecoder = new LineBasedFrameDecoder(MAX_LINE_LENGTH);
    StringDecoder stringDecoder = new StringDecoder(CHARSET); //FIXME: Should only split on CRLF, not on LF alone
    MessageDecoder messageDecoder = new MessageDecoder();
    MessageHandler messageHandler = new MessageHandler(handler);

    StringEncoder stringEncoder = new StringEncoder(CHARSET);
    MessageEncoder messageEncoder = new MessageEncoder();

    IdleStateHandler idleHandler = new IdleStateHandler(IDLE_TIMEOUT, 0, 0);

    // Inbound goes from first to last, outbound goes from last to first.
    // i.e. the outside is on the left/top, the inside is on the right/bottom
    ch.pipeline().addLast(lineDecoder).addLast(stringDecoder).addLast(messageDecoder).addLast(idleHandler).addLast(messageHandler)
            .addLast(stringEncoder).addLast(messageEncoder);

}

项目：carbon-transports 文件：HTTPServerChannelInitializer.java

@Override
public void initChannel(SocketChannel ch) throws Exception {
    if (log.isDebugEnabled()) {
        log.debug("Initializing source channel pipeline");
    }

    ChannelPipeline pipeline = ch.pipeline();

    if (sslConfig != null) {
        pipeline.addLast(Constants.SSL_HANDLER, new SslHandler(new SSLHandlerFactory(sslConfig).build()));
    }

    pipeline.addLast("encoder", new HttpResponseEncoder());
    configureHTTPPipeline(pipeline);

    if (socketIdleTimeout > 0) {
        pipeline.addBefore(
                Constants.HTTP_SOURCE_HANDLER, Constants.IDLE_STATE_HANDLER,
                new IdleStateHandler(socketIdleTimeout, socketIdleTimeout, socketIdleTimeout,
                                     TimeUnit.MILLISECONDS));
    }
}

项目：carbon-transports 文件：HTTPClientRedirectTestCase.java

/**
 * When the maximum redirect count reached, channel should not do any more redirects.
 *
 * @throws URISyntaxException
 * @throws IOException
 */
@Test
public void unitTestForRedirectLoop() throws URISyntaxException, IOException {
    EmbeddedChannel embeddedChannel = new EmbeddedChannel();
    embeddedChannel.pipeline().addLast(new HttpResponseDecoder());
    embeddedChannel.pipeline().addLast(new HttpRequestEncoder());
    embeddedChannel.pipeline()
            .addLast(Constants.IDLE_STATE_HANDLER, new IdleStateHandler(50000, 50000, 0, TimeUnit.MILLISECONDS));
    embeddedChannel.pipeline().addLast(new RedirectHandler(null, false, 5, false, null, false));
    HttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.TEMPORARY_REDIRECT,
            Unpooled.EMPTY_BUFFER);
    response.headers().set(HttpHeaders.Names.LOCATION, FINAL_DESTINATION);
    embeddedChannel.attr(Constants.ORIGINAL_REQUEST)
            .set(createHttpRequest(Constants.HTTP_GET_METHOD, FINAL_DESTINATION));
    embeddedChannel.attr(Constants.RESPONSE_FUTURE_OF_ORIGINAL_CHANNEL).set(new HttpResponseFutureImpl());
    TargetChannel targetChannel = new TargetChannel(null, null);
    targetChannel.setChannel(embeddedChannel);
    embeddedChannel.attr(Constants.TARGET_CHANNEL_REFERENCE).set(targetChannel);
    embeddedChannel.attr(Constants.REDIRECT_COUNT).set(5);
    embeddedChannel.writeInbound(response);
    embeddedChannel.writeInbound(LastHttpContent.EMPTY_LAST_CONTENT);
    assertNull(embeddedChannel.readOutbound());
}

项目：blynk-server 文件：MQTTHardwareServer.java

public MQTTHardwareServer(Holder holder) {
    super(holder.props.getProperty("listen.address"),
            holder.props.getIntProperty("hardware.mqtt.port"), holder.transportTypeHolder);

    int hardTimeoutSecs = holder.limits.hardwareIdleTimeout;
    MqttHardwareLoginHandler mqttHardwareLoginHandler = new MqttHardwareLoginHandler(holder);
    HardwareChannelStateHandler hardwareChannelStateHandler =
            new HardwareChannelStateHandler(holder.sessionDao, holder.gcmWrapper);

    channelInitializer = new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline()
                .addLast("MqttIdleStateHandler", new IdleStateHandler(hardTimeoutSecs, hardTimeoutSecs, 0))
                .addLast(hardwareChannelStateHandler)
                .addLast(new MqttDecoder())
                .addLast(MqttEncoder.INSTANCE)
                .addLast(mqttHardwareLoginHandler)
                .addLast(new HardwareNotLoggedHandler());
        }
    };

    log.debug("hard.socket.idle.timeout = {}", hardTimeoutSecs);
}

项目：blynk-server 文件：BlynkInternalLogic.java

private void parseHardwareInfo(ChannelHandlerContext ctx, String[] messageParts,
                               HardwareStateHolder state, int msgId) {
    HardwareInfo hardwareInfo = new HardwareInfo(messageParts);
    int newHardwareInterval = hardwareInfo.heartbeatInterval;

    log.trace("Info command. heartbeat interval {}", newHardwareInterval);

    if (hardwareIdleTimeout != 0 && newHardwareInterval > 0) {
        int newReadTimeout = (int) Math.ceil(newHardwareInterval * 2.3D);
        log.debug("Changing read timeout interval to {}", newReadTimeout);
        ctx.pipeline().replace(IdleStateHandler.class,
                "H_IdleStateHandler_Replaced", new IdleStateHandler(newReadTimeout, newReadTimeout, 0));
    }

    DashBoard dashBoard = state.dash;
    Device device = state.device;

    if (device != null) {
        otaManager.initiateHardwareUpdate(ctx, state.userKey, hardwareInfo, dashBoard, device);
        device.hardwareInfo = hardwareInfo;
        dashBoard.updatedAt = System.currentTimeMillis();
    }

    ctx.writeAndFlush(ok(msgId), ctx.voidPromise());
}

项目：blynk-server 文件：HardwareServer.java

public HardwareServer(Holder holder) {
    super(holder.props.getProperty("listen.address"),
            holder.props.getIntProperty("hardware.default.port"), holder.transportTypeHolder);

    final int hardTimeoutSecs = holder.limits.hardwareIdleTimeout;
    final HardwareLoginHandler hardwareLoginHandler = new HardwareLoginHandler(holder, port);
    final HardwareChannelStateHandler hardwareChannelStateHandler =
            new HardwareChannelStateHandler(holder.sessionDao, holder.gcmWrapper);
    final AlreadyLoggedHandler alreadyLoggedHandler = new AlreadyLoggedHandler();

    channelInitializer = new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline()
                    .addLast("H_IdleStateHandler", new IdleStateHandler(hardTimeoutSecs, hardTimeoutSecs, 0))
                    .addLast("H_ChannelState", hardwareChannelStateHandler)
                    .addLast("H_MessageDecoder", new MessageDecoder(holder.stats))
                    .addLast("H_MessageEncoder", new MessageEncoder(holder.stats))
                    .addLast("H_Login", hardwareLoginHandler)
                    .addLast("H_NotLogged", new HardwareNotLoggedHandler())
                    .addLast("H_AlreadyLogged", alreadyLoggedHandler);
        }
    };

    log.debug("hard.socket.idle.timeout = {}", hardTimeoutSecs);
}

项目：blynk-server 文件：HardwareSSLServer.java

public HardwareSSLServer(Holder holder) {
    super(holder.props.getProperty("listen.address"),
            holder.props.getIntProperty("hardware.ssl.port"), holder.transportTypeHolder);

    HardwareLoginHandler hardwareLoginHandler = new HardwareLoginHandler(holder, port);
    HardwareChannelStateHandler hardwareChannelStateHandler =
            new HardwareChannelStateHandler(holder.sessionDao, holder.gcmWrapper);
    AlreadyLoggedHandler alreadyLoggedHandler = new AlreadyLoggedHandler();

    final int hardTimeoutSecs = holder.limits.hardwareIdleTimeout;

    this.channelInitializer = new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline()
                .addLast("HSSL_ReadTimeout", new IdleStateHandler(hardTimeoutSecs, hardTimeoutSecs, 0))
                .addLast("HSSL", holder.sslContextHolder.sslCtx.newHandler(ch.alloc()))
                .addLast("HSSLChannelState", hardwareChannelStateHandler)
                .addLast("HSSLMessageDecoder", new MessageDecoder(holder.stats))
                .addLast("HSSLMessageEncoder", new MessageEncoder(holder.stats))
                .addLast("HSSLLogin", hardwareLoginHandler)
                .addLast("HSSLNotLogged", new HardwareNotLoggedHandler())
                .addLast("HSSLAlreadyLogged", alreadyLoggedHandler);
        }
    };
}

项目：blynk-server 文件：BlynkInternalLogicTest.java

@Test
public void testCorrectBehavior() {
    BlynkInternalLogic logic = new BlynkInternalLogic(otaManager, props.getIntProperty("hard.socket.idle.timeout", 0));

    when(ctx.pipeline()).thenReturn(pipeline);
    when(ctx.alloc()).thenReturn(allocator);
    when(allocator.ioBuffer(anyInt())).thenReturn(byteBuf);
    when(byteBuf.writeByte(eq(0))).thenReturn(byteBuf);
    when(byteBuf.writeShort(eq(1))).thenReturn(byteBuf);
    when(byteBuf.writeShort(eq(200))).thenReturn(byteBuf);

    User user = new User();
    user.profile = new Profile();
    DashBoard dashBoard = new DashBoard();
    dashBoard.id = 1;
    user.profile.dashBoards = new DashBoard[] {dashBoard};
    Device device = new Device();
    HardwareStateHolder hardwareStateHolder = new HardwareStateHolder(user, dashBoard, device);

    BlynkInternalMessage hardwareInfoLogic = new BlynkInternalMessage(1, "ver 0.3.2-beta h-beat 60 buff-in 256 dev ESP8266".replaceAll(" ", "\0"));
    logic.messageReceived(ctx, hardwareStateHolder, hardwareInfoLogic);

    verify(pipeline).replace(eq(IdleStateHandler.class), eq("H_IdleStateHandler_Replaced"), any());
    verify(ctx).writeAndFlush(any(), any());
}

项目：tsdblite 文件：Server.java

/**
     * {@inheritDoc}
     * @see io.netty.channel.ChannelInitializer#initChannel(io.netty.channel.Channel)
     */
    @Override
    protected void initChannel(final SocketChannel ch) throws Exception {
        createdChannels.increment();
        channelGroup.add(ch);       
        ch.closeFuture().addListener(new GenericFutureListener<Future<? super Void>>() {
            @Override
            public void operationComplete(Future<? super Void> future) throws Exception {
                log.info("\n\t==============================\n\tChannel Closed [{}]\n\t==============================", ch.id());
//              log.error("Close Back trace", new Exception());
//              if(future.cause()!=null) {
//                  log.error("Close fail", future.cause());                    
//              }
            }
        });
        ch.pipeline().addLast("IdleState", new IdleStateHandler(0, 0, 60));
        ch.pipeline().addLast("ProtocolSwitch", new ProtocolSwitch());
    }

项目：remote-procedure-call 文件：ClientChannelInitializer.java

@Override
protected void initChannel(SocketChannel channel) throws Exception {
    ChannelPipeline pipeline = channel.pipeline();

    // use the IdleStateHandler to get notified if you haven't received or sent data for dozens of seconds.
    // If this is the case, a heartbeat will be written to the remote peer, and if this fails the connection is closed.
    pipeline.addLast(this.executorGroup, "idleStateHandler", new IdleStateHandler(0, 0, Constants.HEARTBEAT_PERIOD, TimeUnit.SECONDS));
    pipeline.addLast(this.executorGroup, "heartbeatHandler", heartbeatHandler);

    if (this.compression) {
        // Enable stream compression
        pipeline.addLast(this.executorGroup, "deflater", ZlibCodecFactory.newZlibEncoder(ZlibWrapper.GZIP));
        pipeline.addLast(this.executorGroup, "inflater", ZlibCodecFactory.newZlibDecoder(ZlibWrapper.GZIP));
    }

    // NUL (0x00) is a message delimiter
    pipeline.addLast(this.executorGroup, "framer", new DelimiterBasedFrameDecoder(8192, Delimiters.nulDelimiter()));

    // string encoder / decoder are responsible for encoding / decoding an UTF-8 string
    pipeline.addLast(this.executorGroup, "encoder", utf8Encoder);
    pipeline.addLast(this.executorGroup, "decoder", utf8Decoder);

    // client hander is responsible for as a remoting call stub
    pipeline.addLast(this.executorGroup, "clientHandler", clientHandler);
}

项目：aj8 文件：JagGrabChannelHandler.java

@Override
protected void initChannel(SocketChannel channel) {
    ChannelPipeline pipeline = channel.pipeline();

    // decoders
    pipeline.addLast("framer", new DelimiterBasedFrameDecoder(MAX_REQUEST_LENGTH, DOUBLE_LINE_FEED_DELIMITER));
    pipeline.addLast("string-decoder", new StringDecoder(JAGGRAB_CHARSET));
    pipeline.addLast("jaggrab-decoder", new JagGrabRequestDecoder());

    // encoders
    pipeline.addLast("jaggrab-encoder", new JagGrabResponseEncoder());

    // handler
    pipeline.addLast("timeout", new IdleStateHandler(NetworkConstants.IDLE_TIME, 0, 0));
    pipeline.addLast("handler", handler);
}

项目：aj8 文件：JagGrabChannelHandler.java

@Override
protected void initChannel(SocketChannel channel) {
    ChannelPipeline pipeline = channel.pipeline();

    // decoders
    pipeline.addLast("framer", new DelimiterBasedFrameDecoder(MAX_REQUEST_LENGTH, DOUBLE_LINE_FEED_DELIMITER));
    pipeline.addLast("string-decoder", new StringDecoder(JAGGRAB_CHARSET));
    pipeline.addLast("jaggrab-decoder", new JagGrabRequestDecoder());

    // encoders
    pipeline.addLast("jaggrab-encoder", new JagGrabResponseEncoder());

    // handler
    pipeline.addLast("timeout", new IdleStateHandler(NetworkConstants.IDLE_TIME, 0, 0));
    pipeline.addLast("handler", handler);
}