Merge pull request #620 from RongtongJin/polish_filter_example

[ISSUE #608] Polish filter example

example/src/main/java/org/apache/rocketmq/example/filter/SqlConsumer.java → example/src/main/java/org/apache/rocketmq/example/filter/SqlFilterConsumer.java

@@ -17,28 +17,24 @@
 
 package org.apache.rocketmq.example.filter;
 
+import java.util.List;
 import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
 import org.apache.rocketmq.client.consumer.MessageSelector;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
-import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.message.MessageExt;
 
-import java.util.List;
+public class SqlFilterConsumer {
+
+    public static void main(String[] args) throws Exception {
 
-public class SqlConsumer {
+        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
 
-    public static void main(String[] args) {
-        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
-        try {
-            consumer.subscribe("TopicTest",
-                MessageSelector.bySql("(TAGS is not null and TAGS in ('TagA', 'TagB'))" +
-                    "and (a is not null and a between 0  3)"));
-        } catch (MQClientException e) {
-            e.printStackTrace();
-            return;
-        }
+        // Don't forget to set enablePropertyFilter=true in broker
+        consumer.subscribe("SqlFilterTest",
+            MessageSelector.bySql("(TAGS is not null and TAGS in ('TagA', 'TagB'))" +
+                "and (a is not null and a between 0 and 3)"));
 
         consumer.registerMessageListener(new MessageListenerConcurrently() {
 
@@ -50,12 +46,7 @@ public class SqlConsumer {
             }
         });
 
-        try {
-            consumer.start();
-        } catch (MQClientException e) {
-            e.printStackTrace();
-            return;
-        }
+        consumer.start();
         System.out.printf("Consumer Started.%n");
     }
 }

example/src/main/java/org/apache/rocketmq/example/filter/Producer.java → example/src/main/java/org/apache/rocketmq/example/filter/SqlFilterProducer.java

@@ -14,33 +14,35 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.apache.rocketmq.example.filter;
 
-import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.client.producer.DefaultMQProducer;
 import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.common.message.Message;
 import org.apache.rocketmq.remoting.common.RemotingHelper;
 
-public class Producer {
-    public static void main(String[] args) throws MQClientException, InterruptedException {
-        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
+public class SqlFilterProducer {
+
+    public static void main(String[] args) throws Exception {
+
+        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
+
         producer.start();
 
-        try {
-            for (int i = 0; i < 6000000; i++) {
-                Message msg = new Message("TopicFilter7",
-                    "TagA",
-                    "OrderID001",
-                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
-
-                msg.putUserProperty("SequenceId", String.valueOf(i));
-                SendResult sendResult = producer.send(msg);
-                System.out.printf("%s%n", sendResult);
-            }
-        } catch (Exception e) {
-            e.printStackTrace();
+        String[] tags = new String[] {"TagA", "TagB", "TagC"};
+
+        for (int i = 0; i < 10; i++) {
+            Message msg = new Message("SqlFilterTest",
+                tags[i % tags.length],
+                ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)
+            );
+            msg.putUserProperty("a", String.valueOf(i));
+
+            SendResult sendResult = producer.send(msg);
+            System.out.printf("%s%n", sendResult);
         }
+
         producer.shutdown();
     }
 }

example/src/main/java/org/apache/rocketmq/example/filter/Consumer.java → example/src/main/java/org/apache/rocketmq/example/filter/TagFilterConsumer.java

@@ -16,7 +16,6 @@
  */
 package org.apache.rocketmq.example.filter;
 
-import java.io.File;
 import java.io.IOException;
 import java.util.List;
 import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
@@ -24,20 +23,15 @@ import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
 import org.apache.rocketmq.client.exception.MQClientException;
-import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.message.MessageExt;
 
-public class Consumer {
+public class TagFilterConsumer {
 
     public static void main(String[] args) throws InterruptedException, MQClientException, IOException {
-        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupNamecc4");
 
-        ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
-        File classFile = new File(classLoader.getResource("MessageFilterImpl.java").getFile());
+        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
 
-        String filterCode = MixAll.file2String(classFile);
-        consumer.subscribe("TopicTest", "org.apache.rocketmq.example.filter.MessageFilterImpl",
-            filterCode);
+        consumer.subscribe("TagFilterTest", "TagA || TagC");
 
         consumer.registerMessageListener(new MessageListenerConcurrently() {
 

example/src/main/java/org/apache/rocketmq/example/filter/SqlProducer.java → example/src/main/java/org/apache/rocketmq/example/filter/TagFilterProducer.java

@@ -14,54 +14,31 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.apache.rocketmq.example.filter;
 
-import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.client.producer.DefaultMQProducer;
 import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.common.message.Message;
 import org.apache.rocketmq.remoting.common.RemotingHelper;
 
-public class SqlProducer {
+public class TagFilterProducer {
+
+    public static void main(String[] args) throws Exception {
 
-    public static void main(String[] args) {
         DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
-        try {
-            producer.start();
-        } catch (MQClientException e) {
-            e.printStackTrace();
-            return;
-        }
+        producer.start();
 
-        for (int i = 0; i < 10; i++) {
-            try {
-                String tag;
-                int div = i % 3;
-                if (div == 0) {
-                    tag = "TagA";
-                } else if (div == 1) {
-                    tag = "TagB";
-                } else {
-                    tag = "TagC";
-                }
-                Message msg = new Message("TopicTest",
-                    tag,
-                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)
-                );
-                msg.putUserProperty("a", String.valueOf(i));
+        String[] tags = new String[] {"TagA", "TagB", "TagC"};
 
-                SendResult sendResult = producer.send(msg);
-                System.out.printf("%s%n", sendResult);
-            } catch (Exception e) {
-                e.printStackTrace();
-                try {
-                    Thread.sleep(1000);
-                } catch (InterruptedException e1) {
-                    e1.printStackTrace();
-                }
-            }
+        for (int i = 0; i < 60; i++) {
+            Message msg = new Message("TagFilterTest",
+                tags[i % tags.length],
+                "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
+
+            SendResult sendResult = producer.send(msg);
+            System.out.printf("%s%n", sendResult);
         }
+
         producer.shutdown();
     }
 }