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

[ISSUE #3949] do the code refactoring work for readability.

Browse Source
This commit is contained in:
Jixiang.jjx
2022-07-13 11:29:21 +08:00
committed by zhouxiang
parent 1cc1d10434
commit c08fee78af
3 changed files with 8 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/main/java/org/apache/rocketmq/proxy/ProxyStartup.java
+1 -1
View File
@@ -81,7 +81,7 @@ public class ProxyStartup {
System.exit(1);
}
System.out.printf("%s%n", new Date() + " rmq-proxy startup successfully");
System.out.println(new Date() + " rmq-proxy startup successfully");
log.info(new Date() + " rmq-proxy startup successfully");
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/adapter/GrpcConverter.java
+6 -5
View File
@@ -89,7 +89,6 @@ import org.apache.rocketmq.common.sysflag.MessageSysFlag;
import org.apache.rocketmq.common.utils.BinaryUtil;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.proxy.common.DelayPolicy;
import org.apache.rocketmq.proxy.common.utils.ProxyUtils;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.connector.transaction.TransactionId;
@@ -444,22 +443,24 @@ public class GrpcConverter {
}
public static List<org.apache.rocketmq.common.message.Message> buildMessage(List<Message> protoMessageList,
Resource topic, String producerGroup) {
Resource topic) {
String topicName = wrapResourceWithNamespace(topic);
List<org.apache.rocketmq.common.message.Message> messages = new ArrayList<>();
for (Message protoMessage : protoMessageList) {
if (!protoMessage.getTopic().equals(topic)) {
throw new ProxyException(Code.MESSAGE_CORRUPTED, "topic in message is not same");
}
messages.add(buildMessage(protoMessage, producerGroup));
// here use topicName as producerGroup for transactional checker.
messages.add(buildMessage(protoMessage, topicName));
}
return messages;
}
public static org.apache.rocketmq.common.message.Message buildMessage(Message protoMessage, String producerGroup) {
String topic = wrapResourceWithNamespace(protoMessage.getTopic());
String topicName = wrapResourceWithNamespace(protoMessage.getTopic());
org.apache.rocketmq.common.message.Message message =
new org.apache.rocketmq.common.message.Message(topic, protoMessage.getBody().toByteArray());
new org.apache.rocketmq.common.message.Message(topicName, protoMessage.getBody().toByteArray());
Map<String, String> messageProperty = buildMessageProperty(protoMessage, producerGroup);
MessageAccessor.setProperties(message, messageProperty);
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/service/cluster/ProducerService.java
+1 -1
View File
@@ -108,7 +108,7 @@ public class ProducerService extends BaseService {
// use topic name as group
Resource topic = request.getMessages(0).getTopic();
String topicName = GrpcConverter.wrapResourceWithNamespace(topic);
return GrpcConverter.buildMessage(request.getMessagesList(), topic, topicName);
return GrpcConverter.buildMessage(request.getMessagesList(), topic);
}
protected SendMessageResponse convertToSendMessageResponse(Context ctx, SendMessageRequest request,