Java 类io.netty.channel.oio.OioEventLoopGroup 实例源码

public static void main(String[] args)
{
    Executor executor = Executors.newFixedThreadPool(200);
    ServerBootstrap bootstrap = new ServerBootstrap();
    EventLoopGroup bossGroup = new OioEventLoopGroup();
    EventLoopGroup childGroup = new OioEventLoopGroup();
    bootstrap.group(bossGroup, childGroup);
    bootstrap.channel(OioServerSocketChannel.class);

    bootstrap.childHandler(new RPCServerSessionPipelineFactory(
            new RPCServerMixinPipelineFactory(executor, childGroup)));

    // Bind and start to accept incoming connections.
    bootstrap.bind(new InetSocketAddress(8080));

}

public void init(ChannelPipelineFactory factory) throws Exception
{
    id = String.format("%1$020d",
            Math.abs(new Random(System.currentTimeMillis()).nextLong()))
            .getBytes();

    group = new OioEventLoopGroup();
    connectionlessBootstrap = new Bootstrap();
    connectionlessBootstrap.group(group);
    connectionlessBootstrap.option(ChannelOption.SO_BROADCAST, true);
    connectionlessBootstrap.handler(factory);
    connectionlessBootstrap.channel(OioDatagramChannel.class);
    ;
    datagramChannel = (DatagramChannel) connectionlessBootstrap
            .bind(new InetSocketAddress(mcastGroupPort)).sync().channel();
    multicastAddress = new InetSocketAddress(mcastGroupIp, mcastGroupPort);
    NetworkInterface networkInterface = NetworkInterface
            .getByInetAddress(InetAddress.getByName(bindAddress));
    // for (Enumeration nifs = NetworkInterface.getNetworkInterfaces();
    // nifs.hasMoreElements(); )
    datagramChannel.joinGroup(multicastAddress, null);// (NetworkInterface)
                                                        // nifs.nextElement());
    init = true;
    if (debug)
        factory.debug();
}

/**
 * Instantiates a new controller.
 * 
 * @param wrappedJavaProcess
 *            the wrapped java process
 */
public JVMController(WrappedProcess wrappedJavaProcess)
{
    super(wrappedJavaProcess);
    _bossGroup = new OioEventLoopGroup();
    _workerGroup = new OioEventLoopGroup();
    ControllerPipelineFactory pipelineFactory = new ControllerPipelineFactory(
            this);

    setDebug(((WrappedJavaProcess)wrappedJavaProcess).getDebug());
    pipelineFactory.setDebug(_debug > 2);
    _acceptor = new ServerBootstrap().group(_bossGroup, _workerGroup)
            .channel(OioServerSocketChannel.class)
            .childOption(ChannelOption.TCP_NODELAY, true)
            // .option(ChannelOption.SO_BACKLOG, 128)
            .childHandler(pipelineFactory);

}

public static void main(String[] args) throws Exception {
    EventLoopGroup group = new OioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group)
         .channel(RxtxChannel.class)
         .handler(new ChannelInitializer<RxtxChannel>() {
             @Override
             public void initChannel(RxtxChannel ch) throws Exception {
                 ch.pipeline().addLast(
                     new LineBasedFrameDecoder(32768),
                     new StringEncoder(),
                     new StringDecoder(),
                     new RxtxClientHandler()
                 );
             }
         });

        ChannelFuture f = b.connect(new RxtxDeviceAddress(PORT)).sync();

        f.channel().closeFuture().sync();
    } finally {
        group.shutdownGracefully();
    }
}

public static void main(String[] args) throws Exception {
    EventLoopGroup group = new OioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group)
         .channel(RxtxChannel.class)
         .handler(new ChannelInitializer<RxtxChannel>() {
             @Override
             public void initChannel(RxtxChannel ch) throws Exception {
                 ch.pipeline().addLast(
                     new LineBasedFrameDecoder(32768),
                     new StringEncoder(),
                     new StringDecoder(),
                     new RxtxClientHandler()
                 );
             }
         });

        ChannelFuture f = b.connect(new RxtxDeviceAddress(PORT)).sync();

        f.channel().closeFuture().sync();
    } finally {
        group.shutdownGracefully();
    }
}

