Java 类io.netty.channel.epoll.EpollEventLoopGroup 实例源码

public void start() throws InterruptedException {
    final EventLoopGroup workerGroup = Epoll.isAvailable() ? new EpollEventLoopGroup() : new NioEventLoopGroup();

    try {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(workerGroup)
                .channel(Epoll.isAvailable() ? EpollSocketChannel.class : NioSocketChannel.class)
                .handler(new OpenCloudChannelInitializer(this))
                .connect(this.host, this.port).sync().channel().closeFuture().syncUninterruptibly();
    } catch (Exception ex) {
        if (ex.getClass().getSimpleName().equals("AnnotatedConnectException")) {
            System.err.println("Cannot connect to master!");
            channel.close();
        } else {
            ex.printStackTrace();
        }
    } finally {
        workerGroup.shutdownGracefully();
        System.out.println("Netty client stopped");
        Runtime.getRuntime().halt(0);
    }
}

public PluginGrpcServer(int port) {
    this.pluginConnections = new HashMap<>();

    PlayerEvents playerEvents = new PlayerEvents(this);

    this.server = NettyServerBuilder.forPort(port)
            .keepAliveTime(1, TimeUnit.MINUTES)
            .keepAliveTimeout(5, TimeUnit.SECONDS)

            .addService(playerEvents)

            .directExecutor()
            .channelType(EpollServerSocketChannel.class)
            .bossEventLoopGroup(new EpollEventLoopGroup())
            .workerEventLoopGroup(new EpollEventLoopGroup())
            .build();

    // demoPluginConnections();
}

public void connectPlugin(String host, int port) {
    ManagedChannel channel = NettyChannelBuilder.forAddress(host, port)
            .negotiationType(NegotiationType.PLAINTEXT) // TODO: gRPC encryption
            .keepAliveTime(1, TimeUnit.MINUTES)
            .keepAliveTimeout(5, TimeUnit.SECONDS)
            .directExecutor()
            .channelType(EpollSocketChannel.class)
            .eventLoopGroup(new EpollEventLoopGroup())
            .build();

    PluginManagerGrpc.PluginManagerBlockingStub blocking = PluginManagerGrpc.newBlockingStub(channel);
    PluginManagerGrpc.PluginManagerStub async = PluginManagerGrpc.newStub(channel);

    ServiceConnection connection = ServiceConnection.builder()
            .channel(channel)
            .blockingStub(blocking)
            .asyncStub(async)
            .build();

    this.pluginConnections.put(PLUGIN_MANAGER, connection);
}

/**
 * Attempts to determine the {@link Channel} class that corresponds to the given
 * event loop group.
 *
 * @param eventLoopGroup the event loop group to determine the {@link Channel} for
 * @return A {@link Channel} class for the given event loop group.
 */
public static Class<? extends Channel> resolveSocketChannelClass(EventLoopGroup eventLoopGroup) {
    if (eventLoopGroup instanceof DelegatingEventLoopGroup) {
        return resolveSocketChannelClass(((DelegatingEventLoopGroup) eventLoopGroup).getDelegate());
    }
    if (eventLoopGroup instanceof NioEventLoopGroup) {
        return NioSocketChannel.class;
    }
    if (eventLoopGroup instanceof EpollEventLoopGroup) {
        return EpollSocketChannel.class;
    }
    String socketFqcn = KNOWN_EL_GROUPS.get(eventLoopGroup.getClass().getName());
    if (socketFqcn == null) {
        throw new IllegalArgumentException("Unknown event loop group : " + eventLoopGroup.getClass());
    }
    return invokeSafely(() -> (Class<? extends Channel>) Class.forName(socketFqcn));
}

