jiulinxiri/rocketmq-dashboard
1
0
Fork 0
mirror of https://github.com/apache/rocketmq-dashboard.git synced 2025-09-11 12:05:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge de0fb99e0b into e57d423268

Browse Source
This commit is contained in:
为了吃方便面
2024-11-26 11:08:48 +08:00
committed by GitHub
parent e57d423268 de0fb99e0b
commit 5215184d67
1 changed files with 14 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
src/main/java/org/apache/rocketmq/dashboard/service/impl/MessageServiceImpl.java
View File

@@ -37,9 +37,10 @@ import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.message.MessageClientIDSetter; import org.apache.rocketmq.common.message.MessageClientIDSetter;
import org.apache.rocketmq.common.message.MessageExt; import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue; import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.protocol.body.Connection; import org.apache.rocketmq.common.protocol.body.Connection;
import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult; import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection; import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.dashboard.config.RMQConfigure; import org.apache.rocketmq.dashboard.config.RMQConfigure;
import org.apache.rocketmq.dashboard.exception.ServiceException; import org.apache.rocketmq.dashboard.exception.ServiceException;
import org.apache.rocketmq.dashboard.model.QueueOffsetInfo; import org.apache.rocketmq.dashboard.model.QueueOffsetInfo;
@@ -67,6 +68,7 @@ import java.util.ArrayList;
import java.util.Set; import java.util.Set;
import java.util.Collection; import java.util.Collection;
import java.util.Date; import java.util.Date;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@@ -218,12 +220,19 @@ public class MessageServiceImpl implements MessageService {
try { try {
ConsumerConnection consumerConnection = mqAdminExt.examineConsumerConnectionInfo(consumerGroup); ConsumerConnection consumerConnection = mqAdminExt.examineConsumerConnectionInfo(consumerGroup);
MessageExt messageExt = mqAdminExt.viewMessage(topic, msgId);
for (Connection connection : consumerConnection.getConnectionSet()) { for (Connection connection : consumerConnection.getConnectionSet()) {
if (StringUtils.isBlank(connection.getClientId())) { if (StringUtils.isBlank(connection.getClientId())) {
continue; continue;
} }
logger.info("clientId={}", connection.getClientId()); ConcurrentMap<String, SubscriptionData> subscriptionTable = consumerConnection.getSubscriptionTable();
return mqAdminExt.consumeMessageDirectly(consumerGroup, connection.getClientId(), topic, msgId); SubscriptionData subscriptionData = subscriptionTable.get(topic);
if (subscriptionData.getSubString().equals(SubscriptionData.SUB_ALL)
|| subscriptionData.getTagsSet().contains(messageExt.getTags())) {
logger.info("clientId={}", connection.getClientId());
return mqAdminExt.consumeMessageDirectly(consumerGroup, connection.getClientId(), topic, msgId);
}
throw new IllegalStateException("CONSUMER NOT SUBSCRIPT THIS TAG");
} }
} catch (Exception e) { } catch (Exception e) {
Throwables.throwIfUnchecked(e); Throwables.throwIfUnchecked(e);