public static void main(String[] args) throws Exception {
    EventLoopGroup bossGroup = new OioEventLoopGroup(1);
    EventLoopGroup workerGroup = new OioEventLoopGroup();
    try {
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
        .channel(OioServerSocketChannel.class)
        .childHandler(new ChannelInitializer<SocketChannel>() {
           @Override
           public void initChannel(SocketChannel ch) {
               ChannelPipeline p = ch.pipeline();
               p.addLast(new EchoServerHandler());
           }
        });

        ChannelFuture f = b.bind(8888).sync();

        f.channel().closeFuture().sync();
    }
    finally {
        workerGroup.shutdownGracefully();
        bossGroup.shutdownGracefully();
    }
}

public static void main(String[] args) throws Exception {
    EventLoopGroup group = new OioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group)
                .channel(RxtxChannel.class)
                .handler(new ChannelInitializer<RxtxChannel>() {
                    @Override
                    public void initChannel(RxtxChannel ch) throws Exception {
                        ch.pipeline().addLast(
                                new LineBasedFrameDecoder(32768),
                                new StringEncoder(),
                                new StringDecoder(),
                                new RxtxClientHandler()
                        );
                    }
                });

        ChannelFuture f = b.connect(new RxtxDeviceAddress(PORT)).sync();

        f.channel().closeFuture().sync();
    } finally {
        group.shutdownGracefully();
    }
}

public static void main(String[] args) throws Exception {
    EventLoopGroup group = new OioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group)
         .channel(RxtxChannel.class)
         .handler(new ChannelInitializer<RxtxChannel>() {
             @Override
             public void initChannel(RxtxChannel ch) throws Exception {
                 ch.pipeline().addLast(
                     new LineBasedFrameDecoder(32768),
                     new StringEncoder(),
                     new StringDecoder(),
                     new RxtxClientHandler()
                 );
             }
         });

        ChannelFuture f = b.connect(new RxtxDeviceAddress("/dev/ttyUSB0")).sync();

        f.channel().closeFuture().sync();
    } finally {
        group.shutdownGracefully();
    }
}

public static Bootstrap createBootstrap(final ChannelType channelType) throws UnsupportedOperationException {
    Bootstrap bootstrap = new Bootstrap();

    switch (channelType) {
        case NIO:

            bootstrap.group(new NioEventLoopGroup());
            bootstrap.channel(NioSocketChannel.class);
            return bootstrap;

        case OIO:

            bootstrap.group(new OioEventLoopGroup());
            bootstrap.channel(OioSocketChannel.class);
            return bootstrap;

        default:
            throw new UnsupportedOperationException("Failed to create Bootstrap,  " + channelType + " not supported!");
    }
}

public static Bootstrap createUDPBootstrap(final ChannelType channelType) throws UnsupportedOperationException {
    Bootstrap bootstrap = new Bootstrap();

    switch (channelType) {
        case NIO:

            bootstrap.group(new NioEventLoopGroup());
            bootstrap.channel(NioDatagramChannel.class);
            return bootstrap;

        case OIO:

            bootstrap.group(new OioEventLoopGroup());
            bootstrap.channel(OioDatagramChannel.class);
            return bootstrap;

        default:
            throw new UnsupportedOperationException("Failed to create Bootstrap,  " + channelType + " not supported!");
    }
}

public static Bootstrap createServerBootstrap(final ChannelType channelType) throws UnsupportedOperationException {

    Bootstrap serverBootstrap = new Bootstrap();

    switch (channelType) {
        case NIO:
            serverBootstrap.group(new NioEventLoopGroup(Runtime.getRuntime().availableProcessors()));
            serverBootstrap.channel(NioDatagramChannel.class);
           // serverBootstrap.localAddress(new InetSocketAddress(port))
           // .handler(packetHandler);

            return serverBootstrap;

        case OIO:
             serverBootstrap.group(new OioEventLoopGroup(Runtime.getRuntime().availableProcessors()));
             serverBootstrap.channel(OioDatagramChannel.class);


            return serverBootstrap;

        default:
            throw new UnsupportedOperationException("Failed to create ServerBootstrap,  " + channelType + " not supported!");
    }
}

