opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-28 20:09:14 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #7493] Introduce a new event NettyEventType.ACTIVE (#7494)

Browse Source 
* [ISSUE #7493] Introduce a new event NettyEventType.ACTIVE for ChannelEventListener

* introduce a new event NettyEventType.ACTIVE,

* implement channelActive interface for NettyRemotingClient#NettyConnectManageHandler

* add onChannelActive for ChannelEventListener interface.

* Move send heartbeat to onChannelActive
This commit is contained in:
Zhouxiang Zhan
2023-10-24 10:29:43 +08:00
committed by GitHub
parent 8f020b397a
commit 2043dd5034
10 changed files with 59 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/client/ClientHousekeepingService.java
+5
View File
@@ -87,4 +87,9 @@ public class ClientHousekeepingService implements ChannelEventListener {
this.brokerController.getConsumerManager().doChannelCloseEvent(remoteAddr, channel);
this.brokerController.getBrokerStatsManager().incChannelIdleNum();
}
@Override
public void onChannelActive(String remoteAddr, Channel channel) {
}
}
client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
+12 -8
View File
@@ -159,14 +159,6 @@ public class MQClientInstance {
private final ConcurrentMap<String, HashMap<Long, String>> brokerAddrTable = MQClientInstance.this.brokerAddrTable;
@Override
public void onChannelConnect(String remoteAddr, Channel channel) {
for (Map.Entry<String, HashMap<Long, String>> addressEntry : brokerAddrTable.entrySet()) {
for (String address : addressEntry.getValue().values()) {
if (address.equals(remoteAddr)) {
sendHeartbeatToAllBrokerWithLockV2(false);
break;
}
}
}
}
@Override
@@ -180,6 +172,18 @@ public class MQClientInstance {
@Override
public void onChannelIdle(String remoteAddr, Channel channel) {
}
@Override
public void onChannelActive(String remoteAddr, Channel channel) {
for (Map.Entry<String, HashMap<Long, String>> addressEntry : brokerAddrTable.entrySet()) {
for (String address : addressEntry.getValue().values()) {
if (address.equals(remoteAddr)) {
sendHeartbeatToAllBrokerWithLockV2(false);
break;
}
}
}
}
};
} else {
channelEventListener = null;
container/src/main/java/org/apache/rocketmq/container/ContainerClientHouseKeepingService.java
+10 -1
View File
@@ -49,6 +49,11 @@ public class ContainerClientHouseKeepingService implements ChannelEventListener
onChannelOperation(CallbackCode.IDLE, remoteAddr, channel);
}
@Override
public void onChannelActive(String remoteAddr, Channel channel) {
onChannelOperation(CallbackCode.ACTIVE, remoteAddr, channel);
}
private void onChannelOperation(CallbackCode callbackCode, String remoteAddr, Channel channel) {
Collection<InnerBrokerController> masterBrokers = this.brokerContainer.getMasterBrokers();
Collection<InnerSalveBrokerController> slaveBrokers = this.brokerContainer.getSlaveBrokers();
@@ -103,6 +108,10 @@ public class ContainerClientHouseKeepingService implements ChannelEventListener
/**
* onChannelIdle
*/
IDLE
IDLE,
/**
* onChannelActive
*/
ACTIVE
}
}
controller/src/main/java/org/apache/rocketmq/controller/BrokerHousekeepingService.java
+5
View File
@@ -48,4 +48,9 @@ public class BrokerHousekeepingService implements ChannelEventListener {
public void onChannelIdle(String remoteAddr, Channel channel) {
this.controllerManager.getHeartbeatManager().onBrokerChannelClose(channel);
}
@Override
public void onChannelActive(String remoteAddr, Channel channel) {
}
}
namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/BrokerHousekeepingService.java
+5
View File
@@ -46,4 +46,9 @@ public class BrokerHousekeepingService implements ChannelEventListener {
public void onChannelIdle(String remoteAddr, Channel channel) {
this.namesrvController.getRouteInfoManager().onChannelDestroy(channel);
}
@Override
public void onChannelActive(String remoteAddr, Channel channel) {
}
}
proxy/src/main/java/org/apache/rocketmq/proxy/remoting/ClientHousekeepingService.java
+4
View File
@@ -49,5 +49,9 @@ public class ClientHousekeepingService implements ChannelEventListener {
this.clientManagerActivity.doChannelCloseEvent(remoteAddr, channel);
}
@Override
public void onChannelActive(String remoteAddr, Channel channel) {
}
}
remoting/src/main/java/org/apache/rocketmq/remoting/ChannelEventListener.java
+2
View File
@@ -26,4 +26,6 @@ public interface ChannelEventListener {
void onChannelException(final String remoteAddr, final Channel channel);
void onChannelIdle(final String remoteAddr, final Channel channel);
void onChannelActive(final String remoteAddr, final Channel channel);
}
remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyEventType.java
+2 -1
View File
@@ -20,5 +20,6 @@ public enum NettyEventType {
CONNECT,
CLOSE,
IDLE,
EXCEPTION
EXCEPTION,
ACTIVE
}
remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java
+3
View File
@@ -701,6 +701,9 @@ public abstract class NettyRemotingAbstract {
case EXCEPTION:
listener.onChannelException(event.getRemoteAddr(), event.getChannel());
break;
case ACTIVE:
listener.onChannelActive(event.getRemoteAddr(), event.getChannel());
break;
default:
break;
remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java
+11
View File
@@ -1106,6 +1106,17 @@ public class NettyRemotingClient extends NettyRemotingAbstract implements Remoti
}
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
LOGGER.info("NETTY CLIENT PIPELINE: ACTIVE, {}", remoteAddress);
super.channelActive(ctx);
if (NettyRemotingClient.this.channelEventListener != null) {
NettyRemotingClient.this.putNettyEvent(new NettyEvent(NettyEventType.ACTIVE, remoteAddress, ctx.channel()));
}
}
@Override
public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());