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] v2 support

Browse Source
This commit is contained in:
kaiyi.lk
2022-07-13 11:29:34 +08:00
committed by zhouxiang
parent c04c3185e5
commit 1451bf5558
2 changed files with 12 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/v2/consumer/ReceiveMessageActivity.java
+11 -1
View File
@@ -31,6 +31,7 @@ import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.config.ConfigurationManager;
import org.apache.rocketmq.proxy.grpc.v2.AbstractMessingActivity;
import org.apache.rocketmq.proxy.grpc.v2.GrpcContextConstants;
import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager;
import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter;
import org.apache.rocketmq.proxy.processor.MessagingProcessor;
@@ -65,6 +66,15 @@ public class ReceiveMessageActivity extends AbstractMessingActivity {
writer.write(proxyContext, Code.MESSAGE_NOT_FOUND, "no new message");
return;
}
long invisibleTime = Durations.toMillis(request.getInvisibleDuration());
if (request.getAutoRenew()) {
invisibleTime = Durations.toMillis(
this.grpcClientSettingsManager.getClientSettings(proxyContext.getVal(GrpcContextConstants.CLIENT_ID))
.getSubscription().getLongPollingTimeout()
);
}
String topic = GrpcConverter.wrapResourceWithNamespace(request.getMessageQueue().getTopic());
String group = GrpcConverter.wrapResourceWithNamespace(request.getGroup());
FilterExpression filterExpression = request.getFilterExpression();
@@ -85,7 +95,7 @@ public class ReceiveMessageActivity extends AbstractMessingActivity {
group,
topic,
request.getBatchSize(),
Durations.toMillis(request.getInvisibleDuration()),
invisibleTime,
pollTime,
ConsumeInitMode.MAX,
subscriptionData,
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageResponseStreamWriter.java
+1 -1
View File
@@ -56,7 +56,7 @@ public class ReceiveMessageResponseStreamWriter {
case FOUND:
if (messageFoundList.isEmpty()) {
streamObserver.onNext(ReceiveMessageResponse.newBuilder()
.setStatus(ResponseBuilder.buildStatus(Code.MESSAGE_NOT_FOUND, "no new message"))
.setStatus(ResponseBuilder.buildStatus(Code.MESSAGE_NOT_FOUND, "no match message"))
.build());
} else {
streamObserver.onNext(ReceiveMessageResponse.newBuilder()