public static ServerBootstrap createServerBootstrap(ChannelType channelType,boolean isUDP) throws UnsupportedOperationException {

    ServerBootstrap serverBootstrap = new ServerBootstrap();

    switch (channelType) {
        case NIO:
            serverBootstrap.group(new NioEventLoopGroup(), new NioEventLoopGroup());
            serverBootstrap.channel(NioServerSocketChannel.class);
            return serverBootstrap;

        case OIO:
            serverBootstrap.group(new OioEventLoopGroup(), new OioEventLoopGroup());
            serverBootstrap.channel(OioServerSocketChannel.class);
            return serverBootstrap;

        default:
            throw new UnsupportedOperationException("Failed to create ServerBootstrap,  " + channelType + " not supported!");
    }
}

public void start() {
        apiBootstrap = new ServerBootstrap();
        try {
            // the hub will only have a few connections, so OIO is likely to be faster than NIO in this case!
            ThreadFactory threadFactory = new NamedThreadFactory("kha-rest-api");
            EventLoopGroup commonGroup = new OioEventLoopGroup(0, threadFactory);
            apiBootstrap.group(commonGroup, commonGroup)
                    .channel(OioServerSocketChannel.class)
                    .localAddress(port)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childHandler(channelInitializer);

            apiBootstrap.bind();
//            ChannelFuture f = apiBootstrap.bind().sync();
            LOGGER.info("REST API available on http://{}:{}", InetAddress.getLocalHost().getCanonicalHostName(), port);
            LOGGER.info("WebSockets API available on ws://{}:{}", InetAddress.getLocalHost().getCanonicalHostName(), port);
//            f.channel().closeFuture().sync();
        } catch (Exception e) {
            LOGGER.error("Can't start API server", e);
        }
    }

public static void main(String[] args) throws Exception {
    EventLoopGroup group = new OioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group)
         .channel(RxtxChannel.class)
         .handler(new ChannelInitializer<RxtxChannel>() {
             @Override
             public void initChannel(RxtxChannel ch) throws Exception {
                 ch.pipeline().addLast(
                     new LineBasedFrameDecoder(32768),
                     new StringEncoder(),
                     new StringDecoder(),
                     new RxtxClientHandler()
                 );
             }
         });

        ChannelFuture f = b.connect(new RxtxDeviceAddress("/dev/ttyUSB0")).sync();

        f.channel().closeFuture().sync();
    } finally {
        group.shutdownGracefully();
    }
}

public SimpleLineBasedSerialChannel(String port, final SimpleStringChannelHandler stringHandler) {
    group = new OioEventLoopGroup();
       Bootstrap b = new Bootstrap();
       b.group(group)
        .channel(JsscChannel.class)
        .handler(new ChannelInitializer<JsscChannel>() {
            @Override
            public void initChannel(JsscChannel ch) throws Exception {
                ch.pipeline().addLast(
                    new LineBasedFrameDecoder(Integer.MAX_VALUE),
                    new StringDecoder(),
                    new SimpleChannelInboundHandler<String>() {
                     @Override
                     protected void channelRead0(io.netty.channel.ChannelHandlerContext ctx, String msg) throws Exception {
                        stringHandler.channelRead(ctx, msg); 
                     }
                 }
                );
            }
        });

        f = b.connect(new JsscDeviceAddress(port)).syncUninterruptibly();
}

/**
 * Initializes a new UDPMulticastBeacon.
 * 
 * @param serverConfig
 *            a configuration to use for initializing
 * @return the new UDPMulticastBeacon
 */
private UDPMulticastBeacon initializeBeacon(final AddressBasedServerConfig serverConfig) {
    LOGGER.entry();
    final OioEventLoopGroup networkEventLoopGroup = new OioEventLoopGroup();
    eventExecutorGroups.add(networkEventLoopGroup);

    final UDPMulticastBeacon beacon =
            new UDPMulticastBeacon(OIO_DATAGRAM_CHANNEL_FACTORY, networkEventLoopGroup, scheduledExecutorService,
                    serverConfig.getModuleID(), serverConfig.getAnnounceInterval(), TimeUnit.SECONDS);
    beacon.addListener((TCPConnectionManager) getConnectionManager());
    beacon.setAnnounceAddresses(new ArrayList<InetSocketAddress>(serverConfig.getAnnounceAddresses()));

    for (final NetConnection netConnection : serverConfig.getMulticastAddresses()) {
        LOGGER.debug("adding address {} to beacon", netConnection);
        beacon.addAddress(netConnection.getInterface(), netConnection.getAddress());
    }

    return LOGGER.exit(beacon);
}

