java类io.netty.channel.socket.SocketChannel的实例源码

public void startSocket() throws InterruptedException {
    EventLoopGroup boss = new NioEventLoopGroup();
    EventLoopGroup worker = new NioEventLoopGroup();

    try {
        ServerBootstrap boot = new ServerBootstrap();
        boot.group(boss,worker);

        boot.channel(NioServerSocketChannel.class);
        boot.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024*1024,0,4,-4,0,false));
                ch.pipeline().addLast(new ByteToPacketCodec());
                //ch.pipeline().addLast(new LoginChannelHandler(listener));
                ch.pipeline().addLast(new PacketChannelHandler(listener));
            }
        });

        boot.option(ChannelOption.SO_BACKLOG,128);
        boot.childOption(ChannelOption.SO_KEEPALIVE,true);
        channelFuture = boot.bind(port).sync();
        System.out.println("服务器"+port+"开启成功...");
        channelFuture.channel().closeFuture().sync();
    }finally {
        boss.shutdownGracefully().sync();
        worker.shutdownGracefully().sync();
        channelFuture = null;
        System.out.println("服务器关闭成功...");
    }
}

@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
    ChannelPipeline cp = socketChannel.pipeline();
    cp.addLast(new HttpServerCodec());  //添加服务端http编、解码器
    cp.addLast(new HttpObjectAggregator(512*1024));  //http消息聚合
    cp.addLast(new HttpContentCompressor());   //开启压缩
    cp.addLast(new HttpServerHandler(kurdran));
}

@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
    socketChannel.pipeline().addLast(
            new HttpServerCodec(),
            new HttpServerExpectContinueHandler(),
            new HttpObjectAggregator(Integer.MAX_VALUE),
            new ChunkedWriteHandler(),
            new HttpRequestHandler()
    );
}

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

    pipeline.addLast(new DelimiterBasedFrameDecoder(1048576 * 2, Delimiters.lineDelimiter()));
    pipeline.addLast(new StringDecoder());
    pipeline.addLast(new StringEncoder());

    pipeline.addLast(new ServerHandler());
}

/**
 * The Method that will initialize the channel.
 *
 * @param socketChannel The channel.
 *
 * @throws Exception Codec exception.
 */
protected void initChannel(SocketChannel socketChannel) throws Exception {
    ChannelPipeline pipeline = socketChannel.pipeline();

    pipeline.addLast(new HTTPDecoder());
    pipeline.addLast(new HTTPEncoder());
    pipeline.addLast(new EchidnaConnection(socketChannel, server));
}

public Client ( final SocketAddress address, final ConnectionStateListener listener, final ProtocolOptions options, final List<ClientModule> modules )
{
    this.address = address;
    this.options = options;

    this.listener = listener;

    this.manager = new MessageManager ( options );

    this.group = new NioEventLoopGroup ();

    this.bootstrap = new Bootstrap ();
    this.bootstrap.group ( this.group );
    this.bootstrap.channel ( NioSocketChannel.class );

    this.bootstrap.handler ( new ChannelInitializer<SocketChannel> () {

        @Override
        protected void initChannel ( final SocketChannel ch ) throws Exception
        {
            handleInitChannel ( ch );
        }
    } );

    this.modules = modules.toArray ( new ClientModule[modules.size ()] );
    this.executor = Executors.newSingleThreadExecutor ( new NamedThreadFactory ( "IEC60870Client/" + address ) );

    for ( final ClientModule module : modules )
    {
        module.initializeClient ( this, this.manager );
    }
}

@Before
public void setUp()
throws Exception {

    serverMock = new NioConnDroppingServer(DEFAULT_PORT, FAIL_EVERY_CONN_ATTEMPT);

    final Semaphore concurrencyThrottle = new Semaphore(CONCURRENCY);
    group = new NioEventLoopGroup();
    final Bootstrap bootstrap = new Bootstrap()
        .group(group)
        .channel(NioSocketChannel.class)
        .handler(
            new ChannelInitializer<SocketChannel>() {
                @Override
                protected final void initChannel(final SocketChannel conn)
                throws Exception {
                    conn.pipeline().addLast(new DummyClientChannelHandler());
                }
            }
        )
        .option(ChannelOption.SO_KEEPALIVE, true)
        .option(ChannelOption.SO_REUSEADDR, true)
        .option(ChannelOption.TCP_NODELAY, true);
    connPool = new BasicMultiNodeConnPool(
        concurrencyThrottle, NODES, bootstrap, CPH, DEFAULT_PORT, 0
    );
    connPool.preCreateConnections(CONCURRENCY);
}

@Override
public void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast("encoder", new MqttMessageEncoder());
    pipeline.addLast("decoder", new MqttMessageDecoder());
    pipeline.addLast("handler", new MqttMessageHandler());
}

protected ChannelHandler setupHttpChannel(Configuration config, SslContext sslCtx) {

        return new ChannelInitializer<SocketChannel>() {

            @Override
            protected void initChannel(SocketChannel ch) throws Exception {

                ch.pipeline().addLast("ssl", new NonSslRedirectHandler(config, sslCtx));
                ch.pipeline().addLast("encoder", new HttpResponseEncoder());
                ch.pipeline().addLast("decoder", new HttpRequestDecoder());
                ch.pipeline().addLast("compressor", new HttpContentCompressor());
                ch.pipeline().addLast("decompressor", new HttpContentDecompressor());
                ch.pipeline().addLast("aggregator", new HttpObjectAggregator(8192));
                ch.pipeline().addLast("chunker", new ChunkedWriteHandler());
                final Configuration.Cors corsCfg = config.getHttp().getCors();
                final CorsConfig.Builder ccb;
                if (corsCfg.isAllowAnyOrigin()) {
                    ccb = new CorsConfig.Builder();
                } else {
                    ccb = new CorsConfig.Builder(corsCfg.getAllowedOrigins().stream().toArray(String[]::new));
                }
                if (corsCfg.isAllowNullOrigin()) {
                    ccb.allowNullOrigin();
                }
                if (corsCfg.isAllowCredentials()) {
                    ccb.allowCredentials();
                }
                corsCfg.getAllowedMethods().stream().map(HttpMethod::valueOf).forEach(ccb::allowedRequestMethods);
                corsCfg.getAllowedHeaders().forEach(ccb::allowedRequestHeaders);
                CorsConfig cors = ccb.build();
                LOG.trace("Cors configuration: {}", cors);
                ch.pipeline().addLast("cors", new CorsHandler(cors));
                ch.pipeline().addLast("queryDecoder", new qonduit.netty.http.HttpRequestDecoder(config));
                ch.pipeline().addLast("strict", new StrictTransportHandler(config));
                ch.pipeline().addLast("login", new X509LoginRequestHandler(config));
                ch.pipeline().addLast("doLogin", new BasicAuthLoginRequestHandler(config));
                ch.pipeline().addLast("error", new HttpExceptionHandler());
            }
        };
    }

protected void initChannel(SocketChannel socketChannel) {
    socketChannel.pipeline()
                 .addLast(new AmqpDecoder())
                 .addLast(new AmqpEncoder())
                 .addLast(new AmqpConnectionHandler(configuration, broker))
                 .addLast(ioExecutors, new AmqpMessageWriter())
                 .addLast(ioExecutors, new BlockingTaskHandler());
}