io.netty.channel.kqueue.KQueueDatagramChannel#io.netty.channel.unix.ServerDomainSocketChannel源码实例Demo

@Override
@SuppressWarnings("unchecked")
public <CHANNEL extends Channel> CHANNEL getChannel(Class<CHANNEL> channelClass) {
	if (channelClass.equals(SocketChannel.class)) {
		return (CHANNEL) new EpollSocketChannel();
	}
	if (channelClass.equals(ServerSocketChannel.class)) {
		return (CHANNEL) new EpollServerSocketChannel();
	}
	if (channelClass.equals(DatagramChannel.class)) {
		return (CHANNEL) new EpollDatagramChannel();
	}
	if (channelClass.equals(DomainSocketChannel.class)) {
		return (CHANNEL) new EpollDomainSocketChannel();
	}
	if (channelClass.equals(ServerDomainSocketChannel.class)) {
		return (CHANNEL) new EpollServerDomainSocketChannel();
	}
	throw new IllegalArgumentException("Unsupported channel type: " + channelClass.getSimpleName());
}
 

@Override
@SuppressWarnings("unchecked")
public <CHANNEL extends Channel> CHANNEL getChannel(Class<CHANNEL> channelClass) {
	if (channelClass.equals(SocketChannel.class)) {
		return (CHANNEL) new KQueueSocketChannel();
	}
	if (channelClass.equals(ServerSocketChannel.class)) {
		return (CHANNEL) new KQueueServerSocketChannel();
	}
	if (channelClass.equals(DatagramChannel.class)) {
		return (CHANNEL) new KQueueDatagramChannel();
	}
	if (channelClass.equals(DomainSocketChannel.class)) {
		return (CHANNEL) new KQueueDomainSocketChannel();
	}
	if (channelClass.equals(ServerDomainSocketChannel.class)) {
		return (CHANNEL) new KQueueServerDomainSocketChannel();
	}
	throw new IllegalArgumentException("Unsupported channel type: " + channelClass.getSimpleName());
}
 

@Override
protected Class<? extends Channel> channelType(boolean isDomainSocket) {
	return isDomainSocket ? ServerDomainSocketChannel.class : ServerSocketChannel.class;
}
 

@Before
public void setupUnixDomainSocketServer() throws IOException, URISyntaxException {
  Class<? extends ServerDomainSocketChannel> channelType = null;

  if (Epoll.isAvailable()) {
    group = new EpollEventLoopGroup(2);
    channelType = EpollServerDomainSocketChannel.class;
  } else if (KQueue.isAvailable()) {
    group = new KQueueEventLoopGroup(2);
    channelType = KQueueServerDomainSocketChannel.class;
  }

  assumeThat(group).isNotNull();
  assumeThat(channelType).isNotNull();

  ServerBootstrap bootstrap = new ServerBootstrap()
    .group(group)
    .option(ChannelOption.SO_BACKLOG, 1024)
    .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
    .channel(channelType)
    .childHandler(new ChannelInitializer<DomainSocketChannel>() {
      @Override
      protected void initChannel(DomainSocketChannel ch) {
        ch.pipeline()
          .addLast("http-codec", new HttpServerCodec())
          .addLast("http-keep-alive", new HttpServerKeepAliveHandler())
          .addLast("http-aggregator", new HttpObjectAggregator(Integer.MAX_VALUE))
          .addLast(new SimpleChannelInboundHandler<FullHttpRequest>() {
            @Override
            protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest req) {
              byte[] bytes = responseText.get().getBytes(UTF_8);
              ByteBuf text = Unpooled.wrappedBuffer(bytes);
              FullHttpResponse res = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, text);
              res.headers().set(CONTENT_TYPE, MediaType.PLAIN_TEXT_UTF_8.toString());
              res.headers().set(CONTENT_LENGTH, bytes.length);

              ctx.writeAndFlush(res);
            }
          });
      }
    });

  Path temp = Files.createTempFile("domain-socket-test", "socket");
  Files.deleteIfExists(temp);

  SocketAddress address = new DomainSocketAddress(temp.toFile());
  future = bootstrap.bind(address);

  socket = new URI("unix", null, null, 0, temp.toString(), null, null);
}