/**
 * Returns a server socket channel class suitable for specified event loop group.
 *
 * @param eventLoopGroup the event loop group for which to identify an appropriate socket channel class; must not
 * be {@code null}
 *
 * @return a server socket channel class suitable for use with the given event loop group
 *
 * @throws IllegalArgumentException in case of null or unrecognized event loop group
 */
@SuppressWarnings("unchecked")
static Class<? extends ServerChannel> getServerSocketChannelClass(final EventLoopGroup eventLoopGroup) {
    Objects.requireNonNull(eventLoopGroup);

    final Class<? extends ServerChannel> serverSocketChannelClass;

    if (eventLoopGroup instanceof NioEventLoopGroup) {
        serverSocketChannelClass = NioServerSocketChannel.class;
    } else if (eventLoopGroup instanceof OioEventLoopGroup) {
        serverSocketChannelClass = OioServerSocketChannel.class;
    } else if (EPOLL_EVENT_LOOP_GROUP_CLASS.equals(eventLoopGroup.getClass().getName())) {
        serverSocketChannelClass = (Class<? extends ServerChannel>) loadSocketChannelClass(EPOLL_SERVER_SOCKET_CHANNEL_CLASS);
    } else if (KQUEUE_EVENT_LOOP_GROUP_CLASS.equals(eventLoopGroup.getClass().getName())) {
        serverSocketChannelClass = (Class<? extends ServerChannel>) loadSocketChannelClass(KQUEUE_SERVER_SOCKET_CHANNEL_CLASS);
    } else {
        throw new IllegalArgumentException("Could not find server socket class for event loop group class: " + eventLoopGroup.getClass().getName());
    }

    return serverSocketChannelClass;
}

/**
 * Returns a socket channel class suitable for specified event loop group.
 *
 * @param eventLoopGroup the event loop group for which to identify an appropriate socket channel class; must not
 * be {@code null}
 *
 * @return a socket channel class suitable for use with the given event loop group
 *
 * @throws IllegalArgumentException in case of null or unrecognized event loop group
 */
static Class<? extends Channel> getSocketChannelClass(final EventLoopGroup eventLoopGroup) {
    Objects.requireNonNull(eventLoopGroup);

    final Class<? extends Channel> socketChannelClass;

    if (eventLoopGroup instanceof NioEventLoopGroup) {
        socketChannelClass = NioSocketChannel.class;
    } else if (eventLoopGroup instanceof OioEventLoopGroup) {
        socketChannelClass = OioSocketChannel.class;
    } else if (EPOLL_EVENT_LOOP_GROUP_CLASS.equals(eventLoopGroup.getClass().getName())) {
        socketChannelClass = loadSocketChannelClass(EPOLL_SOCKET_CHANNEL_CLASS);
    } else if (KQUEUE_EVENT_LOOP_GROUP_CLASS.equals(eventLoopGroup.getClass().getName())) {
        socketChannelClass = loadSocketChannelClass(KQUEUE_SOCKET_CHANNEL_CLASS);
    } else {
        throw new IllegalArgumentException("Could not find socket class for event loop group class: " + eventLoopGroup.getClass().getName());
    }

    return socketChannelClass;
}

@Nonnull
public static BuiltInServer startNioOrOio(int workerCount,
                                          int firstPort,
                                          int portsCount,
                                          boolean tryAnyPort,
                                          @Nullable NotNullProducer<ChannelHandler> handler) throws Exception {
  BuiltInServerThreadFactory threadFactory = new BuiltInServerThreadFactory();
  NioEventLoopGroup nioEventLoopGroup;
  try {
    nioEventLoopGroup = new NioEventLoopGroup(workerCount, threadFactory);
  }
  catch (IllegalStateException e) {
    Logger.getInstance(BuiltInServer.class).warn(e);
    return start(new OioEventLoopGroup(1, threadFactory), true, 6942, 50, false, handler);
  }
  return start(nioEventLoopGroup, true, firstPort, portsCount, tryAnyPort, handler);
}

