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

[ISSUE #7833] Fix invokeImpl() in RemotingAbstract

Browse Source
This commit is contained in:
guyinyou
2024-02-19 17:33:01 +08:00
committed by GitHub
parent 1b46d189db
commit 7dc0e5aed8
2 changed files with 8 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java
+1 -7
View File
@@ -496,13 +496,7 @@ public abstract class NettyRemotingAbstract {
public CompletableFuture<ResponseFuture> invokeImpl(final Channel channel, final RemotingCommand request,
final long timeoutMillis) {
String channelRemoteAddr = RemotingHelper.parseChannelRemoteAddr(channel);
doBeforeRpcHooks(channelRemoteAddr, request);
return invoke0(channel, request, timeoutMillis).whenComplete((v, t) -> {
if (t == null) {
doAfterRpcHooks(channelRemoteAddr, request, v.getResponseCommand());
}
});
return invoke0(channel, request, timeoutMillis);
}
protected CompletableFuture<ResponseFuture> invoke0(final Channel channel, final RemotingCommand request,
remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java
+7
View File
@@ -804,6 +804,9 @@ public class NettyRemotingClient extends NettyRemotingAbstract implements Remoti
public CompletableFuture<ResponseFuture> invokeImpl(final Channel channel, final RemotingCommand request,
final long timeoutMillis) {
Stopwatch stopwatch = Stopwatch.createStarted();
String channelRemoteAddr = RemotingHelper.parseChannelRemoteAddr(channel);
doBeforeRpcHooks(channelRemoteAddr, request);
return super.invokeImpl(channel, request, timeoutMillis).thenCompose(responseFuture -> {
RemotingCommand response = responseFuture.getResponseCommand();
if (response.getCode() == ResponseCode.GO_AWAY) {
@@ -839,6 +842,10 @@ public class NettyRemotingClient extends NettyRemotingAbstract implements Remoti
}
}
return CompletableFuture.completedFuture(responseFuture);
}).whenComplete((v, t) -> {
if (t == null) {
doAfterRpcHooks(channelRemoteAddr, request, v.getResponseCommand());
}
});
}