public void start() throws Exception {
    UnknownPandaServer.getLogger().info("Loading protocol");
    Protocol protocol = ProtocolSpecification.getProtocol();
    protocol.load();

    UnknownPandaServer.getLogger().info("Binding UniverseServer at *::" + port + " [tcp]");
    this.channel = new ServerBootstrap()
            .group(Epoll.isAvailable() ? new EpollEventLoopGroup() : new NioEventLoopGroup())
            .channel(Epoll.isAvailable() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
            //.childOption(ChannelOption.TCP_NODELAY, true)
            .childHandler(new ConnectionInitializer(this))
            .localAddress("", port)
            .bind()
            .addListeners(this)
            .sync()
            .channel();
}

private static Pair<EventLoopGroup, Class<? extends Channel>> createEventLoopGroup(
    Configuration conf) {
  // Max amount of threads to use. 0 lets Netty decide based on amount of cores
  int maxThreads = conf.getInt(CLIENT_MAX_THREADS, 0);

  // Config to enable native transport. Does not seem to be stable at time of implementation
  // although it is not extensively tested.
  boolean epollEnabled = conf.getBoolean(USE_NATIVE_TRANSPORT, false);

  // Use the faster native epoll transport mechanism on linux if enabled
  if (epollEnabled && JVM.isLinux() && JVM.isAmd64()) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Create EpollEventLoopGroup with maxThreads = " + maxThreads);
    }
    return new Pair<EventLoopGroup, Class<? extends Channel>>(new EpollEventLoopGroup(maxThreads,
        Threads.newDaemonThreadFactory("AsyncRpcChannel")), EpollSocketChannel.class);
  } else {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Create NioEventLoopGroup with maxThreads = " + maxThreads);
    }
    return new Pair<EventLoopGroup, Class<? extends Channel>>(new NioEventLoopGroup(maxThreads,
        Threads.newDaemonThreadFactory("AsyncRpcChannel")), NioSocketChannel.class);
  }
}

private void groups(ServerBootstrap b) {
    if (StandardSystemProperty.OS_NAME.value().equals("Linux")) {
        bossGroup = new EpollEventLoopGroup(1);
        workerGroup = new EpollEventLoopGroup();
        b.channel(EpollServerSocketChannel.class)
                .group(bossGroup, workerGroup)
                .option(EpollChannelOption.TCP_CORK, true);
    } else {
        bossGroup = new NioEventLoopGroup(1);
        workerGroup = new NioEventLoopGroup();
        b.channel(NioServerSocketChannel.class)
                .group(bossGroup, workerGroup);
    }
    b.option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.SO_REUSEADDR, true)
            .option(ChannelOption.SO_BACKLOG, 100);
    logger.info("Bootstrap configuration: " + b.toString());
}

public static EventLoopGroup newEventLoop(int numThreads) {
    ThreadFactory threadFactory = new ThreadFactory() {
        private final AtomicInteger counter = new AtomicInteger(1);

        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, String.format("Netty IO Thread #%1$d", counter.getAndIncrement()));
        }
    };
    ExecutorService executor = Executors.newFixedThreadPool(numThreads, threadFactory);
    if (useEpoll) {
        return new EpollEventLoopGroup(0, executor);
    } else {
        return new NioEventLoopGroup(0, executor);
    }
}

public void run() throws Exception {
    ServerBootstrap b = new ServerBootstrap();
    try {
        if (isEpollAvailable) {
            b.group(new EpollEventLoopGroup(this.conf.getEventLoopThreadCount()))
             .channel(EpollServerSocketChannel.class);
        } else {
            b.group(new NioEventLoopGroup(this.conf.getEventLoopThreadCount()))
             .channel(NioServerSocketChannel.class);
        }
        b.childHandler(new DefaultServerInitializer(conf, context))
         .option(ChannelOption.SO_BACKLOG, conf.getBacklog())
         .option(ChannelOption.SO_REUSEADDR, true);

        Channel ch = b.bind(conf.getPort()).sync().channel();
        ch.closeFuture().sync();
    } finally {

    }
}