public void server(int port) throws Exception {
    final ByteBuf buf = Unpooled.unreleasableBuffer(
            Unpooled.copiedBuffer("Hi!\r\n", Charset.forName("UTF-8")));
    EventLoopGroup group = new OioEventLoopGroup();
    try {
        ServerBootstrap b = new ServerBootstrap();

        b.group(group)
         .channel(OioServerSocketChannel.class)
         .localAddress(new InetSocketAddress(port))
         .childHandler(new ChannelInitializer<SocketChannel>() {
             @Override
             public void initChannel(SocketChannel ch) 
                 throws Exception {
                 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                     @Override
                     public void channelActive(ChannelHandlerContext ctx) throws Exception {
                         ctx.writeAndFlush(buf.duplicate()).addListener(ChannelFutureListener.CLOSE);
                     }
                 });
             }
         });
        ChannelFuture f = b.bind().sync();
        f.channel().closeFuture().sync();
    } finally {
        group.shutdownGracefully().sync();
    }
}

/**
 * A communication service that manages the serial connection.
 * It can receive and send serial messages via RXTX.
 *
 * @param bootstrap The bootstrap, not null.
 * @param channelInitializer The channel initializer, not null.
 */
public RxtxCommunicationService(Bootstrap bootstrap, RxtxChannelInitializer channelInitializer) {
    requireNonNull(bootstrap);
    requireNonNull(channelInitializer);

    this.bootstrap = bootstrap;
    this.bootstrap.group(new OioEventLoopGroup());
    this.bootstrap.channel(RxtxChannel.class);

    channelInitializer.setRxTxSerialHandler(serialHandler);

    this.bootstrap.handler(channelInitializer);
}

public void run() {
    // Configure the server.
    worker = new OioEventLoopGroup();
    Bootstrap b = new Bootstrap();
    b.group(worker)
            .channel(OioSocketChannel.class)
            .option(ChannelOption.SO_KEEPALIVE, true)
            .option(ChannelOption.SO_REUSEADDR, true)
            .handler(new SocketClientInitializer());
    // Start the client.
    channel = b.connect(host, port).channel();
}

public void serve(int port) throws IOException, InterruptedException {
    final ByteBuf buf = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("Hi!\r\n", Charset.forName("UTF-8")));
    //1. Create ServerBootstrap to allow bootstrap to server instance
    ServerBootstrap bootstrap = new ServerBootstrap();
    //2. Use OioEventLoopGroup Ito allow blocking mode (Old-IO)
    EventLoopGroup group = new OioEventLoopGroup();
    try {
        bootstrap.group(group)
                .channel(OioServerSocketChannel.class)
                .localAddress(new InetSocketAddress(port))
                .childHandler(new ChannelInitializer<SocketChannel>() { //3. Specify ChannelInitializer that will be called for each accepted connection
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        //4. Add ChannelHandler to intercept events and allow to react on them
                        ch.pipeline().addLast(new ChannelHandlerAdapter() {
                            @Override
                            public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                //5. Write message to client and add ChannelFutureListener to close connection once message written
                                ctx.write(buf.duplicate()).addListener(ChannelFutureListener.CLOSE);
                            }
                        });
                    }
                });
        //6. Bind server to accept connections
        ChannelFuture future = bootstrap.bind().sync();
        future.channel().closeFuture().sync();
    } finally {
        //7. Release all resources
        group.shutdownGracefully().sync();
    }

}

public void run() throws Exception {
    // Configure the client.
    EventLoopGroup group = new OioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group)
         .channel(OioSctpChannel.class)
         .option(SctpChannelOption.SCTP_NODELAY, true)
         .handler(new ChannelInitializer<SctpChannel>() {
             @Override
             public void initChannel(SctpChannel ch) throws Exception {
                 ch.pipeline().addLast(
                         new LoggingHandler(LogLevel.INFO),
                         new SctpEchoClientHandler(firstMessageSize));
             }
         });

        // Start the client.
        ChannelFuture f = b.connect(host, port).sync();

        // Wait until the connection is closed.
        f.channel().closeFuture().sync();
    } finally {
        // Shut down the event loop to terminate all threads.
        group.shutdownGracefully();
    }
}

