io.netty.channel.oio.OioEventLoopGroup#io.netty.channel.socket.oio.OioSocketChannel源码实例Demo

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

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

@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 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 channelOptionsTest_oio() {
  Channel channel = new OioSocketChannel();
  SocketOptions socketOptions = setAndValidateGeneric(channel);
  assertEquals(250, (int) socketOptions.soTimeoutMillis);
}
 

public void connect() {
	initConnect();
	bootstrap.option(ChannelOption.SO_KEEPALIVE, keepAlive);
	bootstrap.option(ChannelOption.TCP_NODELAY, tcpNoDelay);
	bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeOutMillis);
	if (SocketModel.BLOCK.equals(socketModel)) {
		NettyLog.info("block socket");
		bootstrap.group(group).channel(OioSocketChannel.class);
	} else {
		bootstrap.group(group).channel(NioSocketChannel.class);
	}
	bootstrap.remoteAddress(this.getHost(), this.getPort());

	bootstrap.handler(new ChannelInitializer<SocketChannel>() {
		@Override
		protected void initChannel(SocketChannel ch) throws Exception {
			initSocketChannel(ch);
			
			if (sslCtx != null) {
		        SSLEngine sslEngine = sslCtx.newEngine(ch.alloc());
		        sslEngine.setUseClientMode(true);
		        ch.pipeline().addFirst(NettyConstant.HANDLER_NAME_SSL, new SslHandler(sslEngine));  
			}

			if (self().isCheckHeartbeat()) {
				NettyLog.info("checkHeartBeat.....");
				ch.pipeline().addLast(
						new IdleStateHandler(readerIdleTimeSeconds, writerIdleTimeSeconds, allIdleTimeSeconds));
				ch.pipeline().addLast(NettyConstant.HANDLER_NAME_HEARTCHECK, new HeartbeatClientHandler());
			}
		}
	});
	NettyLog.debug("connect start");
	doConnect();

	if (checkConnectFlag && checkConnectSeconds > 1) {
		if (reConnectService == null) {
			reConnectService = Executors.newSingleThreadScheduledExecutor();
			
			// 第二个参数为首次执行的延时时间，第三个参数为定时执行的间隔时间
			reConnectService.scheduleWithFixedDelay(new ConnectCheckClient(this), 20, checkConnectSeconds,
					TimeUnit.SECONDS);
		}  
	}
}
 

@Test(timeout = 3000)
public void testAddInListenerOio() throws Throwable {
    testAddInListener(new OioSocketChannel(), new OioEventLoopGroup(1));
}
 

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

@Test
public void worksWithOioEventLoopGroupFactory() {
    assertThat(resolveSocketChannelFactory(new OioEventLoopGroup()).newChannel()).isInstanceOf(OioSocketChannel.class);
}
 

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