From 5c6fccacc805c63e26ba99a885a0a275b5f00cfb Mon Sep 17 00:00:00 2001 From: "Jixiang.jjx" Date: Wed, 16 Mar 2022 15:09:49 +0800 Subject: [PATCH] [ISSUE #3949] Do some renaming work. --- ...heLoader.java => AbstractCacheLoader.java} | 9 +++++---- .../AbstractForwardClient.java | 4 ++-- .../ConnectorManager.java} | 11 +++++----- .../DefaultForwardClient.java | 13 +++++++----- .../ForwardProducer.java | 4 ++-- .../ForwardReadConsumer.java | 4 ++-- .../ForwardWriteConsumer.java | 4 ++-- .../factory/AbstractClientFactory.java} | 14 ++++++------- .../factory/ForwardClientFactory.java | 8 ++++---- .../factory/MQClientFactory.java | 12 +++++------ .../factory/TransactionProducerFactory.java} | 16 +++++++-------- .../DoNothingClientRemotingProcessor.java | 2 +- .../ProxyClientRemotingProcessor.java | 8 ++++---- .../route/MessageQueueSelector.java | 2 +- .../route/MessageQueueWrapper.java | 2 +- .../route/SelectableMessageQueue.java | 2 +- .../route}/TopicRouteCache.java | 12 +++++------ .../route/TopicRouteHelper.java} | 20 ++++++++----------- .../route/TopicRouteWrapper.java | 2 +- .../transaction/TransactionId.java | 2 +- .../TransactionStateCheckRequest.java | 2 +- .../transaction/TransactionStateChecker.java | 2 +- .../grpc/service/ClusterGrpcService.java | 6 +++--- .../grpc/service/cluster/BaseService.java | 6 +++--- .../grpc/service/cluster/ConsumerService.java | 4 ++-- .../grpc/service/cluster/ProducerService.java | 6 +++--- .../grpc/service/cluster/RouteService.java | 12 +++++------ .../ForwardClientManagerTest.java | 6 +++--- .../grpc/service/cluster/BaseServiceTest.java | 14 ++++++------- .../service/cluster/ProducerServiceTest.java | 2 +- .../service/cluster/RouteServiceTest.java | 3 +-- 31 files changed, 106 insertions(+), 108 deletions(-) rename proxy/src/main/java/org/apache/rocketmq/proxy/common/{RetainCacheLoader.java => AbstractCacheLoader.java} (83%) rename proxy/src/main/java/org/apache/rocketmq/proxy/{client => connector}/AbstractForwardClient.java (94%) rename proxy/src/main/java/org/apache/rocketmq/proxy/{client/ForwardClientManager.java => connector/ConnectorManager.java} (86%) rename proxy/src/main/java/org/apache/rocketmq/proxy/{client => connector}/DefaultForwardClient.java (87%) rename proxy/src/main/java/org/apache/rocketmq/proxy/{client => connector}/ForwardProducer.java (96%) rename proxy/src/main/java/org/apache/rocketmq/proxy/{client => connector}/ForwardReadConsumer.java (95%) rename proxy/src/main/java/org/apache/rocketmq/proxy/{client => connector}/ForwardWriteConsumer.java (96%) rename proxy/src/main/java/org/apache/rocketmq/proxy/{client/factory/AbstractMQClientFactory.java => connector/factory/AbstractClientFactory.java} (86%) rename proxy/src/main/java/org/apache/rocketmq/proxy/{client => connector}/factory/ForwardClientFactory.java (89%) rename proxy/src/main/java/org/apache/rocketmq/proxy/{client => connector}/factory/MQClientFactory.java (76%) rename proxy/src/main/java/org/apache/rocketmq/proxy/{client/factory/TransactionalProducerFactory.java => connector/factory/TransactionProducerFactory.java} (71%) rename proxy/src/main/java/org/apache/rocketmq/proxy/{client => connector}/processor/DoNothingClientRemotingProcessor.java (96%) rename proxy/src/main/java/org/apache/rocketmq/proxy/{client => connector}/processor/ProxyClientRemotingProcessor.java (92%) rename proxy/src/main/java/org/apache/rocketmq/proxy/{client => connector}/route/MessageQueueSelector.java (99%) rename proxy/src/main/java/org/apache/rocketmq/proxy/{client => connector}/route/MessageQueueWrapper.java (97%) rename proxy/src/main/java/org/apache/rocketmq/proxy/{client => connector}/route/SelectableMessageQueue.java (97%) rename proxy/src/main/java/org/apache/rocketmq/proxy/{client => connector/route}/TopicRouteCache.java (93%) rename proxy/src/main/java/org/apache/rocketmq/proxy/{common/RocketMQHelper.java => connector/route/TopicRouteHelper.java} (72%) rename proxy/src/main/java/org/apache/rocketmq/proxy/{client => connector}/route/TopicRouteWrapper.java (98%) rename proxy/src/main/java/org/apache/rocketmq/proxy/{client => connector}/transaction/TransactionId.java (99%) rename proxy/src/main/java/org/apache/rocketmq/proxy/{client => connector}/transaction/TransactionStateCheckRequest.java (97%) rename proxy/src/main/java/org/apache/rocketmq/proxy/{client => connector}/transaction/TransactionStateChecker.java (94%) rename proxy/src/test/java/org/apache/rocketmq/proxy/{client => connector}/ForwardClientManagerTest.java (90%) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/common/RetainCacheLoader.java b/proxy/src/main/java/org/apache/rocketmq/proxy/common/AbstractCacheLoader.java similarity index 83% rename from proxy/src/main/java/org/apache/rocketmq/proxy/common/RetainCacheLoader.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/common/AbstractCacheLoader.java index 80d49ed8e3..581caffdbd 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/common/RetainCacheLoader.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/common/AbstractCacheLoader.java @@ -20,16 +20,17 @@ import com.google.common.cache.CacheLoader; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListenableFutureTask; import java.util.concurrent.ThreadPoolExecutor; +import javax.annotation.Nonnull; -public abstract class RetainCacheLoader extends CacheLoader { +public abstract class AbstractCacheLoader extends CacheLoader { private final ThreadPoolExecutor cacheRefreshExecutor; - public RetainCacheLoader(ThreadPoolExecutor cacheRefreshExecutor) { + public AbstractCacheLoader(ThreadPoolExecutor cacheRefreshExecutor) { this.cacheRefreshExecutor = cacheRefreshExecutor; } @Override - public ListenableFuture reload(K key, V oldValue) throws Exception { + public ListenableFuture reload(@Nonnull K key, @Nonnull V oldValue) throws Exception { ListenableFutureTask task = ListenableFutureTask.create(() -> { try { return getDirectly(key); @@ -43,7 +44,7 @@ public abstract class RetainCacheLoader extends CacheLoader { } @Override - public V load(K key) throws Exception { + public V load(@Nonnull K key) throws Exception { return getDirectly(key); } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/client/AbstractForwardClient.java b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/AbstractForwardClient.java similarity index 94% rename from proxy/src/main/java/org/apache/rocketmq/proxy/client/AbstractForwardClient.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/connector/AbstractForwardClient.java index 480554d6b8..0fa3c3746b 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/client/AbstractForwardClient.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/AbstractForwardClient.java @@ -14,11 +14,11 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.rocketmq.proxy.client; +package org.apache.rocketmq.proxy.connector; import java.util.concurrent.ThreadLocalRandom; import org.apache.rocketmq.client.impl.MQClientAPIExtImpl; -import org.apache.rocketmq.proxy.client.factory.ForwardClientFactory; +import org.apache.rocketmq.proxy.connector.factory.ForwardClientFactory; import org.apache.rocketmq.proxy.common.StartAndShutdown; public abstract class AbstractForwardClient implements StartAndShutdown { diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/client/ForwardClientManager.java b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/ConnectorManager.java similarity index 86% rename from proxy/src/main/java/org/apache/rocketmq/proxy/client/ForwardClientManager.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/connector/ConnectorManager.java index 85dc298a8b..01ec55fb62 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/client/ForwardClientManager.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/ConnectorManager.java @@ -14,13 +14,14 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.rocketmq.proxy.client; +package org.apache.rocketmq.proxy.connector; -import org.apache.rocketmq.proxy.client.factory.ForwardClientFactory; -import org.apache.rocketmq.proxy.client.transaction.TransactionStateChecker; +import org.apache.rocketmq.proxy.connector.factory.ForwardClientFactory; +import org.apache.rocketmq.proxy.connector.route.TopicRouteCache; +import org.apache.rocketmq.proxy.connector.transaction.TransactionStateChecker; import org.apache.rocketmq.proxy.common.AbstractStartAndShutdown; -public class ForwardClientManager extends AbstractStartAndShutdown { +public class ConnectorManager extends AbstractStartAndShutdown { private final ForwardClientFactory forwardClientFactory; private final DefaultForwardClient defaultForwardClient; private final ForwardProducer forwardProducer; @@ -29,7 +30,7 @@ public class ForwardClientManager extends AbstractStartAndShutdown { private final TopicRouteCache topicRouteCache; - public ForwardClientManager(TransactionStateChecker transactionStateChecker) { + public ConnectorManager(TransactionStateChecker transactionStateChecker) { this.forwardClientFactory = new ForwardClientFactory(transactionStateChecker); this.defaultForwardClient = new DefaultForwardClient(this.forwardClientFactory); this.forwardProducer = new ForwardProducer(this.forwardClientFactory); diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/client/DefaultForwardClient.java b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/DefaultForwardClient.java similarity index 87% rename from proxy/src/main/java/org/apache/rocketmq/proxy/client/DefaultForwardClient.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/connector/DefaultForwardClient.java index ede03992f9..1c298311f7 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/client/DefaultForwardClient.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/DefaultForwardClient.java @@ -14,7 +14,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.rocketmq.proxy.client; +package org.apache.rocketmq.proxy.connector; import java.util.List; import java.util.concurrent.CompletableFuture; @@ -22,15 +22,15 @@ import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.client.impl.MQClientAPIExtImpl; import org.apache.rocketmq.common.protocol.header.GetConsumerListByGroupRequestHeader; import org.apache.rocketmq.common.protocol.route.TopicRouteData; -import org.apache.rocketmq.proxy.client.factory.ForwardClientFactory; import org.apache.rocketmq.proxy.config.ConfigurationManager; +import org.apache.rocketmq.proxy.connector.factory.ForwardClientFactory; import org.apache.rocketmq.remoting.exception.RemotingException; public class DefaultForwardClient extends AbstractForwardClient { private static final String CID_PREFIX = "CID_RMQ_PROXY_DEFAULT_"; - public DefaultForwardClient(ForwardClientFactory clientFactory) { - super(clientFactory); + public DefaultForwardClient(ForwardClientFactory forwardClientFactory) { + super(forwardClientFactory); } @Override @@ -52,7 +52,10 @@ public class DefaultForwardClient extends AbstractForwardClient { } public CompletableFuture> getConsumerListByGroup( - String brokerAddr, GetConsumerListByGroupRequestHeader requestHeader, long timeoutMillis) { + String brokerAddr, + GetConsumerListByGroupRequestHeader requestHeader, + long timeoutMillis + ) { return getClient().getConsumerListByGroup(brokerAddr, requestHeader, timeoutMillis); } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/client/ForwardProducer.java b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/ForwardProducer.java similarity index 96% rename from proxy/src/main/java/org/apache/rocketmq/proxy/client/ForwardProducer.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/connector/ForwardProducer.java index 734af13455..cc0e5ba9ce 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/client/ForwardProducer.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/ForwardProducer.java @@ -14,7 +14,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.rocketmq.proxy.client; +package org.apache.rocketmq.proxy.connector; import java.util.concurrent.CompletableFuture; import org.apache.rocketmq.client.impl.MQClientAPIExtImpl; @@ -23,7 +23,7 @@ import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.common.protocol.header.ConsumerSendMsgBackRequestHeader; import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader; import org.apache.rocketmq.common.protocol.heartbeat.HeartbeatData; -import org.apache.rocketmq.proxy.client.factory.ForwardClientFactory; +import org.apache.rocketmq.proxy.connector.factory.ForwardClientFactory; import org.apache.rocketmq.proxy.config.ConfigurationManager; import org.apache.rocketmq.remoting.protocol.RemotingCommand; diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/client/ForwardReadConsumer.java b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/ForwardReadConsumer.java similarity index 95% rename from proxy/src/main/java/org/apache/rocketmq/proxy/client/ForwardReadConsumer.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/connector/ForwardReadConsumer.java index 37dce5a86a..0aa0b30d70 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/client/ForwardReadConsumer.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/ForwardReadConsumer.java @@ -14,7 +14,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.rocketmq.proxy.client; +package org.apache.rocketmq.proxy.connector; import java.util.concurrent.CompletableFuture; import org.apache.rocketmq.client.consumer.PopResult; @@ -22,7 +22,7 @@ import org.apache.rocketmq.client.consumer.PullResult; import org.apache.rocketmq.client.impl.MQClientAPIExtImpl; import org.apache.rocketmq.common.protocol.header.PopMessageRequestHeader; import org.apache.rocketmq.common.protocol.header.PullMessageRequestHeader; -import org.apache.rocketmq.proxy.client.factory.ForwardClientFactory; +import org.apache.rocketmq.proxy.connector.factory.ForwardClientFactory; import org.apache.rocketmq.proxy.config.ConfigurationManager; public class ForwardReadConsumer extends AbstractForwardClient { diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/client/ForwardWriteConsumer.java b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/ForwardWriteConsumer.java similarity index 96% rename from proxy/src/main/java/org/apache/rocketmq/proxy/client/ForwardWriteConsumer.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/connector/ForwardWriteConsumer.java index 446d9f023f..5d3f5c2035 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/client/ForwardWriteConsumer.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/ForwardWriteConsumer.java @@ -14,7 +14,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.rocketmq.proxy.client; +package org.apache.rocketmq.proxy.connector; import java.util.concurrent.CompletableFuture; import org.apache.rocketmq.client.consumer.AckResult; @@ -22,7 +22,7 @@ import org.apache.rocketmq.client.impl.MQClientAPIExtImpl; import org.apache.rocketmq.common.protocol.header.AckMessageRequestHeader; import org.apache.rocketmq.common.protocol.header.ChangeInvisibleTimeRequestHeader; import org.apache.rocketmq.common.protocol.header.UpdateConsumerOffsetRequestHeader; -import org.apache.rocketmq.proxy.client.factory.ForwardClientFactory; +import org.apache.rocketmq.proxy.connector.factory.ForwardClientFactory; import org.apache.rocketmq.proxy.config.ConfigurationManager; import org.apache.rocketmq.remoting.exception.RemotingException; diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/client/factory/AbstractMQClientFactory.java b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/factory/AbstractClientFactory.java similarity index 86% rename from proxy/src/main/java/org/apache/rocketmq/proxy/client/factory/AbstractMQClientFactory.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/connector/factory/AbstractClientFactory.java index fc28f8ba32..7e0519097b 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/client/factory/AbstractMQClientFactory.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/factory/AbstractClientFactory.java @@ -14,7 +14,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.rocketmq.proxy.client.factory; +package org.apache.rocketmq.proxy.connector.factory; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @@ -23,21 +23,21 @@ import org.apache.rocketmq.remoting.netty.NettyClientConfig; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -public abstract class AbstractMQClientFactory { - private static final Logger LOGGER = LoggerFactory.getLogger(AbstractMQClientFactory.class); +public abstract class AbstractClientFactory { + private static final Logger LOGGER = LoggerFactory.getLogger(AbstractClientFactory.class); protected Map cacheTable = new ConcurrentHashMap<>(); protected RPCHook rpcHook; - public AbstractMQClientFactory(RPCHook rpcHook) { + public AbstractClientFactory(RPCHook rpcHook) { this.rpcHook = rpcHook; } - abstract T newOne(String instanceName, RPCHook rpcHook, int bootstrapWorkerThreads) throws Throwable; + protected abstract T newOne(String instanceName, RPCHook rpcHook, int bootstrapWorkerThreads) throws Throwable; - abstract boolean tryStart(T t); + protected abstract boolean tryStart(T t); - abstract void shutdown(T t); + protected abstract void shutdown(T t); protected static NettyClientConfig createNettyClientConfig(int bootstrapWorkerThreads) { NettyClientConfig nettyClientConfig = new NettyClientConfig(); diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/client/factory/ForwardClientFactory.java b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/factory/ForwardClientFactory.java similarity index 89% rename from proxy/src/main/java/org/apache/rocketmq/proxy/client/factory/ForwardClientFactory.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/connector/factory/ForwardClientFactory.java index f27383c9f9..08b43acf2c 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/client/factory/ForwardClientFactory.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/factory/ForwardClientFactory.java @@ -14,13 +14,13 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.rocketmq.proxy.client.factory; +package org.apache.rocketmq.proxy.connector.factory; import org.apache.commons.lang3.StringUtils; import org.apache.rocketmq.client.ClientConfig; import org.apache.rocketmq.client.impl.MQClientAPIExtImpl; import org.apache.rocketmq.common.MixAll; -import org.apache.rocketmq.proxy.client.transaction.TransactionStateChecker; +import org.apache.rocketmq.proxy.connector.transaction.TransactionStateChecker; import org.apache.rocketmq.proxy.common.StartAndShutdown; import org.apache.rocketmq.proxy.config.ConfigurationManager; import org.apache.rocketmq.remoting.RPCHook; @@ -30,13 +30,13 @@ public class ForwardClientFactory implements StartAndShutdown { private RPCHook rpcHook = null; private final MQClientFactory mqClientFactory; - private final TransactionalProducerFactory transactionalProducerFactory; + private final TransactionProducerFactory transactionalProducerFactory; public ForwardClientFactory(TransactionStateChecker transactionStateChecker) { this.init(); this.mqClientFactory = new MQClientFactory(this.rpcHook); - this.transactionalProducerFactory = new TransactionalProducerFactory(this.rpcHook, transactionStateChecker); + this.transactionalProducerFactory = new TransactionProducerFactory(this.rpcHook, transactionStateChecker); } private void init() { diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/client/factory/MQClientFactory.java b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/factory/MQClientFactory.java similarity index 76% rename from proxy/src/main/java/org/apache/rocketmq/proxy/client/factory/MQClientFactory.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/connector/factory/MQClientFactory.java index d95667dea9..3b9a75f33b 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/client/factory/MQClientFactory.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/factory/MQClientFactory.java @@ -14,21 +14,21 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.rocketmq.proxy.client.factory; +package org.apache.rocketmq.proxy.connector.factory; import org.apache.rocketmq.client.ClientConfig; import org.apache.rocketmq.client.impl.MQClientAPIExtImpl; -import org.apache.rocketmq.proxy.client.processor.DoNothingClientRemotingProcessor; +import org.apache.rocketmq.proxy.connector.processor.DoNothingClientRemotingProcessor; import org.apache.rocketmq.remoting.RPCHook; -public class MQClientFactory extends AbstractMQClientFactory { +public class MQClientFactory extends AbstractClientFactory { public MQClientFactory(RPCHook rpcHook) { super(rpcHook); } @Override - MQClientAPIExtImpl newOne(String instanceName, RPCHook rpcHook, int bootstrapWorkerThreads) { + protected MQClientAPIExtImpl newOne(String instanceName, RPCHook rpcHook, int bootstrapWorkerThreads) { ClientConfig clientConfig = new ClientConfig(); clientConfig.setInstanceName(instanceName); @@ -41,13 +41,13 @@ public class MQClientFactory extends AbstractMQClientFactory } @Override - boolean tryStart(MQClientAPIExtImpl client) { + protected boolean tryStart(MQClientAPIExtImpl client) { client.start(); return true; } @Override - void shutdown(MQClientAPIExtImpl client) { + protected void shutdown(MQClientAPIExtImpl client) { client.shutdown(); } } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/client/factory/TransactionalProducerFactory.java b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/factory/TransactionProducerFactory.java similarity index 71% rename from proxy/src/main/java/org/apache/rocketmq/proxy/client/factory/TransactionalProducerFactory.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/connector/factory/TransactionProducerFactory.java index 6902929418..a97727c11d 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/client/factory/TransactionalProducerFactory.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/factory/TransactionProducerFactory.java @@ -14,24 +14,24 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.rocketmq.proxy.client.factory; +package org.apache.rocketmq.proxy.connector.factory; import org.apache.rocketmq.client.ClientConfig; import org.apache.rocketmq.client.impl.MQClientAPIExtImpl; -import org.apache.rocketmq.proxy.client.processor.ProxyClientRemotingProcessor; -import org.apache.rocketmq.proxy.client.transaction.TransactionStateChecker; +import org.apache.rocketmq.proxy.connector.processor.ProxyClientRemotingProcessor; +import org.apache.rocketmq.proxy.connector.transaction.TransactionStateChecker; import org.apache.rocketmq.remoting.RPCHook; -public class TransactionalProducerFactory extends AbstractMQClientFactory { +public class TransactionProducerFactory extends AbstractClientFactory { private final TransactionStateChecker transactionStateChecker; - public TransactionalProducerFactory(RPCHook rpcHook, TransactionStateChecker transactionStateChecker) { + public TransactionProducerFactory(RPCHook rpcHook, TransactionStateChecker transactionStateChecker) { super(rpcHook); this.transactionStateChecker = transactionStateChecker; } @Override - MQClientAPIExtImpl newOne(String instanceName, RPCHook rpcHook, int bootstrapWorkerThreads) { + public MQClientAPIExtImpl newOne(String instanceName, RPCHook rpcHook, int bootstrapWorkerThreads) { ClientConfig clientConfig = new ClientConfig(); clientConfig.setInstanceName(instanceName); @@ -44,13 +44,13 @@ public class TransactionalProducerFactory extends AbstractMQClientFactory { + protected abstract class AbstractTopicRouteCacheLoader extends AbstractCacheLoader { public AbstractTopicRouteCacheLoader() { super(cacheRefreshExecutor); @@ -124,7 +122,7 @@ public class TopicRouteCache { } return MessageQueueWrapper.WRAPPED_EMPTY_QUEUE; } catch (Exception e) { - if (RocketMQHelper.isTopicNotExistError(e)) { + if (TopicRouteHelper.isTopicNotExistError(e)) { return MessageQueueWrapper.WRAPPED_EMPTY_QUEUE; } throw e; diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/common/RocketMQHelper.java b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/route/TopicRouteHelper.java similarity index 72% rename from proxy/src/main/java/org/apache/rocketmq/proxy/common/RocketMQHelper.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/connector/route/TopicRouteHelper.java index a1488cf017..97b345d703 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/common/RocketMQHelper.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/route/TopicRouteHelper.java @@ -14,14 +14,14 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.rocketmq.proxy.common; +package org.apache.rocketmq.proxy.connector.route; import org.apache.rocketmq.client.common.ClientErrorCode; import org.apache.rocketmq.client.exception.MQBrokerException; import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.common.protocol.ResponseCode; -public class RocketMQHelper { +public class TopicRouteHelper { public static boolean isTopicNotExistError(Throwable e) { if (e instanceof MQBrokerException) { @@ -29,24 +29,20 @@ public class RocketMQHelper { return true; } } - if (e instanceof MQClientException) { - if (((MQClientException) e).getResponseCode() == ResponseCode.TOPIC_NOT_EXIST) { - return true; - } - if (((MQClientException) e).getResponseCode() == ClientErrorCode.NOT_FOUND_TOPIC_EXCEPTION) { + if (e instanceof MQClientException) { + int code = ((MQClientException) e).getResponseCode(); + if (code == ResponseCode.TOPIC_NOT_EXIST || code == ClientErrorCode.NOT_FOUND_TOPIC_EXCEPTION) { return true; } Throwable cause = e.getCause(); if (cause instanceof MQClientException) { - if (((MQClientException) cause).getResponseCode() == ResponseCode.TOPIC_NOT_EXIST) { - return true; - } - - return ((MQClientException) cause).getResponseCode() == ClientErrorCode.NOT_FOUND_TOPIC_EXCEPTION; + int causeCode = ((MQClientException) cause).getResponseCode(); + return causeCode == ResponseCode.TOPIC_NOT_EXIST || causeCode == ClientErrorCode.NOT_FOUND_TOPIC_EXCEPTION; } } + return false; } } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/client/route/TopicRouteWrapper.java b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/route/TopicRouteWrapper.java similarity index 98% rename from proxy/src/main/java/org/apache/rocketmq/proxy/client/route/TopicRouteWrapper.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/connector/route/TopicRouteWrapper.java index 9a3c69cbc5..a08bd3aa52 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/client/route/TopicRouteWrapper.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/route/TopicRouteWrapper.java @@ -14,7 +14,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.rocketmq.proxy.client.route; +package org.apache.rocketmq.proxy.connector.route; import java.util.HashMap; import java.util.List; diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/client/transaction/TransactionId.java b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/transaction/TransactionId.java similarity index 99% rename from proxy/src/main/java/org/apache/rocketmq/proxy/client/transaction/TransactionId.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/connector/transaction/TransactionId.java index 05da6904cd..63eb8b3f08 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/client/transaction/TransactionId.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/transaction/TransactionId.java @@ -14,7 +14,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.rocketmq.proxy.client.transaction; +package org.apache.rocketmq.proxy.connector.transaction; import java.net.InetAddress; import java.net.InetSocketAddress; diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/client/transaction/TransactionStateCheckRequest.java b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/transaction/TransactionStateCheckRequest.java similarity index 97% rename from proxy/src/main/java/org/apache/rocketmq/proxy/client/transaction/TransactionStateCheckRequest.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/connector/transaction/TransactionStateCheckRequest.java index 73cab80586..ad6b5ac0d6 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/client/transaction/TransactionStateCheckRequest.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/transaction/TransactionStateCheckRequest.java @@ -14,7 +14,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.rocketmq.proxy.client.transaction; +package org.apache.rocketmq.proxy.connector.transaction; import org.apache.rocketmq.common.message.MessageExt; diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/client/transaction/TransactionStateChecker.java b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/transaction/TransactionStateChecker.java similarity index 94% rename from proxy/src/main/java/org/apache/rocketmq/proxy/client/transaction/TransactionStateChecker.java rename to proxy/src/main/java/org/apache/rocketmq/proxy/connector/transaction/TransactionStateChecker.java index 6cea826cb2..506aa0154c 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/client/transaction/TransactionStateChecker.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/connector/transaction/TransactionStateChecker.java @@ -14,7 +14,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.rocketmq.proxy.client.transaction; +package org.apache.rocketmq.proxy.connector.transaction; public interface TransactionStateChecker { void checkTransactionState(TransactionStateCheckRequest checkData); diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/ClusterGrpcService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/ClusterGrpcService.java index e39740fc2b..59a0df7f1c 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/ClusterGrpcService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/ClusterGrpcService.java @@ -54,7 +54,7 @@ import apache.rocketmq.v1.SendMessageResponse; import io.grpc.Context; import java.util.concurrent.CompletableFuture; import org.apache.rocketmq.common.constant.LoggerName; -import org.apache.rocketmq.proxy.client.ForwardClientManager; +import org.apache.rocketmq.proxy.connector.ConnectorManager; import org.apache.rocketmq.proxy.common.AbstractStartAndShutdown; import org.apache.rocketmq.proxy.grpc.service.cluster.ProducerService; import org.apache.rocketmq.proxy.grpc.service.cluster.RouteService; @@ -64,12 +64,12 @@ import org.slf4j.LoggerFactory; public class ClusterGrpcService extends AbstractStartAndShutdown implements GrpcForwardService { private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.GRPC_LOGGER_NAME); - private final ForwardClientManager forwardClientManager; + private final ConnectorManager forwardClientManager; private final ProducerService producerService; private final RouteService routeService; public ClusterGrpcService() { - this.forwardClientManager = new ForwardClientManager(checkData -> { + this.forwardClientManager = new ConnectorManager(checkData -> { }); this.producerService = new ProducerService(forwardClientManager); this.routeService = new RouteService(forwardClientManager); diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/BaseService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/BaseService.java index b9837ef2a1..9178a1cea7 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/BaseService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/BaseService.java @@ -16,13 +16,13 @@ */ package org.apache.rocketmq.proxy.grpc.service.cluster; -import org.apache.rocketmq.proxy.client.ForwardClientManager; +import org.apache.rocketmq.proxy.connector.ConnectorManager; public class BaseService { - protected final ForwardClientManager clientManager; + protected final ConnectorManager clientManager; - public BaseService(ForwardClientManager clientManager) { + public BaseService(ConnectorManager clientManager) { this.clientManager = clientManager; } } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/ConsumerService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/ConsumerService.java index cc898d16ab..9f091f9fb9 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/ConsumerService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/ConsumerService.java @@ -20,11 +20,11 @@ import apache.rocketmq.v1.ReceiveMessageRequest; import apache.rocketmq.v1.ReceiveMessageResponse; import io.grpc.Context; import java.util.concurrent.CompletableFuture; -import org.apache.rocketmq.proxy.client.ForwardClientManager; +import org.apache.rocketmq.proxy.connector.ConnectorManager; public class ConsumerService extends BaseService { - public ConsumerService(ForwardClientManager clientManager) { + public ConsumerService(ConnectorManager clientManager) { super(clientManager); } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/ProducerService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/ProducerService.java index 8cabdf0204..5da401fba1 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/ProducerService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/ProducerService.java @@ -25,8 +25,8 @@ import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.common.message.MessageConst; import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader; -import org.apache.rocketmq.proxy.client.ForwardClientManager; -import org.apache.rocketmq.proxy.client.route.SelectableMessageQueue; +import org.apache.rocketmq.proxy.connector.ConnectorManager; +import org.apache.rocketmq.proxy.connector.route.SelectableMessageQueue; import org.apache.rocketmq.proxy.common.utils.ProxyUtils; import org.apache.rocketmq.proxy.grpc.common.Converter; import org.apache.rocketmq.proxy.grpc.common.ProxyException; @@ -42,7 +42,7 @@ public class ProducerService extends BaseService { private volatile ProducerServiceHook producerServiceHook = null; private volatile MessageQueueSelector messageQueueSelector = new DefaultMessageQueueSelector(); - public ProducerService(ForwardClientManager clientManager) { + public ProducerService(ConnectorManager clientManager) { super(clientManager); } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/RouteService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/RouteService.java index 3809e8073e..3032458311 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/RouteService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/RouteService.java @@ -34,10 +34,10 @@ import java.util.concurrent.CompletableFuture; import org.apache.rocketmq.common.constant.PermName; import org.apache.rocketmq.common.protocol.route.QueueData; import org.apache.rocketmq.common.protocol.route.TopicRouteData; -import org.apache.rocketmq.proxy.client.ForwardClientManager; -import org.apache.rocketmq.proxy.client.route.SelectableMessageQueue; -import org.apache.rocketmq.proxy.client.route.MessageQueueWrapper; -import org.apache.rocketmq.proxy.common.RocketMQHelper; +import org.apache.rocketmq.proxy.connector.ConnectorManager; +import org.apache.rocketmq.proxy.connector.route.SelectableMessageQueue; +import org.apache.rocketmq.proxy.connector.route.MessageQueueWrapper; +import org.apache.rocketmq.proxy.connector.route.TopicRouteHelper; import org.apache.rocketmq.proxy.grpc.common.Converter; import org.apache.rocketmq.proxy.grpc.common.ResponseBuilder; @@ -47,7 +47,7 @@ public class RouteService extends BaseService { private volatile QueryRouteHook queryRouteHook = null; private volatile QueryAssignmentHook queryAssignmentHook = null; - public RouteService(ForwardClientManager clientManager) { + public RouteService(ConnectorManager clientManager) { super(clientManager); } @@ -126,7 +126,7 @@ public class RouteService extends BaseService { .build(); future.complete(response); } catch (Throwable t) { - if (RocketMQHelper.isTopicNotExistError(t)) { + if (TopicRouteHelper.isTopicNotExistError(t)) { future.complete(QueryRouteResponse.newBuilder() .setCommon(ResponseBuilder.buildCommon(Code.NOT_FOUND, t.getMessage())) .build()); diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/client/ForwardClientManagerTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/connector/ForwardClientManagerTest.java similarity index 90% rename from proxy/src/test/java/org/apache/rocketmq/proxy/client/ForwardClientManagerTest.java rename to proxy/src/test/java/org/apache/rocketmq/proxy/connector/ForwardClientManagerTest.java index 710affa028..b2061744cb 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/client/ForwardClientManagerTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/connector/ForwardClientManagerTest.java @@ -15,9 +15,9 @@ * limitations under the License. */ -package org.apache.rocketmq.proxy.client; +package org.apache.rocketmq.proxy.connector; -import org.apache.rocketmq.proxy.client.transaction.TransactionStateChecker; +import org.apache.rocketmq.proxy.connector.transaction.TransactionStateChecker; import org.apache.rocketmq.proxy.config.ConfigurationManager; import org.apache.rocketmq.proxy.config.InitConfigAndLoggerTest; import org.junit.Test; @@ -30,7 +30,7 @@ public class ForwardClientManagerTest extends InitConfigAndLoggerTest { @Test public void testClientManager() throws Exception { TransactionStateChecker mockedTransactionStateChecker = Mockito.mock(TransactionStateChecker.class); - ForwardClientManager clientManager = new ForwardClientManager(mockedTransactionStateChecker); + ConnectorManager clientManager = new ConnectorManager(mockedTransactionStateChecker); clientManager.start(); assertThat(clientManager.getDefaultForwardClient()).isNotNull(); diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/service/cluster/BaseServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/service/cluster/BaseServiceTest.java index 9a6ab623ae..b29a248cb8 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/service/cluster/BaseServiceTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/service/cluster/BaseServiceTest.java @@ -16,12 +16,12 @@ */ package org.apache.rocketmq.proxy.grpc.service.cluster; -import org.apache.rocketmq.proxy.client.ForwardClientManager; -import org.apache.rocketmq.proxy.client.DefaultForwardClient; -import org.apache.rocketmq.proxy.client.ForwardProducer; -import org.apache.rocketmq.proxy.client.ForwardReadConsumer; -import org.apache.rocketmq.proxy.client.TopicRouteCache; -import org.apache.rocketmq.proxy.client.ForwardWriteConsumer; +import org.apache.rocketmq.proxy.connector.ConnectorManager; +import org.apache.rocketmq.proxy.connector.DefaultForwardClient; +import org.apache.rocketmq.proxy.connector.ForwardProducer; +import org.apache.rocketmq.proxy.connector.ForwardReadConsumer; +import org.apache.rocketmq.proxy.connector.route.TopicRouteCache; +import org.apache.rocketmq.proxy.connector.ForwardWriteConsumer; import org.junit.Before; import org.junit.Ignore; import org.junit.runner.RunWith; @@ -35,7 +35,7 @@ import static org.mockito.Mockito.when; public abstract class BaseServiceTest { @Mock - protected ForwardClientManager clientManager; + protected ConnectorManager clientManager; @Mock protected DefaultForwardClient defaultClient; @Mock diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/service/cluster/ProducerServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/service/cluster/ProducerServiceTest.java index 2ec8bb4602..90a9a3e6ca 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/service/cluster/ProducerServiceTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/service/cluster/ProducerServiceTest.java @@ -35,7 +35,7 @@ import org.apache.rocketmq.client.producer.SendStatus; import org.apache.rocketmq.common.message.MessageConst; import org.apache.rocketmq.common.message.MessageQueue; import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader; -import org.apache.rocketmq.proxy.client.route.SelectableMessageQueue; +import org.apache.rocketmq.proxy.connector.route.SelectableMessageQueue; import org.apache.rocketmq.proxy.grpc.common.ProxyException; import org.apache.rocketmq.proxy.grpc.common.ProxyResponseCode; import org.junit.Test; diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/service/cluster/RouteServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/service/cluster/RouteServiceTest.java index 39ee582e6a..458dcb32b6 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/service/cluster/RouteServiceTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/service/cluster/RouteServiceTest.java @@ -19,13 +19,12 @@ import org.apache.rocketmq.common.protocol.ResponseCode; import org.apache.rocketmq.common.protocol.route.BrokerData; import org.apache.rocketmq.common.protocol.route.QueueData; import org.apache.rocketmq.common.protocol.route.TopicRouteData; -import org.apache.rocketmq.proxy.client.route.MessageQueueWrapper; +import org.apache.rocketmq.proxy.connector.route.MessageQueueWrapper; import org.apache.rocketmq.proxy.grpc.common.ResponseBuilder; import org.junit.Test; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; -import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.when; public class RouteServiceTest extends BaseServiceTest {