public void run() throws Exception {
    // Configure the server.
    EventLoopGroup bossGroup = new OioEventLoopGroup();
    EventLoopGroup workerGroup = new OioEventLoopGroup();
    try {
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
         .channel(OioSctpServerChannel.class)
         .option(ChannelOption.SO_BACKLOG, 100)
         .handler(new LoggingHandler(LogLevel.INFO))
         .childHandler(new ChannelInitializer<SctpChannel>() {
             @Override
             public void initChannel(SctpChannel ch) throws Exception {
                 ch.pipeline().addLast(
                         new LoggingHandler(LogLevel.INFO),
                         new SctpEchoServerHandler());
             }
         });

        // Start the server.
        ChannelFuture f = b.bind(port).sync();

        // Wait until the server socket is closed.
        f.channel().closeFuture().sync();
    } finally {
        // Shut down all event loops to terminate all threads.
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }
}

public void start() {
        apiBootstrap = new ServerBootstrap();
        ThreadFactory threadFactory = new NamedThreadFactory("kha-webapp");
        EventLoopGroup commonGroup = new OioEventLoopGroup(0, threadFactory);
        try {
            // the hub will only have a few connections, so OIO is likely to be faster than NIO in this case!
            apiBootstrap.group(commonGroup, commonGroup)
                    .channel(OioServerSocketChannel.class)
                    .localAddress(port)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();

                            pipeline.addLast("http-request-decoder", new HttpRequestDecoder());
                            pipeline.addLast("http-object-aggregator", new HttpObjectAggregator(1048576));
                            pipeline.addLast("http-response-encoder", new HttpResponseEncoder());
//                            pipeline.addLast("deflater", new HttpContentDecompressor());
//                            pipeline.addLast("inflater", new HttpContentCompressor());
                            pipeline.addLast("chunkedWriter", new ChunkedWriteHandler());
                            pipeline.addLast("cors", new CorsHandler(corsConfig));
                            pipeline.addLast("file-handler", new HttpStaticFileServerHandler(hubSiteDirectory, true));
                        }
                    });

            ChannelFuture f = apiBootstrap.bind().sync();
            LOGGER.info("WebApp available on http://{}:{}", InetAddress.getLocalHost().getCanonicalHostName(), port);
            f.channel().closeFuture().sync();
        } catch (Exception e) {
            LOGGER.error("Can't start WebApp server", e);
        }
    }

public void start() {
        apiBootstrap = new ServerBootstrap();
        parentGroup = new OioEventLoopGroup();
        childGroup = new OioEventLoopGroup();
        try {
            // the cloudPlatform will only have a few connections, so OIO is likely to be faster than NIO in this case!
            apiBootstrap.group(parentGroup, childGroup)
                    .channel(OioServerSocketChannel.class)
                    .localAddress(port)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();

                            pipeline.addLast("http-request-decoder", new HttpRequestDecoder());
//                            pipeline.addLast("deflater", new HttpContentDecompressor());
                            pipeline.addLast("http-object-aggregator", new HttpObjectAggregator(1048576));
                            pipeline.addLast("http-response-encoder", new HttpResponseEncoder());
//                            pipeline.addLast("inflater", new HttpContentCompressor());
                            pipeline.addLast("chunkedWriter", new ChunkedWriteHandler());
                            pipeline.addLast("file-handler", new HttpStaticFileServerHandler(System.getProperty("user.dir"), true));
                        }
                    });

            ChannelFuture f = apiBootstrap.bind().sync();
            LOGGER.info("WebApp available on port {}.", port);
            f.channel().closeFuture().sync();
        } catch (Exception e) {
            LOGGER.error("Can't start WebApp server", e);
        }
    }

public void run() throws Exception {
    // Configure the client.
    EventLoopGroup group = new OioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group)
         .channel(OioSctpChannel.class)
         .option(SctpChannelOption.SCTP_NODELAY, true)
         .handler(new ChannelInitializer<SctpChannel>() {
             @Override
             public void initChannel(SctpChannel ch) throws Exception {
                 ch.pipeline().addLast(
                         new LoggingHandler(LogLevel.INFO),
                         new SctpEchoClientHandler(firstMessageSize));
             }
         });

        // Start the client.
        ChannelFuture f = b.connect(host, port).sync();

        // Wait until the connection is closed.
        f.channel().closeFuture().sync();
    } finally {
        // Shut down the event loop to terminate all threads.
        group.shutdownGracefully();
    }
}

