From c04c3185e5891d806fb1d8b0862d97b06cd698ce Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=B0=98=E5=A4=AE?= Date: Tue, 17 May 2022 17:28:12 +0800 Subject: [PATCH] Rename ProxyOutResult to ProxyRelayResult --- .../proxy/grpc/v2/channel/GrpcChannelManager.java | 12 ++++++------ .../proxy/grpc/v2/channel/GrpcClientChannel.java | 6 +++--- .../proxy/grpc/v2/client/ClientActivity.java | 14 +++++++------- .../service/relay/ClusterProxyRelayService.java | 4 ++-- .../service/relay/LocalProxyRelayService.java | 6 +++--- .../rocketmq/proxy/service/relay/ProxyChannel.java | 4 ++-- .../{ProxyOutResult.java => ProxyRelayResult.java} | 4 ++-- .../proxy/service/relay/ProxyRelayService.java | 4 ++-- 8 files changed, 27 insertions(+), 27 deletions(-) rename proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/{ProxyOutResult.java => ProxyRelayResult.java} (93%) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/channel/GrpcChannelManager.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/channel/GrpcChannelManager.java index bb7f4b203f..262c83fd01 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/channel/GrpcChannelManager.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/channel/GrpcChannelManager.java @@ -33,7 +33,7 @@ import org.apache.rocketmq.common.protocol.ResponseCode; import org.apache.rocketmq.proxy.common.StartAndShutdown; import org.apache.rocketmq.proxy.config.ConfigurationManager; import org.apache.rocketmq.proxy.config.ProxyConfig; -import org.apache.rocketmq.proxy.service.relay.ProxyOutResult; +import org.apache.rocketmq.proxy.service.relay.ProxyRelayResult; import org.apache.rocketmq.proxy.service.relay.ProxyRelayService; public class GrpcChannelManager implements StartAndShutdown { @@ -89,13 +89,13 @@ public class GrpcChannelManager implements StartAndShutdown { return channelRef.get(); } - public String addResponseFuture(CompletableFuture> responseFuture) { + public String addResponseFuture(CompletableFuture> responseFuture) { String nonce = this.nextNonce(); this.resultNonceFutureMap.put(nonce, new ResultFuture<>(responseFuture)); return nonce; } - public CompletableFuture> getAndRemoveResponseFuture(String nonce) { + public CompletableFuture> getAndRemoveResponseFuture(String nonce) { ResultFuture resultFuture = this.resultNonceFutureMap.remove(nonce); if (resultFuture != null) { return resultFuture.future; @@ -120,7 +120,7 @@ public class GrpcChannelManager implements StartAndShutdown { if (System.currentTimeMillis() - resultFuture.createTime > timeOutMs) { resultFuture = this.resultNonceFutureMap.remove(nonce); if (resultFuture != null) { - resultFuture.future.complete(new ProxyOutResult<>(ResponseCode.SYSTEM_BUSY, "call remote timeout", null)); + resultFuture.future.complete(new ProxyRelayResult<>(ResponseCode.SYSTEM_BUSY, "call remote timeout", null)); } } } @@ -137,10 +137,10 @@ public class GrpcChannelManager implements StartAndShutdown { } protected static class ResultFuture { - public CompletableFuture> future; + public CompletableFuture> future; public long createTime = System.currentTimeMillis(); - public ResultFuture(CompletableFuture> future) { + public ResultFuture(CompletableFuture> future) { this.future = future; } } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/channel/GrpcClientChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/channel/GrpcClientChannel.java index 2ec136c5b0..eb3e8c9b47 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/channel/GrpcClientChannel.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/channel/GrpcClientChannel.java @@ -36,7 +36,7 @@ import org.apache.rocketmq.common.protocol.header.GetConsumerRunningInfoRequestH import org.apache.rocketmq.proxy.grpc.interceptor.InterceptorConstants; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter; import org.apache.rocketmq.proxy.service.relay.ProxyChannel; -import org.apache.rocketmq.proxy.service.relay.ProxyOutResult; +import org.apache.rocketmq.proxy.service.relay.ProxyRelayResult; import org.apache.rocketmq.proxy.service.relay.ProxyRelayService; import org.apache.rocketmq.proxy.service.transaction.TransactionId; import org.apache.rocketmq.remoting.common.RemotingUtil; @@ -154,7 +154,7 @@ public class GrpcClientChannel extends ProxyChannel { @Override protected CompletableFuture processGetConsumerRunningInfo(RemotingCommand command, GetConsumerRunningInfoRequestHeader header, - CompletableFuture> responseFuture) { + CompletableFuture> responseFuture) { if (!header.isJstackEnable()) { return CompletableFuture.completedFuture(null); } @@ -169,7 +169,7 @@ public class GrpcClientChannel extends ProxyChannel { @Override protected CompletableFuture processConsumeMessageDirectly(RemotingCommand command, ConsumeMessageDirectlyResultRequestHeader header, - MessageExt messageExt, CompletableFuture> responseFuture) { + MessageExt messageExt, CompletableFuture> responseFuture) { this.getTelemetryCommandStreamObserver().onNext(TelemetryCommand.newBuilder() .setVerifyMessageCommand(VerifyMessageCommand.newBuilder() .setNonce(this.grpcChannelManager.addResponseFuture(responseFuture)) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java index b707351ed6..854507370b 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/client/ClientActivity.java @@ -64,7 +64,7 @@ import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcProxyException; import org.apache.rocketmq.proxy.grpc.v2.common.ResponseBuilder; import org.apache.rocketmq.proxy.processor.MessagingProcessor; -import org.apache.rocketmq.proxy.service.relay.ProxyOutResult; +import org.apache.rocketmq.proxy.service.relay.ProxyRelayResult; import org.apache.rocketmq.remoting.protocol.LanguageCode; public class ClientActivity extends AbstractMessingActivity { @@ -247,17 +247,17 @@ public class ClientActivity extends AbstractMessingActivity { protected void reportThreadStackTrace(Context ctx, Status status, ThreadStackTrace request) { String nonce = request.getNonce(); String threadStack = request.getThreadStackTrace(); - CompletableFuture> responseFuture = this.grpcChannelManager.getAndRemoveResponseFuture(nonce); + CompletableFuture> responseFuture = this.grpcChannelManager.getAndRemoveResponseFuture(nonce); if (responseFuture != null) { try { if (status.getCode().equals(Code.OK)) { ConsumerRunningInfo runningInfo = new ConsumerRunningInfo(); runningInfo.setJstack(threadStack); - responseFuture.complete(new ProxyOutResult<>(ResponseCode.SUCCESS, "", runningInfo)); + responseFuture.complete(new ProxyRelayResult<>(ResponseCode.SUCCESS, "", runningInfo)); } else if (status.getCode().equals(Code.VERIFY_MESSAGE_FORBIDDEN)) { - responseFuture.complete(new ProxyOutResult<>(ResponseCode.NO_PERMISSION, "forbidden to verify message", null)); + responseFuture.complete(new ProxyRelayResult<>(ResponseCode.NO_PERMISSION, "forbidden to verify message", null)); } else { - responseFuture.complete(new ProxyOutResult<>(ResponseCode.SYSTEM_ERROR, "verify message failed", null)); + responseFuture.complete(new ProxyRelayResult<>(ResponseCode.SYSTEM_ERROR, "verify message failed", null)); } } catch (Throwable t) { responseFuture.completeExceptionally(t); @@ -267,11 +267,11 @@ public class ClientActivity extends AbstractMessingActivity { protected void reportVerifyMessageResult(Context ctx, Status status, VerifyMessageResult request) { String nonce = request.getNonce(); - CompletableFuture> responseFuture = this.grpcChannelManager.getAndRemoveResponseFuture(nonce); + CompletableFuture> responseFuture = this.grpcChannelManager.getAndRemoveResponseFuture(nonce); if (responseFuture != null) { try { ConsumeMessageDirectlyResult result = this.buildConsumeMessageDirectlyResult(status, request); - responseFuture.complete(new ProxyOutResult<>(ResponseCode.SUCCESS, "", result)); + responseFuture.complete(new ProxyRelayResult<>(ResponseCode.SUCCESS, "", result)); } catch (Throwable t) { responseFuture.completeExceptionally(t); } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ClusterProxyRelayService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ClusterProxyRelayService.java index 9b1bba2c79..4db1a84540 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ClusterProxyRelayService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ClusterProxyRelayService.java @@ -29,12 +29,12 @@ import org.apache.rocketmq.remoting.protocol.RemotingCommand; public class ClusterProxyRelayService implements ProxyRelayService { @Override - public CompletableFuture> processGetConsumerRunningInfo(RemotingCommand command, + public CompletableFuture> processGetConsumerRunningInfo(RemotingCommand command, GetConsumerRunningInfoRequestHeader header) { return null; } - @Override public CompletableFuture> processConsumeMessageDirectly( + @Override public CompletableFuture> processConsumeMessageDirectly( RemotingCommand command, ConsumeMessageDirectlyResultRequestHeader header) { return null; } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/LocalProxyRelayService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/LocalProxyRelayService.java index e821fb60b4..8ecb6988d2 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/LocalProxyRelayService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/LocalProxyRelayService.java @@ -36,9 +36,9 @@ public class LocalProxyRelayService implements ProxyRelayService { } @Override - public CompletableFuture> processGetConsumerRunningInfo(RemotingCommand command, + public CompletableFuture> processGetConsumerRunningInfo(RemotingCommand command, GetConsumerRunningInfoRequestHeader header) { - CompletableFuture> future = new CompletableFuture<>(); + CompletableFuture> future = new CompletableFuture<>(); future.thenAccept(proxyOutResult -> { if (proxyOutResult.getCode() == ResponseCode.SUCCESS && proxyOutResult.getResult() != null) { ConsumerRunningInfo consumerRunningInfo = proxyOutResult.getResult(); @@ -59,7 +59,7 @@ public class LocalProxyRelayService implements ProxyRelayService { } @Override - public CompletableFuture> processConsumeMessageDirectly(RemotingCommand command, + public CompletableFuture> processConsumeMessageDirectly(RemotingCommand command, ConsumeMessageDirectlyResultRequestHeader header) { return null; } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyChannel.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyChannel.java index 1dcea7aa37..153ffb709f 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyChannel.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyChannel.java @@ -116,13 +116,13 @@ public abstract class ProxyChannel extends AbstractChannel { protected abstract CompletableFuture processGetConsumerRunningInfo( RemotingCommand command, GetConsumerRunningInfoRequestHeader header, - CompletableFuture> responseFuture); + CompletableFuture> responseFuture); protected abstract CompletableFuture processConsumeMessageDirectly( RemotingCommand command, ConsumeMessageDirectlyResultRequestHeader header, MessageExt messageExt, - CompletableFuture> responseFuture); + CompletableFuture> responseFuture); @Override public ChannelConfig config() { diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyOutResult.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyRelayResult.java similarity index 93% rename from proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyOutResult.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyRelayResult.java index 985481a17f..95b98d4d6b 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyOutResult.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyRelayResult.java @@ -17,12 +17,12 @@ package org.apache.rocketmq.proxy.service.relay; -public class ProxyOutResult { +public class ProxyRelayResult { private int code; private String remark; private T result; - public ProxyOutResult(int code, String remark, T result) { + public ProxyRelayResult(int code, String remark, T result) { this.code = code; this.remark = remark; this.result = result; diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyRelayService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyRelayService.java index 9e905d3f5b..1e800206a3 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyRelayService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/relay/ProxyRelayService.java @@ -25,12 +25,12 @@ import org.apache.rocketmq.remoting.protocol.RemotingCommand; public interface ProxyRelayService { - CompletableFuture> processGetConsumerRunningInfo( + CompletableFuture> processGetConsumerRunningInfo( RemotingCommand command, GetConsumerRunningInfoRequestHeader header ); - CompletableFuture> processConsumeMessageDirectly( + CompletableFuture> processConsumeMessageDirectly( RemotingCommand command, ConsumeMessageDirectlyResultRequestHeader header );