diff --git a/net/src/main/java/com/zfoo/net/core/AbstractClient.java b/net/src/main/java/com/zfoo/net/core/AbstractClient.java index a7cfb82e..81228f68 100644 --- a/net/src/main/java/com/zfoo/net/core/AbstractClient.java +++ b/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 extends ChannelInitializer 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 channelChannelInitializer(); - @Override public synchronized Session start() { - return doStart(channelChannelInitializer()); + return doStart(); } - private synchronized Session doStart(ChannelInitializer 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(); diff --git a/net/src/main/java/com/zfoo/net/core/AbstractServer.java b/net/src/main/java/com/zfoo/net/core/AbstractServer.java index 885d5427..480b5c54 100644 --- a/net/src/main/java/com/zfoo/net/core/AbstractServer.java +++ b/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 extends ChannelInitializer implements IServer { private static final Logger logger = LoggerFactory.getLogger(AbstractServer.class); // 所有的服务器都可以在这个列表中取到 - protected static final List allServers = new ArrayList<>(1); + protected static final List> 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 channelChannelInitializer(); - @Override public void start() { - doStart(channelChannelInitializer()); + doStart(); } - protected synchronized void doStart(ChannelInitializer 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(); // 等待服务端监听端口关闭 diff --git a/net/src/main/java/com/zfoo/net/core/event/ClientSessionActiveEvent.java b/net/src/main/java/com/zfoo/net/core/event/ClientSessionActiveEvent.java new file mode 100644 index 00000000..eb4d1980 --- /dev/null +++ b/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; + } +} diff --git a/net/src/main/java/com/zfoo/net/core/event/ServerSessionActiveEvent.java b/net/src/main/java/com/zfoo/net/core/event/ServerSessionActiveEvent.java new file mode 100644 index 00000000..a4f4603a --- /dev/null +++ b/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; + } +} diff --git a/net/src/main/java/com/zfoo/net/core/gateway/GatewayServer.java b/net/src/main/java/com/zfoo/net/core/gateway/GatewayServer.java index 11692b45..d6d47bba 100644 --- a/net/src/main/java/com/zfoo/net/core/gateway/GatewayServer.java +++ b/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 { - private BiFunction packetFilter; + private final GatewayRouteHandler gatewayRouteHandler; public GatewayServer(HostAndPort host, @Nullable BiFunction packetFilter) { + this(host, packetFilter, null); + } + + public GatewayServer(HostAndPort host, + @Nullable BiFunction packetFilter, + GatewayRouteHandler gatewayRouteHandler) { super(host); - this.packetFilter = packetFilter; + this.gatewayRouteHandler = MoreObjects.firstNonNull(gatewayRouteHandler, new GatewayRouteHandler(packetFilter)); } @Override - public ChannelInitializer channelChannelInitializer() { - return new ChannelHandlerInitializer(packetFilter); - } - - - private static class ChannelHandlerInitializer extends ChannelInitializer { - - private BiFunction packetFilter; - - public ChannelHandlerInitializer(BiFunction 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); } } diff --git a/net/src/main/java/com/zfoo/net/core/gateway/WebsocketGatewayServer.java b/net/src/main/java/com/zfoo/net/core/gateway/WebsocketGatewayServer.java index 56913cc0..307c224a 100644 --- a/net/src/main/java/com/zfoo/net/core/gateway/WebsocketGatewayServer.java +++ b/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 { - private BiFunction packetFilter; + private final GatewayRouteHandler gatewayRouteHandler; public WebsocketGatewayServer(HostAndPort host, @Nullable BiFunction packetFilter) { - super(host); - this.packetFilter = packetFilter; + this(host, packetFilter, null); } + public WebsocketGatewayServer(HostAndPort host, + @Nullable BiFunction packetFilter, + GatewayRouteHandler gatewayRouteHandler) { + super(host); + this.gatewayRouteHandler = MoreObjects.firstNonNull(gatewayRouteHandler, new GatewayRouteHandler(packetFilter)); + } + + @Override - public ChannelInitializer 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 { - - private BiFunction packetFilter; - - public ChannelHandlerInitializer(BiFunction 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); } } diff --git a/net/src/main/java/com/zfoo/net/core/gateway/WebsocketSslGatewayServer.java b/net/src/main/java/com/zfoo/net/core/gateway/WebsocketSslGatewayServer.java index 1124ba66..3d1e5dfe 100644 --- a/net/src/main/java/com/zfoo/net/core/gateway/WebsocketSslGatewayServer.java +++ b/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 { private static final Logger logger = LoggerFactory.getLogger(WebsocketSslGatewayServer.class); private SslContext sslContext; - private BiFunction packetFilter; + private final GatewayRouteHandler gatewayRouteHandler; - public WebsocketSslGatewayServer(HostAndPort host, InputStream pem, InputStream key, BiFunction packetFilter) { + public WebsocketSslGatewayServer(HostAndPort host, + InputStream pem, + InputStream key, + BiFunction packetFilter) { + this(host, pem, key, packetFilter, null); + } + + public WebsocketSslGatewayServer(HostAndPort host, + InputStream pem, + InputStream key, + BiFunction 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 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 { - - private SslContext sslContext; - private BiFunction packetFilter; - - public ChannelHandlerInitializer(SslContext sslContext, BiFunction 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); } } diff --git a/net/src/main/java/com/zfoo/net/core/http/HttpServer.java b/net/src/main/java/com/zfoo/net/core/http/HttpServer.java index 2a5d70bc..2021459d 100644 --- a/net/src/main/java/com/zfoo/net/core/http/HttpServer.java +++ b/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 { /** * http的地址解析器 */ - private Function uriResolver; + private final Function uriResolver; + + private final ServerRouteHandler serverRouteHandler; public HttpServer(HostAndPort host, Function uriResolver) { + this(host, uriResolver, null); + } + + + public HttpServer(HostAndPort host, + Function uriResolver, + ServerRouteHandler serverRouteHandler) { super(host); this.uriResolver = uriResolver; + this.serverRouteHandler = MoreObjects.firstNonNull(serverRouteHandler, new ServerRouteHandler()); } @Override - public ChannelInitializer channelChannelInitializer() { - return new ChannelHandlerInitializer(); - } - - - private class ChannelHandlerInitializer extends ChannelInitializer { - @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); } } diff --git a/net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufGatewayServer.java b/net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufGatewayServer.java index 47683fa7..6eb5ddf5 100644 --- a/net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufGatewayServer.java +++ b/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 { - - private final BiFunction packetFilter; + private final GatewayRouteHandler gatewayRouteHandler; public JProtobufGatewayServer(HostAndPort host, @Nullable BiFunction packetFilter) { + this(host, packetFilter, null); + } + + + public JProtobufGatewayServer(HostAndPort host, + @Nullable BiFunction packetFilter, + GatewayRouteHandler gatewayRouteHandler) { super(host); - this.packetFilter = packetFilter; + this.gatewayRouteHandler = MoreObjects.firstNonNull(gatewayRouteHandler, new GatewayRouteHandler(packetFilter)); } @Override - public ChannelInitializer channelChannelInitializer() { - return new ChannelHandlerInitializer(packetFilter); - } - - - private static class ChannelHandlerInitializer extends ChannelInitializer { - - private final BiFunction packetFilter; - - public ChannelHandlerInitializer(BiFunction 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); } } diff --git a/net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufTcpClient.java b/net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufTcpClient.java index ff792935..8c69aa9d 100644 --- a/net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufTcpClient.java +++ b/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 { + + 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 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 { - @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()); - } - } - - } diff --git a/net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufTcpServer.java b/net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufTcpServer.java index 1c19cd30..2a6b3b6f 100644 --- a/net/src/main/java/com/zfoo/net/core/jprotobuf/JProtobufTcpServer.java +++ b/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 { + + 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 channelChannelInitializer() { - return new ChannelHandlerInitializer(); - } - - - private static class ChannelHandlerInitializer extends ChannelInitializer { - @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); } } diff --git a/net/src/main/java/com/zfoo/net/core/json/JsonWebsocketClient.java b/net/src/main/java/com/zfoo/net/core/json/JsonWebsocketClient.java index abc58d44..95418db7 100644 --- a/net/src/main/java/com/zfoo/net/core/json/JsonWebsocketClient.java +++ b/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 { 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 channelChannelInitializer() { - return new ChannelHandlerInitializer(); - } - - - public class ChannelHandlerInitializer extends ChannelInitializer { - @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); } } diff --git a/net/src/main/java/com/zfoo/net/core/json/JsonWebsocketServer.java b/net/src/main/java/com/zfoo/net/core/json/JsonWebsocketServer.java index 70a587b3..82b0da7b 100644 --- a/net/src/main/java/com/zfoo/net/core/json/JsonWebsocketServer.java +++ b/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 { + + 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 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 { - - @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()); - } - } - } diff --git a/net/src/main/java/com/zfoo/net/core/tcp/TcpClient.java b/net/src/main/java/com/zfoo/net/core/tcp/TcpClient.java index fc31cc21..fd747778 100644 --- a/net/src/main/java/com/zfoo/net/core/tcp/TcpClient.java +++ b/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 { + + 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 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 { - @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()); - } - } - - } diff --git a/net/src/main/java/com/zfoo/net/core/tcp/TcpServer.java b/net/src/main/java/com/zfoo/net/core/tcp/TcpServer.java index 59f9bf31..2990736d 100644 --- a/net/src/main/java/com/zfoo/net/core/tcp/TcpServer.java +++ b/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 { + + 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 channelChannelInitializer() { - return new ChannelHandlerInitializer(); - } - - - private static class ChannelHandlerInitializer extends ChannelInitializer { - @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); } } diff --git a/net/src/main/java/com/zfoo/net/core/udp/UdpClient.java b/net/src/main/java/com/zfoo/net/core/udp/UdpClient.java index de683eba..7ac8353a 100644 --- a/net/src/main/java/com/zfoo/net/core/udp/UdpClient.java +++ b/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 { + + 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 channelChannelInitializer() { - return new ChannelHandlerInitializer(); + protected void initChannel(Channel channel) { + channel.pipeline().addLast(new UdpCodecHandler()); + channel.pipeline().addLast(clientRouteHandler); } - - - private static class ChannelHandlerInitializer extends ChannelInitializer { - @Override - protected void initChannel(Channel channel) { - channel.pipeline().addLast(new UdpCodecHandler()); - channel.pipeline().addLast(new ClientRouteHandler()); - } - } - - } diff --git a/net/src/main/java/com/zfoo/net/core/udp/UdpServer.java b/net/src/main/java/com/zfoo/net/core/udp/UdpServer.java index e65429d0..539ee7b2 100644 --- a/net/src/main/java/com/zfoo/net/core/udp/UdpServer.java +++ b/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 { 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 channelChannelInitializer() { - return new ChannelHandlerInitializer(); - } - - - private static class ChannelHandlerInitializer extends ChannelInitializer { - @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); } } diff --git a/net/src/main/java/com/zfoo/net/core/websocket/WebsocketClient.java b/net/src/main/java/com/zfoo/net/core/websocket/WebsocketClient.java index c10db65b..e7d538d4 100644 --- a/net/src/main/java/com/zfoo/net/core/websocket/WebsocketClient.java +++ b/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 { - 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 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 { - @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()); - } - } - } diff --git a/net/src/main/java/com/zfoo/net/core/websocket/WebsocketServer.java b/net/src/main/java/com/zfoo/net/core/websocket/WebsocketServer.java index 88cd2b8c..956f487c 100644 --- a/net/src/main/java/com/zfoo/net/core/websocket/WebsocketServer.java +++ b/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 { + + 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 channelChannelInitializer() { - return new ChannelHandlerInitializer(); - } - - - public static class ChannelHandlerInitializer extends ChannelInitializer { - - @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); } } diff --git a/net/src/main/java/com/zfoo/net/handler/BaseRouteHandler.java b/net/src/main/java/com/zfoo/net/handler/BaseRouteHandler.java index 1e59079d..ae6b8103 100644 --- a/net/src/main/java/com/zfoo/net/handler/BaseRouteHandler.java +++ b/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_KEY = AttributeKey.valueOf("session"); + protected final Consumer sessionActiveConsumer; + protected final Consumer sessionInactiveConsumer; + + protected BaseRouteHandler(Consumer sessionActiveConsumer, Consumer 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); diff --git a/net/src/main/java/com/zfoo/net/handler/ClientRouteHandler.java b/net/src/main/java/com/zfoo/net/handler/ClientRouteHandler.java index baf35096..04bf110c 100644 --- a/net/src/main/java/com/zfoo/net/handler/ClientRouteHandler.java +++ b/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 sessionActiveConsumer, Consumer 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) { diff --git a/net/src/main/java/com/zfoo/net/handler/GatewayRouteHandler.java b/net/src/main/java/com/zfoo/net/handler/GatewayRouteHandler.java index 1a2caf8f..8ab9b955 100644 --- a/net/src/main/java/com/zfoo/net/handler/GatewayRouteHandler.java +++ b/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 packetFilter; public GatewayRouteHandler(BiFunction packetFilter) { + this(packetFilter, null, null); + } + + public GatewayRouteHandler(BiFunction packetFilter, + Consumer sessionActiveConsumer, + Consumer sessionInactiveConsumer) { + super(sessionActiveConsumer, sessionInactiveConsumer); this.packetFilter = packetFilter; } diff --git a/net/src/main/java/com/zfoo/net/handler/ServerRouteHandler.java b/net/src/main/java/com/zfoo/net/handler/ServerRouteHandler.java index 7ce38b1a..cf4ff810 100644 --- a/net/src/main/java/com/zfoo/net/handler/ServerRouteHandler.java +++ b/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 sessionActiveConsumer, Consumer 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); } }