diff --git a/net/src/main/java/com/zfoo/net/core/broker/BrokerClientRouteHandler.java b/net/src/main/java/com/zfoo/net/core/broker/BrokerClientRouteHandler.java new file mode 100644 index 00000000..5194f873 --- /dev/null +++ b/net/src/main/java/com/zfoo/net/core/broker/BrokerClientRouteHandler.java @@ -0,0 +1,51 @@ +/* + * 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.broker; + +import com.zfoo.net.NetContext; +import com.zfoo.net.handler.BaseRouteHandler; +import com.zfoo.net.util.SessionUtils; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author jaysunxiao + */ +@ChannelHandler.Sharable +public class BrokerClientRouteHandler extends BaseRouteHandler { + + private static final Logger logger = LoggerFactory.getLogger(BrokerClientRouteHandler.class); + + @Override + public void channelActive(ChannelHandlerContext ctx) throws Exception { + super.channelActive(ctx); + + BrokerTcpClient.brokers.add(ctx.channel()); + var session = SessionUtils.getSession(ctx); + NetContext.getRouter().send(session, new BrokerRegisterAsk()); + logger.info("broker client activate in sid:[{}]", session.getSid()); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + super.channelInactive(ctx); + + var session = SessionUtils.getSession(ctx); + BrokerTcpClient.brokers.remove(ctx.channel()); + logger.warn("broker client inactivate in sid:[{}]", session.getSid()); + } + +} diff --git a/net/src/main/java/com/zfoo/net/core/broker/BrokerRegisterAsk.java b/net/src/main/java/com/zfoo/net/core/broker/BrokerRegisterAsk.java new file mode 100644 index 00000000..9e51bc66 --- /dev/null +++ b/net/src/main/java/com/zfoo/net/core/broker/BrokerRegisterAsk.java @@ -0,0 +1,44 @@ +/* + * 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.broker; + +import com.zfoo.protocol.anno.Protocol; + +/** + * @author jaysunxiao + */ +@Protocol(id = 150) +public class BrokerRegisterAsk { + + // 不带gpu的服务器 + public static final int HOME = 1; + // 带了gpu的服务器 + public static final int HOME_GPU = 2; + + private int brokerType; + + public BrokerRegisterAsk() { + } + + public BrokerRegisterAsk(int brokerType) { + this.brokerType = brokerType; + } + + public int getBrokerType() { + return brokerType; + } + + public void setBrokerType(int brokerType) { + this.brokerType = brokerType; + } +} diff --git a/net/src/main/java/com/zfoo/net/core/broker/BrokerServerRouteHandler.java b/net/src/main/java/com/zfoo/net/core/broker/BrokerServerRouteHandler.java new file mode 100644 index 00000000..e07d0649 --- /dev/null +++ b/net/src/main/java/com/zfoo/net/core/broker/BrokerServerRouteHandler.java @@ -0,0 +1,59 @@ +/* + * 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.broker; + +import com.zfoo.net.NetContext; +import com.zfoo.net.handler.BaseRouteHandler; +import com.zfoo.net.packet.DecodedPacketInfo; +import com.zfoo.net.util.SessionUtils; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author jaysunxiao + */ +@ChannelHandler.Sharable +public class BrokerServerRouteHandler extends BaseRouteHandler { + + private static final Logger logger = LoggerFactory.getLogger(BrokerServerRouteHandler.class); + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) { + var session = SessionUtils.getSession(ctx); + if (session == null) { + return; + } + DecodedPacketInfo decodedPacketInfo = (DecodedPacketInfo) msg; + if (decodedPacketInfo.getPacket() instanceof BrokerRegisterAsk) { + BrokerTcpServer.brokers.add(ctx.channel()); + return; + } + NetContext.getRouter().receive(session, decodedPacketInfo.getPacket(), decodedPacketInfo.getAttachment()); + } + + @Override + public void channelActive(ChannelHandlerContext ctx) throws Exception { + super.channelActive(ctx); + logger.info("broker server channel is active {}", SessionUtils.sessionInfo(ctx)); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + super.channelInactive(ctx); + BrokerTcpServer.brokers.remove(ctx.channel()); + logger.warn("broker server channel is inactive {}", SessionUtils.sessionSimpleInfo(ctx)); + } +} diff --git a/net/src/main/java/com/zfoo/net/core/broker/BrokerTcpClient.java b/net/src/main/java/com/zfoo/net/core/broker/BrokerTcpClient.java new file mode 100644 index 00000000..66966076 --- /dev/null +++ b/net/src/main/java/com/zfoo/net/core/broker/BrokerTcpClient.java @@ -0,0 +1,46 @@ +/* + * 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.broker; + +import com.zfoo.net.core.AbstractClient; +import com.zfoo.net.core.HostAndPort; +import com.zfoo.net.handler.codec.tcp.TcpCodecHandler; +import com.zfoo.net.handler.idle.ClientIdleHandler; +import io.netty.channel.Channel; +import io.netty.channel.socket.SocketChannel; +import io.netty.handler.timeout.IdleStateHandler; + +import java.util.concurrent.CopyOnWriteArrayList; + +/** + * @author jaysunxiao + */ +public class BrokerTcpClient extends AbstractClient { + + public static final CopyOnWriteArrayList brokers = new CopyOnWriteArrayList<>(); + + public BrokerTcpClient(HostAndPort host) { + super(host); + } + + @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 BrokerClientRouteHandler()); + } +} diff --git a/net/src/main/java/com/zfoo/net/core/broker/BrokerTcpServer.java b/net/src/main/java/com/zfoo/net/core/broker/BrokerTcpServer.java new file mode 100644 index 00000000..b2d53e02 --- /dev/null +++ b/net/src/main/java/com/zfoo/net/core/broker/BrokerTcpServer.java @@ -0,0 +1,44 @@ +/* + * 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.broker; + +import com.zfoo.net.core.AbstractServer; +import com.zfoo.net.core.HostAndPort; +import com.zfoo.net.handler.codec.tcp.TcpCodecHandler; +import com.zfoo.net.handler.idle.ServerIdleHandler; +import io.netty.channel.Channel; +import io.netty.channel.socket.SocketChannel; +import io.netty.handler.timeout.IdleStateHandler; + +import java.util.concurrent.CopyOnWriteArrayList; + +/** + * @author jaysunxiao + */ +public class BrokerTcpServer extends AbstractServer { + + public static final CopyOnWriteArrayList brokers = new CopyOnWriteArrayList<>(); + + public BrokerTcpServer(HostAndPort host) { + super(host); + } + + @Override + protected void initChannel(SocketChannel channel) throws Exception { + channel.pipeline().addLast(new IdleStateHandler(0, 0, 180)); + channel.pipeline().addLast(new ServerIdleHandler()); + channel.pipeline().addLast(new TcpCodecHandler()); + channel.pipeline().addLast(new BrokerServerRouteHandler()); + } +}