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 getConsumerGroupInfo

Browse Source
This commit is contained in:
zhouxiang
2022-07-13 11:30:53 +08:00
parent c3059933e5
commit 67f92fbdf3
3 changed files with 14 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/main/java/org/apache/rocketmq/proxy/processor/ClientProcessor.java
+5
View File
@@ -19,6 +19,7 @@ package org.apache.rocketmq.proxy.processor;
import io.netty.channel.Channel;
import java.util.Set;
import org.apache.rocketmq.broker.client.ClientChannelInfo;
import org.apache.rocketmq.broker.client.ConsumerGroupInfo;
import org.apache.rocketmq.broker.client.ConsumerIdsChangeListener;
import org.apache.rocketmq.broker.client.ProducerChangeListener;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
@@ -101,4 +102,8 @@ public class ClientProcessor extends AbstractProcessor {
public void registerConsumerIdsChangeListener(ConsumerIdsChangeListener listener) {
this.serviceManager.getConsumerManager().appendConsumerIdsChangeListener(listener);
}
public ConsumerGroupInfo getConsumerGroupInfo(String consumerGroup) {
return this.serviceManager.getConsumerManager().getConsumerGroupInfo(consumerGroup);
}
}
proxy/src/main/java/org/apache/rocketmq/proxy/processor/DefaultMessagingProcessor.java
+6
View File
@@ -24,6 +24,7 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.client.ClientChannelInfo;
import org.apache.rocketmq.broker.client.ConsumerGroupInfo;
import org.apache.rocketmq.broker.client.ConsumerIdsChangeListener;
import org.apache.rocketmq.broker.client.ProducerChangeListener;
import org.apache.rocketmq.client.consumer.AckResult;
@@ -217,6 +218,11 @@ public class DefaultMessagingProcessor extends AbstractStartAndShutdown implemen
this.clientProcessor.registerConsumerIdsChangeListener(consumerIdsChangeListener);
}
@Override
public ConsumerGroupInfo getConsumerGroupInfo(String consumerGroup) {
return this.clientProcessor.getConsumerGroupInfo(consumerGroup);
}
@Override
public void addTransactionSubscription(ProxyContext ctx, String producerGroup, String topic) {
this.transactionProcessor.addTransactionSubscription(ctx, producerGroup, topic);
proxy/src/main/java/org/apache/rocketmq/proxy/processor/MessagingProcessor.java
+3
View File
@@ -22,6 +22,7 @@ import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import org.apache.rocketmq.broker.client.ClientChannelInfo;
import org.apache.rocketmq.broker.client.ConsumerGroupInfo;
import org.apache.rocketmq.broker.client.ConsumerIdsChangeListener;
import org.apache.rocketmq.broker.client.ProducerChangeListener;
import org.apache.rocketmq.client.consumer.AckResult;
@@ -222,6 +223,8 @@ public interface MessagingProcessor extends StartAndShutdown {
ConsumerIdsChangeListener consumerIdsChangeListener
);
ConsumerGroupInfo getConsumerGroupInfo(String consumerGroup);
void addTransactionSubscription(
ProxyContext ctx,
String producerGroup,