dkulp commented on a change in pull request #887:
URL: https://github.com/apache/avro/pull/887#discussion_r430351987
##########
File path:
lang/java/ipc-netty/src/main/java/org/apache/avro/ipc/netty/NettyServer.java
##########
@@ -59,55 +55,32 @@
private final Responder responder;
private final Channel serverChannel;
- private final ChannelGroup allChannels = new
DefaultChannelGroup("avro-netty-server");
- private final ChannelFactory channelFactory;
+ private final EventLoopGroup bossGroup = new NioEventLoopGroup(1);
+ private final EventLoopGroup workerGroup = new NioEventLoopGroup(10);
+ private final EventLoopGroup callerGroup = new DefaultEventLoopGroup(16);
private final CountDownLatch closed = new CountDownLatch(1);
+ private final AtomicInteger activeCount = new AtomicInteger(0);
- public NettyServer(Responder responder, InetSocketAddress addr) {
- this(responder, addr,
- new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
Executors.newCachedThreadPool()));
- }
-
- public NettyServer(Responder responder, InetSocketAddress addr,
ChannelFactory channelFactory) {
- this(responder, addr, channelFactory, null);
+ public NettyServer(Responder responder, InetSocketAddress addr) throws
InterruptedException {
+ this(responder, addr, null);
}
- /**
- * @param executionHandler if not null, will be inserted into the Netty
- * pipeline. Use this when your responder does long,
- * non-cpu bound processing (see Netty's
- * ExecutionHandler javadoc).
- * @param pipelineFactory Avro-related handlers will be added on top of what
- * this factory creates
- */
- public NettyServer(Responder responder, InetSocketAddress addr,
ChannelFactory channelFactory,
- final ChannelPipelineFactory pipelineFactory, final ExecutionHandler
executionHandler) {
+ public NettyServer(Responder responder, InetSocketAddress addr, final
Consumer<SocketChannel> initializer)
+ throws InterruptedException {
this.responder = responder;
- this.channelFactory = channelFactory;
- ServerBootstrap bootstrap = new ServerBootstrap(channelFactory);
- bootstrap.setPipelineFactory(() -> {
- ChannelPipeline p = pipelineFactory.getPipeline();
- p.addLast("frameDecoder", new NettyFrameDecoder());
- p.addLast("frameEncoder", new NettyFrameEncoder());
- if (executionHandler != null) {
- p.addLast("executionHandler", executionHandler);
- }
- p.addLast("handler", new NettyServerAvroHandler());
- return p;
- });
- serverChannel = bootstrap.bind(addr);
- allChannels.add(serverChannel);
- }
-
- /**
- * @param executionHandler if not null, will be inserted into the Netty
- * pipeline. Use this when your responder does long,
- * non-cpu bound processing (see Netty's
- * ExecutionHandler javadoc).
- */
- public NettyServer(Responder responder, InetSocketAddress addr,
ChannelFactory channelFactory,
- final ExecutionHandler executionHandler) {
- this(responder, addr, channelFactory, Channels::pipeline,
executionHandler);
+ ServerBootstrap bootstrap = new ServerBootstrap().group(bossGroup,
workerGroup)
+ .channel(NioServerSocketChannel.class).childHandler(new
ChannelInitializer<SocketChannel>() {
+ @Override
+ public void initChannel(SocketChannel ch) throws Exception {
+ if (initializer != null) {
+ initializer.accept(ch);
+ }
+ ch.pipeline().addLast("frameDecoder", new NettyFrameDecoder())
+ .addLast("frameEncoder", new
NettyFrameEncoder()).addLast("handler", new NettyServerAvroHandler());
+ }
+ }).option(ChannelOption.SO_BACKLOG,
128).childOption(ChannelOption.SO_KEEPALIVE, true);
Review comment:
Done for both.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