public void run() throws Exception {
    // Configure the server.
    EventLoopGroup bossGroup = new OioEventLoopGroup();
    EventLoopGroup workerGroup = new OioEventLoopGroup();
    try {
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
         .channel(OioSctpServerChannel.class)
         .option(ChannelOption.SO_BACKLOG, 100)
         .handler(new LoggingHandler(LogLevel.INFO))
         .childHandler(new ChannelInitializer<SctpChannel>() {
             @Override
             public void initChannel(SctpChannel ch) throws Exception {
                 ch.pipeline().addLast(
                         new LoggingHandler(LogLevel.INFO),
                         new SctpEchoServerHandler());
             }
         });

        // Start the server.
        ChannelFuture f = b.bind(port).sync();

        // Wait until the server socket is closed.
        f.channel().closeFuture().sync();
    } finally {
        // Shut down all event loops to terminate all threads.
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }
}

/**
 * Initializes a new TCPConnectionManager.
 * 
 * @param serverConfig
 *            a configuration to use for initializing
 * @return the new ConnectionManager
 */
private ConnectionManager initializeConnectionManager(final AddressBasedServerConfig serverConfig) {
    LOGGER.entry();
    final EventLoopGroup applicationEventLoopGroup = new OioEventLoopGroup();
    final EventLoopGroup networkEventLoopGroup = new OioEventLoopGroup();
    eventExecutorGroups.add(applicationEventLoopGroup);
    eventExecutorGroups.add(networkEventLoopGroup);

    final ServerBootstrap serverBootstrap = new ServerBootstrap();
    serverBootstrap.group(networkEventLoopGroup, applicationEventLoopGroup);
    serverBootstrap.channel(OioServerSocketChannel.class);
    final ServerBootstrapChannelFactory serverChannelFactory = new ServerBootstrapChannelFactory(serverBootstrap);

    final Bootstrap clientBootstrap = new Bootstrap();
    clientBootstrap.group(applicationEventLoopGroup);
    clientBootstrap.channel(OioSocketChannel.class);
    final ClientBootstrapChannelFactory clientChannelFactory = new ClientBootstrapChannelFactory(clientBootstrap);

    final TCPConnectionManager connectionManager =
            new TCPConnectionManager(serverChannelFactory, clientChannelFactory, scheduledExecutorService,
                    serverConfig.getModuleID());

    new TCPProtocol().initialize(connectionManager);

    for (final InetSocketAddress address : serverConfig.getListenAddresses()) {
        connectionManager.startListening(address);
    }

    return LOGGER.exit(connectionManager);
}

private Future<?> startServerInPooledThread() {
  if (!started.compareAndSet(false, true)) {
    return null;
  }

  return ApplicationManager.getApplication().executeOnPooledThread(() -> {
    try {
      BuiltInServer mainServer = StartupUtil.getServer();
      if (mainServer == null || mainServer.getEventLoopGroup() instanceof OioEventLoopGroup) {
        server = BuiltInServer.start(1, getDefaultPort(), PORTS_COUNT, false, null);
      }
      else {
        server = BuiltInServer.start(mainServer.getEventLoopGroup(), false, getDefaultPort(), PORTS_COUNT, true, null);
      }
      bindCustomPorts(server);
    }
    catch (Throwable e) {
      LOG.info(e);
      NOTIFICATION_GROUP.getValue().createNotification("Cannot start internal HTTP server. Git integration, Some plugins may operate with errors. " +
                                                       "Please check your firewall settings and restart " + ApplicationNamesInfo.getInstance().getFullProductName(),
                                                       NotificationType.ERROR).notify(null);
      return;
    }

    LOG.info("built-in server started, port " + server.getPort());

    Disposer.register(ApplicationManager.getApplication(), server);
  });
}

public RpcClient(OioEventLoopGroup eventLoopGroup, EventExecutorGroup eventExecutor) {
    this(eventLoopGroup, eventExecutor, OioSocketChannel.class);
}

public RpcServer(OioEventLoopGroup eventLoopGroup, EventExecutorGroup eventExecutor, @Assisted SocketAddress address) {
    this(eventLoopGroup, eventExecutor, OioServerSocketChannel.class, address);
}

@Test
public void worksWithOioEventLoopGroup() {
    assertThat(resolveSocketChannelClass(new OioEventLoopGroup())).isEqualTo(OioSocketChannel.class);
}

