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

[ISSUE #3949] Add Converter.buildPopMessageRequestHeader

Browse Source
This commit is contained in:
zhouxiang
2022-07-13 11:29:09 +08:00
parent f5df799397
commit 85d4a368c8
1 changed files with 47 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/common/Converter.java
+47
View File
@@ -27,7 +27,9 @@ import apache.rocketmq.v1.FilterType;
import apache.rocketmq.v1.HeartbeatRequest;
import apache.rocketmq.v1.Message;
import apache.rocketmq.v1.MessageType;
import apache.rocketmq.v1.Partition;
import apache.rocketmq.v1.ProducerData;
import apache.rocketmq.v1.ReceiveMessageRequest;
import apache.rocketmq.v1.Resource;
import apache.rocketmq.v1.SendMessageRequest;
import apache.rocketmq.v1.SubscriptionEntry;
@@ -41,6 +43,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.rocketmq.common.constant.ConsumeInitMode;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.filter.ExpressionType;
@@ -49,6 +52,7 @@ import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.protocol.NamespaceUtil;
import org.apache.rocketmq.common.protocol.header.PopMessageRequestHeader;
import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader;
import org.apache.rocketmq.common.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.common.protocol.heartbeat.HeartbeatData;
@@ -97,6 +101,39 @@ public class Converter {
return requestHeader;
}
public static PopMessageRequestHeader buildPopMessageRequestHeader(ReceiveMessageRequest request, long pollTime) {
Resource group = request.getGroup();
String groupName = Converter.getResourceNameWithNamespace(group);
Partition partition = request.getPartition();
Resource topic = partition.getTopic();
String topicName = Converter.getResourceNameWithNamespace(topic);
int queueId = partition.getId();
int maxMessageNumbers = request.getBatchSize();
long invisibleTime = Durations.toMillis(request.getInvisibleDuration());
long bornTime = Timestamps.toMillis(request.getInitializationTimestamp());
ConsumePolicy policy = request.getConsumePolicy();
int initMode = Converter.buildConsumeInitMode(policy);
FilterExpression filterExpression = request.getFilterExpression();
String expression = filterExpression.getExpression();
String expressionType = Converter.buildExpressionType(filterExpression.getType());
PopMessageRequestHeader requestHeader = new PopMessageRequestHeader();
requestHeader.setConsumerGroup(groupName);
requestHeader.setTopic(topicName);
requestHeader.setQueueId(queueId);
requestHeader.setMaxMsgNums(maxMessageNumbers);
requestHeader.setInvisibleTime(invisibleTime);
requestHeader.setPollTime(pollTime);
requestHeader.setBornTime(bornTime);
requestHeader.setInitMode(initMode);
requestHeader.setExpType(expressionType);
requestHeader.setExp(expression);
requestHeader.setOrder(request.getFifoFlag());
return requestHeader;
}
public static Map<String, String> buildMessageProperty(Message message) {
org.apache.rocketmq.common.message.Message messageWithHeader = new org.apache.rocketmq.common.message.Message();
// set user properties
@@ -262,4 +299,14 @@ public class Converter {
}
return subscriptionDataSet;
}
public static int buildConsumeInitMode(ConsumePolicy policy) {
switch (policy) {
case PLAYBACK:
return ConsumeInitMode.MIN;
case RESUME:
default:
return ConsumeInitMode.MAX;
}
}
}