fix

Change-Id: I309419c125475aec5ceeac2ce95f47cb9f45693d

Author: qianye1001

broker/src/main/java/org/apache/rocketmq/broker/processor/NotificationProcessor.java

@@ -21,6 +21,9 @@
 import java.util.Map;
 import java.util.Random;
 import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.broker.filter.ConsumerFilterData;
+import org.apache.rocketmq.broker.filter.ConsumerFilterManager;
+import org.apache.rocketmq.broker.filter.ExpressionMessageFilter;
 import org.apache.rocketmq.broker.longpolling.PollingHeader;
 import org.apache.rocketmq.broker.longpolling.PollingResult;
 import org.apache.rocketmq.broker.longpolling.PopLongPollingService;
@@ -29,6 +32,7 @@
 import org.apache.rocketmq.common.TopicConfig;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.constant.PermName;
+import org.apache.rocketmq.common.filter.ExpressionType;
 import org.apache.rocketmq.common.help.FAQUrl;
 import org.apache.rocketmq.logging.org.slf4j.Logger;
 import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
@@ -37,8 +41,10 @@
 import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.filter.FilterAPI;
 import org.apache.rocketmq.remoting.protocol.header.NotificationRequestHeader;
 import org.apache.rocketmq.remoting.protocol.header.NotificationResponseHeader;
+import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
 import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
 import org.apache.rocketmq.store.exception.ConsumeQueueException;
 
@@ -154,7 +160,49 @@ public RemotingCommand processRequest(final ChannelHandlerContext ctx,
         }
 
         if (!hasMsg) {
-            PollingResult pollingResult = popLongPollingService.polling(ctx, request, new PollingHeader(requestHeader));
+            SubscriptionData subscriptionData = null;
+            ExpressionMessageFilter messageFilter = null;
+            if (requestHeader.getExp() != null && !requestHeader.getExp().isEmpty()) {
+                try {
+                    // origin topic
+                    subscriptionData = FilterAPI.build(
+                        requestHeader.getTopic(), requestHeader.getExp(), requestHeader.getExpType());
+
+                    ConsumerFilterData consumerFilterData = null;
+                    if (!ExpressionType.isTagType(subscriptionData.getExpressionType())) {
+                        consumerFilterData = ConsumerFilterManager.build(
+                            requestHeader.getTopic(), requestHeader.getConsumerGroup(), requestHeader.getExp(),
+                            requestHeader.getExpType(), System.currentTimeMillis());
+                        if (consumerFilterData == null) {
+                            POP_LOGGER.warn("Parse the consumer's subscription[{}] failed, group: {}",
+                                requestHeader.getExp(), requestHeader.getConsumerGroup());
+                            response.setCode(ResponseCode.SUBSCRIPTION_PARSE_FAILED);
+                            response.setRemark("parse the consumer's subscription failed");
+                            return response;
+                        }
+                    }
+                    messageFilter = new ExpressionMessageFilter(
+                        subscriptionData, consumerFilterData, brokerController.getConsumerFilterManager());
+                } catch (Exception e) {
+                    POP_LOGGER.warn("Parse the consumer's subscription[{}] error, group: {}", requestHeader.getExp(),
+                        requestHeader.getConsumerGroup());
+                    response.setCode(ResponseCode.SUBSCRIPTION_PARSE_FAILED);
+                    response.setRemark("parse the consumer's subscription failed");
+                    return response;
+                }
+            } else {
+                try {
+                    // origin topic
+                    subscriptionData = FilterAPI.build(requestHeader.getTopic(), "*", ExpressionType.TAG);
+                } catch (Exception e) {
+                    POP_LOGGER.warn("Build default subscription error, group: {}", requestHeader.getConsumerGroup());
+                }
+            }
+
+            ExpressionMessageFilter finalMessageFilter = messageFilter;
+            SubscriptionData finalSubscriptionData = subscriptionData;
+
+            PollingResult pollingResult = popLongPollingService.polling(ctx, request, new PollingHeader(requestHeader), finalSubscriptionData, finalMessageFilter);
             if (pollingResult == PollingResult.POLLING_SUC) {
                 return null;
             } else if (pollingResult == PollingResult.POLLING_FULL) {

remoting/src/main/java/org/apache/rocketmq/remoting/protocol/header/NotificationRequestHeader.java

@@ -44,6 +44,9 @@ public class NotificationRequestHeader extends TopicQueueRequestHeader {
     private Boolean order = Boolean.FALSE;
     private String attemptId;
 
+    private String expType;
+    private String exp;
+
     @CFNotNull
     @Override
     public void checkFields() throws RemotingCommandException {
@@ -108,6 +111,22 @@ public void setAttemptId(String attemptId) {
         this.attemptId = attemptId;
     }
 
+    public String getExpType() {
+        return expType;
+    }
+
+    public void setExpType(String expType) {
+        this.expType = expType;
+    }
+
+    public String getExp() {
+        return exp;
+    }
+
+    public void setExp(String exp) {
+        this.exp = exp;
+    }
+
     @Override
     public String toString() {
         return MoreObjects.toStringHelper(this)

test/src/main/java/org/apache/rocketmq/test/client/rmq/RMQPopClient.java

@@ -199,6 +199,12 @@ public CompletableFuture<Boolean> notification(String brokerAddr, String topic,
 
     public CompletableFuture<Boolean> notification(String brokerAddr, String topic,
         String consumerGroup, int queueId, Boolean order, String attemptId, long pollTime, long bornTime, long timeoutMillis) {
+        return notification(brokerAddr, topic, consumerGroup, queueId, order, attemptId, pollTime, bornTime, timeoutMillis, null, null);
+    }
+
+
+    public CompletableFuture<Boolean> notification(String brokerAddr, String topic,
+        String consumerGroup, int queueId, Boolean order, String attemptId, long pollTime, long bornTime, long timeoutMillis, String expType, String exp) {
         NotificationRequestHeader requestHeader = new NotificationRequestHeader();
         requestHeader.setConsumerGroup(consumerGroup);
         requestHeader.setTopic(topic);
@@ -207,6 +213,8 @@ public CompletableFuture<Boolean> notification(String brokerAddr, String topic,
         requestHeader.setBornTime(bornTime);
         requestHeader.setOrder(order);
         requestHeader.setAttemptId(attemptId);
+        requestHeader.setExpType(expType);
+        requestHeader.setExp(exp);
         return this.mqClientAPI.notification(brokerAddr, requestHeader, timeoutMillis);
     }
 }