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

Add expression filtering capability to the pullBlockIfNotFound method of pull consumer (#8024)

Browse Source
This commit is contained in:
rongtong
2024-04-19 13:41:43 +08:00
committed by RongtongJin
parent 6e5cdf4a5b
commit e502e315fb
3 changed files with 50 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/main/java/org/apache/rocketmq/client/consumer/DefaultMQPullConsumer.java
+16 -2
View File
@@ -36,8 +36,8 @@ import org.apache.rocketmq.remoting.protocol.NamespaceUtil;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
/**
* @deprecated Default pulling consumer. This class will be removed in 2022, and a better implementation {@link
* DefaultLitePullConsumer} is recommend to use in the scenario of actively pulling messages.
* @deprecated Default pulling consumer. This class will be removed in 2022, and a better implementation
* {@link DefaultLitePullConsumer} is recommend to use in the scenario of actively pulling messages.
*/
@Deprecated
public class DefaultMQPullConsumer extends ClientConfig implements MQPullConsumer {
@@ -375,6 +375,20 @@ public class DefaultMQPullConsumer extends ClientConfig implements MQPullConsume
this.defaultMQPullConsumerImpl.pullBlockIfNotFound(queueWithNamespace(mq), subExpression, offset, maxNums, pullCallback);
}
@Override
public void pullBlockIfNotFoundWithMessageSelector(MessageQueue mq, MessageSelector selector,
long offset, int maxNums,
PullCallback pullCallback) throws MQClientException, RemotingException, InterruptedException {
this.defaultMQPullConsumerImpl.pullBlockIfNotFoundWithMessageSelector(mq, selector, offset, maxNums, pullCallback);
}
@Override
public PullResult pullBlockIfNotFoundWithMessageSelector(MessageQueue mq, MessageSelector selector,
long offset,
int maxNums) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
return this.defaultMQPullConsumerImpl.pullBlockIfNotFoundWithMessageSelector(mq, selector, offset, maxNums);
}
@Override
public void updateConsumeOffset(MessageQueue mq, long offset) throws MQClientException {
this.defaultMQPullConsumerImpl.updateConsumeOffset(queueWithNamespace(mq), offset);
client/src/main/java/org/apache/rocketmq/client/consumer/MQPullConsumer.java
+19 -3
View File
@@ -47,8 +47,7 @@ public interface MQPullConsumer extends MQConsumer {
*
* @param mq from which message queue
* @param subExpression subscription expression.it only support or operation such as "tag1 || tag2 || tag3" <br> if
* null or * expression,meaning subscribe
* all
* null or * expression,meaning subscribe all
* @param offset from where to pull
* @param maxNums max pulling numbers
* @return The resulting {@code PullRequest}
@@ -121,7 +120,7 @@ public interface MQPullConsumer extends MQConsumer {
InterruptedException;
/**
* Pulling the messages in a async. way. Support message selection
* Pulling the messages in a async way. Support message selection
*/
void pull(final MessageQueue mq, final MessageSelector selector, final long offset, final int maxNums,
final PullCallback pullCallback) throws MQClientException, RemotingException,
@@ -150,6 +149,23 @@ public interface MQPullConsumer extends MQConsumer {
final int maxNums, final PullCallback pullCallback) throws MQClientException, RemotingException,
InterruptedException;
/**
* Pulling the messages through callback function,if no message arrival,blocking. Support message selection
*/
void pullBlockIfNotFoundWithMessageSelector(final MessageQueue mq, final MessageSelector selector,
final long offset, final int maxNums,
final PullCallback pullCallback) throws MQClientException, RemotingException,
InterruptedException;
/**
* Pulling the messages,if no message arrival,blocking some time. Support message selection
*
* @return The resulting {@code PullRequest}
*/
PullResult pullBlockIfNotFoundWithMessageSelector(final MessageQueue mq, final MessageSelector selector,
final long offset, final int maxNums) throws MQClientException, RemotingException,
MQBrokerException, InterruptedException;
/**
* Update the offset
*/
client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java
+15
View File
@@ -589,6 +589,21 @@ public class DefaultMQPullConsumerImpl implements MQConsumerInner {
this.getDefaultMQPullConsumer().getConsumerPullTimeoutMillis());
}
public void pullBlockIfNotFoundWithMessageSelector(MessageQueue mq, MessageSelector messageSelector, long offset, int maxNums,
PullCallback pullCallback)
throws MQClientException, RemotingException, InterruptedException {
SubscriptionData subscriptionData = getSubscriptionData(mq, messageSelector);
this.pullAsyncImpl(mq, subscriptionData, offset, maxNums, pullCallback, true,
this.getDefaultMQPullConsumer().getConsumerPullTimeoutMillis());
}
public PullResult pullBlockIfNotFoundWithMessageSelector(MessageQueue mq, MessageSelector messageSelector, long offset, int maxNums)
throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
SubscriptionData subscriptionData = getSubscriptionData(mq, messageSelector);
return this.pullSyncImpl(mq, subscriptionData, offset, maxNums, true, this.getDefaultMQPullConsumer().getConsumerPullTimeoutMillis());
}
public QueryResult queryMessage(String topic, String key, int maxNum, long begin, long end)
throws MQClientException, InterruptedException {
this.isRunning();