Java 类io.netty.channel.socket.oio.OioSocketChannel 实例源码

public AHessianJmxClient(String discoveryName, int port, boolean debug,
        InternalLogger logger) throws Exception
{
    _discoveryName = discoveryName;
    _port = port;
    _debug = debug;
    _logger = logger;
    Map options = new HashMap();
    options.put("sync", true);
    options.put("timeout", (long) 5000);

    final ChannelPipelineFactoryBuilder<MBeanServerConnection> builder = new ChannelPipelineFactoryBuilder<MBeanServerConnection>()
            .serviceThreads(10).reconnect(10)
            .rpcServiceInterface(MBeanServerConnection.class)
            .serviceOptions(options);

    builder.debug();
    builder.serializerFactory(new JmxSerializerFactory());

    final Set<String> channelOptions = new HashSet();
    channelOptions.add("SO_REUSE");
    channelOptions.add("TCP_NODELAY");
    client = new DefaultClient<MBeanServerConnection>(
            OioSocketChannel.class, builder, channelOptions);

}

public List<BootstrapFactory<Bootstrap>> clientSocket() {
    return Arrays.asList(
            new BootstrapFactory<Bootstrap>() {
                @Override
                public Bootstrap newInstance() {
                    return new Bootstrap().group(nioWorkerGroup).channel(NioSocketChannel.class);
                }
            },
            new BootstrapFactory<Bootstrap>() {
                @Override
                public Bootstrap newInstance() {
                    return new Bootstrap().group(oioWorkerGroup).channel(OioSocketChannel.class)
                            .option(ChannelOption.SO_TIMEOUT, OIO_SO_TIMEOUT);
                }
            }
    );
}

static List<Factory<Bootstrap>> clientSocket() {
    List<Factory<Bootstrap>> list = new ArrayList<Factory<Bootstrap>>();
    list.add(new Factory<Bootstrap>() {
        @Override
        public Bootstrap newInstance() {
            return new Bootstrap().group(nioWorkerGroup).channel(NioSocketChannel.class);
        }
    });
    list.add(new Factory<Bootstrap>() {
        @Override
        public Bootstrap newInstance() {
            return new Bootstrap().group(oioWorkerGroup).channel(OioSocketChannel.class);
        }
    });
    return list;
}

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!");
    }
}

static List<Factory<Bootstrap>> clientSocket() {
    List<Factory<Bootstrap>> list = new ArrayList<Factory<Bootstrap>>();
    list.add(new Factory<Bootstrap>() {
        @Override
        public Bootstrap newInstance() {
            return new Bootstrap().group(nioWorkerGroup).channel(NioSocketChannel.class);
        }
    });
    list.add(new Factory<Bootstrap>() {
        @Override
        public Bootstrap newInstance() {
            return new Bootstrap().group(oioWorkerGroup).channel(OioSocketChannel.class);
        }
    });
    return list;
}

/**
 * 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;
}

@Test
public void testTooManyClientChannels() throws Exception {
    EventLoopGroup g = new OioEventLoopGroup(1);
    ServerBootstrap sb = new ServerBootstrap();
    sb.channel(OioServerSocketChannel.class);
    sb.group(g);
    sb.childHandler(new ChannelInboundHandlerAdapter());
    ChannelFuture f1 = sb.bind(0);
    f1.sync();

    Bootstrap cb = new Bootstrap();
    cb.channel(OioSocketChannel.class);
    cb.group(g);
    cb.handler(new ChannelInboundHandlerAdapter());
    ChannelFuture f2 = cb.connect(NetUtil.LOCALHOST, ((InetSocketAddress) f1.channel().localAddress()).getPort());
    f2.await();

    assertThat(f2.cause(), is(instanceOf(ChannelException.class)));
    assertThat(f2.cause().getMessage().toLowerCase(), containsString("too many channels"));

    final CountDownLatch notified = new CountDownLatch(1);
    f2.addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            notified.countDown();
        }
    });

    notified.await();
    g.shutdownGracefully();
}

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

@Test
public void testTooManyClientChannels() throws Exception {
    EventLoopGroup g = new OioEventLoopGroup(1);
    ServerBootstrap sb = new ServerBootstrap();
    sb.channel(OioServerSocketChannel.class);
    sb.group(g);
    sb.childHandler(new ChannelInboundHandlerAdapter());
    ChannelFuture f1 = sb.bind(0);
    f1.sync();

    Bootstrap cb = new Bootstrap();
    cb.channel(OioSocketChannel.class);
    cb.group(g);
    cb.handler(new ChannelInboundHandlerAdapter());
    ChannelFuture f2 = cb.connect(NetUtil.LOCALHOST, ((InetSocketAddress) f1.channel().localAddress()).getPort());
    f2.await();

    assertThat(f2.cause(), is(instanceOf(ChannelException.class)));
    assertThat(f2.cause().getMessage().toLowerCase(), containsString("too many channels"));

    final CountDownLatch notified = new CountDownLatch(1);
    f2.addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            notified.countDown();
        }
    });

    notified.await();
    g.shutdownGracefully();
}

@Test
public void testTooManyClientChannels() throws Exception {
    EventLoopGroup g = new OioEventLoopGroup(1);
    ServerBootstrap sb = new ServerBootstrap();
    sb.channel(OioServerSocketChannel.class);
    sb.group(g);
    sb.childHandler(new ChannelHandlerAdapter());
    ChannelFuture f1 = sb.bind(0);
    f1.sync();

    Bootstrap cb = new Bootstrap();
    cb.channel(OioSocketChannel.class);
    cb.group(g);
    cb.handler(new ChannelHandlerAdapter());
    ChannelFuture f2 = cb.connect(NetUtil.LOCALHOST, ((InetSocketAddress) f1.channel().localAddress()).getPort());
    f2.await();

    assertThat(f2.cause(), is(instanceOf(ChannelException.class)));
    assertThat(f2.cause().getMessage().toLowerCase(), containsString("too many channels"));

    final CountDownLatch notified = new CountDownLatch(1);
    f2.addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            notified.countDown();
        }
    });

    notified.await();
    g.shutdownGracefully();
}

private ChannelFuture connnect(RpcRequest request) {
    final Client client = this;
    final RpcRequest frequest = request;
    Bootstrap b = new Bootstrap();
    try {
        b.group(new OioEventLoopGroup()).channel(OioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1000 * 10)
                .remoteAddress(new InetSocketAddress(host, port))
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch)
                            throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast("encoder", new RpcEncoder(End.CLIENT));
                        pipeline.addLast("decoder", new RpcDecoder(End.CLIENT));
                        pipeline.addLast("handler", new ClientChannelHandler(client, frequest));
                    }
                });

        ChannelFuture future = b.connect().syncUninterruptibly();
        future.awaitUninterruptibly(1000 * 10);
        return future;
    } catch (Exception e) {
        e.printStackTrace();
        return null;
    }
}

/**
 * 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);
}

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

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

@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;
}

/**
 * Create a new {@link AbstractEndpoint}.
 *
 * @param hostname the hostname/ipaddr of the remote channel.
 * @param bucket the name of the bucket.
 * @param username the user authorized for bucket access.
 * @param password the password of the user.
 * @param port the port of the remote channel.
 * @param environment the environment of the core.
 * @param responseBuffer the response buffer for passing responses up the stack.
 */