public void startServers() throws Exception {
    int bossThreadCount = Settings.INSTANCE.getInt("netty.bossThreadCount", 0);
    int workerThreadCount = Settings.INSTANCE.getInt("netty.workerThreadCount", 0);

    ThreadFactory bossThreadFactory = new DefaultThreadFactory("lannister/boss");
    ThreadFactory workerThreadFactory = new DefaultThreadFactory("lannister/worker");

    if (Literals.NETTY_EPOLL.equals(Settings.INSTANCE.nettyTransportMode())) {
        bossGroup = new EpollEventLoopGroup(bossThreadCount, bossThreadFactory);
        workerGroup = new EpollEventLoopGroup(workerThreadCount, workerThreadFactory);
    }
    else {
        bossGroup = new NioEventLoopGroup(bossThreadCount, bossThreadFactory);
        workerGroup = new NioEventLoopGroup(workerThreadCount, workerThreadFactory);
    }

    mqttServer = new MqttServer(bossGroup, workerGroup);
    mqttServer.start();

    webServer = new WebServer(bossGroup, workerGroup);
    webServer.start("net.anyflow");
}

/**
 * Construct a Gremlin Server instance from the {@link ServerGremlinExecutor} which internally carries some
 * pre-constructed objects used by the server as well as the {@link Settings} object itself.  This constructor
 * is useful when Gremlin Server is being used in an embedded style and there is a need to share thread pools
 * with the hosting application.
 *
 * @deprecated As of release 3.1.1-incubating, not replaced.
 * @see <a href="https://issues.apache.org/jira/browse/TINKERPOP-912">TINKERPOP-912</a>
 */
@Deprecated
public GremlinServer(final ServerGremlinExecutor<EventLoopGroup> serverGremlinExecutor) {
    this.serverGremlinExecutor = serverGremlinExecutor;
    this.settings = serverGremlinExecutor.getSettings();
    this.isEpollEnabled = settings.useEpollEventLoop && SystemUtils.IS_OS_LINUX;
    if(settings.useEpollEventLoop && !SystemUtils.IS_OS_LINUX){
        logger.warn("cannot use epoll in non-linux env, falling back to NIO");
    }

    Runtime.getRuntime().addShutdownHook(new Thread(() -> this.stop().join(), SERVER_THREAD_PREFIX + "shutdown"));

    final ThreadFactory threadFactoryBoss = ThreadFactoryUtil.create("boss-%d");
    if(isEpollEnabled) {
        bossGroup = new EpollEventLoopGroup(settings.threadPoolBoss, threadFactoryBoss);
    } else{
        bossGroup = new NioEventLoopGroup(settings.threadPoolBoss, threadFactoryBoss);
    }
    workerGroup = serverGremlinExecutor.getScheduledExecutorService();
    gremlinExecutorService = serverGremlinExecutor.getGremlinExecutorService();
}

public void connect(String apiKey) {
  Bootstrap bootstrap = new Bootstrap();
  Class<? extends Channel> channelClazz;

  if (Epoll.isAvailable()) {
    channelClazz = EpollSocketChannel.class;
    eventLoopGroup = new EpollEventLoopGroup();
  } else {
    channelClazz = NioSocketChannel.class;
    eventLoopGroup = new NioEventLoopGroup();
  }

  bootstrap.group(eventLoopGroup)
      .channel(channelClazz)
      .option(ChannelOption.SO_KEEPALIVE, true)
      // TODO: add function to get data class by topic and add handler
      .remoteAddress(host, port)
      .connect();
}

