You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by vi...@apache.org on 2017/01/07 15:31:20 UTC
incubator-rocketmq git commit: [ROCKETMQ-17] openmessaging: add
PushConsumer and filter ASF JIRA:
https://issues.apache.org/jira/browse/ROCKETMQ-17
Repository: incubator-rocketmq
Updated Branches:
refs/heads/spec 89945900c -> b99ce053d
[ROCKETMQ-17] openmessaging: add PushConsumer and filter
ASF JIRA: https://issues.apache.org/jira/browse/ROCKETMQ-17
Project: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/commit/b99ce053
Tree: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/tree/b99ce053
Diff: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/diff/b99ce053
Branch: refs/heads/spec
Commit: b99ce053dbb406a226fd8e84890fdd35af8c9543
Parents: 8994590
Author: vintagewang <vi...@apache.org>
Authored: Sat Jan 7 23:30:56 2017 +0800
Committer: vintagewang <vi...@apache.org>
Committed: Sat Jan 7 23:30:56 2017 +0800
----------------------------------------------------------------------
.../java/org/apache/openmessaging/Filters.java | 21 ++++++++
.../apache/openmessaging/MessageListener.java | 22 ++++++++
.../apache/openmessaging/MessagingEndPoint.java | 2 +
.../apache/openmessaging/OnMessageContext.java | 21 ++++++++
.../org/apache/openmessaging/PushConsumer.java | 10 ++++
.../samples/simple/ConsumerQueueApp.java | 53 ++++++++++++++++++
.../samples/simple/ConsumerTopicApp.java | 56 ++++++++++++++++++++
.../samples/simple/ProducerApp.java | 4 +-
8 files changed, 187 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/b99ce053/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/Filters.java
----------------------------------------------------------------------
diff --git a/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/Filters.java b/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/Filters.java
new file mode 100644
index 0000000..8020253
--- /dev/null
+++ b/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/Filters.java
@@ -0,0 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.openmessaging;
+
+public interface Filters {
+ Filters addFilter(final String sql);
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/b99ce053/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/MessageListener.java
----------------------------------------------------------------------
diff --git a/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/MessageListener.java b/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/MessageListener.java
new file mode 100644
index 0000000..241e4cc
--- /dev/null
+++ b/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/MessageListener.java
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.openmessaging;
+
+public interface MessageListener {
+ void onMessage(final Message message, final OnMessageContext context);
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/b99ce053/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/MessagingEndPoint.java
----------------------------------------------------------------------
diff --git a/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/MessagingEndPoint.java b/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/MessagingEndPoint.java
index 21e20e6..3fe76b5 100644
--- a/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/MessagingEndPoint.java
+++ b/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/MessagingEndPoint.java
@@ -34,4 +34,6 @@ public interface MessagingEndPoint {
PullConsumer createPullConsumer(final KeyValue properties);
ResourceManager createResourceManager();
+
+ Filters createFilters();
}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/b99ce053/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/OnMessageContext.java
----------------------------------------------------------------------
diff --git a/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/OnMessageContext.java b/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/OnMessageContext.java
new file mode 100644
index 0000000..076b0bf
--- /dev/null
+++ b/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/OnMessageContext.java
@@ -0,0 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.openmessaging;
+
+public interface OnMessageContext {
+ KeyValue properties();
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/b99ce053/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/PushConsumer.java
----------------------------------------------------------------------
diff --git a/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/PushConsumer.java b/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/PushConsumer.java
index 78ea718..354f5bb 100644
--- a/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/PushConsumer.java
+++ b/spec/code/messaging-user-level-api/java/src/main/java/org/apache/openmessaging/PushConsumer.java
@@ -17,5 +17,15 @@
package org.apache.openmessaging;
public interface PushConsumer {
+ void start();
+ void shutdown();
+
+ void resume();
+
+ void suspend();
+
+ void attachQueue(final String queueName, final MessageListener listener);
+
+ void attachQueue(final String queueName, final Filters filters, final MessageListener listener);
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/b99ce053/spec/code/messaging-user-level-samples/java/src/main/java/org/apache/openmessaging/samples/simple/ConsumerQueueApp.java
----------------------------------------------------------------------
diff --git a/spec/code/messaging-user-level-samples/java/src/main/java/org/apache/openmessaging/samples/simple/ConsumerQueueApp.java b/spec/code/messaging-user-level-samples/java/src/main/java/org/apache/openmessaging/samples/simple/ConsumerQueueApp.java
new file mode 100644
index 0000000..e2a9572
--- /dev/null
+++ b/spec/code/messaging-user-level-samples/java/src/main/java/org/apache/openmessaging/samples/simple/ConsumerQueueApp.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.openmessaging.samples.simple;
+
+import org.apache.openmessaging.Message;
+import org.apache.openmessaging.MessageListener;
+import org.apache.openmessaging.MessagingEndPoint;
+import org.apache.openmessaging.MessagingEndPointManager;
+import org.apache.openmessaging.OnMessageContext;
+import org.apache.openmessaging.PushConsumer;
+
+public class ConsumerQueueApp {
+ public static void main(String[] args) {
+ final MessagingEndPoint messagingEndPoint = MessagingEndPointManager.getMessagingEndPoint("openmessaging:rocketmq://localhost:10911/namespace");
+
+ final PushConsumer consumer = messagingEndPoint.createPushConsumer();
+
+ consumer.attachQueue("HELLO_QUEUE", new MessageListener() {
+ @Override public void onMessage(Message message, OnMessageContext context) {
+ System.out.println("receive one message: " + message);
+ }
+ });
+
+ messagingEndPoint.start();
+ System.out.println("messagingEndPoint startup OK");
+
+ consumer.start();
+ System.out.println("consumer startup OK");
+
+ Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
+ @Override
+ public void run() {
+ consumer.shutdown();
+ messagingEndPoint.shutdown();
+ }
+ }));
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/b99ce053/spec/code/messaging-user-level-samples/java/src/main/java/org/apache/openmessaging/samples/simple/ConsumerTopicApp.java
----------------------------------------------------------------------
diff --git a/spec/code/messaging-user-level-samples/java/src/main/java/org/apache/openmessaging/samples/simple/ConsumerTopicApp.java b/spec/code/messaging-user-level-samples/java/src/main/java/org/apache/openmessaging/samples/simple/ConsumerTopicApp.java
new file mode 100644
index 0000000..e219378
--- /dev/null
+++ b/spec/code/messaging-user-level-samples/java/src/main/java/org/apache/openmessaging/samples/simple/ConsumerTopicApp.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.openmessaging.samples.simple;
+
+import org.apache.openmessaging.Message;
+import org.apache.openmessaging.MessageListener;
+import org.apache.openmessaging.MessagingEndPoint;
+import org.apache.openmessaging.MessagingEndPointManager;
+import org.apache.openmessaging.OnMessageContext;
+import org.apache.openmessaging.PushConsumer;
+
+public class ConsumerTopicApp {
+ public static void main(String[] args) {
+ final MessagingEndPoint messagingEndPoint = MessagingEndPointManager.getMessagingEndPoint("openmessaging:rocketmq://localhost:10911/namespace");
+
+ final PushConsumer consumer = messagingEndPoint.createPushConsumer();
+
+ consumer.attachQueue("HELLO_QUEUE", messagingEndPoint.createFilters()//
+ .addFilter("TOPIC='HELLO_TOPIC1'")//
+ .addFilter("TOPIC='HELLO_TOPIC2' AND KEY2 > 199"),//
+ new MessageListener() {
+ @Override public void onMessage(Message message, OnMessageContext context) {
+ System.out.println("receive one message: " + message);
+ }
+ });
+
+ messagingEndPoint.start();
+ System.out.println("messagingEndPoint startup OK");
+
+ consumer.start();
+ System.out.println("consumer startup OK");
+
+ Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
+ @Override
+ public void run() {
+ consumer.shutdown();
+ messagingEndPoint.shutdown();
+ }
+ }));
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/b99ce053/spec/code/messaging-user-level-samples/java/src/main/java/org/apache/openmessaging/samples/simple/ProducerApp.java
----------------------------------------------------------------------
diff --git a/spec/code/messaging-user-level-samples/java/src/main/java/org/apache/openmessaging/samples/simple/ProducerApp.java b/spec/code/messaging-user-level-samples/java/src/main/java/org/apache/openmessaging/samples/simple/ProducerApp.java
index e14d6e0..a833842 100644
--- a/spec/code/messaging-user-level-samples/java/src/main/java/org/apache/openmessaging/samples/simple/ProducerApp.java
+++ b/spec/code/messaging-user-level-samples/java/src/main/java/org/apache/openmessaging/samples/simple/ProducerApp.java
@@ -42,10 +42,10 @@ public class ProducerApp {
}
}));
- producer.send(producer.createBytesMessageToTopic("HELLO_TOPIC", "HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
+ producer.send(producer.createBytesMessageToTopic("HELLO_TOPIC1", "HELLO_BODY1".getBytes(Charset.forName("UTF-8"))));
System.out.println("send first message to topic OK");
- producer.send(producer.createBytesMessageToTopic("HELLO_TOPIC", "HELLO_BODY".getBytes(Charset.forName("UTF-8")))
+ producer.send(producer.createBytesMessageToTopic("HELLO_TOPIC2", "HELLO_BODY2".getBytes(Charset.forName("UTF-8")))
.putProperties("KEY1", 100)//
.putProperties("KEY2", 200L)//
.putProperties("KEY3", 3.14)//