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] pollCommand for cluster mode

Browse Source
This commit is contained in:
kaiyi.lk
2022-07-13 11:29:11 +08:00
committed by zhouxiang
parent c456ea7883
commit 67440ebe27
1 changed files with 14 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/ClusterGrpcService.java
+14 -2
View File
@@ -31,6 +31,7 @@ import apache.rocketmq.v1.HeartbeatRequest;
import apache.rocketmq.v1.HeartbeatResponse;
import apache.rocketmq.v1.NackMessageRequest;
import apache.rocketmq.v1.NackMessageResponse;
import apache.rocketmq.v1.NoopCommand;
import apache.rocketmq.v1.NotifyClientTerminationRequest;
import apache.rocketmq.v1.NotifyClientTerminationResponse;
import apache.rocketmq.v1.PollCommandRequest;
@@ -168,17 +169,28 @@ public class ClusterGrpcService extends AbstractStartAndShutdown implements Grpc
public CompletableFuture<PollCommandResponse> pollCommand(Context ctx, PollCommandRequest request) {
CompletableFuture<PollCommandResponse> future = new CompletableFuture<>();
String clientId = request.getClientId();
PollCommandResponse noopCommandResponse = PollCommandResponse.newBuilder().setNoopCommand(NoopCommand.newBuilder().build()).build();
switch (request.getGroupCase()) {
case PRODUCER_GROUP:
Resource producerGroup = request.getProducerGroup();
String producerGroupName = Converter.getResourceNameWithNamespace(producerGroup);
GrpcClientChannel.addClientObserver(this.channelManager, producerGroupName, clientId, future);
GrpcClientChannel producerChannel = GrpcClientChannel.getChannel(this.channelManager, producerGroupName, clientId);
if (producerChannel == null) {
future.complete(noopCommandResponse);
} else {
producerChannel.addClientObserver(future);
}
break;
case CONSUMER_GROUP:
Resource consumerGroup = request.getConsumerGroup();
String consumerGroupName = Converter.getResourceNameWithNamespace(consumerGroup);
GrpcClientChannel.addClientObserver(this.channelManager, consumerGroupName, clientId, future);
GrpcClientChannel consumerChannel = GrpcClientChannel.getChannel(this.channelManager, consumerGroupName, clientId);
if (consumerChannel == null) {
future.complete(noopCommandResponse);
} else {
consumerChannel.addClientObserver(future);
}
break;
default:
break;