@Override
public void initialise(NetworkChannelHandler channelHandler) {
    this.channelHandler = channelHandler;

    final boolean useEpoll = this.configuration.getBoolean("epoll") && Epoll.isAvailable();

    EventLoopGroup acceptGroup = useEpoll ? new EpollEventLoopGroup(this.configuration.getInt("acceptGroup")) :
            new NioEventLoopGroup(this.configuration.getInt("acceptGroup"));

    EventLoopGroup ioGroup = useEpoll ? new EpollEventLoopGroup(this.configuration.getInt("ioGroup")) :
            new NioEventLoopGroup(this.configuration.getInt("ioGroup"));

    EventLoopGroup channelGroup = useEpoll ? new EpollEventLoopGroup(this.configuration.getInt("channelGroup")) :
            new NioEventLoopGroup(this.configuration.getInt("channelGroup"));

    this.serverBootstrap = new ServerBootstrap()
            .group(acceptGroup, ioGroup)
            .channel(useEpoll ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
            .childHandler(new ChannelInitialiser(channelGroup, this.channelHandler, null))
            .option(ChannelOption.SO_BACKLOG, this.configuration.getInt("backlog"))
            .option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .option(ChannelOption.MESSAGE_SIZE_ESTIMATOR, DefaultMessageSizeEstimator.DEFAULT)
            .childOption(ChannelOption.TCP_NODELAY, this.configuration.getBoolean("tcpNoDelay"))
            .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
}

public McpeOverRakNetNetworkListener(VoxelwindServer voxelwindServer, String host, int port, boolean useSoReuseport) {
    this.server = voxelwindServer;
    this.address = new InetSocketAddress(host, port);
    this.useSoReuseport = useSoReuseport;
    if (Epoll.isAvailable()) {
        bootstrap = new Bootstrap()
                .channel(EpollDatagramChannel.class)
                .group(new EpollEventLoopGroup(0, new ThreadFactoryBuilder().setDaemon(true).setNameFormat("Voxelwind MCPE Listener - #%d").build()))
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .handler(this);
        if (useSoReuseport) {
            bootstrap.option(EpollChannelOption.SO_REUSEPORT, true);
        }
    } else {
        bootstrap = new Bootstrap()
                .channel(NioDatagramChannel.class)
                .group(new NioEventLoopGroup(0, new ThreadFactoryBuilder().setDaemon(true).setNameFormat("Voxelwind MCPE Listener - #%d").build()))
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .handler(this);
    }
}

private void initEventLoopGroup() {
    // try Epoll first and if that does work, use nio.
    try {
        clientGroup = new EpollEventLoopGroup();
        serverGroup = new EpollEventLoopGroup();
        serverChannelClass = EpollServerSocketChannel.class;
        clientChannelClass = EpollSocketChannel.class;
        return;
    } catch (Throwable t) {
        log.warn("Failed to initialize native (epoll) transport. Reason: {}. Proceeding with nio.", t.getMessage());
    }
    clientGroup = new NioEventLoopGroup();
    serverGroup = new NioEventLoopGroup();
    serverChannelClass = NioServerSocketChannel.class;
    clientChannelClass = NioSocketChannel.class;
}

public static void main(String[] args) throws Exception {
    EventLoopGroup bossGroup = new EpollEventLoopGroup(1);
    EventLoopGroup workerGroup = new EpollEventLoopGroup();
    try {
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
         .channel(EpollServerSocketChannel.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 void initChannel() throws Exception {
    bossGroup = new EpollEventLoopGroup();
    workerGroup = new EpollEventLoopGroup(pushListenerWorkerNum,
            new ThreadFactoryWithName(NettyPushListener.class));
    serverBootstarp = new ServerBootstrap()
            .group(bossGroup, workerGroup)
            .channel(EpollServerSocketChannel.class)
            .option(ChannelOption.SO_TIMEOUT, sockTimout)
            .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .option(ChannelOption.TCP_NODELAY, true)
            .childOption(ChannelOption.ALLOCATOR,
                    PooledByteBufAllocator.DEFAULT)
            .childOption(ChannelOption.TCP_NODELAY, true)
            .childHandler(pushListenerChannelInitializer);

    serverBootstarp.bind(port).sync();

    logger.info("Netty TCP Push Listener nio provider: {} with {} workers", serverBootstarp
            .getClass().getCanonicalName(), pushListenerWorkerNum);
}

@Override
    public ServerBootstrap createBootstrap() {
        bootstrap = new ServerBootstrap();
        if (isEpollAvailable) {
            this.parentGroup = new EpollEventLoopGroup();
            this.childGroup = new EpollEventLoopGroup();
            bootstrap.channel(EpollServerSocketChannel.class);
        } else {
            this.parentGroup = new NioEventLoopGroup();
            this.childGroup = new NioEventLoopGroup();
            bootstrap.channel(NioServerSocketChannel.class);
        }
        bootstrap.group(parentGroup(), childGroup());
        bootstrap.childHandler(newChannelInitializer());

        bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
//        bootstrap.option(ChannelOption.SO_REUSEADDR, true);
//        bootstrap.childOption(ChannelOption.TCP_NODELAY, true);
        bootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);

        return bootstrap;
    }

private void initEventLoopGroup() {
  // try Epoll first and if that does work, use nio.
  try {
    clientGroup = new EpollEventLoopGroup(0, namedThreads("netty-messaging-event-epoll-client-%d", log));
    serverGroup = new EpollEventLoopGroup(0, namedThreads("netty-messaging-event-epoll-server-%d", log));
    serverChannelClass = EpollServerSocketChannel.class;
    clientChannelClass = EpollSocketChannel.class;
    return;
  } catch (Throwable e) {
    log.debug("Failed to initialize native (epoll) transport. "
        + "Reason: {}. Proceeding with nio.", e.getMessage());
  }
  clientGroup = new NioEventLoopGroup(0, namedThreads("netty-messaging-event-nio-client-%d", log));
  serverGroup = new NioEventLoopGroup(0, namedThreads("netty-messaging-event-nio-server-%d", log));
  serverChannelClass = NioServerSocketChannel.class;
  clientChannelClass = NioSocketChannel.class;
}

public static ChannelConfiguration clientConfig(EventLoopGroup workerGroup) {
  EventLoopGroup parent = workerGroup;
  if (parent instanceof EventLoop) {
    parent = ((EventLoop) workerGroup).parent();
  }
  Class<? extends Channel> channelClass;
  if (parent instanceof EpollEventLoopGroup) {
    channelClass = EpollSocketChannel.class;
  } else if (parent instanceof NioEventLoopGroup) {
    channelClass = NioSocketChannel.class;
  } else {
    throw new RuntimeException("Unsupported EventLoopGroup " + workerGroup.getClass());
  }

  return new ChannelConfiguration(workerGroup, channelClass);
}

/**
 * This method will configure a worker EventLoopGroup and a Channel for use by a client. It will
 * try to use the correct SocketChannel for the provided workerGroup.
 *
 * @param workerGroup uses EventLoopGroup in the ClientChannelConfiguration
 * @return ClientChannelConfiguration
 */
public static ClientChannelConfiguration clientConfig(EventLoopGroup workerGroup) {
  EventLoopGroup parent = workerGroup;
  if (parent instanceof EventLoop) {
    parent = ((EventLoop) workerGroup).parent();
  }
  Class<? extends Channel> channelClass;
  if (parent instanceof EpollEventLoopGroup) {
    channelClass = EpollSocketChannel.class;
  } else if (parent instanceof NioEventLoopGroup) {
    channelClass = NioSocketChannel.class;
  } else {
    throw new RuntimeException("Unsupported EventLoopGroup " + workerGroup.getClass());
  }

  return new ClientChannelConfiguration(workerGroup, channelClass);
}

private void groups(ServerBootstrap b) {
    if (StandardSystemProperty.OS_NAME.value().equals("Linux")) {
        bossGroup = new EpollEventLoopGroup(1);
        workerGroup = new EpollEventLoopGroup();
        b.channel(EpollServerSocketChannel.class)
                .group(bossGroup, workerGroup)
                .option(EpollChannelOption.TCP_CORK, true);
    } else {
        bossGroup = new NioEventLoopGroup(1);
        workerGroup = new NioEventLoopGroup();
        b.channel(NioServerSocketChannel.class)
                .group(bossGroup, workerGroup);
    }
    b.option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.SO_REUSEADDR, true)
            .option(ChannelOption.SO_BACKLOG, 100);
    logger.info("Bootstrap configuration: " + b.toString());
}

private void assertEpoll(String message, boolean expected, Transport transport) throws Exception {
    Field group = null;
    Class<?> transportType = transport.getClass();

    while (transportType != null && group == null) {
        try {
            group = transportType.getDeclaredField("group");
        } catch (NoSuchFieldException error) {
            transportType = transportType.getSuperclass();
            if (Object.class.equals(transportType)) {
                transportType = null;
            }
        }
    }

    assertNotNull("Transport implementation unknown", group);

    group.setAccessible(true);
    if (expected) {
        assertTrue(message, group.get(transport) instanceof EpollEventLoopGroup);
    } else {
        assertFalse(message, group.get(transport) instanceof EpollEventLoopGroup);
    }
}

public EventLoopGroup epollGroup() {
    EventLoopGroup epollEventLoopGroup = new EpollEventLoopGroup(0, new DefaultThreadFactory(threadPrefix));

    ChannelFactory<EpollDomainSocketChannel> factory = new ChannelFactory<EpollDomainSocketChannel>() {
        @Override
        public EpollDomainSocketChannel newChannel() {
            return configure(new EpollDomainSocketChannel());
        }
    };

    bootstrap.group(epollEventLoopGroup).channelFactory(factory).handler(new ChannelInitializer<UnixChannel>() {
        @Override
        protected void initChannel(final UnixChannel channel) throws Exception {
            channel.pipeline().addLast(new HttpClientCodec());
        }
    });
    return epollEventLoopGroup;
}

public EventLoopGroup epollGroup() {
    EventLoopGroup epollEventLoopGroup = new EpollEventLoopGroup(0, new DefaultThreadFactory(threadPrefix));

    ChannelFactory<EpollDomainSocketChannel> factory = new ChannelFactory<EpollDomainSocketChannel>() {
        @Override
        public EpollDomainSocketChannel newChannel() {
            return configure(new EpollDomainSocketChannel());
        }
    };

    bootstrap.group(epollEventLoopGroup).channelFactory(factory).handler(new ChannelInitializer<UnixChannel>() {
        @Override
        protected void initChannel(final UnixChannel channel) throws Exception {
            channel.pipeline().addLast(new HttpClientCodec());
        }
    });
    return epollEventLoopGroup;
}

private void initEventLoopGroup() {
    // try Epoll first and if that does work, use nio.
    try {
        clientGroup = new EpollEventLoopGroup(0, groupedThreads("NettyMessagingEvt", "epollC-%d", log));
        serverGroup = new EpollEventLoopGroup(0, groupedThreads("NettyMessagingEvt", "epollS-%d", log));
        serverChannelClass = EpollServerSocketChannel.class;
        clientChannelClass = EpollSocketChannel.class;
        return;
    } catch (Throwable e) {
        log.debug("Failed to initialize native (epoll) transport. "
                + "Reason: {}. Proceeding with nio.", e.getMessage());
    }
    clientGroup = new NioEventLoopGroup(0, groupedThreads("NettyMessagingEvt", "nioC-%d", log));
    serverGroup = new NioEventLoopGroup(0, groupedThreads("NettyMessagingEvt", "nioS-%d", log));
    serverChannelClass = NioServerSocketChannel.class;
    clientChannelClass = NioSocketChannel.class;
}

/** Creates a Netty EventLoopGroup based on the IOMode. */
public static EventLoopGroup createEventLoop(IOMode mode, int numThreads, String threadPrefix) {
  ThreadFactory threadFactory = createThreadFactory(threadPrefix);

  switch (mode) {
    case NIO:
      return new NioEventLoopGroup(numThreads, threadFactory);
    case EPOLL:
      return new EpollEventLoopGroup(numThreads, threadFactory);
    default:
      throw new IllegalArgumentException("Unknown io mode: " + mode);
  }
}

@SuppressWarnings("unchecked")
@Override
public void addConnection(ServiceType serviceType, String host, int port) {
    ManagedChannel channel = NettyChannelBuilder.forAddress(host, port)
            .negotiationType(NegotiationType.PLAINTEXT) // TODO: gRPC encryption
            .keepAliveTime(1, TimeUnit.MINUTES)
            .keepAliveTimeout(5, TimeUnit.SECONDS)
            .directExecutor()
            .channelType(EpollSocketChannel.class)
            .eventLoopGroup(new EpollEventLoopGroup())
            .build();

    AbstractStub blocking;
    AbstractStub async;

    switch (serviceType) {
        case WORLD: {
            blocking = WorldServiceGrpc.newBlockingStub(channel);
            async = WorldServiceGrpc.newStub(channel);
            break;
        }
        case PLUGIN_MANAGER: {
            blocking = PluginManagerGrpc.newBlockingStub(channel);
            async = PluginManagerGrpc.newStub(channel);
            break;
        }
        default: {
            throw new RuntimeException("Service type not handled: " + serviceType.name());
        }
    }

    ServiceConnection connection = ServiceConnection.builder()
            .channel(channel)
            .blockingStub(blocking)
            .asyncStub(async)
            .build();

    this.connections.put(serviceType, connection);
}

private EventLoopGroup newEventLoop(int size, String threadNamePrefix) {
    switch (transport) {
        case EPOLL: {
            return new EpollEventLoopGroup(size, new HekateThreadFactory(threadNamePrefix));
        }
        case NIO: {
            return new NioEventLoopGroup(size, new HekateThreadFactory(threadNamePrefix));
        }
        case AUTO: // <-- Fail since AUTO must be resolved in the constructor.
        default: {
            throw new IllegalArgumentException("Unexpected transport type: " + transport);
        }
    }
}

protected EventLoopGroup newEventLoop(int thread) {
    if (context().transport() == NetworkTransportType.EPOLL) {
        return new EpollEventLoopGroup(thread);
    } else {
        return new NioEventLoopGroup(thread);
    }
}

@Before
public void setUp()
throws Exception {

    serverMock = new EpollConnDroppingServer(DEFAULT_PORT, FAIL_EVERY_CONN_ATTEMPT);

    final Semaphore concurrencyThrottle = new Semaphore(CONCURRENCY);
    group = new EpollEventLoopGroup();
    final Bootstrap bootstrap = new Bootstrap()
        .group(group)
        .channel(EpollSocketChannel.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);
}

public EpollConnDroppingServer(final int port, final int dropEveryRequest)
throws InterruptedException {
    dispatchGroup = new EpollEventLoopGroup();
    workerGroup = new EpollEventLoopGroup();
    final ServerBootstrap bootstrap = new ServerBootstrap()
        .group(dispatchGroup, workerGroup)
        .channel(EpollServerSocketChannel.class)
        .childHandler(
            new ChannelInitializer<SocketChannel>() {
                @Override
                public final void initChannel(final SocketChannel ch) {
                    if(dropEveryRequest > 0) {
                        ch.pipeline().addLast(
                            new SimpleChannelInboundHandler<Object>() {
                                @Override
                                protected final void channelRead0(
                                    final ChannelHandlerContext ctx, final Object msg
                                ) throws Exception {
                                    if(0 == reqCounter.incrementAndGet() % dropEveryRequest) {
                                        final Channel conn = ctx.channel();
                                        System.out.println("Dropping the connection " + conn);
                                        conn.close();
                                    }
                                }
                            }
                        );
                    }
                }
            }
        );

    bindFuture = bootstrap.bind(port).sync();
}

public static EventLoopGroup createEventLoopGroup(int nThreads, String prefix) {
   if(SUPPORTS_EPOLL){
     return new EpollEventLoopGroup(nThreads, new NamedThreadFactory(prefix));
   }else{
     return new NioEventLoopGroup(nThreads, new NamedThreadFactory(prefix));
   }
}

/** Creates a Netty EventLoopGroup based on the IOMode. */
public static EventLoopGroup createEventLoop(IOMode mode, int numThreads, String threadPrefix) {
  ThreadFactory threadFactory = createThreadFactory(threadPrefix);

  switch (mode) {
    case NIO:
      return new NioEventLoopGroup(numThreads, threadFactory);
    case EPOLL:
      return new EpollEventLoopGroup(numThreads, threadFactory);
    default:
      throw new IllegalArgumentException("Unknown io mode: " + mode);
  }
}

@Bean(name = "bossGroup")
public EventLoopGroup getBossGroup() {
    if (isLinux) {
        return new EpollEventLoopGroup(0x1,
                new NettyThreadFactory("@+DNBossThread", Thread.NORM_PRIORITY));
    } else {
        return new NioEventLoopGroup(0x1, new NettyThreadFactory("@+DNBossThread", Thread.NORM_PRIORITY));
    }
}

@Bean(name = "workerGroup")
public EventLoopGroup getWorkerGroup() {
    if (isLinux) {
        return new EpollEventLoopGroup(Runtime.getRuntime().availableProcessors() + 0x1,
                new NettyThreadFactory("@+DNWorkerThread", Thread.NORM_PRIORITY));
    } else {
        return new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() + 0x1,
                new NettyThreadFactory("@+DNWorkerThread", Thread.NORM_PRIORITY));
    }
}

@Bean(name = "bossGroup")
public EventLoopGroup getBossGroup() {
    if (isLinux) {
        return new EpollEventLoopGroup(0x1,
                new NettyThreadFactory("@+DNCSBossThread", Thread.NORM_PRIORITY));
    } else {
        return new NioEventLoopGroup(0x1, new NettyThreadFactory("@+DNCSBossThread", Thread.NORM_PRIORITY));
    }
}

@Bean(name = "workerGroup")
public EventLoopGroup getWorkerGroup() {
    if (isLinux) {
        return new EpollEventLoopGroup(Runtime.getRuntime().availableProcessors() + 0x1,
                new NettyThreadFactory("@+DNCSWorkerThread", Thread.NORM_PRIORITY));
    } else {
        return new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() + 0x1,
                new NettyThreadFactory("@+DNCSWorkerThread", Thread.NORM_PRIORITY));
    }
}

