You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by yu...@apache.org on 2017/08/01 07:23:33 UTC
[35/50] [abbrv] incubator-rocketmq git commit: [ROCKETMQ-186]
Implement the OpenMessaging specification 0.1.0-alpha version
[ROCKETMQ-186] Implement the OpenMessaging specification 0.1.0-alpha version
Project: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/commit/c8e84adf
Tree: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/tree/c8e84adf
Diff: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/diff/c8e84adf
Branch: refs/heads/develop
Commit: c8e84adf5a2911c21faea80320d3fd5fd012f4a9
Parents: 2a5a6d6
Author: yukon <yu...@apache.org>
Authored: Wed May 24 16:50:51 2017 +0800
Committer: dongeforever <zh...@yeah.net>
Committed: Tue Jun 6 11:37:29 2017 +0800
----------------------------------------------------------------------
distribution/release-client.xml | 1 +
distribution/release.xml | 1 +
example/pom.xml | 9 +
.../example/openmessaging/SimpleProducer.java | 76 +++++++
.../openmessaging/SimplePullConsumer.java | 58 +++++
.../openmessaging/SimplePushConsumer.java | 59 +++++
openmessaging/pom.xml | 42 ++++
.../rocketmq/MessagingAccessPointImpl.java | 132 +++++++++++
.../rocketmq/config/ClientConfig.java | 194 ++++++++++++++++
.../rocketmq/consumer/LocalMessageCache.java | 213 +++++++++++++++++
.../rocketmq/consumer/PullConsumerImpl.java | 166 ++++++++++++++
.../rocketmq/consumer/PushConsumerImpl.java | 181 +++++++++++++++
.../rocketmq/domain/BytesMessageImpl.java | 108 +++++++++
.../rocketmq/domain/ConsumeRequest.java | 55 +++++
.../rocketmq/domain/NonStandardKeys.java | 30 +++
.../rocketmq/domain/SendResultImpl.java | 40 ++++
.../rocketmq/producer/AbstractOMSProducer.java | 138 +++++++++++
.../rocketmq/producer/ProducerImpl.java | 124 ++++++++++
.../rocketmq/producer/SequenceProducerImpl.java | 95 ++++++++
.../rocketmq/promise/DefaultPromise.java | 227 +++++++++++++++++++
.../rocketmq/promise/FutureState.java | 51 +++++
.../openmessaging/rocketmq/utils/BeanUtils.java | 185 +++++++++++++++
.../openmessaging/rocketmq/utils/OMSUtil.java | 182 +++++++++++++++
.../consumer/LocalMessageCacheTest.java | 89 ++++++++
.../rocketmq/consumer/PullConsumerImplTest.java | 96 ++++++++
.../rocketmq/consumer/PushConsumerImplTest.java | 87 +++++++
.../rocketmq/producer/ProducerImplTest.java | 101 +++++++++
.../producer/SequenceProducerImplTest.java | 86 +++++++
.../rocketmq/promise/DefaultPromiseTest.java | 136 +++++++++++
.../rocketmq/utils/BeanUtilsTest.java | 110 +++++++++
pom.xml | 6 +
31 files changed, 3078 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/c8e84adf/distribution/release-client.xml
----------------------------------------------------------------------
diff --git a/distribution/release-client.xml b/distribution/release-client.xml
index 46563eb..84d33a0 100644
--- a/distribution/release-client.xml
+++ b/distribution/release-client.xml
@@ -47,6 +47,7 @@
<useAllReactorProjects>true</useAllReactorProjects>
<includes>
<include>org.apache.rocketmq:rocketmq-client</include>
+ <include>org.apache.rocketmq:rocketmq-openmessaging</include>
</includes>
<binaries>
<outputDirectory>./</outputDirectory>
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/c8e84adf/distribution/release.xml
----------------------------------------------------------------------
diff --git a/distribution/release.xml b/distribution/release.xml
index 9e4ef2a..c67d23e 100644
--- a/distribution/release.xml
+++ b/distribution/release.xml
@@ -68,6 +68,7 @@
<include>org.apache.rocketmq:rocketmq-filtersrv</include>
<include>org.apache.rocketmq:rocketmq-example</include>
<include>org.apache.rocketmq:rocketmq-filter</include>
+ <include>org.apache.rocketmq:rocketmq-openmessaging</include>
</includes>
<binaries>
<outputDirectory>lib/</outputDirectory>
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/c8e84adf/example/pom.xml
----------------------------------------------------------------------
diff --git a/example/pom.xml b/example/pom.xml
index 785a4ca..840fa36 100644
--- a/example/pom.xml
+++ b/example/pom.xml
@@ -48,5 +48,14 @@
<groupId>org.javassist</groupId>
<artifactId>javassist</artifactId>
</dependency>
+ <dependency>
+ <groupId>io.openmessaging</groupId>
+ <artifactId>openmessaging-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-openmessaging</artifactId>
+ <version>4.1.0-incubating-SNAPSHOT</version>
+ </dependency>
</dependencies>
</project>
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/c8e84adf/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimpleProducer.java
----------------------------------------------------------------------
diff --git a/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimpleProducer.java b/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimpleProducer.java
new file mode 100644
index 0000000..9d162ac
--- /dev/null
+++ b/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimpleProducer.java
@@ -0,0 +1,76 @@
+/*
+ * 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.rocketmq.example.openmessaging;
+
+import io.openmessaging.Message;
+import io.openmessaging.MessagingAccessPoint;
+import io.openmessaging.MessagingAccessPointFactory;
+import io.openmessaging.Producer;
+import io.openmessaging.Promise;
+import io.openmessaging.PromiseListener;
+import io.openmessaging.SendResult;
+import java.nio.charset.Charset;
+
+public class SimpleProducer {
+ public static void main(String[] args) {
+ final MessagingAccessPoint messagingAccessPoint = MessagingAccessPointFactory
+ .getMessagingAccessPoint("openmessaging:rocketmq://IP1:9876,IP2:9876/namespace");
+
+ final Producer producer = messagingAccessPoint.createProducer();
+
+ messagingAccessPoint.startup();
+ System.out.printf("MessagingAccessPoint startup OK%n");
+
+ producer.startup();
+ System.out.printf("Producer startup OK%n");
+
+ Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
+ @Override
+ public void run() {
+ producer.shutdown();
+ messagingAccessPoint.shutdown();
+ }
+ }));
+
+ {
+ Message message = producer.createBytesMessageToTopic("OMS_HELLO_TOPIC", "OMS_HELLO_BODY".getBytes(Charset.forName("UTF-8")));
+ SendResult sendResult = producer.send(message);
+ //final Void aVoid = result.get(3000L);
+ System.out.printf("Send async message OK, msgId: %s%n", sendResult.messageId());
+ }
+
+ {
+ final Promise<SendResult> result = producer.sendAsync(producer.createBytesMessageToTopic("OMS_HELLO_TOPIC", "OMS_HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
+ result.addListener(new PromiseListener<SendResult>() {
+ @Override
+ public void operationCompleted(Promise<SendResult> promise) {
+ System.out.printf("Send async message OK, msgId: %s%n", promise.get().messageId());
+ }
+
+ @Override
+ public void operationFailed(Promise<SendResult> promise) {
+ System.out.printf("Send async message Failed, error: %s%n", promise.getThrowable().getMessage());
+ }
+ });
+ }
+
+ {
+ producer.sendOneway(producer.createBytesMessageToTopic("OMS_HELLO_TOPIC", "OMS_HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
+ System.out.printf("Send oneway message OK%n");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/c8e84adf/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimplePullConsumer.java
----------------------------------------------------------------------
diff --git a/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimplePullConsumer.java b/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimplePullConsumer.java
new file mode 100644
index 0000000..8e06772
--- /dev/null
+++ b/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimplePullConsumer.java
@@ -0,0 +1,58 @@
+/*
+ * 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.rocketmq.example.openmessaging;
+
+import io.openmessaging.Message;
+import io.openmessaging.MessageHeader;
+import io.openmessaging.MessagingAccessPoint;
+import io.openmessaging.MessagingAccessPointFactory;
+import io.openmessaging.OMS;
+import io.openmessaging.PullConsumer;
+import io.openmessaging.rocketmq.domain.NonStandardKeys;
+
+public class SimplePullConsumer {
+ public static void main(String[] args) {
+ final MessagingAccessPoint messagingAccessPoint = MessagingAccessPointFactory
+ .getMessagingAccessPoint("openmessaging:rocketmq://IP1:9876,IP2:9876/namespace");
+
+ final PullConsumer consumer = messagingAccessPoint.createPullConsumer("OMS_HELLO_TOPIC",
+ OMS.newKeyValue().put(NonStandardKeys.CONSUMER_GROUP, "OMS_CONSUMER"));
+
+ messagingAccessPoint.startup();
+ System.out.printf("MessagingAccessPoint startup OK%n");
+
+ Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
+ @Override
+ public void run() {
+ consumer.shutdown();
+ messagingAccessPoint.shutdown();
+ }
+ }));
+
+ consumer.startup();
+ System.out.printf("Consumer startup OK%n");
+
+ while (true) {
+ Message message = consumer.poll();
+ if (message != null) {
+ String msgId = message.headers().getString(MessageHeader.MESSAGE_ID);
+ System.out.printf("Received one message: %s%n", msgId);
+ consumer.ack(msgId);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/c8e84adf/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimplePushConsumer.java
----------------------------------------------------------------------
diff --git a/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimplePushConsumer.java b/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimplePushConsumer.java
new file mode 100644
index 0000000..b0935d4
--- /dev/null
+++ b/example/src/main/java/org/apache/rocketmq/example/openmessaging/SimplePushConsumer.java
@@ -0,0 +1,59 @@
+/*
+ * 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.rocketmq.example.openmessaging;
+
+import io.openmessaging.Message;
+import io.openmessaging.MessageHeader;
+import io.openmessaging.MessageListener;
+import io.openmessaging.MessagingAccessPoint;
+import io.openmessaging.MessagingAccessPointFactory;
+import io.openmessaging.OMS;
+import io.openmessaging.PushConsumer;
+import io.openmessaging.ReceivedMessageContext;
+import io.openmessaging.rocketmq.domain.NonStandardKeys;
+
+public class SimplePushConsumer {
+ public static void main(String[] args) {
+ final MessagingAccessPoint messagingAccessPoint = MessagingAccessPointFactory
+ .getMessagingAccessPoint("openmessaging:rocketmq://IP1:9876,IP2:9876/namespace");
+
+ final PushConsumer consumer = messagingAccessPoint.
+ createPushConsumer(OMS.newKeyValue().put(NonStandardKeys.CONSUMER_GROUP, "OMS_CONSUMER"));
+
+ messagingAccessPoint.startup();
+ System.out.printf("MessagingAccessPoint startup OK%n");
+
+ Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
+ @Override
+ public void run() {
+ consumer.shutdown();
+ messagingAccessPoint.shutdown();
+ }
+ }));
+
+ consumer.attachQueue("OMS_HELLO_TOPIC", new MessageListener() {
+ @Override
+ public void onMessage(final Message message, final ReceivedMessageContext context) {
+ System.out.printf("Received one message: %s%n", message.headers().getString(MessageHeader.MESSAGE_ID));
+ context.ack();
+ }
+ });
+
+ consumer.startup();
+ System.out.printf("Consumer startup OK%n");
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/c8e84adf/openmessaging/pom.xml
----------------------------------------------------------------------
diff --git a/openmessaging/pom.xml b/openmessaging/pom.xml
new file mode 100644
index 0000000..e853642
--- /dev/null
+++ b/openmessaging/pom.xml
@@ -0,0 +1,42 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>rocketmq-all</artifactId>
+ <groupId>org.apache.rocketmq</groupId>
+ <version>4.1.0-incubating-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>rocketmq-openmessaging</artifactId>
+ <name>rocketmq-openmessaging ${project.version}</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>io.openmessaging</groupId>
+ <artifactId>openmessaging-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-client</artifactId>
+ </dependency>
+ </dependencies>
+</project>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/c8e84adf/openmessaging/src/main/java/io/openmessaging/rocketmq/MessagingAccessPointImpl.java
----------------------------------------------------------------------
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/MessagingAccessPointImpl.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/MessagingAccessPointImpl.java
new file mode 100644
index 0000000..65caf84
--- /dev/null
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/MessagingAccessPointImpl.java
@@ -0,0 +1,132 @@
+/*
+ * 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 io.openmessaging.rocketmq;
+
+import io.openmessaging.IterableConsumer;
+import io.openmessaging.KeyValue;
+import io.openmessaging.MessagingAccessPoint;
+import io.openmessaging.Producer;
+import io.openmessaging.PullConsumer;
+import io.openmessaging.PushConsumer;
+import io.openmessaging.ResourceManager;
+import io.openmessaging.SequenceProducer;
+import io.openmessaging.ServiceEndPoint;
+import io.openmessaging.exception.OMSNotSupportedException;
+import io.openmessaging.observer.Observer;
+import io.openmessaging.rocketmq.consumer.PullConsumerImpl;
+import io.openmessaging.rocketmq.consumer.PushConsumerImpl;
+import io.openmessaging.rocketmq.producer.ProducerImpl;
+import io.openmessaging.rocketmq.producer.SequenceProducerImpl;
+import io.openmessaging.rocketmq.utils.OMSUtil;
+
+public class MessagingAccessPointImpl implements MessagingAccessPoint {
+ private final KeyValue accessPointProperties;
+
+ public MessagingAccessPointImpl(final KeyValue accessPointProperties) {
+ this.accessPointProperties = accessPointProperties;
+ }
+
+ @Override
+ public KeyValue properties() {
+ return accessPointProperties;
+ }
+
+ @Override
+ public Producer createProducer() {
+ return new ProducerImpl(this.accessPointProperties);
+ }
+
+ @Override
+ public Producer createProducer(KeyValue properties) {
+ return new ProducerImpl(OMSUtil.buildKeyValue(this.accessPointProperties, properties));
+ }
+
+ @Override
+ public SequenceProducer createSequenceProducer() {
+ return new SequenceProducerImpl(this.accessPointProperties);
+ }
+
+ @Override
+ public SequenceProducer createSequenceProducer(KeyValue properties) {
+ return new SequenceProducerImpl(OMSUtil.buildKeyValue(this.accessPointProperties, properties));
+ }
+
+ @Override
+ public PushConsumer createPushConsumer() {
+ return new PushConsumerImpl(accessPointProperties);
+ }
+
+ @Override
+ public PushConsumer createPushConsumer(KeyValue properties) {
+ return new PushConsumerImpl(OMSUtil.buildKeyValue(this.accessPointProperties, properties));
+ }
+
+ @Override
+ public PullConsumer createPullConsumer(String queueName) {
+ return new PullConsumerImpl(queueName, accessPointProperties);
+ }
+
+ @Override
+ public PullConsumer createPullConsumer(String queueName, KeyValue properties) {
+ return new PullConsumerImpl(queueName, OMSUtil.buildKeyValue(this.accessPointProperties, properties));
+ }
+
+ @Override
+ public IterableConsumer createIterableConsumer(String queueName) {
+ throw new OMSNotSupportedException("-1", "IterableConsumer is not supported in current version");
+ }
+
+ @Override
+ public IterableConsumer createIterableConsumer(String queueName, KeyValue properties) {
+ throw new OMSNotSupportedException("-1", "IterableConsumer is not supported in current version");
+ }
+
+ @Override
+ public ResourceManager getResourceManager() {
+ throw new OMSNotSupportedException("-1", "ResourceManager is not supported in current version.");
+ }
+
+ @Override
+ public ServiceEndPoint createServiceEndPoint() {
+ throw new OMSNotSupportedException("-1", "ServiceEndPoint is not supported in current version.");
+ }
+
+ @Override
+ public ServiceEndPoint createServiceEndPoint(KeyValue properties) {
+ throw new OMSNotSupportedException("-1", "ServiceEndPoint is not supported in current version.");
+ }
+
+ @Override
+ public void addObserver(Observer observer) {
+ //Ignore
+ }
+
+ @Override
+ public void deleteObserver(Observer observer) {
+ //Ignore
+ }
+
+ @Override
+ public void startup() {
+ //Ignore
+ }
+
+ @Override
+ public void shutdown() {
+ //Ignore
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/c8e84adf/openmessaging/src/main/java/io/openmessaging/rocketmq/config/ClientConfig.java
----------------------------------------------------------------------
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/config/ClientConfig.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/config/ClientConfig.java
new file mode 100644
index 0000000..7077c6d
--- /dev/null
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/config/ClientConfig.java
@@ -0,0 +1,194 @@
+/*
+ * 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 io.openmessaging.rocketmq.config;
+
+import io.openmessaging.PropertyKeys;
+import io.openmessaging.rocketmq.domain.NonStandardKeys;
+
+public class ClientConfig implements PropertyKeys, NonStandardKeys {
+ private String omsDriverImpl;
+ private String omsAccessPoints;
+ private String omsNamespace;
+ private String omsProducerId;
+ private String omsConsumerId;
+ private int omsOperationTimeout = 5000;
+ private String omsRoutingName;
+ private String omsOperatorName;
+ private String omsDstQueue;
+ private String omsSrcTopic;
+ private String rmqConsumerGroup;
+ private String rmqProducerGroup = "__OMS_PRODUCER_DEFAULT_GROUP";
+ private int rmqMaxRedeliveryTimes = 16;
+ private int rmqMessageConsumeTimeout = 15; //In minutes
+ private int rmqMaxConsumeThreadNums = 64;
+ private int rmqMinConsumeThreadNums = 20;
+ private String rmqMessageDestination;
+ private int rmqPullMessageBatchNums = 32;
+ private int rmqPullMessageCacheCapacity = 1000;
+
+ public String getOmsDriverImpl() {
+ return omsDriverImpl;
+ }
+
+ public void setOmsDriverImpl(final String omsDriverImpl) {
+ this.omsDriverImpl = omsDriverImpl;
+ }
+
+ public String getOmsAccessPoints() {
+ return omsAccessPoints;
+ }
+
+ public void setOmsAccessPoints(final String omsAccessPoints) {
+ this.omsAccessPoints = omsAccessPoints;
+ }
+
+ public String getOmsNamespace() {
+ return omsNamespace;
+ }
+
+ public void setOmsNamespace(final String omsNamespace) {
+ this.omsNamespace = omsNamespace;
+ }
+
+ public String getOmsProducerId() {
+ return omsProducerId;
+ }
+
+ public void setOmsProducerId(final String omsProducerId) {
+ this.omsProducerId = omsProducerId;
+ }
+
+ public String getOmsConsumerId() {
+ return omsConsumerId;
+ }
+
+ public void setOmsConsumerId(final String omsConsumerId) {
+ this.omsConsumerId = omsConsumerId;
+ }
+
+ public int getOmsOperationTimeout() {
+ return omsOperationTimeout;
+ }
+
+ public void setOmsOperationTimeout(final int omsOperationTimeout) {
+ this.omsOperationTimeout = omsOperationTimeout;
+ }
+
+ public String getOmsRoutingName() {
+ return omsRoutingName;
+ }
+
+ public void setOmsRoutingName(final String omsRoutingName) {
+ this.omsRoutingName = omsRoutingName;
+ }
+
+ public String getOmsOperatorName() {
+ return omsOperatorName;
+ }
+
+ public void setOmsOperatorName(final String omsOperatorName) {
+ this.omsOperatorName = omsOperatorName;
+ }
+
+ public String getOmsDstQueue() {
+ return omsDstQueue;
+ }
+
+ public void setOmsDstQueue(final String omsDstQueue) {
+ this.omsDstQueue = omsDstQueue;
+ }
+
+ public String getOmsSrcTopic() {
+ return omsSrcTopic;
+ }
+
+ public void setOmsSrcTopic(final String omsSrcTopic) {
+ this.omsSrcTopic = omsSrcTopic;
+ }
+
+ public String getRmqConsumerGroup() {
+ return rmqConsumerGroup;
+ }
+
+ public void setRmqConsumerGroup(final String rmqConsumerGroup) {
+ this.rmqConsumerGroup = rmqConsumerGroup;
+ }
+
+ public String getRmqProducerGroup() {
+ return rmqProducerGroup;
+ }
+
+ public void setRmqProducerGroup(final String rmqProducerGroup) {
+ this.rmqProducerGroup = rmqProducerGroup;
+ }
+
+ public int getRmqMaxRedeliveryTimes() {
+ return rmqMaxRedeliveryTimes;
+ }
+
+ public void setRmqMaxRedeliveryTimes(final int rmqMaxRedeliveryTimes) {
+ this.rmqMaxRedeliveryTimes = rmqMaxRedeliveryTimes;
+ }
+
+ public int getRmqMessageConsumeTimeout() {
+ return rmqMessageConsumeTimeout;
+ }
+
+ public void setRmqMessageConsumeTimeout(final int rmqMessageConsumeTimeout) {
+ this.rmqMessageConsumeTimeout = rmqMessageConsumeTimeout;
+ }
+
+ public int getRmqMaxConsumeThreadNums() {
+ return rmqMaxConsumeThreadNums;
+ }
+
+ public void setRmqMaxConsumeThreadNums(final int rmqMaxConsumeThreadNums) {
+ this.rmqMaxConsumeThreadNums = rmqMaxConsumeThreadNums;
+ }
+
+ public int getRmqMinConsumeThreadNums() {
+ return rmqMinConsumeThreadNums;
+ }
+
+ public void setRmqMinConsumeThreadNums(final int rmqMinConsumeThreadNums) {
+ this.rmqMinConsumeThreadNums = rmqMinConsumeThreadNums;
+ }
+
+ public String getRmqMessageDestination() {
+ return rmqMessageDestination;
+ }
+
+ public void setRmqMessageDestination(final String rmqMessageDestination) {
+ this.rmqMessageDestination = rmqMessageDestination;
+ }
+
+ public int getRmqPullMessageBatchNums() {
+ return rmqPullMessageBatchNums;
+ }
+
+ public void setRmqPullMessageBatchNums(final int rmqPullMessageBatchNums) {
+ this.rmqPullMessageBatchNums = rmqPullMessageBatchNums;
+ }
+
+ public int getRmqPullMessageCacheCapacity() {
+ return rmqPullMessageCacheCapacity;
+ }
+
+ public void setRmqPullMessageCacheCapacity(final int rmqPullMessageCacheCapacity) {
+ this.rmqPullMessageCacheCapacity = rmqPullMessageCacheCapacity;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/c8e84adf/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/LocalMessageCache.java
----------------------------------------------------------------------
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/LocalMessageCache.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/LocalMessageCache.java
new file mode 100644
index 0000000..90f9e03
--- /dev/null
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/LocalMessageCache.java
@@ -0,0 +1,213 @@
+/*
+ * 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 io.openmessaging.rocketmq.consumer;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.PropertyKeys;
+import io.openmessaging.ServiceLifecycle;
+import io.openmessaging.rocketmq.config.ClientConfig;
+import io.openmessaging.rocketmq.domain.ConsumeRequest;
+import java.util.Collections;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReadWriteLock;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.impl.consumer.ProcessQueue;
+import org.apache.rocketmq.client.log.ClientLogger;
+import org.apache.rocketmq.common.ThreadFactoryImpl;
+import org.apache.rocketmq.common.message.MessageAccessor;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.common.utils.ThreadUtils;
+import org.slf4j.Logger;
+
+class LocalMessageCache implements ServiceLifecycle {
+ private final BlockingQueue<ConsumeRequest> consumeRequestCache;
+ private final Map<String, ConsumeRequest> consumedRequest;
+ private final ConcurrentHashMap<MessageQueue, Long> pullOffsetTable;
+ private final DefaultMQPullConsumer rocketmqPullConsumer;
+ private final ClientConfig clientConfig;
+ private final ScheduledExecutorService cleanExpireMsgExecutors;
+
+ private final static Logger log = ClientLogger.getLog();
+
+ LocalMessageCache(final DefaultMQPullConsumer rocketmqPullConsumer, final ClientConfig clientConfig) {
+ consumeRequestCache = new LinkedBlockingQueue<>(clientConfig.getRmqPullMessageCacheCapacity());
+ this.consumedRequest = new ConcurrentHashMap<>();
+ this.pullOffsetTable = new ConcurrentHashMap<>();
+ this.rocketmqPullConsumer = rocketmqPullConsumer;
+ this.clientConfig = clientConfig;
+ this.cleanExpireMsgExecutors = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl(
+ "OMS_CleanExpireMsgScheduledThread_"));
+ }
+
+ int nextPullBatchNums() {
+ return Math.min(clientConfig.getRmqPullMessageBatchNums(), consumeRequestCache.remainingCapacity());
+ }
+
+ long nextPullOffset(MessageQueue remoteQueue) {
+ if (!pullOffsetTable.containsKey(remoteQueue)) {
+ try {
+ pullOffsetTable.putIfAbsent(remoteQueue,
+ rocketmqPullConsumer.fetchConsumeOffset(remoteQueue, false));
+ } catch (MQClientException e) {
+ log.error("A error occurred in fetch consume offset process.", e);
+ }
+ }
+ return pullOffsetTable.get(remoteQueue);
+ }
+
+ void updatePullOffset(MessageQueue remoteQueue, long nextPullOffset) {
+ pullOffsetTable.put(remoteQueue, nextPullOffset);
+ }
+
+ void submitConsumeRequest(ConsumeRequest consumeRequest) {
+ try {
+ consumeRequestCache.put(consumeRequest);
+ } catch (InterruptedException ignore) {
+ }
+ }
+
+ MessageExt poll() {
+ return poll(clientConfig.getOmsOperationTimeout());
+ }
+
+ MessageExt poll(final KeyValue properties) {
+ int currentPollTimeout = clientConfig.getOmsOperationTimeout();
+ if (properties.containsKey(PropertyKeys.OPERATION_TIMEOUT)) {
+ currentPollTimeout = properties.getInt(PropertyKeys.OPERATION_TIMEOUT);
+ }
+ return poll(currentPollTimeout);
+ }
+
+ private MessageExt poll(long timeout) {
+ try {
+ ConsumeRequest consumeRequest = consumeRequestCache.poll(timeout, TimeUnit.MILLISECONDS);
+ if (consumeRequest != null) {
+ MessageExt messageExt = consumeRequest.getMessageExt();
+ consumeRequest.setStartConsumeTimeMillis(System.currentTimeMillis());
+ MessageAccessor.setConsumeStartTimeStamp(messageExt, String.valueOf(consumeRequest.getStartConsumeTimeMillis()));
+ consumedRequest.put(messageExt.getMsgId(), consumeRequest);
+ return messageExt;
+ }
+ } catch (InterruptedException ignore) {
+ }
+ return null;
+ }
+
+ void ack(final String messageId) {
+ ConsumeRequest consumeRequest = consumedRequest.remove(messageId);
+ if (consumeRequest != null) {
+ long offset = consumeRequest.getProcessQueue().removeMessage(Collections.singletonList(consumeRequest.getMessageExt()));
+ try {
+ rocketmqPullConsumer.updateConsumeOffset(consumeRequest.getMessageQueue(), offset);
+ } catch (MQClientException e) {
+ log.error("A error occurred in update consume offset process.", e);
+ }
+ }
+ }
+
+ void ack(final MessageQueue messageQueue, final ProcessQueue processQueue, final MessageExt messageExt) {
+ consumedRequest.remove(messageExt.getMsgId());
+ long offset = processQueue.removeMessage(Collections.singletonList(messageExt));
+ try {
+ rocketmqPullConsumer.updateConsumeOffset(messageQueue, offset);
+ } catch (MQClientException e) {
+ log.error("A error occurred in update consume offset process.", e);
+ }
+ }
+
+ @Override
+ public void startup() {
+ this.cleanExpireMsgExecutors.scheduleAtFixedRate(new Runnable() {
+ @Override
+ public void run() {
+ cleanExpireMsg();
+ }
+ }, clientConfig.getRmqMessageConsumeTimeout(), clientConfig.getRmqMessageConsumeTimeout(), TimeUnit.MINUTES);
+ }
+
+ @Override
+ public void shutdown() {
+ ThreadUtils.shutdownGracefully(cleanExpireMsgExecutors, 5000, TimeUnit.MILLISECONDS);
+ }
+
+ private void cleanExpireMsg() {
+ for (final Map.Entry<MessageQueue, ProcessQueue> next : rocketmqPullConsumer.getDefaultMQPullConsumerImpl()
+ .getRebalanceImpl().getProcessQueueTable().entrySet()) {
+ ProcessQueue pq = next.getValue();
+ MessageQueue mq = next.getKey();
+ ReadWriteLock lockTreeMap = getLockInProcessQueue(pq);
+ if (lockTreeMap == null) {
+ log.error("Gets tree map lock in process queue error, may be has compatibility issue");
+ return;
+ }
+
+ TreeMap<Long, MessageExt> msgTreeMap = pq.getMsgTreeMap();
+
+ int loop = msgTreeMap.size();
+ for (int i = 0; i < loop; i++) {
+ MessageExt msg = null;
+ try {
+ lockTreeMap.readLock().lockInterruptibly();
+ try {
+ if (!msgTreeMap.isEmpty()) {
+ msg = msgTreeMap.firstEntry().getValue();
+ if (System.currentTimeMillis() - Long.parseLong(MessageAccessor.getConsumeStartTimeStamp(msg))
+ > clientConfig.getRmqMessageConsumeTimeout() * 60 * 1000) {
+ //Expired, ack and remove it.
+ } else {
+ break;
+ }
+ } else {
+ break;
+ }
+ } finally {
+ lockTreeMap.readLock().unlock();
+ }
+ } catch (InterruptedException e) {
+ log.error("Gets expired message exception", e);
+ }
+
+ try {
+ rocketmqPullConsumer.sendMessageBack(msg, 3);
+ log.info("Send expired msg back. topic={}, msgId={}, storeHost={}, queueId={}, queueOffset={}",
+ msg.getTopic(), msg.getMsgId(), msg.getStoreHost(), msg.getQueueId(), msg.getQueueOffset());
+ ack(mq, pq, msg);
+ } catch (Exception e) {
+ log.error("Send back expired msg exception", e);
+ }
+ }
+ }
+ }
+
+ private ReadWriteLock getLockInProcessQueue(ProcessQueue pq) {
+ try {
+ return (ReadWriteLock) FieldUtils.readDeclaredField(pq, "lockTreeMap", true);
+ } catch (IllegalAccessException e) {
+ return null;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/c8e84adf/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/PullConsumerImpl.java
----------------------------------------------------------------------
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/PullConsumerImpl.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/PullConsumerImpl.java
new file mode 100644
index 0000000..8d396d4
--- /dev/null
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/PullConsumerImpl.java
@@ -0,0 +1,166 @@
+/*
+ * 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 io.openmessaging.rocketmq.consumer;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.Message;
+import io.openmessaging.PropertyKeys;
+import io.openmessaging.PullConsumer;
+import io.openmessaging.exception.OMSRuntimeException;
+import io.openmessaging.rocketmq.config.ClientConfig;
+import io.openmessaging.rocketmq.domain.ConsumeRequest;
+import io.openmessaging.rocketmq.utils.BeanUtils;
+import io.openmessaging.rocketmq.utils.OMSUtil;
+import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
+import org.apache.rocketmq.client.consumer.MQPullConsumer;
+import org.apache.rocketmq.client.consumer.MQPullConsumerScheduleService;
+import org.apache.rocketmq.client.consumer.PullResult;
+import org.apache.rocketmq.client.consumer.PullTaskCallback;
+import org.apache.rocketmq.client.consumer.PullTaskContext;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.impl.consumer.ProcessQueue;
+import org.apache.rocketmq.client.log.ClientLogger;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.slf4j.Logger;
+
+public class PullConsumerImpl implements PullConsumer {
+ private final DefaultMQPullConsumer rocketmqPullConsumer;
+ private final KeyValue properties;
+ private boolean started = false;
+ private String targetQueueName;
+ private final MQPullConsumerScheduleService pullConsumerScheduleService;
+ private final LocalMessageCache localMessageCache;
+ private final ClientConfig clientConfig;
+
+ final static Logger log = ClientLogger.getLog();
+
+ public PullConsumerImpl(final String queueName, final KeyValue properties) {
+ this.properties = properties;
+ this.targetQueueName = queueName;
+
+ this.clientConfig = BeanUtils.populate(properties, ClientConfig.class);
+
+ String consumerGroup = clientConfig.getRmqConsumerGroup();
+ if (null == consumerGroup || consumerGroup.isEmpty()) {
+ throw new OMSRuntimeException("-1", "Consumer Group is necessary for RocketMQ, please set it.");
+ }
+ pullConsumerScheduleService = new MQPullConsumerScheduleService(consumerGroup);
+
+ this.rocketmqPullConsumer = pullConsumerScheduleService.getDefaultMQPullConsumer();
+
+ String accessPoints = clientConfig.getOmsAccessPoints();
+ if (accessPoints == null || accessPoints.isEmpty()) {
+ throw new OMSRuntimeException("-1", "OMS AccessPoints is null or empty.");
+ }
+ this.rocketmqPullConsumer.setNamesrvAddr(accessPoints.replace(',', ';'));
+
+ this.rocketmqPullConsumer.setConsumerGroup(consumerGroup);
+
+ int maxReDeliveryTimes = clientConfig.getRmqMaxRedeliveryTimes();
+ this.rocketmqPullConsumer.setMaxReconsumeTimes(maxReDeliveryTimes);
+
+ String consumerId = OMSUtil.buildInstanceName();
+ this.rocketmqPullConsumer.setInstanceName(consumerId);
+ properties.put(PropertyKeys.CONSUMER_ID, consumerId);
+
+ this.localMessageCache = new LocalMessageCache(this.rocketmqPullConsumer, clientConfig);
+ }
+
+ @Override
+ public KeyValue properties() {
+ return properties;
+ }
+
+ @Override
+ public Message poll() {
+ MessageExt rmqMsg = localMessageCache.poll();
+ return rmqMsg == null ? null : OMSUtil.msgConvert(rmqMsg);
+ }
+
+ @Override
+ public Message poll(final KeyValue properties) {
+ MessageExt rmqMsg = localMessageCache.poll(properties);
+ return rmqMsg == null ? null : OMSUtil.msgConvert(rmqMsg);
+ }
+
+ @Override
+ public void ack(final String messageId) {
+ localMessageCache.ack(messageId);
+ }
+
+ @Override
+ public void ack(final String messageId, final KeyValue properties) {
+ localMessageCache.ack(messageId);
+ }
+
+ @Override
+ public synchronized void startup() {
+ if (!started) {
+ try {
+ registerPullTaskCallback();
+ this.pullConsumerScheduleService.start();
+ this.localMessageCache.startup();
+ } catch (MQClientException e) {
+ throw new OMSRuntimeException("-1", e);
+ }
+ }
+ this.started = true;
+ }
+
+ private void registerPullTaskCallback() {
+ this.pullConsumerScheduleService.registerPullTaskCallback(targetQueueName, new PullTaskCallback() {
+ @Override
+ public void doPullTask(final MessageQueue mq, final PullTaskContext context) {
+ MQPullConsumer consumer = context.getPullConsumer();
+ try {
+ long offset = localMessageCache.nextPullOffset(mq);
+
+ PullResult pullResult = consumer.pull(mq, "*",
+ offset, localMessageCache.nextPullBatchNums());
+ ProcessQueue pq = rocketmqPullConsumer.getDefaultMQPullConsumerImpl().getRebalanceImpl()
+ .getProcessQueueTable().get(mq);
+ switch (pullResult.getPullStatus()) {
+ case FOUND:
+ if (pq != null) {
+ pq.putMessage(pullResult.getMsgFoundList());
+ for (final MessageExt messageExt : pullResult.getMsgFoundList()) {
+ localMessageCache.submitConsumeRequest(new ConsumeRequest(messageExt, mq, pq));
+ }
+ }
+ break;
+ default:
+ break;
+ }
+ localMessageCache.updatePullOffset(mq, pullResult.getNextBeginOffset());
+ } catch (Exception e) {
+ log.error("A error occurred in pull message process.", e);
+ }
+ }
+ });
+ }
+
+ @Override
+ public synchronized void shutdown() {
+ if (this.started) {
+ this.localMessageCache.shutdown();
+ this.pullConsumerScheduleService.shutdown();
+ this.rocketmqPullConsumer.shutdown();
+ }
+ this.started = false;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/c8e84adf/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/PushConsumerImpl.java
----------------------------------------------------------------------
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/PushConsumerImpl.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/PushConsumerImpl.java
new file mode 100644
index 0000000..f9b8058
--- /dev/null
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/consumer/PushConsumerImpl.java
@@ -0,0 +1,181 @@
+/*
+ * 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 io.openmessaging.rocketmq.consumer;
+
+import io.openmessaging.BytesMessage;
+import io.openmessaging.KeyValue;
+import io.openmessaging.MessageListener;
+import io.openmessaging.OMS;
+import io.openmessaging.PropertyKeys;
+import io.openmessaging.PushConsumer;
+import io.openmessaging.ReceivedMessageContext;
+import io.openmessaging.exception.OMSRuntimeException;
+import io.openmessaging.rocketmq.config.ClientConfig;
+import io.openmessaging.rocketmq.domain.NonStandardKeys;
+import io.openmessaging.rocketmq.utils.BeanUtils;
+import io.openmessaging.rocketmq.utils.OMSUtil;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+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;
+
+public class PushConsumerImpl implements PushConsumer {
+ private final DefaultMQPushConsumer rocketmqPushConsumer;
+ private final KeyValue properties;
+ private boolean started = false;
+ private final Map<String, MessageListener> subscribeTable = new ConcurrentHashMap<>();
+ private final ClientConfig clientConfig;
+
+ public PushConsumerImpl(final KeyValue properties) {
+ this.rocketmqPushConsumer = new DefaultMQPushConsumer();
+ this.properties = properties;
+ this.clientConfig = BeanUtils.populate(properties, ClientConfig.class);
+
+ String accessPoints = clientConfig.getOmsAccessPoints();
+ if (accessPoints == null || accessPoints.isEmpty()) {
+ throw new OMSRuntimeException("-1", "OMS AccessPoints is null or empty.");
+ }
+ this.rocketmqPushConsumer.setNamesrvAddr(accessPoints.replace(',', ';'));
+
+ String consumerGroup = clientConfig.getRmqConsumerGroup();
+ if (null == consumerGroup || consumerGroup.isEmpty()) {
+ throw new OMSRuntimeException("-1", "Consumer Group is necessary for RocketMQ, please set it.");
+ }
+ this.rocketmqPushConsumer.setConsumerGroup(consumerGroup);
+ this.rocketmqPushConsumer.setMaxReconsumeTimes(clientConfig.getRmqMaxRedeliveryTimes());
+ this.rocketmqPushConsumer.setConsumeTimeout(clientConfig.getRmqMessageConsumeTimeout());
+ this.rocketmqPushConsumer.setConsumeThreadMax(clientConfig.getRmqMaxConsumeThreadNums());
+ this.rocketmqPushConsumer.setConsumeThreadMin(clientConfig.getRmqMinConsumeThreadNums());
+
+ String consumerId = OMSUtil.buildInstanceName();
+ this.rocketmqPushConsumer.setInstanceName(consumerId);
+ properties.put(PropertyKeys.CONSUMER_ID, consumerId);
+
+ this.rocketmqPushConsumer.registerMessageListener(new MessageListenerImpl());
+ }
+
+ @Override
+ public KeyValue properties() {
+ return properties;
+ }
+
+ @Override
+ public void resume() {
+ this.rocketmqPushConsumer.resume();
+ }
+
+ @Override
+ public void suspend() {
+ this.rocketmqPushConsumer.suspend();
+ }
+
+ @Override
+ public boolean isSuspended() {
+ return this.rocketmqPushConsumer.getDefaultMQPushConsumerImpl().isPause();
+ }
+
+ @Override
+ public PushConsumer attachQueue(final String queueName, final MessageListener listener) {
+ this.subscribeTable.put(queueName, listener);
+ try {
+ this.rocketmqPushConsumer.subscribe(queueName, "*");
+ } catch (MQClientException e) {
+ throw new OMSRuntimeException("-1", String.format("RocketMQ push consumer can't attach to %s.", queueName));
+ }
+ return this;
+ }
+
+ @Override
+ public synchronized void startup() {
+ if (!started) {
+ try {
+ this.rocketmqPushConsumer.start();
+ } catch (MQClientException e) {
+ throw new OMSRuntimeException("-1", e);
+ }
+ }
+ this.started = true;
+ }
+
+ @Override
+ public synchronized void shutdown() {
+ if (this.started) {
+ this.rocketmqPushConsumer.shutdown();
+ }
+ this.started = false;
+ }
+
+ class MessageListenerImpl implements MessageListenerConcurrently {
+
+ @Override
+ public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> rmqMsgList,
+ ConsumeConcurrentlyContext contextRMQ) {
+ MessageExt rmqMsg = rmqMsgList.get(0);
+ BytesMessage omsMsg = OMSUtil.msgConvert(rmqMsg);
+
+ MessageListener listener = PushConsumerImpl.this.subscribeTable.get(rmqMsg.getTopic());
+
+ if (listener == null) {
+ throw new OMSRuntimeException("-1",
+ String.format("The topic/queue %s isn't attached to this consumer", rmqMsg.getTopic()));
+ }
+
+ final KeyValue contextProperties = OMS.newKeyValue();
+ final CountDownLatch sync = new CountDownLatch(1);
+
+ contextProperties.put(NonStandardKeys.MESSAGE_CONSUME_STATUS, ConsumeConcurrentlyStatus.RECONSUME_LATER.name());
+
+ ReceivedMessageContext context = new ReceivedMessageContext() {
+ @Override
+ public KeyValue properties() {
+ return contextProperties;
+ }
+
+ @Override
+ public void ack() {
+ sync.countDown();
+ contextProperties.put(NonStandardKeys.MESSAGE_CONSUME_STATUS,
+ ConsumeConcurrentlyStatus.CONSUME_SUCCESS.name());
+ }
+
+ @Override
+ public void ack(final KeyValue properties) {
+ sync.countDown();
+ contextProperties.put(NonStandardKeys.MESSAGE_CONSUME_STATUS,
+ properties.getString(NonStandardKeys.MESSAGE_CONSUME_STATUS));
+ }
+ };
+ long begin = System.currentTimeMillis();
+ listener.onMessage(omsMsg, context);
+ long costs = System.currentTimeMillis() - begin;
+ long timeoutMills = clientConfig.getRmqMessageConsumeTimeout() * 60 * 1000;
+ try {
+ sync.await(Math.max(0, timeoutMills - costs), TimeUnit.MILLISECONDS);
+ } catch (InterruptedException ignore) {
+ }
+
+ return ConsumeConcurrentlyStatus.valueOf(contextProperties.getString(NonStandardKeys.MESSAGE_CONSUME_STATUS));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/c8e84adf/openmessaging/src/main/java/io/openmessaging/rocketmq/domain/BytesMessageImpl.java
----------------------------------------------------------------------
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/domain/BytesMessageImpl.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/domain/BytesMessageImpl.java
new file mode 100644
index 0000000..43f80ae
--- /dev/null
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/domain/BytesMessageImpl.java
@@ -0,0 +1,108 @@
+/*
+ * 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 io.openmessaging.rocketmq.domain;
+
+import io.openmessaging.BytesMessage;
+import io.openmessaging.KeyValue;
+import io.openmessaging.Message;
+import io.openmessaging.OMS;
+import org.apache.commons.lang3.builder.ToStringBuilder;
+
+public class BytesMessageImpl implements BytesMessage {
+ private KeyValue headers;
+ private KeyValue properties;
+ private byte[] body;
+
+ public BytesMessageImpl() {
+ this.headers = OMS.newKeyValue();
+ this.properties = OMS.newKeyValue();
+ }
+
+ @Override
+ public byte[] getBody() {
+ return body;
+ }
+
+ @Override
+ public BytesMessage setBody(final byte[] body) {
+ this.body = body;
+ return this;
+ }
+
+ @Override
+ public KeyValue headers() {
+ return headers;
+ }
+
+ @Override
+ public KeyValue properties() {
+ return properties;
+ }
+
+ @Override
+ public Message putHeaders(final String key, final int value) {
+ headers.put(key, value);
+ return this;
+ }
+
+ @Override
+ public Message putHeaders(final String key, final long value) {
+ headers.put(key, value);
+ return this;
+ }
+
+ @Override
+ public Message putHeaders(final String key, final double value) {
+ headers.put(key, value);
+ return this;
+ }
+
+ @Override
+ public Message putHeaders(final String key, final String value) {
+ headers.put(key, value);
+ return this;
+ }
+
+ @Override
+ public Message putProperties(final String key, final int value) {
+ properties.put(key, value);
+ return this;
+ }
+
+ @Override
+ public Message putProperties(final String key, final long value) {
+ properties.put(key, value);
+ return this;
+ }
+
+ @Override
+ public Message putProperties(final String key, final double value) {
+ properties.put(key, value);
+ return this;
+ }
+
+ @Override
+ public Message putProperties(final String key, final String value) {
+ properties.put(key, value);
+ return this;
+ }
+
+ @Override
+ public String toString() {
+ return ToStringBuilder.reflectionToString(this);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/c8e84adf/openmessaging/src/main/java/io/openmessaging/rocketmq/domain/ConsumeRequest.java
----------------------------------------------------------------------
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/domain/ConsumeRequest.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/domain/ConsumeRequest.java
new file mode 100644
index 0000000..7ce4a9b
--- /dev/null
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/domain/ConsumeRequest.java
@@ -0,0 +1,55 @@
+/*
+ * 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 io.openmessaging.rocketmq.domain;
+
+import org.apache.rocketmq.client.impl.consumer.ProcessQueue;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageQueue;
+
+public class ConsumeRequest {
+ private final MessageExt messageExt;
+ private final MessageQueue messageQueue;
+ private final ProcessQueue processQueue;
+ private long startConsumeTimeMillis;
+
+ public ConsumeRequest(final MessageExt messageExt, final MessageQueue messageQueue,
+ final ProcessQueue processQueue) {
+ this.messageExt = messageExt;
+ this.messageQueue = messageQueue;
+ this.processQueue = processQueue;
+ }
+
+ public MessageExt getMessageExt() {
+ return messageExt;
+ }
+
+ public MessageQueue getMessageQueue() {
+ return messageQueue;
+ }
+
+ public ProcessQueue getProcessQueue() {
+ return processQueue;
+ }
+
+ public long getStartConsumeTimeMillis() {
+ return startConsumeTimeMillis;
+ }
+
+ public void setStartConsumeTimeMillis(final long startConsumeTimeMillis) {
+ this.startConsumeTimeMillis = startConsumeTimeMillis;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/c8e84adf/openmessaging/src/main/java/io/openmessaging/rocketmq/domain/NonStandardKeys.java
----------------------------------------------------------------------
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/domain/NonStandardKeys.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/domain/NonStandardKeys.java
new file mode 100644
index 0000000..3639a3f
--- /dev/null
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/domain/NonStandardKeys.java
@@ -0,0 +1,30 @@
+/*
+ * 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 io.openmessaging.rocketmq.domain;
+
+public interface NonStandardKeys {
+ String CONSUMER_GROUP = "rmq.consumer.group";
+ String PRODUCER_GROUP = "rmq.producer.group";
+ String MAX_REDELIVERY_TIMES = "rmq.max.redelivery.times";
+ String MESSAGE_CONSUME_TIMEOUT = "rmq.message.consume.timeout";
+ String MAX_CONSUME_THREAD_NUMS = "rmq.max.consume.thread.nums";
+ String MIN_CONSUME_THREAD_NUMS = "rmq.min.consume.thread.nums";
+ String MESSAGE_CONSUME_STATUS = "rmq.message.consume.status";
+ String MESSAGE_DESTINATION = "rmq.message.destination";
+ String PULL_MESSAGE_BATCH_NUMS = "rmq.pull.message.batch.nums";
+ String PULL_MESSAGE_CACHE_CAPACITY = "rmq.pull.message.cache.capacity";
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/c8e84adf/openmessaging/src/main/java/io/openmessaging/rocketmq/domain/SendResultImpl.java
----------------------------------------------------------------------
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/domain/SendResultImpl.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/domain/SendResultImpl.java
new file mode 100644
index 0000000..228a9f0
--- /dev/null
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/domain/SendResultImpl.java
@@ -0,0 +1,40 @@
+/*
+ * 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 io.openmessaging.rocketmq.domain;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.SendResult;
+
+public class SendResultImpl implements SendResult {
+ private String messageId;
+ private KeyValue properties;
+
+ public SendResultImpl(final String messageId, final KeyValue properties) {
+ this.messageId = messageId;
+ this.properties = properties;
+ }
+
+ @Override
+ public String messageId() {
+ return messageId;
+ }
+
+ @Override
+ public KeyValue properties() {
+ return properties;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/c8e84adf/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/AbstractOMSProducer.java
----------------------------------------------------------------------
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/AbstractOMSProducer.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/AbstractOMSProducer.java
new file mode 100644
index 0000000..8246bcd
--- /dev/null
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/AbstractOMSProducer.java
@@ -0,0 +1,138 @@
+/*
+ * 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 io.openmessaging.rocketmq.producer;
+
+import io.openmessaging.BytesMessage;
+import io.openmessaging.KeyValue;
+import io.openmessaging.Message;
+import io.openmessaging.MessageFactory;
+import io.openmessaging.MessageHeader;
+import io.openmessaging.PropertyKeys;
+import io.openmessaging.ServiceLifecycle;
+import io.openmessaging.exception.OMSMessageFormatException;
+import io.openmessaging.exception.OMSNotSupportedException;
+import io.openmessaging.exception.OMSRuntimeException;
+import io.openmessaging.exception.OMSTimeOutException;
+import io.openmessaging.rocketmq.config.ClientConfig;
+import io.openmessaging.rocketmq.domain.BytesMessageImpl;
+import io.openmessaging.rocketmq.utils.BeanUtils;
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.log.ClientLogger;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.common.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.exception.RemotingConnectException;
+import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
+import org.slf4j.Logger;
+
+import static io.openmessaging.rocketmq.utils.OMSUtil.buildInstanceName;
+
+abstract class AbstractOMSProducer implements ServiceLifecycle, MessageFactory {
+ final static Logger log = ClientLogger.getLog();
+ final KeyValue properties;
+ final DefaultMQProducer rocketmqProducer;
+ private boolean started = false;
+ final ClientConfig clientConfig;
+
+ AbstractOMSProducer(final KeyValue properties) {
+ this.properties = properties;
+ this.rocketmqProducer = new DefaultMQProducer();
+ this.clientConfig = BeanUtils.populate(properties, ClientConfig.class);
+
+ String accessPoints = clientConfig.getOmsAccessPoints();
+ if (accessPoints == null || accessPoints.isEmpty()) {
+ throw new OMSRuntimeException("-1", "OMS AccessPoints is null or empty.");
+ }
+ this.rocketmqProducer.setNamesrvAddr(accessPoints.replace(',', ';'));
+ this.rocketmqProducer.setProducerGroup(clientConfig.getRmqProducerGroup());
+
+ String producerId = buildInstanceName();
+ this.rocketmqProducer.setSendMsgTimeout(clientConfig.getOmsOperationTimeout());
+ this.rocketmqProducer.setInstanceName(producerId);
+ this.rocketmqProducer.setMaxMessageSize(1024 * 1024 * 4);
+ properties.put(PropertyKeys.PRODUCER_ID, producerId);
+ }
+
+ @Override
+ public synchronized void startup() {
+ if (!started) {
+ try {
+ this.rocketmqProducer.start();
+ } catch (MQClientException e) {
+ throw new OMSRuntimeException("-1", e);
+ }
+ }
+ this.started = true;
+ }
+
+ @Override
+ public synchronized void shutdown() {
+ if (this.started) {
+ this.rocketmqProducer.shutdown();
+ }
+ this.started = false;
+ }
+
+ OMSRuntimeException checkProducerException(String topic, String msgId, Throwable e) {
+ if (e instanceof MQClientException) {
+ if (e.getCause() != null) {
+ if (e.getCause() instanceof RemotingTimeoutException) {
+ return new OMSTimeOutException("-1", String.format("Send message to broker timeout, %dms, Topic=%s, msgId=%s",
+ this.rocketmqProducer.getSendMsgTimeout(), topic, msgId), e);
+ } else if (e.getCause() instanceof MQBrokerException || e.getCause() instanceof RemotingConnectException) {
+ MQBrokerException brokerException = (MQBrokerException) e.getCause();
+ return new OMSRuntimeException("-1", String.format("Received a broker exception, Topic=%s, msgId=%s, %s",
+ topic, msgId, brokerException.getErrorMessage()), e);
+ }
+ }
+ // Exception thrown by local.
+ else {
+ MQClientException clientException = (MQClientException) e;
+ if (-1 == clientException.getResponseCode()) {
+ return new OMSRuntimeException("-1", String.format("Topic does not exist, Topic=%s, msgId=%s",
+ topic, msgId), e);
+ } else if (ResponseCode.MESSAGE_ILLEGAL == clientException.getResponseCode()) {
+ return new OMSMessageFormatException("-1", String.format("A illegal message for RocketMQ, Topic=%s, msgId=%s",
+ topic, msgId), e);
+ }
+ }
+ }
+ return new OMSRuntimeException("-1", "Send message to RocketMQ broker failed.", e);
+ }
+
+ protected void checkMessageType(Message message) {
+ if (!(message instanceof BytesMessage)) {
+ throw new OMSNotSupportedException("-1", "Only BytesMessage is supported.");
+ }
+ }
+
+ @Override
+ public BytesMessage createBytesMessageToTopic(final String topic, final byte[] body) {
+ BytesMessage bytesMessage = new BytesMessageImpl();
+ bytesMessage.setBody(body);
+ bytesMessage.headers().put(MessageHeader.TOPIC, topic);
+ return bytesMessage;
+ }
+
+ @Override
+ public BytesMessage createBytesMessageToQueue(final String queue, final byte[] body) {
+ BytesMessage bytesMessage = new BytesMessageImpl();
+ bytesMessage.setBody(body);
+ bytesMessage.headers().put(MessageHeader.QUEUE, queue);
+ return bytesMessage;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/c8e84adf/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/ProducerImpl.java
----------------------------------------------------------------------
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/ProducerImpl.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/ProducerImpl.java
new file mode 100644
index 0000000..2c00c60
--- /dev/null
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/ProducerImpl.java
@@ -0,0 +1,124 @@
+/*
+ * 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 io.openmessaging.rocketmq.producer;
+
+import io.openmessaging.BytesMessage;
+import io.openmessaging.KeyValue;
+import io.openmessaging.Message;
+import io.openmessaging.MessageHeader;
+import io.openmessaging.Producer;
+import io.openmessaging.Promise;
+import io.openmessaging.PropertyKeys;
+import io.openmessaging.SendResult;
+import io.openmessaging.exception.OMSRuntimeException;
+import io.openmessaging.rocketmq.promise.DefaultPromise;
+import io.openmessaging.rocketmq.utils.OMSUtil;
+import org.apache.rocketmq.client.producer.SendCallback;
+import org.apache.rocketmq.client.producer.SendStatus;
+
+import static io.openmessaging.rocketmq.utils.OMSUtil.msgConvert;
+
+public class ProducerImpl extends AbstractOMSProducer implements Producer {
+
+ public ProducerImpl(final KeyValue properties) {
+ super(properties);
+ }
+
+ @Override
+ public KeyValue properties() {
+ return properties;
+ }
+
+ @Override
+ public SendResult send(final Message message) {
+ return send(message, this.rocketmqProducer.getSendMsgTimeout());
+ }
+
+ @Override
+ public SendResult send(final Message message, final KeyValue properties) {
+ long timeout = properties.containsKey(PropertyKeys.OPERATION_TIMEOUT)
+ ? properties.getInt(PropertyKeys.OPERATION_TIMEOUT) : this.rocketmqProducer.getSendMsgTimeout();
+ return send(message, timeout);
+ }
+
+ private SendResult send(final Message message, long timeout) {
+ checkMessageType(message);
+ org.apache.rocketmq.common.message.Message rmqMessage = msgConvert((BytesMessage) message);
+ try {
+ org.apache.rocketmq.client.producer.SendResult rmqResult = this.rocketmqProducer.send(rmqMessage, timeout);
+ if (!rmqResult.getSendStatus().equals(SendStatus.SEND_OK)) {
+ log.error(String.format("Send message to RocketMQ failed, %s", message));
+ throw new OMSRuntimeException("-1", "Send message to RocketMQ broker failed.");
+ }
+ message.headers().put(MessageHeader.MESSAGE_ID, rmqResult.getMsgId());
+ return OMSUtil.sendResultConvert(rmqResult);
+ } catch (Exception e) {
+ log.error(String.format("Send message to RocketMQ failed, %s", message), e);
+ throw checkProducerException(rmqMessage.getTopic(), message.headers().getString(MessageHeader.MESSAGE_ID), e);
+ }
+ }
+
+ @Override
+ public Promise<SendResult> sendAsync(final Message message) {
+ return sendAsync(message, this.rocketmqProducer.getSendMsgTimeout());
+ }
+
+ @Override
+ public Promise<SendResult> sendAsync(final Message message, final KeyValue properties) {
+ long timeout = properties.containsKey(PropertyKeys.OPERATION_TIMEOUT)
+ ? properties.getInt(PropertyKeys.OPERATION_TIMEOUT) : this.rocketmqProducer.getSendMsgTimeout();
+ return sendAsync(message, timeout);
+ }
+
+ private Promise<SendResult> sendAsync(final Message message, long timeout) {
+ checkMessageType(message);
+ org.apache.rocketmq.common.message.Message rmqMessage = msgConvert((BytesMessage) message);
+ final Promise<SendResult> promise = new DefaultPromise<>();
+ try {
+ this.rocketmqProducer.send(rmqMessage, new SendCallback() {
+ @Override
+ public void onSuccess(final org.apache.rocketmq.client.producer.SendResult rmqResult) {
+ message.headers().put(MessageHeader.MESSAGE_ID, rmqResult.getMsgId());
+ promise.set(OMSUtil.sendResultConvert(rmqResult));
+ }
+
+ @Override
+ public void onException(final Throwable e) {
+ promise.setFailure(e);
+ }
+ }, timeout);
+ } catch (Exception e) {
+ promise.setFailure(e);
+ }
+ return promise;
+ }
+
+ @Override
+ public void sendOneway(final Message message) {
+ checkMessageType(message);
+ org.apache.rocketmq.common.message.Message rmqMessage = msgConvert((BytesMessage) message);
+ try {
+ this.rocketmqProducer.sendOneway(rmqMessage);
+ } catch (Exception ignore) { //Ignore the oneway exception.
+ }
+ }
+
+ @Override
+ public void sendOneway(final Message message, final KeyValue properties) {
+ sendOneway(message);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/c8e84adf/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/SequenceProducerImpl.java
----------------------------------------------------------------------
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/SequenceProducerImpl.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/SequenceProducerImpl.java
new file mode 100644
index 0000000..05225cc
--- /dev/null
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/producer/SequenceProducerImpl.java
@@ -0,0 +1,95 @@
+/*
+ * 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 io.openmessaging.rocketmq.producer;
+
+import io.openmessaging.BytesMessage;
+import io.openmessaging.KeyValue;
+import io.openmessaging.Message;
+import io.openmessaging.MessageHeader;
+import io.openmessaging.SequenceProducer;
+import io.openmessaging.rocketmq.utils.OMSUtil;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import org.apache.rocketmq.client.Validators;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.SendResult;
+
+public class SequenceProducerImpl extends AbstractOMSProducer implements SequenceProducer {
+
+ private BlockingQueue<Message> msgCacheQueue;
+
+ public SequenceProducerImpl(final KeyValue properties) {
+ super(properties);
+ this.msgCacheQueue = new LinkedBlockingQueue<>();
+ }
+
+ @Override
+ public KeyValue properties() {
+ return properties;
+ }
+
+ @Override
+ public void send(final Message message) {
+ checkMessageType(message);
+ org.apache.rocketmq.common.message.Message rmqMessage = OMSUtil.msgConvert((BytesMessage) message);
+ try {
+ Validators.checkMessage(rmqMessage, this.rocketmqProducer);
+ } catch (MQClientException e) {
+ throw checkProducerException(rmqMessage.getTopic(), message.headers().getString(MessageHeader.MESSAGE_ID), e);
+ }
+ msgCacheQueue.add(message);
+ }
+
+ @Override
+ public void send(final Message message, final KeyValue properties) {
+ send(message);
+ }
+
+ @Override
+ public synchronized void commit() {
+ List<Message> messages = new ArrayList<>();
+ msgCacheQueue.drainTo(messages);
+
+ List<org.apache.rocketmq.common.message.Message> rmqMessages = new ArrayList<>();
+
+ for (Message message : messages) {
+ rmqMessages.add(OMSUtil.msgConvert((BytesMessage) message));
+ }
+
+ if (rmqMessages.size() == 0) {
+ return;
+ }
+
+ try {
+ SendResult sendResult = this.rocketmqProducer.send(rmqMessages);
+ String[] msgIdArray = sendResult.getMsgId().split(",");
+ for (int i = 0; i < messages.size(); i++) {
+ Message message = messages.get(i);
+ message.headers().put(MessageHeader.MESSAGE_ID, msgIdArray[i]);
+ }
+ } catch (Exception e) {
+ throw checkProducerException("", "", e);
+ }
+ }
+
+ @Override
+ public synchronized void rollback() {
+ msgCacheQueue.clear();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/c8e84adf/openmessaging/src/main/java/io/openmessaging/rocketmq/promise/DefaultPromise.java
----------------------------------------------------------------------
diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/promise/DefaultPromise.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/promise/DefaultPromise.java
new file mode 100644
index 0000000..c863ccf
--- /dev/null
+++ b/openmessaging/src/main/java/io/openmessaging/rocketmq/promise/DefaultPromise.java
@@ -0,0 +1,227 @@
+/*
+ * 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 io.openmessaging.rocketmq.promise;
+
+import io.openmessaging.Promise;
+import io.openmessaging.PromiseListener;
+import io.openmessaging.exception.OMSRuntimeException;
+import java.util.ArrayList;
+import java.util.List;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DefaultPromise<V> implements Promise<V> {
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultPromise.class);
+ private final Object lock = new Object();
+ private volatile FutureState state = FutureState.DOING;
+ private V result = null;
+ private long timeout;
+ private long createTime;
+ private Throwable exception = null;
+ private List<PromiseListener<V>> promiseListenerList;
+
+ public DefaultPromise() {
+ createTime = System.currentTimeMillis();
+ promiseListenerList = new ArrayList<>();
+ timeout = 5000;
+ }
+
+ @Override
+ public boolean cancel(final boolean mayInterruptIfRunning) {
+ return false;
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return state.isCancelledState();
+ }
+
+ @Override
+ public boolean isDone() {
+ return state.isDoneState();
+ }
+
+ @Override
+ public V get() {
+ return result;
+ }
+
+ @Override
+ public V get(final long timeout) {
+ synchronized (lock) {
+ if (!isDoing()) {
+ return getValueOrThrowable();
+ }
+
+ if (timeout <= 0) {
+ try {
+ lock.wait();
+ } catch (Exception e) {
+ cancel(e);
+ }
+ return getValueOrThrowable();
+ } else {
+ long waitTime = timeout - (System.currentTimeMillis() - createTime);
+ if (waitTime > 0) {
+ for (;; ) {
+ try {
+ lock.wait(waitTime);
+ } catch (InterruptedException e) {
+ LOG.error("promise get value interrupted,excepiton:{}", e.getMessage());
+ }
+
+ if (!isDoing()) {
+ break;
+ } else {
+ waitTime = timeout - (System.currentTimeMillis() - createTime);
+ if (waitTime <= 0) {
+ break;
+ }
+ }
+ }
+ }
+
+ if (isDoing()) {
+ timeoutSoCancel();
+ }
+ }
+ return getValueOrThrowable();
+ }
+ }
+
+ @Override
+ public boolean set(final V value) {
+ if (value == null)
+ return false;
+ this.result = value;
+ return done();
+ }
+
+ @Override
+ public boolean setFailure(final Throwable cause) {
+ if (cause == null)
+ return false;
+ this.exception = cause;
+ return done();
+ }
+
+ @Override
+ public void addListener(final PromiseListener<V> listener) {
+ if (listener == null) {
+ throw new NullPointerException("FutureListener is null");
+ }
+
+ boolean notifyNow = false;
+ synchronized (lock) {
+ if (!isDoing()) {
+ notifyNow = true;
+ } else {
+ if (promiseListenerList == null) {
+ promiseListenerList = new ArrayList<>();
+ }
+ promiseListenerList.add(listener);
+ }
+ }
+
+ if (notifyNow) {
+ notifyListener(listener);
+ }
+ }
+
+ @Override
+ public Throwable getThrowable() {
+ return exception;
+ }
+
+ private void notifyListeners() {
+ if (promiseListenerList != null) {
+ for (PromiseListener<V> listener : promiseListenerList) {
+ notifyListener(listener);
+ }
+ }
+ }
+
+ private boolean isSuccess() {
+ return isDone() && (exception == null);
+ }
+
+ private void timeoutSoCancel() {
+ synchronized (lock) {
+ if (!isDoing()) {
+ return;
+ }
+ state = FutureState.CANCELLED;
+ exception = new RuntimeException("Get request result is timeout or interrupted");
+ lock.notifyAll();
+ }
+ notifyListeners();
+ }
+
+ private V getValueOrThrowable() {
+ if (exception != null) {
+ Throwable e = exception.getCause() != null ? exception.getCause() : exception;
+ throw new OMSRuntimeException("-1", e);
+ }
+ notifyListeners();
+ return result;
+ }
+
+ private boolean isDoing() {
+ return state.isDoingState();
+ }
+
+ private boolean done() {
+ synchronized (lock) {
+ if (!isDoing()) {
+ return false;
+ }
+
+ state = FutureState.DONE;
+ lock.notifyAll();
+ }
+
+ notifyListeners();
+ return true;
+ }
+
+ private void notifyListener(final PromiseListener<V> listener) {
+ try {
+ if (exception != null)
+ listener.operationFailed(this);
+ else
+ listener.operationCompleted(this);
+ } catch (Throwable t) {
+ LOG.error("notifyListener {} Error:{}", listener.getClass().getSimpleName(), t);
+ }
+ }
+
+ private boolean cancel(Exception e) {
+ synchronized (lock) {
+ if (!isDoing()) {
+ return false;
+ }
+
+ state = FutureState.CANCELLED;
+ exception = e;
+ lock.notifyAll();
+ }
+
+ notifyListeners();
+ return true;
+ }
+}
+