Merge pull request #38 from tingyanshen/feat/session_lifecycle

refactor[net]: adding session lifecycle consumer support

net/src/main/java/com/zfoo/net/core/AbstractClient.java

@@ -35,7 +35,7 @@ import org.slf4j.LoggerFactory;
  * @author godotg
  * @version 3.0
  */
-public abstract class AbstractClient implements IClient {
+public abstract class AbstractClient<C extends Channel> extends ChannelInitializer<C> implements IClient {
 
     protected static final Logger logger = LoggerFactory.getLogger(AbstractClient.class);
 
@@ -53,20 +53,18 @@ public abstract class AbstractClient implements IClient {
         this.port = host.getPort();
     }
 
-    public abstract ChannelInitializer<? extends Channel> channelChannelInitializer();
-
     @Override
     public synchronized Session start() {
-        return doStart(channelChannelInitializer());
+        return doStart();
     }
 
-    private synchronized Session doStart(ChannelInitializer<? extends Channel> channelChannelInitializer) {
+    private synchronized Session doStart() {
         this.bootstrap = new Bootstrap();
         this.bootstrap.group(nioEventLoopGroup)
                 .channel(Epoll.isAvailable() ? EpollSocketChannel.class : NioSocketChannel.class)
                 .option(ChannelOption.TCP_NODELAY, true)
                 .option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(16 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_MB))
-                .handler(channelChannelInitializer());
+                .handler(this);
         var channelFuture = bootstrap.connect(hostAddress, port);
         channelFuture.syncUninterruptibly();
 

net/src/main/java/com/zfoo/net/core/AbstractServer.java

@@ -35,11 +35,11 @@ import java.util.List;
  * @author godotg
  * @version 3.0
  */
-public abstract class AbstractServer implements IServer {
+public abstract class AbstractServer<C extends Channel> extends ChannelInitializer<C> implements IServer {
     private static final Logger logger = LoggerFactory.getLogger(AbstractServer.class);
 
     // 所有的服务器都可以在这个列表中取到
-    protected static final List<AbstractServer> allServers = new ArrayList<>(1);
+    protected static final List<AbstractServer<? extends Channel>> allServers = new ArrayList<>(1);
 
     protected String hostAddress;
     protected int port;
@@ -60,14 +60,12 @@ public abstract class AbstractServer implements IServer {
         this.port = host.getPort();
     }
 
-    public abstract ChannelInitializer<? extends Channel> channelChannelInitializer();
-
     @Override
     public void start() {
-        doStart(channelChannelInitializer());
+        doStart();
     }
 
-    protected synchronized void doStart(ChannelInitializer<? extends Channel> channelChannelInitializer) {
+    protected synchronized void doStart() {
         var cpuNum = Runtime.getRuntime().availableProcessors();
         // 一条线程持有一个端口对应的selector，如果我们启动不仅仅是一个服务器端口的话，为了更好的性能需要修改对应的bossGroup数量
         bossGroup = Epoll.isAvailable()
@@ -84,7 +82,7 @@ public abstract class AbstractServer implements IServer {
                 .option(ChannelOption.SO_REUSEADDR, true)
                 .childOption(ChannelOption.TCP_NODELAY, true)
                 .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(16 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_MB))
-                .childHandler(channelChannelInitializer);
+                .childHandler(this);
         // 绑定端口，同步等待成功
         // channelFuture = bootstrap.bind(hostAddress, port).sync();
         // 等待服务端监听端口关闭

net/src/main/java/com/zfoo/net/core/event/ClientSessionActiveEvent.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright (C) 2020 The zfoo Authors
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and limitations under the License.
+ */
+
+package com.zfoo.net.core.event;
+
+import com.zfoo.event.model.event.IEvent;
+import com.zfoo.net.session.Session;
+
+/**
+ * @author tingyanshen
+ * @version 3.0
+ */
+public class ClientSessionActiveEvent implements IEvent {
+
+    private Session session;
+
+    public static ClientSessionActiveEvent valueOf(Session session) {
+        var event = new ClientSessionActiveEvent();
+        event.session = session;
+        return event;
+    }
+
+    public Session getSession() {
+        return session;
+    }
+
+    public void setSession(Session session) {
+        this.session = session;
+    }
+}

net/src/main/java/com/zfoo/net/core/event/ServerSessionActiveEvent.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright (C) 2020 The zfoo Authors
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and limitations under the License.
+ */
+
+package com.zfoo.net.core.event;
+
+import com.zfoo.event.model.event.IEvent;
+import com.zfoo.net.session.Session;
+
+/**
+ * @author tingyanshen
+ * @version 3.0
+ */
+public class ServerSessionActiveEvent implements IEvent {
+
+    private Session session;
+
+    public static ServerSessionActiveEvent valueOf(Session session) {
+        var event = new ServerSessionActiveEvent();
+        event.session = session;
+        return event;
+    }
+
+    public Session getSession() {
+        return session;
+    }
+
+    public void setSession(Session session) {
+        this.session = session;
+    }
+}

net/src/main/java/com/zfoo/net/core/gateway/GatewayServer.java

@@ -20,9 +20,9 @@ import com.zfoo.net.handler.idle.ServerIdleHandler;
 import com.zfoo.net.session.Session;
 import com.zfoo.protocol.IPacket;
 import com.zfoo.util.net.HostAndPort;
-import io.netty.channel.ChannelInitializer;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.timeout.IdleStateHandler;
+import org.apache.curator.shaded.com.google.common.base.MoreObjects;
 import org.springframework.lang.Nullable;
 
 import java.util.function.BiFunction;
@@ -31,35 +31,26 @@ import java.util.function.BiFunction;
  * @author godotg
  * @version 3.0
  */
-public class GatewayServer extends AbstractServer {
+public class GatewayServer extends AbstractServer<SocketChannel> {
 
-    private BiFunction<Session, IPacket, Boolean> packetFilter;
+    private final GatewayRouteHandler gatewayRouteHandler;
 
     public GatewayServer(HostAndPort host, @Nullable BiFunction<Session, IPacket, Boolean> packetFilter) {
+        this(host, packetFilter, null);
+    }
+
+    public GatewayServer(HostAndPort host,
+                         @Nullable BiFunction<Session, IPacket, Boolean> packetFilter,
+                         GatewayRouteHandler gatewayRouteHandler) {
         super(host);
-        this.packetFilter = packetFilter;
+        this.gatewayRouteHandler = MoreObjects.firstNonNull(gatewayRouteHandler, new GatewayRouteHandler(packetFilter));
     }
 
     @Override
-    public ChannelInitializer<SocketChannel> channelChannelInitializer() {
-        return new ChannelHandlerInitializer(packetFilter);
-    }
-
-
-    private static class ChannelHandlerInitializer extends ChannelInitializer<SocketChannel> {
-
-        private BiFunction<Session, IPacket, Boolean> packetFilter;
-
-        public ChannelHandlerInitializer(BiFunction<Session, IPacket, Boolean> packetFilter) {
-            this.packetFilter = packetFilter;
-        }
-
-        @Override
-        protected void initChannel(SocketChannel channel) {
-            channel.pipeline().addLast(new IdleStateHandler(0, 0, 180));
-            channel.pipeline().addLast(new ServerIdleHandler());
-            channel.pipeline().addLast(new TcpCodecHandler());
-            channel.pipeline().addLast(new GatewayRouteHandler(packetFilter));
-        }
+    protected void initChannel(SocketChannel channel) {
+        channel.pipeline().addLast(new IdleStateHandler(0, 0, 180));
+        channel.pipeline().addLast(new ServerIdleHandler());
+        channel.pipeline().addLast(new TcpCodecHandler());
+        channel.pipeline().addLast(gatewayRouteHandler);
     }
 }

net/src/main/java/com/zfoo/net/core/gateway/WebsocketGatewayServer.java

@@ -21,13 +21,13 @@ import com.zfoo.net.session.Session;
 import com.zfoo.protocol.IPacket;
 import com.zfoo.protocol.util.IOUtils;
 import com.zfoo.util.net.HostAndPort;
-import io.netty.channel.ChannelInitializer;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.codec.http.HttpObjectAggregator;
 import io.netty.handler.codec.http.HttpServerCodec;
 import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
 import io.netty.handler.stream.ChunkedWriteHandler;
 import io.netty.handler.timeout.IdleStateHandler;
+import org.apache.curator.shaded.com.google.common.base.MoreObjects;
 import org.springframework.lang.Nullable;
 
 import java.util.function.BiFunction;
@@ -36,40 +36,32 @@ import java.util.function.BiFunction;
  * @author godotg
  * @version 3.0
  */
-public class WebsocketGatewayServer extends AbstractServer {
+public class WebsocketGatewayServer extends AbstractServer<SocketChannel> {
 
-    private BiFunction<Session, IPacket, Boolean> packetFilter;
+    private final GatewayRouteHandler gatewayRouteHandler;
 
     public WebsocketGatewayServer(HostAndPort host, @Nullable BiFunction<Session, IPacket, Boolean> packetFilter) {
-        super(host);
-        this.packetFilter = packetFilter;
+        this(host, packetFilter, null);
     }
 
+    public WebsocketGatewayServer(HostAndPort host,
+                                  @Nullable BiFunction<Session, IPacket, Boolean> packetFilter,
+                                  GatewayRouteHandler gatewayRouteHandler) {
+        super(host);
+        this.gatewayRouteHandler = MoreObjects.firstNonNull(gatewayRouteHandler, new GatewayRouteHandler(packetFilter));
+    }
+
+
     @Override
-    public ChannelInitializer<SocketChannel> channelChannelInitializer() {
-        return new ChannelHandlerInitializer(packetFilter);
-    }
+    protected void initChannel(SocketChannel channel) {
+        channel.pipeline().addLast(new IdleStateHandler(0, 0, 180));
+        channel.pipeline().addLast(new ServerIdleHandler());
 
-
-    private static class ChannelHandlerInitializer extends ChannelInitializer<SocketChannel> {
-
-        private BiFunction<Session, IPacket, Boolean> packetFilter;
-
-        public ChannelHandlerInitializer(BiFunction<Session, IPacket, Boolean> packetFilter) {
-            this.packetFilter = packetFilter;
-        }
-
-        @Override
-        protected void initChannel(SocketChannel channel) {
-            channel.pipeline().addLast(new IdleStateHandler(0, 0, 180));
-            channel.pipeline().addLast(new ServerIdleHandler());
-
-            channel.pipeline().addLast(new HttpServerCodec(8 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB));
-            channel.pipeline().addLast(new HttpObjectAggregator(16 * IOUtils.BYTES_PER_MB));
-            channel.pipeline().addLast(new WebSocketServerProtocolHandler("/websocket"));
-            channel.pipeline().addLast(new ChunkedWriteHandler());
-            channel.pipeline().addLast(new WebSocketCodecHandler());
-            channel.pipeline().addLast(new GatewayRouteHandler(packetFilter));
-        }
+        channel.pipeline().addLast(new HttpServerCodec(8 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB));
+        channel.pipeline().addLast(new HttpObjectAggregator(16 * IOUtils.BYTES_PER_MB));
+        channel.pipeline().addLast(new WebSocketServerProtocolHandler("/websocket"));
+        channel.pipeline().addLast(new ChunkedWriteHandler());
+        channel.pipeline().addLast(new WebSocketCodecHandler());
+        channel.pipeline().addLast(gatewayRouteHandler);
     }
 }

net/src/main/java/com/zfoo/net/core/gateway/WebsocketSslGatewayServer.java

@@ -22,7 +22,6 @@ import com.zfoo.protocol.IPacket;
 import com.zfoo.protocol.exception.ExceptionUtils;
 import com.zfoo.protocol.util.IOUtils;
 import com.zfoo.util.net.HostAndPort;
-import io.netty.channel.ChannelInitializer;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.codec.http.HttpObjectAggregator;
 import io.netty.handler.codec.http.HttpServerCodec;
@@ -31,6 +30,7 @@ import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
 import io.netty.handler.stream.ChunkedWriteHandler;
 import io.netty.handler.timeout.IdleStateHandler;
+import org.apache.curator.shaded.com.google.common.base.MoreObjects;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -42,52 +42,46 @@ import java.util.function.BiFunction;
  * @author godotg
  * @version 3.0
  */
-public class WebsocketSslGatewayServer extends AbstractServer {
+public class WebsocketSslGatewayServer extends AbstractServer<SocketChannel> {
 
     private static final Logger logger = LoggerFactory.getLogger(WebsocketSslGatewayServer.class);
 
     private SslContext sslContext;
 
-    private BiFunction<Session, IPacket, Boolean> packetFilter;
+    private final GatewayRouteHandler gatewayRouteHandler;
 
-    public WebsocketSslGatewayServer(HostAndPort host, InputStream pem, InputStream key, BiFunction<Session, IPacket, Boolean> packetFilter) {
+    public WebsocketSslGatewayServer(HostAndPort host,
+                                     InputStream pem,
+                                     InputStream key,
+                                     BiFunction<Session, IPacket, Boolean> packetFilter) {
+        this(host, pem, key, packetFilter, null);
+    }
+
+    public WebsocketSslGatewayServer(HostAndPort host,
+                                     InputStream pem,
+                                     InputStream key,
+                                     BiFunction<Session, IPacket, Boolean> packetFilter,
+                                     GatewayRouteHandler gatewayRouteHandler) {
         super(host);
         try {
             this.sslContext = SslContextBuilder.forServer(pem, key).build();
         } catch (SSLException e) {
             logger.error(ExceptionUtils.getMessage(e));
         }
-        this.packetFilter = packetFilter;
+        this.gatewayRouteHandler = MoreObjects.firstNonNull(gatewayRouteHandler, new GatewayRouteHandler(packetFilter));
     }
 
     @Override
-    public ChannelInitializer<SocketChannel> channelChannelInitializer() {
-        return new ChannelHandlerInitializer(sslContext, packetFilter);
-    }
+    protected void initChannel(SocketChannel channel) {
+        channel.pipeline().addLast(new IdleStateHandler(0, 0, 180));
+        channel.pipeline().addLast(new ServerIdleHandler());
 
-
-    private static class ChannelHandlerInitializer extends ChannelInitializer<SocketChannel> {
-
-        private SslContext sslContext;
-        private BiFunction<Session, IPacket, Boolean> packetFilter;
-
-        public ChannelHandlerInitializer(SslContext sslContext, BiFunction<Session, IPacket, Boolean> packetFilter) {
-            this.sslContext = sslContext;
-            this.packetFilter = packetFilter;
-        }
-
-        @Override
-        protected void initChannel(SocketChannel channel) {
-            channel.pipeline().addLast(new IdleStateHandler(0, 0, 180));
-            channel.pipeline().addLast(new ServerIdleHandler());
-
-            channel.pipeline().addLast(sslContext.newHandler(channel.alloc()));
-            channel.pipeline().addLast(new HttpServerCodec(8 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB));
-            channel.pipeline().addLast(new HttpObjectAggregator(16 * IOUtils.BYTES_PER_MB));
-            channel.pipeline().addLast(new WebSocketServerProtocolHandler("/"));
-            channel.pipeline().addLast(new ChunkedWriteHandler());
-            channel.pipeline().addLast(new WebSocketCodecHandler());
-            channel.pipeline().addLast(new GatewayRouteHandler(packetFilter));
-        }
+        channel.pipeline().addLast(sslContext.newHandler(channel.alloc()));
+        channel.pipeline().addLast(new HttpServerCodec(8 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB));
+        channel.pipeline().addLast(new HttpObjectAggregator(16 * IOUtils.BYTES_PER_MB));
+        channel.pipeline().addLast(new WebSocketServerProtocolHandler("/"));
+        channel.pipeline().addLast(new ChunkedWriteHandler());
+        channel.pipeline().addLast(new WebSocketCodecHandler());
+        channel.pipeline().addLast(gatewayRouteHandler);
     }
 }

net/src/main/java/com/zfoo/net/core/http/HttpServer.java

@@ -19,12 +19,12 @@ import com.zfoo.net.handler.codec.http.HttpCodecHandler;
 import com.zfoo.net.packet.DecodedPacketInfo;
 import com.zfoo.protocol.util.IOUtils;
 import com.zfoo.util.net.HostAndPort;
-import io.netty.channel.ChannelInitializer;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.codec.http.FullHttpRequest;
 import io.netty.handler.codec.http.HttpObjectAggregator;
 import io.netty.handler.codec.http.HttpServerCodec;
 import io.netty.handler.stream.ChunkedWriteHandler;
+import org.apache.curator.shaded.com.google.common.base.MoreObjects;
 
 import java.util.function.Function;
 
@@ -32,32 +32,34 @@ import java.util.function.Function;
  * @author godotg
  * @version 3.0
  */
-public class HttpServer extends AbstractServer {
+public class HttpServer extends AbstractServer<SocketChannel> {
 
     /**
      * http的地址解析器
      */
-    private Function<FullHttpRequest, DecodedPacketInfo> uriResolver;
+    private final Function<FullHttpRequest, DecodedPacketInfo> uriResolver;
+
+    private final ServerRouteHandler serverRouteHandler;
 
     public HttpServer(HostAndPort host, Function<FullHttpRequest, DecodedPacketInfo> uriResolver) {
+        this(host, uriResolver, null);
+    }
+
+
+    public HttpServer(HostAndPort host,
+                      Function<FullHttpRequest, DecodedPacketInfo> uriResolver,
+                      ServerRouteHandler serverRouteHandler) {
         super(host);
         this.uriResolver = uriResolver;
+        this.serverRouteHandler = MoreObjects.firstNonNull(serverRouteHandler, new ServerRouteHandler());
     }
 
     @Override
-    public ChannelInitializer<SocketChannel> channelChannelInitializer() {
-        return new ChannelHandlerInitializer();
-    }
-
-
-    private class ChannelHandlerInitializer extends ChannelInitializer<SocketChannel> {
-        @Override
-        protected void initChannel(SocketChannel channel) {
-            channel.pipeline().addLast(new HttpServerCodec(8 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB));
-            channel.pipeline().addLast(new HttpObjectAggregator(16 * IOUtils.BYTES_PER_MB));
-            channel.pipeline().addLast(new ChunkedWriteHandler());
-            channel.pipeline().addLast(new HttpCodecHandler(uriResolver));
-            channel.pipeline().addLast(new ServerRouteHandler());
-        }
+    protected void initChannel(SocketChannel channel) {
+        channel.pipeline().addLast(new HttpServerCodec(8 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB));
+        channel.pipeline().addLast(new HttpObjectAggregator(16 * IOUtils.BYTES_PER_MB));
+        channel.pipeline().addLast(new ChunkedWriteHandler());
+        channel.pipeline().addLast(new HttpCodecHandler(uriResolver));
+        channel.pipeline().addLast(serverRouteHandler);
     }
 }

net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufGatewayServer.java

@@ -20,9 +20,9 @@ import com.zfoo.net.handler.idle.ServerIdleHandler;
 import com.zfoo.net.session.Session;
 import com.zfoo.protocol.IPacket;
 import com.zfoo.util.net.HostAndPort;
-import io.netty.channel.ChannelInitializer;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.timeout.IdleStateHandler;
+import org.apache.curator.shaded.com.google.common.base.MoreObjects;
 import org.springframework.lang.Nullable;
 
 import java.util.function.BiFunction;
@@ -31,37 +31,28 @@ import java.util.function.BiFunction;
  * @author godotg
  * @version 3.0
  */
-public class JProtobufGatewayServer extends AbstractServer {
+public class JProtobufGatewayServer extends AbstractServer<SocketChannel> {
 
-
-    private final BiFunction<Session, IPacket, Boolean> packetFilter;
+    private final GatewayRouteHandler gatewayRouteHandler;
 
     public JProtobufGatewayServer(HostAndPort host, @Nullable BiFunction<Session, IPacket, Boolean> packetFilter) {
+        this(host, packetFilter, null);
+    }
+
+
+    public JProtobufGatewayServer(HostAndPort host,
+                                  @Nullable BiFunction<Session, IPacket, Boolean> packetFilter,
+                                  GatewayRouteHandler gatewayRouteHandler) {
         super(host);
-        this.packetFilter = packetFilter;
+        this.gatewayRouteHandler = MoreObjects.firstNonNull(gatewayRouteHandler, new GatewayRouteHandler(packetFilter));
     }
 
     @Override
-    public ChannelInitializer<SocketChannel> channelChannelInitializer() {
-        return new ChannelHandlerInitializer(packetFilter);
-    }
-
-
-    private static class ChannelHandlerInitializer extends ChannelInitializer<SocketChannel> {
-
-        private final BiFunction<Session, IPacket, Boolean> packetFilter;
-
-        public ChannelHandlerInitializer(BiFunction<Session, IPacket, Boolean> packetFilter) {
-            this.packetFilter = packetFilter;
-        }
-
-        @Override
-        protected void initChannel(SocketChannel channel) {
-            channel.pipeline().addLast(new IdleStateHandler(0, 0, 180));
-            channel.pipeline().addLast(new ServerIdleHandler());
-            channel.pipeline().addLast(new JProtobufTcpCodecHandler());
-            channel.pipeline().addLast(new GatewayRouteHandler(packetFilter));
-        }
+    protected void initChannel(SocketChannel channel) {
+        channel.pipeline().addLast(new IdleStateHandler(0, 0, 180));
+        channel.pipeline().addLast(new ServerIdleHandler());
+        channel.pipeline().addLast(new JProtobufTcpCodecHandler());
+        channel.pipeline().addLast(gatewayRouteHandler);
     }
 
 }

net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufTcpClient.java

@@ -22,32 +22,30 @@ import io.netty.channel.Channel;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.timeout.IdleStateHandler;
+import org.apache.curator.shaded.com.google.common.base.MoreObjects;
 
 /**
  * @author godotg
  * @version 3.0
  */
-public class JProtobufTcpClient extends AbstractClient {
+public class JProtobufTcpClient extends AbstractClient<SocketChannel> {
+
+    private final ClientRouteHandler clientRouteHandler;
 
     public JProtobufTcpClient(HostAndPort host) {
+        this(host, null);
+    }
+
+    public JProtobufTcpClient(HostAndPort host, ClientRouteHandler clientRouteHandler) {
         super(host);
+        this.clientRouteHandler = MoreObjects.firstNonNull(clientRouteHandler, new ClientRouteHandler());
     }
 
     @Override
-    public ChannelInitializer<? extends Channel> channelChannelInitializer() {
-        return new ChannelHandlerInitializer();
+    protected void initChannel(SocketChannel channel) {
+        channel.pipeline().addLast(new IdleStateHandler(0, 0, 60));
+        channel.pipeline().addLast(new ClientIdleHandler());
+        channel.pipeline().addLast(new JProtobufTcpCodecHandler());
+        channel.pipeline().addLast(clientRouteHandler);
     }
-
-
-    private static class ChannelHandlerInitializer extends ChannelInitializer<SocketChannel> {
-        @Override
-        protected void initChannel(SocketChannel channel) {
-            channel.pipeline().addLast(new IdleStateHandler(0, 0, 60));
-            channel.pipeline().addLast(new ClientIdleHandler());
-            channel.pipeline().addLast(new JProtobufTcpCodecHandler());
-            channel.pipeline().addLast(new ClientRouteHandler());
-        }
-    }
-
-
 }

net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufTcpServer.java

@@ -18,33 +18,32 @@ import com.zfoo.net.handler.ServerRouteHandler;
 import com.zfoo.net.handler.codec.jprotobuf.JProtobufTcpCodecHandler;
 import com.zfoo.net.handler.idle.ServerIdleHandler;
 import com.zfoo.util.net.HostAndPort;
-import io.netty.channel.ChannelInitializer;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.timeout.IdleStateHandler;
+import org.apache.curator.shaded.com.google.common.base.MoreObjects;
 
 /**
  * @author godotg
  * @version 3.0
  */
-public class JProtobufTcpServer extends AbstractServer {
+public class JProtobufTcpServer extends AbstractServer<SocketChannel> {
+
+    private final ServerRouteHandler serverRouteHandler;
 
     public JProtobufTcpServer(HostAndPort host) {
+        this(host, null);
+    }
+
+    public JProtobufTcpServer(HostAndPort host, ServerRouteHandler serverRouteHandler) {
         super(host);
+        this.serverRouteHandler = MoreObjects.firstNonNull(serverRouteHandler, new ServerRouteHandler());
     }
 
     @Override
-    public ChannelInitializer<SocketChannel> channelChannelInitializer() {
-        return new ChannelHandlerInitializer();
-    }
-
-
-    private static class ChannelHandlerInitializer extends ChannelInitializer<SocketChannel> {
-        @Override
-        protected void initChannel(SocketChannel channel) {
-            channel.pipeline().addLast(new IdleStateHandler(0, 0, 180));
-            channel.pipeline().addLast(new ServerIdleHandler());
-            channel.pipeline().addLast(new JProtobufTcpCodecHandler());
-            channel.pipeline().addLast(new ServerRouteHandler());
-        }
+    protected void initChannel(SocketChannel channel) {
+        channel.pipeline().addLast(new IdleStateHandler(0, 0, 180));
+        channel.pipeline().addLast(new ServerIdleHandler());
+        channel.pipeline().addLast(new JProtobufTcpCodecHandler());
+        channel.pipeline().addLast(serverRouteHandler);
     }
 }

net/src/main/java/com/zfoo/net/core/json/JsonWebsocketClient.java

@@ -18,44 +18,43 @@ import com.zfoo.net.handler.ClientRouteHandler;
 import com.zfoo.net.handler.codec.json.JsonWebSocketCodecHandler;
 import com.zfoo.protocol.util.IOUtils;
 import com.zfoo.util.net.HostAndPort;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelInitializer;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.codec.http.HttpClientCodec;
 import io.netty.handler.codec.http.HttpObjectAggregator;
 import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolConfig;
 import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler;
 import io.netty.handler.stream.ChunkedWriteHandler;
+import org.apache.curator.shaded.com.google.common.base.MoreObjects;
 
 /**
  * @author godotg
  * @version 3.0
  */
-public class JsonWebsocketClient extends AbstractClient {
+public class JsonWebsocketClient extends AbstractClient<SocketChannel> {
 
     private WebSocketClientProtocolConfig webSocketClientProtocolConfig;
+    private final ClientRouteHandler clientRouteHandler;
 
     public JsonWebsocketClient(HostAndPort host, WebSocketClientProtocolConfig webSocketClientProtocolConfig) {
+        this(host, webSocketClientProtocolConfig, null);
+    }
+
+    public JsonWebsocketClient(HostAndPort host, WebSocketClientProtocolConfig webSocketClientProtocolConfig,
+                               ClientRouteHandler clientRouteHandler) {
         super(host);
         this.webSocketClientProtocolConfig = webSocketClientProtocolConfig;
+        this.clientRouteHandler = MoreObjects.firstNonNull(clientRouteHandler, new ClientRouteHandler());
     }
 
+
     @Override
-    public ChannelInitializer<? extends Channel> channelChannelInitializer() {
-        return new ChannelHandlerInitializer();
-    }
-
-
-    public class ChannelHandlerInitializer extends ChannelInitializer<SocketChannel> {
-        @Override
-        public void initChannel(SocketChannel channel) {
-            channel.pipeline().addLast(new HttpClientCodec(8 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB));
-            channel.pipeline().addLast(new HttpObjectAggregator(16 * IOUtils.BYTES_PER_MB));
-            channel.pipeline().addLast(new WebSocketClientProtocolHandler(webSocketClientProtocolConfig));
-            channel.pipeline().addLast(new ChunkedWriteHandler());
-            channel.pipeline().addLast(new JsonWebSocketCodecHandler());
-            channel.pipeline().addLast(new ClientRouteHandler());
-        }
+    public void initChannel(SocketChannel channel) {
+        channel.pipeline().addLast(new HttpClientCodec(8 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB));
+        channel.pipeline().addLast(new HttpObjectAggregator(16 * IOUtils.BYTES_PER_MB));
+        channel.pipeline().addLast(new WebSocketClientProtocolHandler(webSocketClientProtocolConfig));
+        channel.pipeline().addLast(new ChunkedWriteHandler());
+        channel.pipeline().addLast(new JsonWebSocketCodecHandler());
+        channel.pipeline().addLast(clientRouteHandler);
     }
 
 }

net/src/main/java/com/zfoo/net/core/json/JsonWebsocketServer.java

@@ -16,49 +16,45 @@ package com.zfoo.net.core.json;
 import com.zfoo.net.core.AbstractServer;
 import com.zfoo.net.handler.ServerRouteHandler;
 import com.zfoo.net.handler.codec.json.JsonWebSocketCodecHandler;
-import com.zfoo.net.handler.codec.websocket.WebSocketCodecHandler;
 import com.zfoo.protocol.util.IOUtils;
 import com.zfoo.util.net.HostAndPort;
-import io.netty.channel.ChannelInitializer;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.codec.http.HttpObjectAggregator;
 import io.netty.handler.codec.http.HttpServerCodec;
 import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
 import io.netty.handler.stream.ChunkedWriteHandler;
+import org.apache.curator.shaded.com.google.common.base.MoreObjects;
 
 /**
  * @author godotg
  * @version 3.0
  */
-public class JsonWebsocketServer extends AbstractServer {
+public class JsonWebsocketServer extends AbstractServer<SocketChannel> {
+
+    private final ServerRouteHandler serverRouteHandler;
 
     public JsonWebsocketServer(HostAndPort host) {
+        this(host, null);
+    }
+
+    public JsonWebsocketServer(HostAndPort host, ServerRouteHandler serverRouteHandler) {
         super(host);
+        this.serverRouteHandler = MoreObjects.firstNonNull(serverRouteHandler, new ServerRouteHandler());
     }
 
     @Override
-    public ChannelInitializer<SocketChannel> channelChannelInitializer() {
-        return new ChannelHandlerInitializer();
+    public void initChannel(SocketChannel channel) {
+        // 编解码 http 请求
+        channel.pipeline().addLast(new HttpServerCodec(8 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB));
+        // 聚合解码 HttpRequest/HttpContent/LastHttpContent 到 FullHttpRequest
+        // 保证接收的 Http 请求的完整性
+        channel.pipeline().addLast(new HttpObjectAggregator(16 * IOUtils.BYTES_PER_MB));
+        // 处理其他的 WebSocketFrame
+        channel.pipeline().addLast(new WebSocketServerProtocolHandler("/websocket"));
+        // 写文件内容，支持异步发送大的码流，一般用于发送文件流
+        channel.pipeline().addLast(new ChunkedWriteHandler());
+        // 编解码WebSocketFrame二进制协议
+        channel.pipeline().addLast(new JsonWebSocketCodecHandler());
+        channel.pipeline().addLast(serverRouteHandler);
     }
-
-
-    public static class ChannelHandlerInitializer extends ChannelInitializer<SocketChannel> {
-
-        @Override
-        public void initChannel(SocketChannel channel) {
-            // 编解码 http 请求
-            channel.pipeline().addLast(new HttpServerCodec(8 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB));
-            // 聚合解码 HttpRequest/HttpContent/LastHttpContent 到 FullHttpRequest
-            // 保证接收的 Http 请求的完整性
-            channel.pipeline().addLast(new HttpObjectAggregator(16 * IOUtils.BYTES_PER_MB));
-            // 处理其他的 WebSocketFrame
-            channel.pipeline().addLast(new WebSocketServerProtocolHandler("/websocket"));
-            // 写文件内容，支持异步发送大的码流，一般用于发送文件流
-            channel.pipeline().addLast(new ChunkedWriteHandler());
-            // 编解码WebSocketFrame二进制协议
-            channel.pipeline().addLast(new JsonWebSocketCodecHandler());
-            channel.pipeline().addLast(new ServerRouteHandler());
-        }
-    }
-
 }

net/src/main/java/com/zfoo/net/core/tcp/TcpClient.java

@@ -18,38 +18,35 @@ import com.zfoo.net.handler.ClientRouteHandler;
 import com.zfoo.net.handler.codec.tcp.TcpCodecHandler;
 import com.zfoo.net.handler.idle.ClientIdleHandler;
 import com.zfoo.util.net.HostAndPort;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelInitializer;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.timeout.IdleStateHandler;
+import org.apache.curator.shaded.com.google.common.base.MoreObjects;
 
 /**
  * @author godotg
  * @version 3.0
  */
-public class TcpClient extends AbstractClient {
+public class TcpClient extends AbstractClient<SocketChannel> {
+
+    private final ClientRouteHandler clientRouteHandler;
 
     public TcpClient(HostAndPort host) {
+        this(host, null);
+    }
+
+    public TcpClient(HostAndPort host, ClientRouteHandler clientRouteHandler) {
         super(host);
+        this.clientRouteHandler = MoreObjects.firstNonNull(clientRouteHandler, new ClientRouteHandler());
     }
 
     @Override
-    public ChannelInitializer<? extends Channel> channelChannelInitializer() {
-        return new ChannelHandlerInitializer();
+    protected void initChannel(SocketChannel channel) {
+        // 可以看出来，这个客户端检测到空闲的时间是60s，相对短一点，这样子就可以发送心跳。
+        // 服务器端则是180s，相对长一点，一旦检测到空闲，则把客户端踢掉。
+        channel.pipeline().addLast(new IdleStateHandler(0, 0, 60));
+        channel.pipeline().addLast(new ClientIdleHandler());
+        channel.pipeline().addLast(new TcpCodecHandler());
+        channel.pipeline().addLast(clientRouteHandler);
     }
 
-
-    private static class ChannelHandlerInitializer extends ChannelInitializer<SocketChannel> {
-        @Override
-        protected void initChannel(SocketChannel channel) {
-            // 可以看出来，这个客户端检测到空闲的时间是60s，相对短一点，这样子就可以发送心跳。
-            // 服务器端则是180s，相对长一点，一旦检测到空闲，则把客户端踢掉。
-            channel.pipeline().addLast(new IdleStateHandler(0, 0, 60));
-            channel.pipeline().addLast(new ClientIdleHandler());
-            channel.pipeline().addLast(new TcpCodecHandler());
-            channel.pipeline().addLast(new ClientRouteHandler());
-        }
-    }
-
-
 }

net/src/main/java/com/zfoo/net/core/tcp/TcpServer.java

@@ -18,33 +18,32 @@ import com.zfoo.net.handler.ServerRouteHandler;
 import com.zfoo.net.handler.codec.tcp.TcpCodecHandler;
 import com.zfoo.net.handler.idle.ServerIdleHandler;
 import com.zfoo.util.net.HostAndPort;
-import io.netty.channel.ChannelInitializer;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.timeout.IdleStateHandler;
+import org.apache.curator.shaded.com.google.common.base.MoreObjects;
 
 /**
  * @author godotg
  * @version 3.0
  */
-public class TcpServer extends AbstractServer {
+public class TcpServer extends AbstractServer<SocketChannel> {
+
+    private final ServerRouteHandler routeHandler;
 
     public TcpServer(HostAndPort host) {
+        this(host, null);
+    }
+
+    public TcpServer(HostAndPort host, ServerRouteHandler serverRouteHandler) {
         super(host);
+        this.routeHandler = MoreObjects.firstNonNull(serverRouteHandler, new ServerRouteHandler());
     }
 
     @Override
-    public ChannelInitializer<SocketChannel> channelChannelInitializer() {
-        return new ChannelHandlerInitializer();
-    }
-
-
-    private static class ChannelHandlerInitializer extends ChannelInitializer<SocketChannel> {
-        @Override
-        protected void initChannel(SocketChannel channel) {
-            channel.pipeline().addLast(new IdleStateHandler(0, 0, 180));
-            channel.pipeline().addLast(new ServerIdleHandler());
-            channel.pipeline().addLast(new TcpCodecHandler());
-            channel.pipeline().addLast(new ServerRouteHandler());
-        }
+    protected void initChannel(SocketChannel socketChannel) throws Exception {
+        channel.pipeline().addLast(new IdleStateHandler(0, 0, 180));
+        channel.pipeline().addLast(new ServerIdleHandler());
+        channel.pipeline().addLast(new TcpCodecHandler());
+        channel.pipeline().addLast(routeHandler);
     }
 }

net/src/main/java/com/zfoo/net/core/udp/UdpClient.java

@@ -23,20 +23,29 @@ import com.zfoo.protocol.exception.ExceptionUtils;
 import com.zfoo.util.net.HostAndPort;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.Channel;
-import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.epoll.Epoll;
 import io.netty.channel.epoll.EpollDatagramChannel;
 import io.netty.channel.socket.nio.NioDatagramChannel;
+import org.apache.curator.shaded.com.google.common.base.MoreObjects;
+
+import java.util.function.Consumer;
 
 /**
  * @author godotg
  * @version 3.0
  */
-public class UdpClient extends AbstractClient {
+public class UdpClient extends AbstractClient<Channel> {
+
+    private final ClientRouteHandler clientRouteHandler;
 
     public UdpClient(HostAndPort host) {
+        this(host, null);
+    }
+
+    public UdpClient(HostAndPort host, ClientRouteHandler clientRouteHandler) {
         super(host);
+        this.clientRouteHandler = MoreObjects.firstNonNull(clientRouteHandler, new ClientRouteHandler());
     }
 
     @Override
@@ -46,7 +55,7 @@ public class UdpClient extends AbstractClient {
             this.bootstrap.group(nioEventLoopGroup)
                     .channel(Epoll.isAvailable() ? EpollDatagramChannel.class : NioDatagramChannel.class)
                     .option(ChannelOption.SO_BROADCAST, true)
-                    .handler(new ChannelHandlerInitializer());
+                    .handler(this);
 
             // bind(0)随机选择一个端口
             var channelFuture = bootstrap.bind(0).sync();
@@ -72,18 +81,8 @@ public class UdpClient extends AbstractClient {
     }
 
     @Override
-    public ChannelInitializer<Channel> channelChannelInitializer() {
-        return new ChannelHandlerInitializer();
+    protected void initChannel(Channel channel) {
+        channel.pipeline().addLast(new UdpCodecHandler());
+        channel.pipeline().addLast(clientRouteHandler);
     }
-
-
-    private static class ChannelHandlerInitializer extends ChannelInitializer<Channel> {
-        @Override
-        protected void initChannel(Channel channel) {
-            channel.pipeline().addLast(new UdpCodecHandler());
-            channel.pipeline().addLast(new ClientRouteHandler());
-        }
-    }
-
-
 }

net/src/main/java/com/zfoo/net/core/udp/UdpServer.java

@@ -18,7 +18,6 @@ import com.zfoo.net.handler.codec.udp.UdpCodecHandler;
 import com.zfoo.util.net.HostAndPort;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.Channel;
-import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.epoll.Epoll;
 import io.netty.channel.epoll.EpollDatagramChannel;
@@ -26,6 +25,7 @@ import io.netty.channel.epoll.EpollEventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.nio.NioDatagramChannel;
 import io.netty.util.concurrent.DefaultThreadFactory;
+import org.apache.curator.shaded.com.google.common.base.MoreObjects;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -33,12 +33,19 @@ import org.slf4j.LoggerFactory;
  * @author godotg
  * @version 3.0
  */
-public class UdpServer extends AbstractServer {
+public class UdpServer extends AbstractServer<Channel> {
 
     private static final Logger logger = LoggerFactory.getLogger(UdpServer.class);
 
+    private final ServerRouteHandler serverRouteHandler;
+
     public UdpServer(HostAndPort host) {
+        this(host, null);
+    }
+
+    public UdpServer(HostAndPort host, ServerRouteHandler serverRouteHandler) {
         super(host);
+        this.serverRouteHandler = MoreObjects.firstNonNull(serverRouteHandler, new ServerRouteHandler());
     }
 
     @Override
@@ -54,7 +61,7 @@ public class UdpServer extends AbstractServer {
         bootstrap.group(workerGroup)
                 .channel(Epoll.isAvailable() ? EpollDatagramChannel.class : NioDatagramChannel.class)
                 .option(ChannelOption.SO_BROADCAST, true)
-                .handler(channelChannelInitializer());
+                .handler(this);
 
         // 异步
         channelFuture = bootstrap.bind(hostAddress, port);
@@ -67,16 +74,8 @@ public class UdpServer extends AbstractServer {
     }
 
     @Override
-    public ChannelInitializer<Channel> channelChannelInitializer() {
-        return new ChannelHandlerInitializer();
-    }
-
-
-    private static class ChannelHandlerInitializer extends ChannelInitializer<Channel> {
-        @Override
-        protected void initChannel(Channel channel) {
-            channel.pipeline().addLast(new UdpCodecHandler());
-            channel.pipeline().addLast(new ServerRouteHandler());
-        }
+    protected void initChannel(Channel channel) {
+        channel.pipeline().addLast(new UdpCodecHandler());
+        channel.pipeline().addLast(serverRouteHandler);
     }
 }

net/src/main/java/com/zfoo/net/core/websocket/WebsocketClient.java

@@ -18,44 +18,43 @@ import com.zfoo.net.handler.ClientRouteHandler;
 import com.zfoo.net.handler.codec.websocket.WebSocketCodecHandler;
 import com.zfoo.protocol.util.IOUtils;
 import com.zfoo.util.net.HostAndPort;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelInitializer;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.codec.http.HttpClientCodec;
 import io.netty.handler.codec.http.HttpObjectAggregator;
 import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolConfig;
 import io.netty.handler.codec.http.websocketx.WebSocketClientProtocolHandler;
 import io.netty.handler.stream.ChunkedWriteHandler;
+import org.apache.curator.shaded.com.google.common.base.MoreObjects;
+
 
 /**
  * @author godotg
  * @version 3.0
  */
-public class WebsocketClient extends AbstractClient {
+public class WebsocketClient extends AbstractClient<SocketChannel> {
 
-    private WebSocketClientProtocolConfig webSocketClientProtocolConfig;
+    private final WebSocketClientProtocolConfig webSocketClientProtocolConfig;
+
+    private final ClientRouteHandler clientRouteHandler;
 
     public WebsocketClient(HostAndPort host, WebSocketClientProtocolConfig webSocketClientProtocolConfig) {
+        this(host, webSocketClientProtocolConfig, null);
+    }
+
+    public WebsocketClient(HostAndPort host, WebSocketClientProtocolConfig webSocketClientProtocolConfig,
+                           ClientRouteHandler clientRouteHandler) {
         super(host);
         this.webSocketClientProtocolConfig = webSocketClientProtocolConfig;
+        this.clientRouteHandler = MoreObjects.firstNonNull(clientRouteHandler, new ClientRouteHandler());
     }
 
     @Override
-    public ChannelInitializer<? extends Channel> channelChannelInitializer() {
-        return new ChannelHandlerInitializer();
+    public void initChannel(SocketChannel channel) {
+        channel.pipeline().addLast(new HttpClientCodec(8 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB));
+        channel.pipeline().addLast(new HttpObjectAggregator(16 * IOUtils.BYTES_PER_MB));
+        channel.pipeline().addLast(new WebSocketClientProtocolHandler(webSocketClientProtocolConfig));
+        channel.pipeline().addLast(new ChunkedWriteHandler());
+        channel.pipeline().addLast(new WebSocketCodecHandler());
+        channel.pipeline().addLast(clientRouteHandler);
     }
-
-
-    public class ChannelHandlerInitializer extends ChannelInitializer<SocketChannel> {
-        @Override
-        public void initChannel(SocketChannel channel) {
-            channel.pipeline().addLast(new HttpClientCodec(8 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB));
-            channel.pipeline().addLast(new HttpObjectAggregator(16 * IOUtils.BYTES_PER_MB));
-            channel.pipeline().addLast(new WebSocketClientProtocolHandler(webSocketClientProtocolConfig));
-            channel.pipeline().addLast(new ChunkedWriteHandler());
-            channel.pipeline().addLast(new WebSocketCodecHandler());
-            channel.pipeline().addLast(new ClientRouteHandler());
-        }
-    }
-
 }

net/src/main/java/com/zfoo/net/core/websocket/WebsocketServer.java

@@ -18,46 +18,45 @@ import com.zfoo.net.handler.ServerRouteHandler;
 import com.zfoo.net.handler.codec.websocket.WebSocketCodecHandler;
 import com.zfoo.protocol.util.IOUtils;
 import com.zfoo.util.net.HostAndPort;
-import io.netty.channel.ChannelInitializer;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.codec.http.HttpObjectAggregator;
 import io.netty.handler.codec.http.HttpServerCodec;
 import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
 import io.netty.handler.stream.ChunkedWriteHandler;
+import org.apache.curator.shaded.com.google.common.base.MoreObjects;
 
 /**
  * @author godotg
  * @version 3.0
  */
-public class WebsocketServer extends AbstractServer {
+public class WebsocketServer extends AbstractServer<SocketChannel> {
+
+    private final ServerRouteHandler serverRouteHandler;
 
     public WebsocketServer(HostAndPort host) {
+        this(host, null);
+    }
+
+    public WebsocketServer(HostAndPort host, ServerRouteHandler serverRouteHandler) {
         super(host);
+        this.serverRouteHandler = MoreObjects.firstNonNull(serverRouteHandler, new ServerRouteHandler());
+
     }
 
     @Override
-    public ChannelInitializer<SocketChannel> channelChannelInitializer() {
-        return new ChannelHandlerInitializer();
-    }
-
-
-    public static class ChannelHandlerInitializer extends ChannelInitializer<SocketChannel> {
-
-        @Override
-        public void initChannel(SocketChannel channel) {
-            // 编解码 http 请求
-            channel.pipeline().addLast(new HttpServerCodec(8 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB));
-            // 聚合解码 HttpRequest/HttpContent/LastHttpContent 到 FullHttpRequest
-            // 保证接收的 Http 请求的完整性
-            channel.pipeline().addLast(new HttpObjectAggregator(16 * IOUtils.BYTES_PER_MB));
-            // 处理其他的 WebSocketFrame
-            channel.pipeline().addLast(new WebSocketServerProtocolHandler("/websocket"));
-            // 写文件内容，支持异步发送大的码流，一般用于发送文件流
-            channel.pipeline().addLast(new ChunkedWriteHandler());
-            // 编解码WebSocketFrame二进制协议
-            channel.pipeline().addLast(new WebSocketCodecHandler());
-            channel.pipeline().addLast(new ServerRouteHandler());
-        }
+    public void initChannel(SocketChannel channel) {
+        // 编解码 http 请求
+        channel.pipeline().addLast(new HttpServerCodec(8 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB, 16 * IOUtils.BYTES_PER_KB));
+        // 聚合解码 HttpRequest/HttpContent/LastHttpContent 到 FullHttpRequest
+        // 保证接收的 Http 请求的完整性
+        channel.pipeline().addLast(new HttpObjectAggregator(16 * IOUtils.BYTES_PER_MB));
+        // 处理其他的 WebSocketFrame
+        channel.pipeline().addLast(new WebSocketServerProtocolHandler("/websocket"));
+        // 写文件内容，支持异步发送大的码流，一般用于发送文件流
+        channel.pipeline().addLast(new ChunkedWriteHandler());
+        // 编解码WebSocketFrame二进制协议
+        channel.pipeline().addLast(new WebSocketCodecHandler());
+        channel.pipeline().addLast(serverRouteHandler);
     }
 
 }

net/src/main/java/com/zfoo/net/handler/BaseRouteHandler.java

@@ -26,17 +26,28 @@ import io.netty.util.AttributeKey;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Objects;
+import java.util.function.Consumer;
+
 /**
  * @author godotg
  * @version 3.0
  */
 @ChannelHandler.Sharable
-public class BaseRouteHandler extends ChannelInboundHandlerAdapter {
+public abstract class BaseRouteHandler extends ChannelInboundHandlerAdapter {
 
     private static final Logger logger = LoggerFactory.getLogger(BaseRouteHandler.class);
 
     public static final AttributeKey<Session> SESSION_KEY = AttributeKey.valueOf("session");
 
+    protected final Consumer<Session> sessionActiveConsumer;
+    protected final Consumer<Session> sessionInactiveConsumer;
+
+    protected BaseRouteHandler(Consumer<Session> sessionActiveConsumer, Consumer<Session> sessionInactiveConsumer) {
+        this.sessionActiveConsumer = Objects.requireNonNull(sessionActiveConsumer);
+        this.sessionInactiveConsumer = Objects.requireNonNull(sessionInactiveConsumer);
+    }
+
     public static Session initChannel(Channel channel) {
         var sessionAttr = channel.attr(SESSION_KEY);
         var session = new Session(channel);
@@ -48,6 +59,22 @@ public class BaseRouteHandler extends ChannelInboundHandlerAdapter {
         return session;
     }
 
+    protected void onSessionActive(Session session) {
+        try {
+            this.sessionActiveConsumer.accept(session);
+        } catch (Throwable ignored) {
+
+        }
+    }
+
+    protected void onSessionInavtive(Session session) {
+        try {
+            this.sessionInactiveConsumer.accept(session);
+        } catch (Throwable ignored) {
+
+        }
+    }
+
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) {
         var session = SessionUtils.getSession(ctx);

net/src/main/java/com/zfoo/net/handler/ClientRouteHandler.java

@@ -15,13 +15,18 @@ package com.zfoo.net.handler;
 
 import com.zfoo.event.manager.EventBus;
 import com.zfoo.net.NetContext;
+import com.zfoo.net.core.event.ClientSessionActiveEvent;
 import com.zfoo.net.core.event.ClientSessionInactiveEvent;
+import com.zfoo.net.session.Session;
 import com.zfoo.net.util.SessionUtils;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
+import org.apache.curator.shaded.com.google.common.base.MoreObjects;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.function.Consumer;
+
 /**
  * @author godotg
  * @version 3.0
@@ -31,6 +36,17 @@ public class ClientRouteHandler extends BaseRouteHandler {
 
     private static final Logger logger = LoggerFactory.getLogger(ClientRouteHandler.class);
 
+    public ClientRouteHandler() {
+        this(null, null);
+    }
+
+    public ClientRouteHandler(Consumer<Session> sessionActiveConsumer, Consumer<Session> sessionInactiveConsumer) {
+        super(MoreObjects.firstNonNull(sessionActiveConsumer,
+                        (session) -> EventBus.submit(ClientSessionActiveEvent.valueOf(session)))
+                , MoreObjects.firstNonNull(sessionInactiveConsumer,
+                        (session) -> EventBus.submit(ClientSessionInactiveEvent.valueOf(session))));
+    }
+
     @Override
     public void channelActive(ChannelHandlerContext ctx) throws Exception {
         super.channelActive(ctx);
@@ -48,7 +64,7 @@ public class ClientRouteHandler extends BaseRouteHandler {
         }
 
         NetContext.getSessionManager().removeClientSession(session);
-        EventBus.submit(ClientSessionInactiveEvent.valueOf(session));
+        onSessionInavtive(session);
 
         // 如果是消费者inactive，还需要触发客户端消费者检查事件，以便重新连接
         if (session.getConsumerAttribute() != null) {

net/src/main/java/com/zfoo/net/handler/GatewayRouteHandler.java

@@ -37,6 +37,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.function.BiFunction;
+import java.util.function.Consumer;
 
 /**
  * @author godotg
@@ -50,6 +51,13 @@ public class GatewayRouteHandler extends ServerRouteHandler {
     private final BiFunction<Session, IPacket, Boolean> packetFilter;
 
     public GatewayRouteHandler(BiFunction<Session, IPacket, Boolean> packetFilter) {
+        this(packetFilter, null, null);
+    }
+
+    public GatewayRouteHandler(BiFunction<Session, IPacket, Boolean> packetFilter,
+                               Consumer<Session> sessionActiveConsumer,
+                               Consumer<Session> sessionInactiveConsumer) {
+        super(sessionActiveConsumer, sessionInactiveConsumer);
         this.packetFilter = packetFilter;
     }
 

net/src/main/java/com/zfoo/net/handler/ServerRouteHandler.java

@@ -15,13 +15,18 @@ package com.zfoo.net.handler;
 
 import com.zfoo.event.manager.EventBus;
 import com.zfoo.net.NetContext;
+import com.zfoo.net.core.event.ServerSessionActiveEvent;
 import com.zfoo.net.core.event.ServerSessionInactiveEvent;
+import com.zfoo.net.session.Session;
 import com.zfoo.net.util.SessionUtils;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
+import org.apache.curator.shaded.com.google.common.base.MoreObjects;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.function.Consumer;
+
 /**
  * @author godotg
  * @version 3.0
@@ -31,12 +36,24 @@ public class ServerRouteHandler extends BaseRouteHandler {
 
     private static final Logger logger = LoggerFactory.getLogger(ServerRouteHandler.class);
 
+    public ServerRouteHandler() {
+        this(null, null);
+    }
+
+    public ServerRouteHandler(Consumer<Session> sessionActiveConsumer, Consumer<Session> sessionInactiveConsumer) {
+        super(MoreObjects.firstNonNull(sessionActiveConsumer,
+                        (session) -> EventBus.submit(ServerSessionActiveEvent.valueOf(session)))
+                , MoreObjects.firstNonNull(sessionInactiveConsumer,
+                        (session) -> EventBus.submit(ServerSessionInactiveEvent.valueOf(session))));
+    }
+
     @Override
     public void channelActive(ChannelHandlerContext ctx) throws Exception {
         super.channelActive(ctx);
         var session = initChannel(ctx.channel());
         NetContext.getSessionManager().addServerSession(session);
         logger.info("server channel is active {}", SessionUtils.sessionInfo(ctx));
+        onSessionActive(session);
     }
 
     @Override
@@ -48,7 +65,7 @@ public class ServerRouteHandler extends BaseRouteHandler {
             return;
         }
         NetContext.getSessionManager().removeServerSession(session);
-        EventBus.submit(ServerSessionInactiveEvent.valueOf(session));
         logger.warn("server channel is inactive {}", SessionUtils.sessionSimpleInfo(ctx));
+        onSessionInavtive(session);
     }
 }