public static EventLoopGroup createEventLoopGroup(int nThreads, String prefix) {
   if(SUPPORTS_EPOLL){
     return new EpollEventLoopGroup(nThreads, new NamedThreadFactory(prefix));
   }else{
     return new NioEventLoopGroup(nThreads, new NamedThreadFactory(prefix));
   }
}

private void setConf(Configuration conf) {
  conf.setBoolean(AsyncRpcClient.USE_NATIVE_TRANSPORT, useNativeTransport);
  conf.setBoolean(AsyncRpcClient.USE_GLOBAL_EVENT_LOOP_GROUP, useGlobalEventLoopGroup);
  if (useGlobalEventLoopGroup && AsyncRpcClient.GLOBAL_EVENT_LOOP_GROUP != null) {
    if (useNativeTransport
        && !(AsyncRpcClient.GLOBAL_EVENT_LOOP_GROUP.getFirst() instanceof EpollEventLoopGroup)
        || (!useNativeTransport
        && !(AsyncRpcClient.GLOBAL_EVENT_LOOP_GROUP.getFirst() instanceof NioEventLoopGroup))) {
      AsyncRpcClient.GLOBAL_EVENT_LOOP_GROUP.getFirst().shutdownGracefully();
      AsyncRpcClient.GLOBAL_EVENT_LOOP_GROUP = null;
    }
  }
}