protected AbstractEndpoint(final String hostname, final String bucket, final String username, final String password, final int port,
    final CoreEnvironment environment, final RingBuffer<ResponseEvent> responseBuffer, boolean isTransient,
    final EventLoopGroup ioPool, final boolean pipeline) {
    super(LifecycleState.DISCONNECTED);
    this.bucket = bucket;
    this.username = username;
    this.password = password;
    this.responseBuffer = responseBuffer;
    this.env = environment;
    this.isTransient = isTransient;
    this.ioPool = ioPool;
    this.pipeline = pipeline;
    this.free = true;
    this.hostname = hostname;
    this.connectCallbackGracePeriod = Integer.parseInt(
        System.getProperty("com.couchbase.connectCallbackGracePeriod", DEFAULT_CONNECT_CALLBACK_GRACE_PERIOD)
    );
    LOGGER.debug("Using a connectCallbackGracePeriod of {} on Endpoint {}:{}", connectCallbackGracePeriod,
        hostname, port);
    if (environment.sslEnabled()) {
        this.sslEngineFactory = new SSLEngineFactory(environment);
    }

    Class<? extends Channel> channelClass = NioSocketChannel.class;
    if (ioPool instanceof EpollEventLoopGroup) {
        channelClass = EpollSocketChannel.class;
    } else if (ioPool instanceof OioEventLoopGroup) {
        channelClass = OioSocketChannel.class;
    }

    ByteBufAllocator allocator = env.bufferPoolingEnabled()
            ? PooledByteBufAllocator.DEFAULT : UnpooledByteBufAllocator.DEFAULT;

    boolean tcpNodelay = environment().tcpNodelayEnabled();
    bootstrap = new BootstrapAdapter(new Bootstrap()
        .remoteAddress(hostname, port)
        .group(ioPool)
        .channel(channelClass)
        .option(ChannelOption.ALLOCATOR, allocator)
        .option(ChannelOption.TCP_NODELAY, tcpNodelay)
        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, env.socketConnectTimeout())
        .handler(new ChannelInitializer<Channel>() {
            @Override
            protected void initChannel(Channel channel) throws Exception {
                ChannelPipeline pipeline = channel.pipeline();
                if (environment.sslEnabled()) {
                    pipeline.addLast(new SslHandler(sslEngineFactory.get()));
                }
                if (LOGGER.isTraceEnabled()) {
                    pipeline.addLast(LOGGING_HANDLER_INSTANCE);
                }
                customEndpointHandlers(pipeline);
            }
        }));
}

@Override
public OioSocketChannel newChannel() {
    return new OioSocketChannel(new Socket(this.proxy));
}

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

private static Channel doConnect(Bootstrap bootstrap,
                                 InetSocketAddress remoteAddress,
                                 AsyncResult<?> asyncResult,
                                 int maxAttemptCount,
                                 @Nullable Condition<Void> stopCondition) throws Throwable {
  int attemptCount = 0;
  if (bootstrap.config().group() instanceof NioEventLoopGroup) {
    return connectNio(bootstrap, remoteAddress, asyncResult, maxAttemptCount, stopCondition, attemptCount);
  }

  bootstrap.validate();

  while (true) {
    try {
      OioSocketChannel channel = new OioSocketChannel(new Socket(remoteAddress.getAddress(), remoteAddress.getPort()));
      BootstrapUtil.initAndRegister(channel, bootstrap).sync();
      return channel;
    }
    catch (IOException e) {
      if (stopCondition != null && stopCondition.value(null) || asyncResult != null && !asyncResult.isProcessed()) {
        return null;
      }
      else if (maxAttemptCount == -1) {
        if (sleep(asyncResult, 300)) {
          return null;
        }
        attemptCount++;
      }
      else if (++attemptCount < maxAttemptCount) {
        if (sleep(asyncResult, attemptCount * NettyUtil.MIN_START_TIME)) {
          return null;
        }
      }
      else {
        if (asyncResult != null) {
          asyncResult.rejectWithThrowable(e);
        }
        return null;
      }
    }
  }
}