public DefaultServer(Class serverChannelClass,
        ChannelPipelineFactoryFactory factory, Set<String> channelOptions,
        int port, InetAddress address)
{
    if (!ServerChannel.class.isAssignableFrom(serverChannelClass))
        throw new RuntimeException(
                "serverChannelClass must implement ServerChannel");

    // Configure the server.
    bootstrap = new ServerBootstrap();
    _port = port;
    _address = address;
    internalGroup = new DefaultEventExecutorGroup(10);

    if (isNio(serverChannelClass))
    {
        bossGroup = new NioEventLoopGroup();
        childGroup = new NioEventLoopGroup();
    }
    else if (isOio(serverChannelClass))
    {
        bossGroup = new OioEventLoopGroup();
        childGroup = new OioEventLoopGroup();
    }
    else
    {
        bossGroup = new NioEventLoopGroup();
        childGroup = new NioEventLoopGroup();
    }
    bootstrap.group(bossGroup, childGroup);
    bootstrap.channel(serverChannelClass);
    // bootstrap.setOption("child.trafficClass", IPTOS_LOWDELAY);
    // bootstrap.setOption("child.tcpNoDelay", false);
    // bootstrap.childOption(ChannelOption.IP_TOS, IPTOS_THROUGHPUT);
    setChannelOptions(channelOptions);
    bootstrap.option(ChannelOption.SO_BACKLOG, 100);
    ChannelPipelineFactory channelPipelineFactory = factory.create(
            internalGroup, bootstrap);
    bootstrap.childHandler(channelPipelineFactory);

}

public MockBookKeeper(ClientConfiguration conf, ZooKeeper zk) throws Exception {
    super(conf, zk, new OioEventLoopGroup());
}

@NotNull
public static Bootstrap oioClientBootstrap() {
  Bootstrap bootstrap = new Bootstrap().group(new OioEventLoopGroup(1, PooledThreadExecutor.INSTANCE)).channel(OioSocketChannel.class);
  bootstrap.option(ChannelOption.TCP_NODELAY, true).option(ChannelOption.SO_KEEPALIVE, true);
  return bootstrap;
}

/**
 * {@inheritDoc}
 */
@Override
public synchronized boolean init() {
    if(this.running.get()) {
        return true;
    }

    // create bootstrap
    Class<? extends ServerChannel> channelType;
       if(useNio) {
        this.workerGroup = new NioEventLoopGroup();
        this.bossGroup = new NioEventLoopGroup();
        channelType = NioServerSocketChannel.class;
       } else {
        this.workerGroup = new OioEventLoopGroup();
        this.bossGroup = new OioEventLoopGroup();
        channelType = OioServerSocketChannel.class;
       }

    bootstrap = new ServerBootstrap();
       bootstrap.group(this.bossGroup, this.workerGroup)
            .option(ChannelOption.SO_SNDBUF, this.sendBufferSize)
            .option(ChannelOption.SO_RCVBUF, this.receiveBufferSize)
            .channel(channelType)
            .handler(new LoggingHandler(LogLevel.INFO))
            .childHandler(new ChannelInitializer<Channel>() { // is used to initialize the ChannelPipeline
                @Override
                protected void initChannel(Channel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();
                    pipeline.addLast("encoder", new RtspEncoder());
                    pipeline.addLast("decoder", new RtspDecoder());
                    pipeline.addLast("aggregator", new HttpObjectAggregator(64*1024));
                    pipeline.addLast("handler", new RtspHandler(SimpleRtspSession.this));
                }
            });
       // create channel
       try {
        ChannelFuture future = bootstrap.bind(this.localAddress);
           this.channel = future.sync().channel(); // wait for future to complete and retrieve channel

       } catch (Exception e) {
           LOG.error("Failed to bind RTSP channel for session with id " + this.id, e);
           this.workerGroup.shutdownGracefully();
           this.bossGroup.shutdownGracefully();

           this.workerGroup.terminationFuture().syncUninterruptibly();
           this.bossGroup.terminationFuture().syncUninterruptibly();
           return false;
       }
       LOG.debug("RTSP channel bound for RtspSession with id {}.", this.id);
       this.running.set(true);
       return true;
}

@Override
protected void configureBootstrap() {
    serverBootstrap.group(new OioEventLoopGroup(workerCount))
            .channel(OioServerSocketChannel.class);
}

@Override
protected void configureBootstrap() {
    _nettyBootstrap.group(new OioEventLoopGroup(workerCount))
            .channel(OioServerSocketChannel.class);
}