You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by di...@apache.org on 2019/01/17 13:20:08 UTC
[rocketmq-spring] 02/37: First release version 2.0.0
This is an automated email from the ASF dual-hosted git repository.
dinglei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-spring.git
commit d7bc7be61a5b8cb26771f46794ff975da6e257fc
Author: walking98 <wi...@gmail.com>
AuthorDate: Thu Dec 6 17:02:42 2018 +0800
First release version 2.0.0
---
.gitignore | 13 +
LICENSE | 202 +++++++
NOTICE | 5 +
README.md | 278 +++++++++-
README_zh_CN.md | 268 +++++++++
pom.xml | 80 +++
rocketmq-spring-boot-parent/pom.xml | 177 ++++++
rocketmq-spring-boot-samples/LICENSE | 202 +++++++
rocketmq-spring-boot-samples/README.md | 38 ++
rocketmq-spring-boot-samples/README_zh_CN.md | 48 ++
rocketmq-spring-boot-samples/pom.xml | 89 +++
.../rocketmq-consume-demo/pom.xml | 31 ++
.../samples/springboot/ConsumerApplication.java | 33 ++
.../samples/springboot/consumer/Checker.java | 44 ++
.../springboot/consumer/MessageExtConsumer.java | 46 ++
.../consumer/OrderPaidEventConsumer.java | 36 ++
.../springboot/consumer/StringConsumer.java | 34 ++
.../consumer/StringTransactionalConsumer.java | 34 ++
.../samples/springboot/domain/OrderPaidEvent.java | 54 ++
.../src/main/resources/application.properties | 7 +
.../rocketmq-produce-demo/pom.xml | 32 ++
.../samples/springboot/ProducerApplication.java | 171 ++++++
.../samples/springboot/domain/OrderPaidEvent.java | 50 ++
.../src/main/resources/application.properties | 6 +
.../style/copyright/Apache.xml | 23 +
.../style/copyright/profiles_settings.xml | 64 +++
.../style/rmq_checkstyle.xml | 135 +++++
.../style/rmq_codeStyle.xml | 157 ++++++
rocketmq-spring-boot-starter/pom.xml | 51 ++
rocketmq-spring-boot/README.md | 7 +
rocketmq-spring-boot/pom.xml | 104 ++++
.../rocketmq/spring/annotation/ConsumeMode.java | 30 +
.../rocketmq/spring/annotation/MessageModel.java | 33 ++
.../spring/annotation/RocketMQMessageListener.java | 72 +++
.../annotation/RocketMQTransactionListener.java | 69 +++
.../rocketmq/spring/annotation/SelectorType.java | 33 ++
.../config/ListenerContainerConfiguration.java | 133 +++++
.../spring/config/RocketMQAutoConfiguration.java | 104 ++++
.../spring/config/RocketMQConfigUtils.java | 29 +
.../rocketmq/spring/config/RocketMQProperties.java | 144 +++++
.../RocketMQTransactionAnnotationProcessor.java | 110 ++++
.../rocketmq/spring/config/TransactionHandler.java | 75 +++
.../spring/config/TransactionHandlerRegistry.java | 50 ++
.../rocketmq/spring/core/RocketMQListener.java | 22 +
.../core/RocketMQLocalTransactionListener.java | 25 +
.../spring/core/RocketMQLocalTransactionState.java | 24 +
.../RocketMQPushConsumerLifecycleListener.java | 24 +
.../rocketmq/spring/core/RocketMQTemplate.java | 609 +++++++++++++++++++++
.../support/DefaultRocketMQListenerContainer.java | 423 ++++++++++++++
.../support/RocketMQConsumerLifecycleListener.java | 22 +
.../rocketmq/spring/support/RocketMQHeaders.java | 34 ++
.../spring/support/RocketMQListenerContainer.java | 30 +
.../rocketmq/spring/support/RocketMQUtil.java | 168 ++++++
.../src/main/resources/META-INF/spring.factories | 2 +
.../config/RocketMQAutoConfigurationTest.java | 92 ++++
.../rocketmq/spring/support/RocketMQUtilTest.java | 37 ++
style/copyright/Apache.xml | 23 +
style/copyright/profiles_settings.xml | 64 +++
style/rmq_checkstyle.xml | 135 +++++
style/rmq_codeStyle.xml | 157 ++++++
60 files changed, 5291 insertions(+), 1 deletion(-)
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..06bb326
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,13 @@
+.idea
+.classpath
+.project
+.settings/
+target/
+*.log*
+*.iml
+.idea/
+*.versionsBackup
+!NOTICE-BIN
+!LICENSE-BIN
+.DS_Store
+.vscode
diff --git a/LICENSE b/LICENSE
new file mode 100644
index 0000000..7a4a3ea
--- /dev/null
+++ b/LICENSE
@@ -0,0 +1,202 @@
+
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ Licensed 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.
\ No newline at end of file
diff --git a/NOTICE b/NOTICE
new file mode 100644
index 0000000..f07078d
--- /dev/null
+++ b/NOTICE
@@ -0,0 +1,5 @@
+Apache RocketMQ
+Copyright 2016-2018 The Apache Software Foundation
+
+This product includes software developed at
+The Apache Software Foundation (http://www.apache.org/).
diff --git a/README.md b/README.md
index 1ce682f..9ca1362 100644
--- a/README.md
+++ b/README.md
@@ -1 +1,277 @@
-# rocketmq-spring
+# rocketmq-spring-boot
+
+[中文](./README_zh_CN.md)
+
+[](https://www.apache.org/licenses/LICENSE-2.0.html)
+
+## Introduction
+
+This is a sub-repo of RocketMQ TLP (http://rocketmq.apache.org/), It aims to help developers quickly integrate [RocketMQ](http://rocketmq.apache.org/) in [Spring Boot](http://projects.spring.io/spring-boot/). Support the Spring Message specification to facilitate developers to quickly switch from other MQ to RocketMQ.
+
+## How To Contribute
+
+We are always very happy to have contributions, whether for trivial cleanups or big new features. Please see the RocketMQ main website to read [details](http://rocketmq.apache.org/docs/how-to-contribute/)
+
+
+## Prerequisites
+- JDK 1.8 and above
+- [Maven](http://maven.apache.org/) 3.0 and above
+
+## Build and Install with local maven repository
+
+```
+ mvn clean install
+```
+
+## Features:
+
+- [x] synchronous transmission
+- [x] synchronous ordered transmission
+- [x] asynchronous transmission
+- [x] asynchronous ordered transmission
+- [x] orderly consume
+- [x] concurrently consume(broadcasting/clustering)
+- [x] one-way transmission
+- [x] transaction transmission
+- [ ] pull consume
+
+## Quick Start
+
+Please see the complete sample [rocketmq-spring-boot-samples](rocketmq-spring-boot-samples)
+
+Note: Current RELEASE.VERSION=2.0.0
+
+```xml
+<!--add dependency in pom.xml-->
+<dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-spring-boot-starter</artifactId>
+ <version>${RELEASE.VERSION}</version>
+</dependency>
+```
+
+### Produce Message
+
+```properties
+## application.properties
+spring.rocketmq.nameServer=127.0.0.1:9876
+spring.rocketmq.producer.group=my-group
+```
+
+> Note:
+>
+> Maybe you need change `127.0.0.1:9876` with your real NameServer address for RocketMQ
+
+```java
+@SpringBootApplication
+public class ProducerApplication implements CommandLineRunner{
+ @Resource
+ private RocketMQTemplate rocketMQTemplate;
+
+ public static void main(String[] args){
+ SpringApplication.run(ProducerApplication.class, args);
+ }
+
+ public void run(String... args) throws Exception {
+ rocketMQTemplate.convertAndSend("test-topic-1", "Hello, World!");
+ rocketMQTemplate.send("test-topic-1", MessageBuilder.withPayload("Hello, World! I'm from spring message").build());
+ rocketMQTemplate.convertAndSend("test-topic-2", new OrderPaidEvent("T_001", new BigDecimal("88.00")));
+
+// rocketMQTemplate.destroy(); // notes: once rocketMQTemplate be destroyed, you can not send any message again with this rocketMQTemplate
+ }
+
+ @Data
+ @AllArgsConstructor
+ public class OrderPaidEvent implements Serializable{
+ private String orderId;
+
+ private BigDecimal paidMoney;
+ }
+}
+```
+
+> More relevant configurations for producing:
+>
+> ```properties
+> spring.rocketmq.producer.retryTimesWhenSendAsyncFailed=0
+> spring.rocketmq.producer.sendMessageTimeout=300000
+> spring.rocketmq.producer.compressMessageBodyOverHowmuch=4096
+> spring.rocketmq.producer.maxMessageSize=4194304
+> spring.rocketmq.producer.retryAnotherBrokerWhenNotStoreOk=false
+> spring.rocketmq.producer.retryTimesWhenSendFailed=2
+> ```
+
+
+### Send message in transaction and implement local check Listener
+```java
+@SpringBootApplication
+public class ProducerApplication implements CommandLineRunner{
+ @Resource
+ private RocketMQTemplate rocketMQTemplate;
+
+ public static void main(String[] args){
+ SpringApplication.run(ProducerApplication.class, args);
+ }
+
+ public void run(String... args) throws Exception {
+ try {
+ // Build a SpringMessage for sending in transaction
+ Message msg = MessageBuilder.withPayload(..)...
+ // In sendMessageInTransaction(), the first parameter transaction name ("test")
+ // must be same with the @RocketMQTransactionListener's member field 'transName'
+ rocketMQTemplate.sendMessageInTransaction("test", "test-topic" msg, null);
+ } catch (MQClientException e) {
+ e.printStackTrace(System.out);
+ }
+ }
+
+ // Define transaction listener with the annotation @RocketMQTransactionListener
+ @RocketMQTransactionListener(transName="test")
+ class TransactionListenerImpl implements RocketMQLocalTransactionListener {
+ @Override
+ public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
+ // ... local transaction process, return bollback, commit or unknown
+ return RocketMQLocalTransactionState.UNKNOWN;
+ }
+
+ @Override
+ public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
+ // ... check transaction status and return bollback, commit or unknown
+ return RocketMQLocalTransactionState.COMMIT;
+ }
+ }
+}
+```
+
+### Consume Message
+
+```properties
+## application.properties
+spring.rocketmq.nameServer=127.0.0.1:9876
+```
+
+> Note:
+>
+> Maybe you need change `127.0.0.1:9876` with your real NameServer address for RocketMQ
+
+```java
+@SpringBootApplication
+public class ConsumerApplication{
+
+ public static void main(String[] args){
+ SpringApplication.run(ConsumerApplication.class, args);
+ }
+
+ @Slf4j
+ @Service
+ @RocketMQMessageListener(topic = "test-topic-1", consumerGroup = "my-consumer_test-topic-1")
+ public class MyConsumer1 implements RocketMQListener<String>{
+ public void onMessage(String message) {
+ log.info("received message: {}", message);
+ }
+ }
+
+ @Slf4j
+ @Service
+ @RocketMQMessageListener(topic = "test-topic-2", consumerGroup = "my-consumer_test-topic-2")
+ public class MyConsumer2 implements RocketMQListener<OrderPaidEvent>{
+ public void onMessage(OrderPaidEvent orderPaidEvent) {
+ log.info("received orderPaidEvent: {}", orderPaidEvent);
+ }
+ }
+}
+```
+
+> More relevant configurations for consuming:
+>
+> see: [RocketMQMessageListener](src/main/java/org/apache/rocketmq/spring/starter/annotation/RocketMQMessageListener.java)
+
+
+## FAQ
+
+1. How to connected many `nameserver` on production environment？
+
+ `spring.rocketmq.nameServer` support the configuration of multiple `nameserver`, separated by `;`. For example: `172.19.0.1: 9876; 172.19.0.2: 9876`
+
+1. When was `rocketMQTemplate` destroyed?
+
+ Developers do not need to manually execute the `rocketMQTemplate.destroy ()` method when using `rocketMQTemplate` to send a message in the project, and` rocketMQTemplate` will be destroyed automatically when the spring container is destroyed.
+
+1. start exception：`Caused by: org.apache.rocketmq.client.exception.MQClientException: The consumer group[xxx] has been created before, specify another name please`
+
+ RocketMQ in the design do not want a consumer to deal with multiple types of messages at the same time, so the same `consumerGroup` consumer responsibility should be the same, do not do different things (that is, consumption of multiple topics). Suggested `consumerGroup` and` topic` one correspondence.
+
+1. How is the message content body being serialized and deserialized?
+
+ RocketMQ's message body is stored as `byte []`. When the business system message content body if it is `java.lang.String` type, unified in accordance with` utf-8` code into `byte []`; If the business system message content is not `java.lang.String` Type, then use [jackson-databind](https://github.com/FasterXML/jackson-databind) serialized into the `JSON` format string, and then unified in accordance with` utf-8` code into `byte [] `.
+
+1. How do I specify the `tags` for topic?
+
+ RocketMQ best practice recommended: an application as much as possible with one Topic, the message sub-type with `tags` to identify,` tags` can be set by the application free.
+
+ When you use `rocketMQTemplate` to send a message, set the destination of the message by setting the` destination` parameter of the send method. The `destination` format is `topicName:tagName`, `:` Precedes the name of the topic, followed by the `tags` name.
+
+ > Note:
+ >
+ > `tags` looks a complex, but when sending a message , the destination can only specify one topic under a `tag`, can not specify multiple.
+
+1. How do I set the message's `key` when sending a message?
+
+ You can send a message by overloading method like `xxxSend(String destination, Message<?> msg, ...)`, setting `headers` of `msg`. for example:
+
+ ```java
+ Message<?> message = MessageBuilder.withPayload(payload).setHeader(MessageConst.PROPERTY_KEYS, msgId).build();
+ rocketMQTemplate.send("topic-test", message);
+ ```
+
+ Similarly, you can also set the message `FLAG`,` WAIT_STORE_MSG_OK` and some other user-defined other header information according to the above method.
+
+ > Note:
+ >
+ > In the case of converting Spring's Message to RocketMQ's Message, to prevent the `header` information from conflicting with RocketMQ's system properties, the prefix `USERS_` was added in front of all `header` names. So if you want to get a custom message header when consuming, please pass through the key at the beginning of `USERS_` in the header.
+
+1. When consume message, in addition to get the message `payload`, but also want to get RocketMQ message of other system attributes, how to do?
+
+ Consumers in the realization of `RocketMQListener` interface, only need to be generic for the` MessageExt` can, so in the `onMessage` method will receive RocketMQ native 'MessageExt` message.
+
+ ```java
+ @Slf4j
+ @Service
+ @RocketMQMessageListener(topic = "test-topic-1", consumerGroup = "my-consumer_test-topic-1")
+ public class MyConsumer2 implements RocketMQListener<MessageExt>{
+ public void onMessage(MessageExt messageExt) {
+ log.info("received messageExt: {}", messageExt);
+ }
+ }
+ ```
+
+1. How do I specify where consumers start consuming messages?
+
+ The default consume offset please refer: [RocketMQ FAQ](http://rocketmq.apache.org/docs/faq/).
+ To customize the consumer's starting location, simply add a `RocketMQPushConsumerLifecycleListener` interface implementation to the consumer class. Examples are as follows:
+
+ ```java
+ @Slf4j
+ @Service
+ @RocketMQMessageListener(topic = "test-topic-1", consumerGroup = "my-consumer_test-topic-1")
+ public class MyConsumer1 implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {
+ @Override
+ public void onMessage(String message) {
+ log.info("received message: {}", message);
+ }
+
+ @Override
+ public void prepareStart(final DefaultMQPushConsumer consumer) {
+ // set consumer consume message from now
+ consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
+ consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis()));
+ }
+ }
+ ```
+
+ Similarly, any other configuration on `DefaultMQPushConsumer` can be done in the same way as above.
+
+
+1. How do I send transactional messages?
+ It needs two steps on client side: a) Define a class which is annotated with @RocketMQTransactionListener and implements RocketMQLocalTransactionListener interface, in which, the executeLocalTransaction() and checkLocalTransaction() methods are implemented;
+ b) Invoke the sendMessageInTransaction() method with the RocketMQTemplate API. Note: The first parameter of this method is correlated with the txProducerGroup attribute of @RocketMQTransactionListener. It can be null if using the default transaction producer group.
diff --git a/README_zh_CN.md b/README_zh_CN.md
new file mode 100644
index 0000000..c7c928a
--- /dev/null
+++ b/README_zh_CN.md
@@ -0,0 +1,268 @@
+# rocketmq-spring-boot
+
+[English](./README.md)
+
+[](https://www.apache.org/licenses/LICENSE-2.0.html)
+
+帮助开发者在[Spring Boot](http://projects.spring.io/spring-boot/)中快速集成[RocketMQ](http://rocketmq.apache.org/)。支持Spring Message规范，方便开发者从其它MQ快速切换到RocketMQ。
+
+## 如何贡献和帮助社区
+
+我们永远欢迎开发者的帮助来使这个项目更加完善，无论是小的文档还是大的功能新特性，请参考RocketMQ的主站了解[细节](http://rocketmq.apache.org/docs/how-to-contribute/)
+
+## 前提条件
+- JDK 1.8 and above
+- [Maven](http://maven.apache.org/) 3.0 and above
+
+功能特性：
+
+- [x] 同步发送
+- [x] 同步顺序发送
+- [x] 异步发送
+- [x] 异步顺序发送
+- [x] 顺序消费
+- [x] 并发消费（广播/集群）
+- [x] one-way方式发送
+- [x] 事务方式发送
+- [ ] pull消费
+
+## Quick Start
+
+下面列出来了一些关键点，完整的示例请参考： [rocketmq-spring-boot-samples](rocketmq-spring-boot-samples)
+
+注意:当前的RELEASE.VERSION=2.0.0
+
+```xml
+<!--在pom.xml中添加依赖-->
+<dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-spring-boot-starter</artifactId>
+ <version>${RELEASE.VERSION}</version>
+</dependency>
+```
+
+### 发送消息
+
+```properties
+## application.properties
+spring.rocketmq.nameServer=127.0.0.1:9876
+spring.rocketmq.producer.group=my-group
+```
+
+> 注意:
+>
+> 请将上述示例配置中的`127.0.0.1:9876`替换成真实RocketMQ的NameServer地址与端口
+
+```java
+@SpringBootApplication
+public class ProducerApplication implements CommandLineRunner{
+ @Resource
+ private RocketMQTemplate rocketMQTemplate;
+
+ public static void main(String[] args){
+ SpringApplication.run(ProducerApplication.class, args);
+ }
+
+ public void run(String... args) throws Exception {
+ rocketMQTemplate.convertAndSend("test-topic-1", "Hello, World!");
+ rocketMQTemplate.send("test-topic-1", MessageBuilder.withPayload("Hello, World! I'm from spring message").build());
+ rocketMQTemplate.convertAndSend("test-topic-2", new OrderPaidEvent("T_001", new BigDecimal("88.00")));
+
+// rocketMQTemplate.destroy(); // notes: once rocketMQTemplate be destroyed, you can not send any message again with this rocketMQTemplate
+ }
+
+ @Data
+ @AllArgsConstructor
+ public class OrderPaidEvent implements Serializable{
+ private String orderId;
+
+ private BigDecimal paidMoney;
+ }
+}
+```
+
+### 在发送客户端发送事务性消息并且实现回查Listener
+```java
+@SpringBootApplication
+public class ProducerApplication implements CommandLineRunner{
+ @Resource
+ private RocketMQTemplate rocketMQTemplate;
+
+ public static void main(String[] args){
+ SpringApplication.run(ProducerApplication.class, args);
+ }
+
+ public void run(String... args) throws Exception {
+ try {
+ // Build a SpringMessage for sending in transaction
+ Message msg = MessageBuilder.withPayload(..)...
+ // In sendMessageInTransaction(), the first parameter transaction name ("test")
+ // must be same with the @RocketMQTransactionListener's member field 'transName'
+ rocketMQTemplate.sendMessageInTransaction("test", "test-topic" msg, null);
+ } catch (MQClientException e) {
+ e.printStackTrace(System.out);
+ }
+ }
+
+ // Define transaction listener with the annotation @RocketMQTransactionListener
+ @RocketMQTransactionListener(transName="test")
+ class TransactionListenerImpl implements RocketMQLocalTransactionListener {
+ @Override
+ public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
+ // ... local transaction process, return bollback, commit or unknown
+ return RocketMQLocalTransactionState.UNKNOWN;
+ }
+
+ @Override
+ public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
+ // ... check transaction status and return bollback, commit or unknown
+ return RocketMQLocalTransactionState.COMMIT;
+ }
+ }
+}
+```
+
+> 更多发送相关配置
+>
+> ```properties
+> spring.rocketmq.producer.retryTimesWhenSendAsyncFailed=0
+> spring.rocketmq.producer.sendMessageTimeout=300000
+> spring.rocketmq.producer.compressMessageBodyOverHowmuch=4096
+> spring.rocketmq.producer.maxMessageSize=4194304
+> spring.rocketmq.producer.retryAnotherBrokerWhenNotStoreOk=false
+> spring.rocketmq.producer.retryTimesWhenSendFailed=2
+> ```
+
+
+### 接收消息
+
+```properties
+## application.properties
+spring.rocketmq.nameServer=127.0.0.1:9876
+```
+
+> 注意:
+>
+> 请将上述示例配置中的`127.0.0.1:9876`替换成真实RocketMQ的NameServer地址与端口
+
+```java
+@SpringBootApplication
+public class ConsumerApplication{
+
+ public static void main(String[] args){
+ SpringApplication.run(ConsumerApplication.class, args);
+ }
+
+ @Slf4j
+ @Service
+ @RocketMQMessageListener(topic = "test-topic-1", consumerGroup = "my-consumer_test-topic-1")
+ public class MyConsumer1 implements RocketMQListener<String>{
+ public void onMessage(String message) {
+ log.info("received message: {}", message);
+ }
+ }
+
+ @Slf4j
+ @Service
+ @RocketMQMessageListener(topic = "test-topic-2", consumerGroup = "my-consumer_test-topic-2")
+ public class MyConsumer2 implements RocketMQListener<OrderPaidEvent>{
+ public void onMessage(OrderPaidEvent orderPaidEvent) {
+ log.info("received orderPaidEvent: {}", orderPaidEvent);
+ }
+ }
+}
+```
+
+
+> 更多消费相关配置
+>
+> see: [RocketMQMessageListener](src/main/java/org/apache/rocketmq/spring/starter/annotation/RocketMQMessageListener.java)
+
+
+## FAQ
+
+1. 生产环境有多个`nameserver`该如何连接？
+
+ `spring.rocketmq.nameServer`支持配置多个`nameserver`地址，采用`;`分隔即可。例如：`172.19.0.1:9876;172.19.0.2:9876`
+
+1. `rocketMQTemplate`在什么时候被销毁？
+
+ 开发者在项目中使用`rocketMQTemplate`发送消息时，不需要手动执行`rocketMQTemplate.destroy()`方法， `rocketMQTemplate`会在spring容器销毁时自动销毁。
+
+1. 启动报错：`Caused by: org.apache.rocketmq.client.exception.MQClientException: The consumer group[xxx] has been created before, specify another name please`
+
+ RocketMQ在设计时就不希望一个消费者同时处理多个类型的消息，因此同一个`consumerGroup`下的consumer职责应该是一样的，不要干不同的事情（即消费多个topic）。建议`consumerGroup`与`topic`一一对应。
+
+1. 发送的消息内容体是如何被序列化与反序列化的？
+
+ RocketMQ的消息体都是以`byte[]`方式存储。当业务系统的消息内容体如果是`java.lang.String`类型时，统一按照`utf-8`编码转成`byte[]`；如果业务系统的消息内容为非`java.lang.String`类型，则采用[jackson-databind](https://github.com/FasterXML/jackson-databind)序列化成`JSON`格式的字符串之后，再统一按照`utf-8`编码转成`byte[]`。
+
+1. 如何指定topic的`tags`?
+
+ RocketMQ的最佳实践中推荐：一个应用尽可能用一个Topic，消息子类型用`tags`来标识，`tags`可以由应用自由设置。
+ 在使用`rocketMQTemplate`发送消息时，通过设置发送方法的`destination`参数来设置消息的目的地，`destination`的格式为`topicName:tagName`，`:`前面表示topic的名称，后面表示`tags`名称。
+
+ > 注意:
+ >
+ > `tags`从命名来看像是一个复数，但发送消息时，目的地只能指定一个topic下的一个`tag`，不能指定多个。
+
+1. 发送消息时如何设置消息的`key`?
+
+ 可以通过重载的`xxxSend(String destination, Message<?> msg, ...)`方法来发送消息，指定`msg`的`headers`来完成。示例：
+
+ ```java
+ Message<?> message = MessageBuilder.withPayload(payload).setHeader(MessageConst.PROPERTY_KEYS, msgId).build();
+ rocketMQTemplate.send("topic-test", message);
+ ```
+
+ 同理还可以根据上面的方式来设置消息的`FLAG`、`WAIT_STORE_MSG_OK`以及一些用户自定义的其它头信息。
+
+ > 注意:
+ >
+ > 在将Spring的Message转化为RocketMQ的Message时，为防止`header`信息与RocketMQ的系统属性冲突，在所有`header`的名称前面都统一添加了前缀`USERS_`。因此在消费时如果想获取自定义的消息头信息，请遍历头信息中以`USERS_`开头的key即可。
+
+1. 消费消息时，除了获取消息`payload`外，还想获取RocketMQ消息的其它系统属性，需要怎么做？
+
+ 消费者在实现`RocketMQListener`接口时，只需要起泛型为`MessageExt`即可，这样在`onMessage`方法将接收到RocketMQ原生的`MessageExt`消息。
+
+ ```java
+ @Slf4j
+ @Service
+ @RocketMQMessageListener(topic = "test-topic-1", consumerGroup = "my-consumer_test-topic-1")
+ public class MyConsumer2 implements RocketMQListener<MessageExt>{
+ public void onMessage(MessageExt messageExt) {
+ log.info("received messageExt: {}", messageExt);
+ }
+ }
+ ```
+
+1. 如何指定消费者从哪开始消费消息，或开始消费的位置？
+
+ 消费者默认开始消费的位置请参考：[RocketMQ FAQ](http://rocketmq.apache.org/docs/faq/)。
+ 若想自定义消费者开始的消费位置，只需在消费者类添加一个`RocketMQPushConsumerLifecycleListener`接口的实现即可。 示例如下：
+
+ ```java
+ @Slf4j
+ @Service
+ @RocketMQMessageListener(topic = "test-topic-1", consumerGroup = "my-consumer_test-topic-1")
+ public class MyConsumer1 implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {
+ @Override
+ public void onMessage(String message) {
+ log.info("received message: {}", message);
+ }
+
+ @Override
+ public void prepareStart(final DefaultMQPushConsumer consumer) {
+ // set consumer consume message from now
+ consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
+ consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis()));
+ }
+ }
+ ```
+
+ 同理，任何关于`DefaultMQPushConsumer`的更多其它其它配置，都可以采用上述方式来完成。
+
+1. 如何发送事务消息(即半消息支持分布式事务)？
+ 在客户端，首先用户需要实现RocketMQLocalTransactionListener接口，并在接口类上注解声明@RocketMQTransactionListener，实现确认和回查方法；然后再使用资源模板RocketMQTemplate，
+ 调用方法sendMessageInTransaction()来进行消息的发布。 注意：这个方法通过指定发送者组名与具体的声明了txProducerGroup的TransactionListener进行关联，您也可以不指定这个值，从而使用默认的事务发送者组。
+
diff --git a/pom.xml b/pom.xml
new file mode 100644
index 0000000..7ca40cc
--- /dev/null
+++ b/pom.xml
@@ -0,0 +1,80 @@
+<?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">
+ <modelVersion>4.0.0</modelVersion>
+
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-spring-boot-build</artifactId>
+ <version>2.0.0</version>
+ <packaging>pom</packaging>
+
+ <name>RocketMQ Spring Boot Build</name>
+ <description>Build for Apache RocketMQ Spring Boot Support</description>
+ <url>https://github.com/apache/rocketmq-spring</url>
+
+ <organization>
+ <name>Apache Software Foundation</name>
+ <url>http://www.apache.org</url>
+ </organization>
+
+ <licenses>
+ <license>
+ <name>Apache License, Version 2.0</name>
+ <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+ <distribution>repo</distribution>
+ <comments>A business-friendly OSS license</comments>
+ </license>
+ </licenses>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-checkstyle-plugin</artifactId>
+ <version>2.17</version>
+ <executions>
+ <execution>
+ <id>validate</id>
+ <phase>validate</phase>
+ <configuration>
+ <excludes>src/main/resources</excludes>
+ <configLocation>style/rmq_checkstyle.xml</configLocation>
+ <encoding>UTF-8</encoding>
+ <consoleOutput>true</consoleOutput>
+ <failsOnError>true</failsOnError>
+ </configuration>
+ <goals>
+ <goal>check</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+ <modules>
+ <module>rocketmq-spring-boot-parent</module>
+ <module>rocketmq-spring-boot</module>
+ <module>rocketmq-spring-boot-starter</module>
+ <!-- Note: The samples need to mvn compiple in its own directory
+ <module>rocketmq-spring-boot-samples</module>
+ -->
+ </modules>
+</project>
diff --git a/rocketmq-spring-boot-parent/pom.xml b/rocketmq-spring-boot-parent/pom.xml
new file mode 100644
index 0000000..005c2cb
--- /dev/null
+++ b/rocketmq-spring-boot-parent/pom.xml
@@ -0,0 +1,177 @@
+<?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">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-spring-boot-build</artifactId>
+ <version>2.0.0</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>rocketmq-spring-boot-parent</artifactId>
+ <packaging>pom</packaging>
+
+ <name>RocketMQ Spring Boot Parent</name>
+ <description>Parent package for Apache RocketMQ Spring Boot Support</description>
+ <url>https://github.com/apache/rocketmq-spring</url>
+
+ <properties>
+ <project.rootdir>${project.basedir}/..</project.rootdir>
+ <spring.boot.version>2.0.5.RELEASE</spring.boot.version>
+ <spring.version>5.1.0.RELEASE</spring.version>
+
+ <rocketmq.spring.boot.version>2.0.0</rocketmq.spring.boot.version>
+
+ <rocketmq-version>4.3.2</rocketmq-version>
+ <slf4j.version>1.6.1</slf4j.version>
+ <jackson.version>2.9.7</jackson.version>
+
+ <java.version>1.8</java.version>
+ <resource.delimiter>@</resource.delimiter>
+ <!-- Delimiter that doesn't clash with Spring ${} placeholders -->
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+ <maven.compiler.source>${java.version}</maven.compiler.source>
+ <maven.compiler.target>${java.version}</maven.compiler.target>
+ <additionalparam>-Xdoclint:none</additionalparam>
+ </properties>
+
+ <dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot</artifactId>
+ <version>${spring.boot.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-autoconfigure</artifactId>
+ <version>${spring.boot.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-autoconfigure-processor</artifactId>
+ <version>${spring.boot.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-test</artifactId>
+ <version>${spring.boot.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter</artifactId>
+ <version>${spring.boot.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-validation</artifactId>
+ <version>${spring.boot.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-spring-boot</artifactId>
+ <version>${rocketmq.spring.boot.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-client</artifactId>
+ <version>${rocketmq-version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-messaging</artifactId>
+ <version>${spring.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-core</artifactId>
+ <version>${spring.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-context</artifactId>
+ <version>${spring.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-aop</artifactId>
+ <version>${spring.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ <version>${slf4j.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>${jackson.version}</version>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-checkstyle-plugin</artifactId>
+ <version>2.17</version>
+ <executions>
+ <execution>
+ <id>validate</id>
+ <phase>validate</phase>
+ <configuration>
+ <excludes>src/main/resources</excludes>
+ <configLocation>style/rmq_checkstyle.xml</configLocation>
+ <encoding>UTF-8</encoding>
+ <consoleOutput>true</consoleOutput>
+ <failsOnError>true</failsOnError>
+ </configuration>
+ <goals>
+ <goal>check</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git a/rocketmq-spring-boot-samples/LICENSE b/rocketmq-spring-boot-samples/LICENSE
new file mode 100644
index 0000000..7a4a3ea
--- /dev/null
+++ b/rocketmq-spring-boot-samples/LICENSE
@@ -0,0 +1,202 @@
+
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ Licensed 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.
\ No newline at end of file
diff --git a/rocketmq-spring-boot-samples/README.md b/rocketmq-spring-boot-samples/README.md
new file mode 100644
index 0000000..555f3bf
--- /dev/null
+++ b/rocketmq-spring-boot-samples/README.md
@@ -0,0 +1,38 @@
+# rocketmq-spring-boot-samples
+
+[中文](./README_zh_CN.md)
+
+[](https://www.apache.org/licenses/LICENSE-2.0.html)
+
+It's a demo project for how to use [rocketmq-spring-boot](https://github.com/apache/rocketmq-spring)
+
+> Note:
+>
+> If `org.apache.rocketmq:rocketmq-spring-boot-starter:${RELEASE.VERSION}` not exists in maven central repository, before you run this demo, you must git clone [rocketmq-spring-boot](https://github.com/apache/rocketmq-spring) and use `mvn clean install` build it by yourself.
+
+Run the test case locally
+1. build and install the rocketmq-spring-boot-starter
+
+2. startup rocketmq according to quick-start, verify the namesvr and broker startup correctly, Note: DON'T do "Shutdown Servers" step.
+http://rocketmq.apache.org/docs/quick-start/
+
+3. create topics for the demo test cases
+```
+bash bin/mqadmin updateTopic -c DefaultCluster -t string-topic
+bash bin/mqadmin updateTopic -c DefaultCluster -t order-paid-topic
+bash bin/mqadmin updateTopic -c DefaultCluster -t message-ext-topic
+bash bin/mqadmin updateTopic -c DefaultCluster -t spring-transaction-topic
+```
+4. run tests
+
+```
+# open a terminal, run produce
+cd rocketmq-produce-demo
+mvn clean package
+java -jar target/rocketmq-produce-demo-0.0.1-SNAPSHOT.jar
+
+# open another terminal, run consume
+cd rocketmq-consume-demo
+mvn clean package
+java -jar target/rocketmq-consume-demo-0.0.1-SNAPSHOT.jar
+```
diff --git a/rocketmq-spring-boot-samples/README_zh_CN.md b/rocketmq-spring-boot-samples/README_zh_CN.md
new file mode 100644
index 0000000..099d3b8
--- /dev/null
+++ b/rocketmq-spring-boot-samples/README_zh_CN.md
@@ -0,0 +1,48 @@
+# rocketmq-spring-boot-samples
+
+[English](./README.md)
+
+[](https://www.apache.org/licenses/LICENSE-2.0.html)
+
+这里是一个使用rocketmq-spring-boot-starter的列子。 [rocketmq-spring-boot](https://github.com/apache/rocketmq-spring)
+
+> 注意:
+>
+> 如果这个dependency: `org.apache.rocketmq:rocketmq-spring-boot-starter:${RELEASE.VERSION}`在maven中心库不存在, 需要开发者在本地build并安装到本地maven库，执行如下步骤:
+```
+ git clone https://github.com/apache/rocketmq-spring
+ cd rocketmq-spring
+ mvn clean install
+```
+
+## 在本地运行这个测试例子
+
+1. 如上面注意项所述，需要开发者在本地build并安装rocketmq-spring-boot-starter
+
+2. 根据RocketMQ官网的quick-start来启动NameServer和Broker，并验证是否启动正确。注意: 测试期间不要停止Broker或者NameServer
+http://rocketmq.apache.org/docs/quick-start/
+
+3. 创建测试例子所需要的Topic
+```
+cd YOUR_ROCKETMQ_HOME
+
+bash bin/mqadmin updateTopic -c DefaultCluster -t string-topic
+bash bin/mqadmin updateTopic -c DefaultCluster -t order-paid-topic
+bash bin/mqadmin updateTopic -c DefaultCluster -t message-ext-topic
+bash bin/mqadmin updateTopic -c DefaultCluster -t spring-transaction-topic
+```
+
+4. 编译并运行测试例子
+
+```
+# 打开一个终端窗口，编译并启动发送端
+cd rocketmq-produce-demo
+mvn clean package
+java -jar target/rocketmq-produce-demo-0.0.1-SNAPSHOT.jar
+
+# 打开另一个终端窗口，编译并启动消费端
+cd rocketmq-consume-demo
+mvn clean package
+java -jar target/rocketmq-consume-demo-0.0.1-SNAPSHOT.jar
+```
+结合测试代码，观察窗口中消息的发送和接收情况
diff --git a/rocketmq-spring-boot-samples/pom.xml b/rocketmq-spring-boot-samples/pom.xml
new file mode 100644
index 0000000..7b5eaa5
--- /dev/null
+++ b/rocketmq-spring-boot-samples/pom.xml
@@ -0,0 +1,89 @@
+<?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">
+ <modelVersion>4.0.0</modelVersion>
+
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-spring-boot-samples</artifactId>
+ <packaging>pom</packaging>
+ <version>0.0.1-SNAPSHOT</version>
+
+ <name>RocketMQ Spring Boot Samples</name>
+ <description>Samples for RocketMQ Spring Boot</description>
+ <url>https://github.com/apache/rocketmq-spring</url>
+
+ <modules>
+ <module>rocketmq-produce-demo</module>
+ <module>rocketmq-consume-demo</module>
+ </modules>
+
+ <properties>
+ <rocketmq-spring-boot-starter-version>2.0.0</rocketmq-spring-boot-starter-version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-spring-boot-starter</artifactId>
+ <version>${rocketmq-spring-boot-starter-version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-checkstyle-plugin</artifactId>
+ <version>2.17</version>
+ <executions>
+ <execution>
+ <id>validate</id>
+ <phase>validate</phase>
+ <configuration>
+ <skip>${disable.checks}</skip>
+ <excludes>src/main/resources</excludes>
+ <configLocation>style/rmq_checkstyle.xml</configLocation>
+ <encoding>UTF-8</encoding>
+ <consoleOutput>true</consoleOutput>
+ <failsOnError>true</failsOnError>
+ </configuration>
+ <goals>
+ <goal>check</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-maven-plugin</artifactId>
+ <version>2.1.0.RELEASE</version>
+ <executions>
+ <execution>
+ <goals>
+ <goal>repackage</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
diff --git a/rocketmq-spring-boot-samples/rocketmq-consume-demo/pom.xml b/rocketmq-spring-boot-samples/rocketmq-consume-demo/pom.xml
new file mode 100644
index 0000000..4093170
--- /dev/null
+++ b/rocketmq-spring-boot-samples/rocketmq-consume-demo/pom.xml
@@ -0,0 +1,31 @@
+<?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">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-spring-boot-samples</artifactId>
+ <version>0.0.1-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>rocketmq-consume-demo</artifactId>
+
+</project>
\ No newline at end of file
diff --git a/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/ConsumerApplication.java b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/ConsumerApplication.java
new file mode 100644
index 0000000..e4fbc8c
--- /dev/null
+++ b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/ConsumerApplication.java
@@ -0,0 +1,33 @@
+/*
+ * 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.samples.springboot;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+/**
+ * ConsumerApplication
+ */
+@SpringBootApplication
+public class ConsumerApplication {
+
+ public static void main(String[] args) {
+ SpringApplication.run(ConsumerApplication.class, args);
+ }
+}
+
diff --git a/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/Checker.java b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/Checker.java
new file mode 100644
index 0000000..608df1b
--- /dev/null
+++ b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/Checker.java
@@ -0,0 +1,44 @@
+/*
+ * 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.samples.springboot.consumer;
+
+import org.apache.rocketmq.client.producer.LocalTransactionState;
+import org.apache.rocketmq.client.producer.TransactionListener;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageExt;
+
+/**
+ * Note: This is a nagitive testing. It aims to tell user the fact that
+ * the @RocketMQTransactionListener can not be used on consumer side!!!
+ *
+ * <p>How to try it? just uncomment the annotation declaration, then compile
+ * and run the consumer, it will fail to start.
+ */
+
+//@RocketMQTransactionListener
+public class Checker implements TransactionListener {
+ @Override
+ public LocalTransactionState executeLocalTransaction(Message message, Object o) {
+ return null;
+ }
+
+ @Override
+ public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
+ return null;
+ }
+}
diff --git a/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/MessageExtConsumer.java b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/MessageExtConsumer.java
new file mode 100644
index 0000000..59ef263
--- /dev/null
+++ b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/MessageExtConsumer.java
@@ -0,0 +1,46 @@
+/*
+ * 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.samples.springboot.consumer;
+
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
+import org.springframework.stereotype.Service;
+
+/**
+ * MessageExtConsumer, consume listener impl class.
+ */
+@Service
+@RocketMQMessageListener(topic = "message-ext-topic", selectorExpression = "tag1", consumerGroup = "${spring.application.name}-message-ext-consumer")
+public class MessageExtConsumer implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {
+ @Override
+ public void onMessage(MessageExt message) {
+ System.out.printf("------- MessageExtConsumer received message, msgId: %s, body:%s \n", message.getMsgId(), new String(message.getBody()));
+ }
+
+ @Override
+ public void prepareStart(DefaultMQPushConsumer consumer) {
+ // set consumer consume message from now
+ consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
+ consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis()));
+ }
+}
diff --git a/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/OrderPaidEventConsumer.java b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/OrderPaidEventConsumer.java
new file mode 100644
index 0000000..774cd23
--- /dev/null
+++ b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/OrderPaidEventConsumer.java
@@ -0,0 +1,36 @@
+/*
+ * 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.samples.springboot.consumer;
+
+import org.apache.rocketmq.samples.springboot.domain.OrderPaidEvent;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.stereotype.Service;
+
+/**
+ * OrderPaidEventConsumer
+ */
+@Service
+@RocketMQMessageListener(topic = "${spring.rocketmq.orderTopic}", consumerGroup = "order-paid-consumer")
+public class OrderPaidEventConsumer implements RocketMQListener<OrderPaidEvent> {
+
+ @Override
+ public void onMessage(OrderPaidEvent orderPaidEvent) {
+ System.out.printf("------- OrderPaidEventConsumer received: %s \n", orderPaidEvent);
+ }
+}
diff --git a/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/StringConsumer.java b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/StringConsumer.java
new file mode 100644
index 0000000..370e847
--- /dev/null
+++ b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/StringConsumer.java
@@ -0,0 +1,34 @@
+/*
+ * 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.samples.springboot.consumer;
+
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.stereotype.Service;
+
+/**
+ * RocketMQMessageListener
+ */
+@Service
+@RocketMQMessageListener(topic = "${spring.rocketmq.topic}", consumerGroup = "string_consumer")
+public class StringConsumer implements RocketMQListener<String> {
+ @Override
+ public void onMessage(String message) {
+ System.out.printf("------- StringConsumer received: %s \n", message);
+ }
+}
diff --git a/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/StringTransactionalConsumer.java b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/StringTransactionalConsumer.java
new file mode 100644
index 0000000..5eebcc8
--- /dev/null
+++ b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/consumer/StringTransactionalConsumer.java
@@ -0,0 +1,34 @@
+/*
+ * 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.samples.springboot.consumer;
+
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.stereotype.Service;
+
+/**
+ * StringTransactionalConsumer
+ */
+@Service
+@RocketMQMessageListener(topic = "${spring.rocketmq.transTopic}", consumerGroup = "string_trans_consumer")
+public class StringTransactionalConsumer implements RocketMQListener<String> {
+ @Override
+ public void onMessage(String message) {
+ System.out.printf("------- StringTransactionalConsumer received: %s \n", message);
+ }
+}
diff --git a/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/domain/OrderPaidEvent.java b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/domain/OrderPaidEvent.java
new file mode 100644
index 0000000..812ab83
--- /dev/null
+++ b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/java/org/apache/rocketmq/samples/springboot/domain/OrderPaidEvent.java
@@ -0,0 +1,54 @@
+/*
+ * 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.samples.springboot.domain;
+
+import java.io.Serializable;
+import java.math.BigDecimal;
+
+/**
+ * OrderPaidEvent
+ */
+public class OrderPaidEvent implements Serializable {
+ private String orderId;
+
+ private BigDecimal paidMoney;
+
+ public OrderPaidEvent() {
+ }
+
+ public OrderPaidEvent(String orderId, BigDecimal paidMoney) {
+ this.orderId = orderId;
+ this.paidMoney = paidMoney;
+ }
+
+ public String getOrderId() {
+ return orderId;
+ }
+
+ public void setOrderId(String orderId) {
+ this.orderId = orderId;
+ }
+
+ public BigDecimal getPaidMoney() {
+ return paidMoney;
+ }
+
+ public void setPaidMoney(BigDecimal paidMoney) {
+ this.paidMoney = paidMoney;
+ }
+}
diff --git a/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/resources/application.properties b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/resources/application.properties
new file mode 100644
index 0000000..bbdeed1
--- /dev/null
+++ b/rocketmq-spring-boot-samples/rocketmq-consume-demo/src/main/resources/application.properties
@@ -0,0 +1,7 @@
+spring.application.name=rocketmq-consume-demo
+
+spring.rocketmq.nameServer=localhost:9876
+spring.rocketmq.topic=string-topic
+spring.rocketmq.orderTopic=order-paid-topic
+spring.rocketmq.msgExtTopic=message-ext-topic
+spring.rocketmq.transTopic=spring-transaction-topic
diff --git a/rocketmq-spring-boot-samples/rocketmq-produce-demo/pom.xml b/rocketmq-spring-boot-samples/rocketmq-produce-demo/pom.xml
new file mode 100644
index 0000000..0ef364d
--- /dev/null
+++ b/rocketmq-spring-boot-samples/rocketmq-produce-demo/pom.xml
@@ -0,0 +1,32 @@
+<?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">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-spring-boot-samples</artifactId>
+ <version>0.0.1-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>rocketmq-produce-demo</artifactId>
+
+
+</project>
\ No newline at end of file
diff --git a/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/java/org/apache/rocketmq/samples/springboot/ProducerApplication.java b/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/java/org/apache/rocketmq/samples/springboot/ProducerApplication.java
new file mode 100644
index 0000000..bec2450
--- /dev/null
+++ b/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/java/org/apache/rocketmq/samples/springboot/ProducerApplication.java
@@ -0,0 +1,171 @@
+/*
+ * 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.samples.springboot;
+
+import org.apache.rocketmq.client.producer.SendCallback;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.samples.springboot.domain.OrderPaidEvent;
+import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
+import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
+import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.apache.rocketmq.spring.support.RocketMQHeaders;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.support.MessageBuilder;
+
+import javax.annotation.Resource;
+import java.math.BigDecimal;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Producer, using RocketMQTemplate sends a variety of messages
+ */
+@SpringBootApplication
+public class ProducerApplication implements CommandLineRunner {
+ private static final String TX_PGROUP_NAME = "myTxProducerGroup";
+ @Resource
+ private RocketMQTemplate rocketMQTemplate;
+ @Value("${spring.rocketmq.transTopic}")
+ private String springTransTopic;
+ @Value("${spring.rocketmq.topic}")
+ private String springTopic;
+ @Value("${spring.rocketmq.orderTopic}")
+ private String orderPaidTopic;
+ @Value("${spring.rocketmq.msgExtTopic}")
+ private String msgExtTopic;
+
+ public static void main(String[] args) {
+ SpringApplication.run(ProducerApplication.class, args);
+ }
+
+ @Override
+ public void run(String... args) throws Exception {
+ // Send string
+ SendResult sendResult = rocketMQTemplate.syncSend(springTopic, "Hello, World!");
+ System.out.printf("syncSend1 to topic %s sendResult=%s %n", springTopic, sendResult);
+
+ // Send string with spring Message
+ sendResult = rocketMQTemplate.syncSend(springTopic, MessageBuilder.withPayload("Hello, World! I'm from spring message").build());
+ System.out.printf("syncSend2 to topic %s sendResult=%s %n", springTopic, sendResult);
+
+ // Send user-defined object
+ rocketMQTemplate.asyncSend(orderPaidTopic, new OrderPaidEvent("T_001", new BigDecimal("88.00")), new SendCallback() {
+ public void onSuccess(SendResult var1) {
+ System.out.printf("async onSucess SendResult=%s %n", var1);
+ }
+
+ public void onException(Throwable var1) {
+ System.out.printf("async onException Throwable=%s %n", var1);
+ }
+
+ });
+
+ // Send message with special tag
+ rocketMQTemplate.convertAndSend(msgExtTopic + ":tag0", "I'm from tag0"); // tag0 will not be consumer-selected
+ System.out.printf("syncSend topic %s tag %s %n", msgExtTopic, "tag0");
+ rocketMQTemplate.convertAndSend(msgExtTopic + ":tag1", "I'm from tag1");
+ System.out.printf("syncSend topic %s tag %s %n", msgExtTopic, "tag1");
+
+ // Send transactional messages
+ testTransaction();
+ }
+
+
+ private void testTransaction() throws MessagingException {
+ String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};
+ for (int i = 0; i < 10; i++) {
+ try {
+
+ Message msg = MessageBuilder.withPayload("Hello RocketMQ " + i).
+ setHeader(RocketMQHeaders.KEYS, "KEY_" + i).build();
+ SendResult sendResult = rocketMQTemplate.sendMessageInTransaction(TX_PGROUP_NAME,
+ springTransTopic + ":" + tags[i % tags.length], msg, null);
+ System.out.printf("------ send Transactional msg body = %s , sendResult=%s %n",
+ msg.getPayload(), sendResult.getSendStatus());
+
+ Thread.sleep(10);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ }
+
+ @RocketMQTransactionListener(txProducerGroup = TX_PGROUP_NAME)
+ class TransactionListenerImpl implements RocketMQLocalTransactionListener {
+ private AtomicInteger transactionIndex = new AtomicInteger(0);
+
+ private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<String, Integer>();
+
+ @Override
+ public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
+ String transId = (String)msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
+ System.out.printf("#### executeLocalTransaction is executed, msgTransactionId=%s %n",
+ transId);
+ int value = transactionIndex.getAndIncrement();
+ int status = value % 3;
+ localTrans.put(transId, status);
+ if (status == 0) {
+ // Return local transaction with success(commit), in this case,
+ // this message will not be checked in checkLocalTransaction()
+ System.out.printf(" # COMMIT # Simulating msg %s related local transaction exec succeeded! ### %n", msg.getPayload());
+ return RocketMQLocalTransactionState.COMMIT;
+ }
+
+ if (status == 1) {
+ // Return local transaction with failure(rollback) , in this case,
+ // this message will not be checked in checkLocalTransaction()
+ System.out.printf(" # ROLLBACK # Simulating %s related local transaction exec failed! %n", msg.getPayload());
+ return RocketMQLocalTransactionState.ROLLBACK;
+ }
+
+ System.out.printf(" # UNKNOW # Simulating %s related local transaction exec UNKNOWN! \n");
+ return RocketMQLocalTransactionState.UNKNOWN;
+ }
+
+ @Override
+ public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
+ String transId = (String)msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
+ RocketMQLocalTransactionState retState = RocketMQLocalTransactionState.COMMIT;
+ Integer status = localTrans.get(transId);
+ if (null != status) {
+ switch (status) {
+ case 0:
+ retState = RocketMQLocalTransactionState.UNKNOWN;
+ break;
+ case 1:
+ retState = RocketMQLocalTransactionState.COMMIT;
+ break;
+ case 2:
+ retState = RocketMQLocalTransactionState.ROLLBACK;
+ break;
+ }
+ }
+ System.out.printf("------ !!! checkLocalTransaction is executed once," +
+ " msgTransactionId=%s, TransactionState=%s status=%s %n",
+ transId, retState, status);
+ return retState;
+ }
+ }
+
+}
diff --git a/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/java/org/apache/rocketmq/samples/springboot/domain/OrderPaidEvent.java b/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/java/org/apache/rocketmq/samples/springboot/domain/OrderPaidEvent.java
new file mode 100644
index 0000000..feaac0d
--- /dev/null
+++ b/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/java/org/apache/rocketmq/samples/springboot/domain/OrderPaidEvent.java
@@ -0,0 +1,50 @@
+/*
+ * 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.samples.springboot.domain;
+
+import java.io.Serializable;
+import java.math.BigDecimal;
+
+public class OrderPaidEvent implements Serializable {
+ private String orderId;
+
+ private BigDecimal paidMoney;
+
+ public OrderPaidEvent() {
+ }
+
+ public OrderPaidEvent(String orderId, BigDecimal paidMoney) {
+ this.orderId = orderId;
+ this.paidMoney = paidMoney;
+ }
+
+ public String getOrderId() {
+ return orderId;
+ }
+
+ public void setOrderId(String orderId) {
+ this.orderId = orderId;
+ }
+
+ public BigDecimal getPaidMoney() {
+ return paidMoney;
+ }
+
+ public void setPaidMoney(BigDecimal paidMoney) {
+ this.paidMoney = paidMoney;
+ }
+}
diff --git a/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/resources/application.properties b/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/resources/application.properties
new file mode 100644
index 0000000..2187442
--- /dev/null
+++ b/rocketmq-spring-boot-samples/rocketmq-produce-demo/src/main/resources/application.properties
@@ -0,0 +1,6 @@
+spring.rocketmq.nameServer=localhost:9876
+spring.rocketmq.producer.group=my-group1
+spring.rocketmq.topic=string-topic
+spring.rocketmq.orderTopic=order-paid-topic
+spring.rocketmq.msgExtTopic=message-ext-topic
+spring.rocketmq.transTopic=spring-transaction-topic
\ No newline at end of file
diff --git a/rocketmq-spring-boot-samples/style/copyright/Apache.xml b/rocketmq-spring-boot-samples/style/copyright/Apache.xml
new file mode 100644
index 0000000..e3e3dec
--- /dev/null
+++ b/rocketmq-spring-boot-samples/style/copyright/Apache.xml
@@ -0,0 +1,23 @@
+<!--
+ 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.
+ -->
+
+<component name="CopyrightManager">
+ <copyright>
+ <option name="myName" value="Apache" />
+ <option name="notice" value="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.a [...]
+ </copyright>
+</component>
\ No newline at end of file
diff --git a/rocketmq-spring-boot-samples/style/copyright/profiles_settings.xml b/rocketmq-spring-boot-samples/style/copyright/profiles_settings.xml
new file mode 100644
index 0000000..747c7e2
--- /dev/null
+++ b/rocketmq-spring-boot-samples/style/copyright/profiles_settings.xml
@@ -0,0 +1,64 @@
+<!--
+ 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.
+ -->
+
+<component name="CopyrightManager">
+ <settings default="Apache">
+ <module2copyright>
+ <element module="All" copyright="Apache"/>
+ </module2copyright>
+ <LanguageOptions name="GSP">
+ <option name="fileTypeOverride" value="3"/>
+ <option name="prefixLines" value="false"/>
+ </LanguageOptions>
+ <LanguageOptions name="HTML">
+ <option name="fileTypeOverride" value="3"/>
+ <option name="prefixLines" value="false"/>
+ </LanguageOptions>
+ <LanguageOptions name="JAVA">
+ <option name="fileTypeOverride" value="3" />
+ <option name="addBlankAfter" value="false" />
+ </LanguageOptions>
+ <LanguageOptions name="JSP">
+ <option name="fileTypeOverride" value="3"/>
+ <option name="prefixLines" value="false"/>
+ </LanguageOptions>
+ <LanguageOptions name="JSPX">
+ <option name="fileTypeOverride" value="3"/>
+ <option name="prefixLines" value="false"/>
+ </LanguageOptions>
+ <LanguageOptions name="MXML">
+ <option name="fileTypeOverride" value="3"/>
+ <option name="prefixLines" value="false"/>
+ </LanguageOptions>
+ <LanguageOptions name="Properties">
+ <option name="fileTypeOverride" value="3"/>
+ <option name="block" value="false"/>
+ </LanguageOptions>
+ <LanguageOptions name="SPI">
+ <option name="fileTypeOverride" value="3"/>
+ <option name="block" value="false"/>
+ </LanguageOptions>
+ <LanguageOptions name="XML">
+ <option name="fileTypeOverride" value="3"/>
+ <option name="prefixLines" value="false"/>
+ </LanguageOptions>
+ <LanguageOptions name="__TEMPLATE__">
+ <option name="separateBefore" value="true"/>
+ <option name="lenBefore" value="1"/>
+ </LanguageOptions>
+ </settings>
+</component>
\ No newline at end of file
diff --git a/rocketmq-spring-boot-samples/style/rmq_checkstyle.xml b/rocketmq-spring-boot-samples/style/rmq_checkstyle.xml
new file mode 100644
index 0000000..2e9658f
--- /dev/null
+++ b/rocketmq-spring-boot-samples/style/rmq_checkstyle.xml
@@ -0,0 +1,135 @@
+<?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.
+ -->
+
+<!DOCTYPE module PUBLIC
+ "-//Puppy Crawl//DTD Check Configuration 1.3//EN"
+ "http://www.puppycrawl.com/dtds/configuration_1_3.dtd">
+<!--Refer http://checkstyle.sourceforge.net/reports/google-java-style.html#s2.2-file-encoding -->
+<module name="Checker">
+
+ <property name="localeLanguage" value="en"/>
+
+ <!--To configure the check to report on the first instance in each file-->
+ <module name="FileTabCharacter"/>
+
+ <!-- header -->
+ <module name="RegexpHeader">
+ <property name="header" value="/\*\nLicensed to the Apache Software Foundation*"/>
+ <property name="fileExtensions" value="java" />
+ </module>
+
+ <module name="RegexpSingleline">
+ <property name="format" value="System\.out\.println"/>
+ <property name="message" value="Prohibit invoking System.out.println in source code !"/>
+ </module>
+
+ <module name="RegexpSingleline">
+ <property name="format" value="//FIXME"/>
+ <property name="message" value="Recommended fix FIXME task !"/>
+ </module>
+
+ <module name="RegexpSingleline">
+ <property name="format" value="//TODO"/>
+ <property name="message" value="Recommended fix TODO task !"/>
+ </module>
+
+ <module name="RegexpSingleline">
+ <property name="format" value="@alibaba"/>
+ <property name="message" value="Recommended remove @alibaba keyword!"/>
+ </module>
+ <module name="RegexpSingleline">
+ <property name="format" value="@taobao"/>
+ <property name="message" value="Recommended remove @taobao keyword!"/>
+ </module>
+ <module name="RegexpSingleline">
+ <property name="format" value="@author"/>
+ <property name="message" value="Recommended remove @author tag in javadoc!"/>
+ </module>
+
+ <module name="RegexpSingleline">
+ <property name="format"
+ value=".*[\u3400-\u4DB5\u4E00-\u9FA5\u9FA6-\u9FBB\uF900-\uFA2D\uFA30-\uFA6A\uFA70-\uFAD9\uFF00-\uFFEF\u2E80-\u2EFF\u3000-\u303F\u31C0-\u31EF]+.*"/>
+ <property name="message" value="Not allow chinese character !"/>
+ </module>
+
+ <module name="FileLength">
+ <property name="max" value="3000"/>
+ </module>
+
+ <module name="TreeWalker">
+
+ <module name="UnusedImports">
+ <property name="processJavadoc" value="true"/>
+ </module>
+ <module name="RedundantImport"/>
+
+ <!--<module name="IllegalImport" />-->
+
+ <!--Checks that classes that override equals() also override hashCode()-->
+ <module name="EqualsHashCode"/>
+ <!--Checks for over-complicated boolean expressions. Currently finds code like if (topic == true), topic || true, !false, etc.-->
+ <module name="SimplifyBooleanExpression"/>
+ <module name="OneStatementPerLine"/>
+ <module name="UnnecessaryParentheses"/>
+ <!--Checks for over-complicated boolean return statements. For example the following code-->
+ <module name="SimplifyBooleanReturn"/>
+
+ <!--Check that the default is after all the cases in producerGroup switch statement-->
+ <module name="DefaultComesLast"/>
+ <!--Detects empty statements (standalone ";" semicolon)-->
+ <module name="EmptyStatement"/>
+ <!--Checks that long constants are defined with an upper ell-->
+ <module name="UpperEll"/>
+ <module name="ConstantName">
+ <property name="format" value="(^[A-Z][A-Z0-9]*(_[A-Z0-9]+)*$)|(^log$)"/>
+ </module>
+ <!--Checks that local, non-final variable names conform to producerGroup format specified by the format property-->
+ <module name="LocalVariableName"/>
+ <!--Validates identifiers for local, final variables, including catch parameters-->
+ <module name="LocalFinalVariableName"/>
+ <!--Validates identifiers for non-static fields-->
+ <module name="MemberName"/>
+ <!--Validates identifiers for class type parameters-->
+ <module name="ClassTypeParameterName">
+ <property name="format" value="^[A-Z0-9]*$"/>
+ </module>
+ <!--Validates identifiers for method type parameters-->
+ <module name="MethodTypeParameterName">
+ <property name="format" value="^[A-Z0-9]*$"/>
+ </module>
+ <module name="PackageName"/>
+ <module name="ParameterName"/>
+ <module name="StaticVariableName"/>
+ <module name="TypeName"/>
+ <!--Checks that there are no import statements that use the * notation-->
+ <module name="AvoidStarImport"/>
+
+ <!--whitespace-->
+ <module name="GenericWhitespace"/>
+ <module name="NoWhitespaceBefore"/>
+ <module name="NoWhitespaceAfter"/>
+ <module name="WhitespaceAround">
+ <property name="allowEmptyConstructors" value="true"/>
+ <property name="allowEmptyMethods" value="true"/>
+ </module>
+ <module name="Indentation"/>
+ <module name="MethodParamPad"/>
+ <module name="ParenPad"/>
+ <module name="TypecastParenPad"/>
+ </module>
+</module>
diff --git a/rocketmq-spring-boot-samples/style/rmq_codeStyle.xml b/rocketmq-spring-boot-samples/style/rmq_codeStyle.xml
new file mode 100644
index 0000000..9db075e
--- /dev/null
+++ b/rocketmq-spring-boot-samples/style/rmq_codeStyle.xml
@@ -0,0 +1,157 @@
+<!--
+ 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.
+ -->
+
+<code_scheme name="rocketmq">
+ <option name="USE_SAME_INDENTS" value="true"/>
+ <option name="IGNORE_SAME_INDENTS_FOR_LANGUAGES" value="true"/>
+ <option name="OTHER_INDENT_OPTIONS">
+ <value>
+ <option name="INDENT_SIZE" value="4"/>
+ <option name="CONTINUATION_INDENT_SIZE" value="4"/>
+ <option name="TAB_SIZE" value="4"/>
+ <option name="USE_TAB_CHARACTER" value="false"/>
+ <option name="SMART_TABS" value="false"/>
+ <option name="LABEL_INDENT_SIZE" value="0"/>
+ <option name="LABEL_INDENT_ABSOLUTE" value="false"/>
+ <option name="USE_RELATIVE_INDENTS" value="false"/>
+ </value>
+ </option>
+ <option name="PREFER_LONGER_NAMES" value="false"/>
+ <option name="CLASS_COUNT_TO_USE_IMPORT_ON_DEMAND" value="1000"/>
+ <option name="NAMES_COUNT_TO_USE_IMPORT_ON_DEMAND" value="1000"/>
+ <option name="PACKAGES_TO_USE_IMPORT_ON_DEMAND">
+ <value/>
+ </option>
+ <option name="IMPORT_LAYOUT_TABLE">
+ <value>
+ <package name="" withSubpackages="true" static="false"/>
+ <emptyLine/>
+ <package name="" withSubpackages="true" static="true"/>
+ </value>
+ </option>
+ <option name="JD_ALIGN_PARAM_COMMENTS" value="false"/>
+ <option name="JD_ALIGN_EXCEPTION_COMMENTS" value="false"/>
+ <option name="JD_P_AT_EMPTY_LINES" value="false"/>
+ <option name="JD_KEEP_INVALID_TAGS" value="false"/>
+ <option name="JD_DO_NOT_WRAP_ONE_LINE_COMMENTS" value="true"/>
+ <option name="KEEP_CONTROL_STATEMENT_IN_ONE_LINE" value="false"/>
+ <option name="KEEP_BLANK_LINES_IN_DECLARATIONS" value="1"/>
+ <option name="KEEP_BLANK_LINES_IN_CODE" value="1"/>
+ <option name="KEEP_BLANK_LINES_BEFORE_RBRACE" value="1"/>
+ <option name="ELSE_ON_NEW_LINE" value="true"/>
+ <option name="WHILE_ON_NEW_LINE" value="true"/>
+ <option name="CATCH_ON_NEW_LINE" value="true"/>
+ <option name="FINALLY_ON_NEW_LINE" value="true"/>
+ <option name="ALIGN_MULTILINE_PARAMETERS" value="false"/>
+ <option name="ALIGN_MULTILINE_FOR" value="false"/>
+ <option name="SPACE_AFTER_TYPE_CAST" value="false"/>
+ <option name="SPACE_BEFORE_ARRAY_INITIALIZER_LBRACE" value="true"/>
+ <option name="METHOD_PARAMETERS_WRAP" value="1"/>
+ <option name="ARRAY_INITIALIZER_LBRACE_ON_NEXT_LINE" value="true"/>
+ <option name="LABELED_STATEMENT_WRAP" value="1"/>
+ <option name="WRAP_COMMENTS" value="true"/>
+ <option name="METHOD_ANNOTATION_WRAP" value="1"/>
+ <option name="CLASS_ANNOTATION_WRAP" value="1"/>
+ <option name="FIELD_ANNOTATION_WRAP" value="1"/>
+ <JavaCodeStyleSettings>
+ <option name="CLASS_NAMES_IN_JAVADOC" value="3"/>
+ </JavaCodeStyleSettings>
+ <XML>
+ <option name="XML_LEGACY_SETTINGS_IMPORTED" value="true"/>
+ </XML>
+ <ADDITIONAL_INDENT_OPTIONS fileType="haml">
+ <option name="INDENT_SIZE" value="2"/>
+ </ADDITIONAL_INDENT_OPTIONS>
+ <codeStyleSettings language="Groovy">
+ <option name="KEEP_CONTROL_STATEMENT_IN_ONE_LINE" value="false"/>
+ <option name="KEEP_BLANK_LINES_IN_DECLARATIONS" value="1"/>
+ <option name="KEEP_BLANK_LINES_IN_CODE" value="1"/>
+ <option name="KEEP_BLANK_LINES_BEFORE_RBRACE" value="1"/>
+ <option name="ELSE_ON_NEW_LINE" value="true"/>
+ <option name="CATCH_ON_NEW_LINE" value="true"/>
+ <option name="FINALLY_ON_NEW_LINE" value="true"/>
+ <option name="ALIGN_MULTILINE_PARAMETERS" value="false"/>
+ <option name="ALIGN_MULTILINE_FOR" value="false"/>
+ <option name="SPACE_AFTER_TYPE_CAST" value="false"/>
+ <option name="METHOD_PARAMETERS_WRAP" value="1"/>
+ <option name="METHOD_ANNOTATION_WRAP" value="1"/>
+ <option name="CLASS_ANNOTATION_WRAP" value="1"/>
+ <option name="FIELD_ANNOTATION_WRAP" value="1"/>
+ <option name="PARENT_SETTINGS_INSTALLED" value="true"/>
+ <indentOptions>
+ <option name="CONTINUATION_INDENT_SIZE" value="4"/>
+ </indentOptions>
+ </codeStyleSettings>
+ <codeStyleSettings language="HOCON">
+ <option name="KEEP_BLANK_LINES_BEFORE_RBRACE" value="1"/>
+ <option name="PARENT_SETTINGS_INSTALLED" value="true"/>
+ </codeStyleSettings>
+ <codeStyleSettings language="JAVA">
+ <option name="KEEP_CONTROL_STATEMENT_IN_ONE_LINE" value="false"/>
+ <option name="KEEP_BLANK_LINES_IN_DECLARATIONS" value="1"/>
+ <option name="KEEP_BLANK_LINES_IN_CODE" value="1"/>
+ <option name="KEEP_BLANK_LINES_BEFORE_RBRACE" value="1"/>
+ <option name="ELSE_ON_NEW_LINE" value="true"/>
+ <option name="WHILE_ON_NEW_LINE" value="true"/>
+ <option name="CATCH_ON_NEW_LINE" value="true"/>
+ <option name="FINALLY_ON_NEW_LINE" value="true"/>
+ <option name="ALIGN_MULTILINE_PARAMETERS" value="false"/>
+ <option name="ALIGN_MULTILINE_FOR" value="false"/>
+ <option name="SPACE_AFTER_TYPE_CAST" value="false"/>
+ <option name="SPACE_BEFORE_ARRAY_INITIALIZER_LBRACE" value="true"/>
+ <option name="METHOD_PARAMETERS_WRAP" value="1"/>
+ <option name="ARRAY_INITIALIZER_LBRACE_ON_NEXT_LINE" value="true"/>
+ <option name="LABELED_STATEMENT_WRAP" value="1"/>
+ <option name="METHOD_ANNOTATION_WRAP" value="1"/>
+ <option name="CLASS_ANNOTATION_WRAP" value="1"/>
+ <option name="FIELD_ANNOTATION_WRAP" value="1"/>
+ <option name="PARENT_SETTINGS_INSTALLED" value="true"/>
+ <indentOptions>
+ <option name="CONTINUATION_INDENT_SIZE" value="4"/>
+ </indentOptions>
+ </codeStyleSettings>
+ <codeStyleSettings language="JSON">
+ <option name="KEEP_BLANK_LINES_IN_CODE" value="1"/>
+ <option name="PARENT_SETTINGS_INSTALLED" value="true"/>
+ </codeStyleSettings>
+ <codeStyleSettings language="Scala">
+ <option name="KEEP_BLANK_LINES_IN_DECLARATIONS" value="1"/>
+ <option name="KEEP_BLANK_LINES_IN_CODE" value="1"/>
+ <option name="KEEP_BLANK_LINES_BEFORE_RBRACE" value="1"/>
+ <option name="ELSE_ON_NEW_LINE" value="true"/>
+ <option name="WHILE_ON_NEW_LINE" value="true"/>
+ <option name="CATCH_ON_NEW_LINE" value="true"/>
+ <option name="FINALLY_ON_NEW_LINE" value="true"/>
+ <option name="ALIGN_MULTILINE_PARAMETERS" value="false"/>
+ <option name="ALIGN_MULTILINE_FOR" value="false"/>
+ <option name="METHOD_PARAMETERS_WRAP" value="1"/>
+ <option name="METHOD_ANNOTATION_WRAP" value="1"/>
+ <option name="CLASS_ANNOTATION_WRAP" value="1"/>
+ <option name="FIELD_ANNOTATION_WRAP" value="1"/>
+ <option name="PARENT_SETTINGS_INSTALLED" value="true"/>
+ <indentOptions>
+ <option name="INDENT_SIZE" value="4"/>
+ <option name="CONTINUATION_INDENT_SIZE" value="4"/>
+ <option name="TAB_SIZE" value="4"/>
+ </indentOptions>
+ </codeStyleSettings>
+ <codeStyleSettings language="XML">
+ <indentOptions>
+ <option name="CONTINUATION_INDENT_SIZE" value="4"/>
+ </indentOptions>
+ </codeStyleSettings>
+</code_scheme>
\ No newline at end of file
diff --git a/rocketmq-spring-boot-starter/pom.xml b/rocketmq-spring-boot-starter/pom.xml
new file mode 100644
index 0000000..cd88b71
--- /dev/null
+++ b/rocketmq-spring-boot-starter/pom.xml
@@ -0,0 +1,51 @@
+<?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">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-spring-boot-parent</artifactId>
+ <version>2.0.0</version>
+ <relativePath>../rocketmq-spring-boot-parent/pom.xml</relativePath>
+ </parent>
+
+ <artifactId>rocketmq-spring-boot-starter</artifactId>
+ <packaging>jar</packaging>
+
+ <name>RocketMQ Spring Boot Starter</name>
+ <description>SRocketMQ Spring Boot Starter</description>
+ <url>https://github.com/apache/rocketmq-spring</url>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-spring-boot</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-validation</artifactId>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/rocketmq-spring-boot/README.md b/rocketmq-spring-boot/README.md
new file mode 100644
index 0000000..a4d3022
--- /dev/null
+++ b/rocketmq-spring-boot/README.md
@@ -0,0 +1,7 @@
+# RocketMQ Spring Boot Support
+This project provides auto-configuration for the following RocketMQ client:
+
+- [Produce and Consume Spring Message](../rocketmq-spring-boot-starter)
+- [Send (halp) Message in Transaction](../rocketmq-spring-boot-starter)
+
+For details, please see sample code in the [rocketmq-spring-boot-samples](../rocketmq-spring-boot-samples)
\ No newline at end of file
diff --git a/rocketmq-spring-boot/pom.xml b/rocketmq-spring-boot/pom.xml
new file mode 100644
index 0000000..94e528c
--- /dev/null
+++ b/rocketmq-spring-boot/pom.xml
@@ -0,0 +1,104 @@
+<?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">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-spring-boot-parent</artifactId>
+ <version>2.0.0</version>
+ <relativePath>../rocketmq-spring-boot-parent/pom.xml</relativePath>
+ </parent>
+
+ <artifactId>rocketmq-spring-boot</artifactId>
+ <packaging>jar</packaging>
+
+ <name>RocketMQ Spring Boot AutoConfigure</name>
+ <description>SRocketMQ Spring Boot AutoConfigure</description>
+ <url>https://github.com/apache/rocketmq-spring</url>
+
+
+ <dependencies>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot</artifactId>
+ <optional>true</optional>
+ </dependency>
+
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-autoconfigure</artifactId>
+ <optional>true</optional>
+ </dependency>
+
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-autoconfigure-processor</artifactId>
+ <optional>true</optional>
+ </dependency>
+
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-test</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-client</artifactId>
+ <exclusions>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-messaging</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-core</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-context</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-aop</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/ConsumeMode.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/ConsumeMode.java
new file mode 100644
index 0000000..1b9f35a
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/ConsumeMode.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 org.apache.rocketmq.spring.annotation;
+
+public enum ConsumeMode {
+ /**
+ * Receive asynchronously delivered messages concurrently
+ */
+ CONCURRENTLY,
+
+ /**
+ * Receive asynchronously delivered messages orderly. one queue, one thread
+ */
+ ORDERLY
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/MessageModel.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/MessageModel.java
new file mode 100644
index 0000000..9bdaa64
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/MessageModel.java
@@ -0,0 +1,33 @@
+/*
+ * 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.spring.annotation;
+
+public enum MessageModel {
+ BROADCASTING("BROADCASTING"),
+ CLUSTERING("CLUSTERING");
+
+ private String modeCN;
+
+ MessageModel(String modeCN) {
+ this.modeCN = modeCN;
+ }
+
+ public String getModeCN() {
+ return this.modeCN;
+ }
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/RocketMQMessageListener.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/RocketMQMessageListener.java
new file mode 100644
index 0000000..762ea82
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/RocketMQMessageListener.java
@@ -0,0 +1,72 @@
+/*
+ * 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.spring.annotation;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+public @interface RocketMQMessageListener {
+
+ /**
+ * Consumers of the same role is required to have exactly same subscriptions and consumerGroup to correctly achieve
+ * load balance. It's required and needs to be globally unique.
+ *
+ *
+ * See <a href="http://rocketmq.apache.org/docs/core-concept/">here</a> for further discussion.
+ */
+ String consumerGroup();
+
+ /**
+ * Topic name.
+ */
+ String topic();
+
+ /**
+ * Control how to selector message.
+ *
+ * @see SelectorType
+ */
+ SelectorType selectorType() default SelectorType.TAG;
+
+ /**
+ * Control which message can be select. Grammar please see {@link SelectorType#TAG} and {@link SelectorType#SQL92}
+ */
+ String selectorExpression() default "*";
+
+ /**
+ * Control consume mode, you can choice receive message concurrently or orderly.
+ */
+ ConsumeMode consumeMode() default ConsumeMode.CONCURRENTLY;
+
+ /**
+ * Control message mode, if you want all subscribers receive message all message, broadcasting is a good choice.
+ */
+ MessageModel messageModel() default MessageModel.CLUSTERING;
+
+ /**
+ * Max consumer thread number.
+ */
+ int consumeThreadMax() default 64;
+
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/RocketMQTransactionListener.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/RocketMQTransactionListener.java
new file mode 100644
index 0000000..b355641
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/RocketMQTransactionListener.java
@@ -0,0 +1,69 @@
+/*
+ * 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.spring.annotation;
+
+import org.apache.rocketmq.spring.config.RocketMQConfigUtils;
+import org.springframework.stereotype.Component;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * This annotation is used over a class which implements interface
+ * org.apache.rocketmq.client.producer.TransactionListener. The class implements
+ * two methods for process callback events after the txProducer sends a transactional message.
+ * <p>Note: The annotation is used only on RocketMQ client producer side, it can not be used
+ * on consumer side.
+ */
+@Target({ElementType.TYPE, ElementType.ANNOTATION_TYPE})
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@Component
+public @interface RocketMQTransactionListener {
+
+ /**
+ * Declare the txProducerGroup that is used to relate callback event to the listener, rocketMQTemplate must send a
+ * transactional message with the declared txProducerGroup.
+ * <p>
+ * <p>It is suggested to use the default txProducerGroup if your system only needs to define a TransactionListener class.
+ */
+ String txProducerGroup() default RocketMQConfigUtils.ROCKETMQ_TRANSACTION_DEFAULT_GLOBAL_NAME;
+
+ /**
+ * Set ExecutorService params -- corePoolSize
+ */
+ int corePoolSize() default 1;
+
+ /**
+ * Set ExecutorService params -- maximumPoolSize
+ */
+ int maximumPoolSize() default 1;
+
+ /**
+ * Set ExecutorService params -- keepAliveTime
+ */
+ long keepAliveTime() default 1000 * 60; //60ms
+
+ /**
+ * Set ExecutorService params -- blockingQueueSize
+ */
+ int blockingQueueSize() default 2000;
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/SelectorType.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/SelectorType.java
new file mode 100644
index 0000000..97e8b7e
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/annotation/SelectorType.java
@@ -0,0 +1,33 @@
+/*
+ * 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.spring.annotation;
+
+import org.apache.rocketmq.common.filter.ExpressionType;
+
+public enum SelectorType {
+
+ /**
+ * @see ExpressionType#TAG
+ */
+ TAG,
+
+ /**
+ * @see ExpressionType#SQL92
+ */
+ SQL92
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/config/ListenerContainerConfiguration.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/config/ListenerContainerConfiguration.java
new file mode 100644
index 0000000..e1fab88
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/config/ListenerContainerConfiguration.java
@@ -0,0 +1,133 @@
+/*
+ * 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.spring.config;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.rocketmq.spring.annotation.ConsumeMode;
+import org.apache.rocketmq.spring.annotation.MessageModel;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.aop.support.AopUtils;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.SmartInitializingSingleton;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.support.BeanDefinitionValidationException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.core.env.StandardEnvironment;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicLong;
+
+
+public class ListenerContainerConfiguration implements ApplicationContextAware, SmartInitializingSingleton {
+ private final static Logger log = LoggerFactory.getLogger(RocketMQAutoConfiguration.class);
+
+ private ConfigurableApplicationContext applicationContext;
+
+ private AtomicLong counter = new AtomicLong(0);
+
+ private StandardEnvironment environment;
+
+ private RocketMQProperties rocketMQProperties;
+
+ private ObjectMapper objectMapper;
+
+ public ListenerContainerConfiguration() {
+ }
+
+ @Autowired
+ public ListenerContainerConfiguration(
+ @Qualifier("rocketMQMessageObjectMapper") ObjectMapper objectMapper,
+ StandardEnvironment environment,
+ RocketMQProperties rocketMQProperties) {
+ this.objectMapper = objectMapper;
+ this.environment = environment;
+ this.rocketMQProperties = rocketMQProperties;
+ }
+
+ @Override
+ public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+ this.applicationContext = (ConfigurableApplicationContext) applicationContext;
+ }
+
+ @Override
+ public void afterSingletonsInstantiated() {
+ Map<String, Object> beans = this.applicationContext.getBeansWithAnnotation(RocketMQMessageListener.class);
+
+ if (Objects.nonNull(beans)) {
+ beans.forEach(this::registerContainer);
+ }
+ }
+
+ private void registerContainer(String beanName, Object bean) {
+ Class<?> clazz = AopUtils.getTargetClass(bean);
+
+ if (!RocketMQListener.class.isAssignableFrom(bean.getClass())) {
+ throw new IllegalStateException(clazz + " is not instance of " + RocketMQListener.class.getName());
+ }
+
+ RocketMQMessageListener annotation = clazz.getAnnotation(RocketMQMessageListener.class);
+ validate(annotation);
+
+ String containerBeanName = String.format("%s_%s", DefaultRocketMQListenerContainer.class.getName(),
+ counter.incrementAndGet());
+ GenericApplicationContext genericApplicationContext = (GenericApplicationContext) applicationContext;
+
+ genericApplicationContext.registerBean(containerBeanName, DefaultRocketMQListenerContainer.class,
+ () -> createRocketMQListenerContainer(bean, annotation));
+ DefaultRocketMQListenerContainer container = genericApplicationContext.getBean(containerBeanName,
+ DefaultRocketMQListenerContainer.class);
+ if (!container.isRunning()) {
+ try {
+ container.start();
+ } catch (Exception e) {
+ log.error("Started container failed. {}", container, e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ log.info("Register the listener to container, listenerBeanName:{}, containerBeanName:{}", beanName, containerBeanName);
+ }
+
+ private DefaultRocketMQListenerContainer createRocketMQListenerContainer(Object bean, RocketMQMessageListener annotation) {
+ DefaultRocketMQListenerContainer container = new DefaultRocketMQListenerContainer();
+
+ container.setNameServer(rocketMQProperties.getNameServer());
+ container.setTopic(environment.resolvePlaceholders(annotation.topic()));
+ container.setConsumerGroup(environment.resolvePlaceholders(annotation.consumerGroup()));
+ container.setRocketMQMessageListener(annotation);
+ container.setRocketMQListener((RocketMQListener) bean);
+ container.setObjectMapper(objectMapper);
+
+ return container;
+ }
+
+ private void validate(RocketMQMessageListener annotation) {
+ if (annotation.consumeMode() == ConsumeMode.ORDERLY &&
+ annotation.messageModel() == MessageModel.BROADCASTING)
+ throw new BeanDefinitionValidationException("Bad annotation definition in @RocketMQMessageListener, messageModel BROADCASTING does not support ORDERLY message!");
+ }
+}
\ No newline at end of file
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/config/RocketMQAutoConfiguration.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/config/RocketMQAutoConfiguration.java
new file mode 100644
index 0000000..c22c6ee
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/config/RocketMQAutoConfiguration.java
@@ -0,0 +1,104 @@
+/*
+ * 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.spring.config;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.context.annotation.Role;
+import org.springframework.util.Assert;
+
+import java.util.Objects;
+
+@Configuration
+@EnableConfigurationProperties(RocketMQProperties.class)
+@ConditionalOnProperty(prefix = "spring.rocketmq", value = "nameServer")
+@Import(ListenerContainerConfiguration.class)
+public class RocketMQAutoConfiguration {
+
+ @Bean
+ @ConditionalOnMissingBean(DefaultMQProducer.class)
+ @ConditionalOnProperty(prefix = "spring.rocketmq", value = {"nameServer", "producer.group"})
+ public DefaultMQProducer defaultMQProducer(RocketMQProperties rocketMQProperties) {
+ RocketMQProperties.Producer producerConfig = rocketMQProperties.getProducer();
+ String nameServer = rocketMQProperties.getNameServer();
+ String groupName = producerConfig.getGroup();
+ Assert.hasText(nameServer, "[spring.rocketmq.nameServer] must not be null");
+ Assert.hasText(groupName, "[spring.rocketmq.producer.group] must not be null");
+
+ DefaultMQProducer producer = new DefaultMQProducer(groupName);
+ producer.setNamesrvAddr(nameServer);
+ producer.setSendMsgTimeout(producerConfig.getSendMessageTimeout());
+ producer.setRetryTimesWhenSendFailed(producerConfig.getRetryTimesWhenSendFailed());
+ producer.setRetryTimesWhenSendAsyncFailed(producerConfig.getRetryTimesWhenSendAsyncFailed());
+ producer.setMaxMessageSize(producerConfig.getMaxMessageSize());
+ producer.setCompressMsgBodyOverHowmuch(producerConfig.getCompressMessageBodyOverHowmuch());
+ producer.setRetryAnotherBrokerWhenNotStoreOK(producerConfig.isRetryAnotherBrokerWhenNotStoreOk());
+
+ return producer;
+ }
+
+ @Bean
+ @ConditionalOnClass(name = "com.fasterxml.jackson.databind.ObjectMapper")
+ @ConditionalOnMissingBean(ObjectMapper.class)
+ public ObjectMapper rocketMQMessageObjectMapper() {
+ return new ObjectMapper();
+ }
+
+ @Bean(destroyMethod = "destroy")
+ @ConditionalOnBean(DefaultMQProducer.class)
+ @ConditionalOnMissingBean(RocketMQTemplate.class)
+ public RocketMQTemplate rocketMQTemplate(DefaultMQProducer mqProducer,
+ @Autowired(required = false)
+ @Qualifier("rocketMQMessageObjectMapper")
+ ObjectMapper objectMapper) {
+ RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
+ rocketMQTemplate.setProducer(mqProducer);
+ if (Objects.nonNull(objectMapper)) {
+ rocketMQTemplate.setObjectMapper(objectMapper);
+ } else {
+ throw new IllegalStateException("Can not inject null objectMapper into RocketMQTemplate!");
+ }
+ return rocketMQTemplate;
+ }
+
+ @Bean
+ @ConditionalOnBean(RocketMQTemplate.class)
+ @ConditionalOnMissingBean(TransactionHandlerRegistry.class)
+ public TransactionHandlerRegistry transactionHandlerRegistry(RocketMQTemplate template) {
+ return new TransactionHandlerRegistry(template);
+ }
+
+ @Bean(name = RocketMQConfigUtils.ROCKETMQ_TRANSACTION_ANNOTATION_PROCESSOR_BEAN_NAME)
+ @ConditionalOnBean(TransactionHandlerRegistry.class)
+ @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
+ public RocketMQTransactionAnnotationProcessor transactionAnnotationProcessor(TransactionHandlerRegistry transactionHandlerRegistry) {
+ return new RocketMQTransactionAnnotationProcessor(transactionHandlerRegistry);
+ }
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/config/RocketMQConfigUtils.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/config/RocketMQConfigUtils.java
new file mode 100644
index 0000000..a9c9f80
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/config/RocketMQConfigUtils.java
@@ -0,0 +1,29 @@
+/*
+ * 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.spring.config;
+
+public class RocketMQConfigUtils {
+ /**
+ * The bean name of the internally managed RocketMQ transaction annotation processor.
+ */
+ public static final String ROCKETMQ_TRANSACTION_ANNOTATION_PROCESSOR_BEAN_NAME =
+ "org.springframework.rocketmq.spring.starter.internalRocketMQTransAnnotationProcessor";
+
+ public static final String ROCKETMQ_TRANSACTION_DEFAULT_GLOBAL_NAME =
+ "rocketmq_transaction_default_global_name";
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/config/RocketMQProperties.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/config/RocketMQProperties.java
new file mode 100644
index 0000000..8590e05
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/config/RocketMQProperties.java
@@ -0,0 +1,144 @@
+/*
+ * 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.spring.config;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+@SuppressWarnings("WeakerAccess")
+@ConfigurationProperties(prefix = "spring.rocketmq")
+public class RocketMQProperties {
+
+ /**
+ * The name server for rocketMQ, formats: `host:port;host:port`.
+ */
+ private String nameServer;
+
+ private Producer producer;
+
+ public String getNameServer() {
+ return nameServer;
+ }
+
+ public void setNameServer(String nameServer) {
+ this.nameServer = nameServer;
+ }
+
+ public RocketMQProperties.Producer getProducer() {
+ return producer;
+ }
+
+ public void setProducer(RocketMQProperties.Producer producer) {
+ this.producer = producer;
+ }
+
+ public static class Producer {
+
+ /**
+ * Name of producer.
+ */
+ private String group;
+
+ /**
+ * Millis of send message timeout.
+ */
+ private int sendMessageTimeout = 3000;
+
+ /**
+ * Compress message body threshold, namely, message body larger than 4k will be compressed on default.
+ */
+ private int compressMessageBodyOverHowmuch = 1024 * 4;
+
+ /**
+ * Maximum number of retry to perform internally before claiming sending failure in synchronous mode.
+ * This may potentially cause message duplication which is up to application developers to resolve.
+ */
+ private int retryTimesWhenSendFailed = 2;
+
+ /**
+ * <p> Maximum number of retry to perform internally before claiming sending failure in asynchronous mode. </p>
+ * This may potentially cause message duplication which is up to application developers to resolve.
+ */
+ private int retryTimesWhenSendAsyncFailed = 2;
+
+ /**
+ * Indicate whether to retry another broker on sending failure internally.
+ */
+ private boolean retryAnotherBrokerWhenNotStoreOk = false;
+
+ /**
+ * Maximum allowed message size in bytes.
+ */
+ private int maxMessageSize = 1024 * 1024 * 4;
+
+ public String getGroup() {
+ return group;
+ }
+
+ public void setGroup(String group) {
+ this.group = group;
+ }
+
+ public int getSendMessageTimeout() {
+ return sendMessageTimeout;
+ }
+
+ public void setSendMessageTimeout(int sendMessageTimeout) {
+ this.sendMessageTimeout = sendMessageTimeout;
+ }
+
+ public int getCompressMessageBodyOverHowmuch() {
+ return compressMessageBodyOverHowmuch;
+ }
+
+ public void setCompressMessageBodyOverHowmuch(int compressMessageBodyOverHowmuch) {
+ this.compressMessageBodyOverHowmuch = compressMessageBodyOverHowmuch;
+ }
+
+ public int getRetryTimesWhenSendFailed() {
+ return retryTimesWhenSendFailed;
+ }
+
+ public void setRetryTimesWhenSendFailed(int retryTimesWhenSendFailed) {
+ this.retryTimesWhenSendFailed = retryTimesWhenSendFailed;
+ }
+
+ public int getRetryTimesWhenSendAsyncFailed() {
+ return retryTimesWhenSendAsyncFailed;
+ }
+
+ public void setRetryTimesWhenSendAsyncFailed(int retryTimesWhenSendAsyncFailed) {
+ this.retryTimesWhenSendAsyncFailed = retryTimesWhenSendAsyncFailed;
+ }
+
+ public boolean isRetryAnotherBrokerWhenNotStoreOk() {
+ return retryAnotherBrokerWhenNotStoreOk;
+ }
+
+ public void setRetryAnotherBrokerWhenNotStoreOk(boolean retryAnotherBrokerWhenNotStoreOk) {
+ this.retryAnotherBrokerWhenNotStoreOk = retryAnotherBrokerWhenNotStoreOk;
+ }
+
+ public int getMaxMessageSize() {
+ return maxMessageSize;
+ }
+
+ public void setMaxMessageSize(int maxMessageSize) {
+ this.maxMessageSize = maxMessageSize;
+ }
+ }
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/config/RocketMQTransactionAnnotationProcessor.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/config/RocketMQTransactionAnnotationProcessor.java
new file mode 100644
index 0000000..cd78fdb
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/config/RocketMQTransactionAnnotationProcessor.java
@@ -0,0 +1,110 @@
+/*
+ * 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.spring.config;
+
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
+import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
+import org.springframework.aop.support.AopUtils;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.core.Ordered;
+import org.springframework.core.annotation.AnnotationUtils;
+
+import java.util.Collections;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RocketMQTransactionAnnotationProcessor
+ implements BeanPostProcessor, Ordered, BeanFactoryAware {
+ private final static Logger log = LoggerFactory.getLogger(RocketMQTransactionAnnotationProcessor.class);
+
+ private BeanFactory beanFactory;
+ private final Set<Class<?>> nonProcessedClasses =
+ Collections.newSetFromMap(new ConcurrentHashMap<Class<?>, Boolean>(64));
+
+ private TransactionHandlerRegistry transactionHandlerRegistry;
+
+ public RocketMQTransactionAnnotationProcessor(TransactionHandlerRegistry transactionHandlerRegistry) {
+ this.transactionHandlerRegistry = transactionHandlerRegistry;
+ }
+
+ @Override
+ public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
+ this.beanFactory = beanFactory;
+ }
+
+ @Override
+ public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
+ return bean;
+ }
+
+ @Override
+ public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+ if (!this.nonProcessedClasses.contains(bean.getClass())) {
+ Class<?> targetClass = AopUtils.getTargetClass(bean);
+ RocketMQTransactionListener listener = AnnotationUtils.findAnnotation(targetClass, RocketMQTransactionListener.class);
+ this.nonProcessedClasses.add(bean.getClass());
+ if (listener == null) { // for quick search
+ log.trace("No @RocketMQTransactionListener annotations found on bean type: {}", bean.getClass());
+ } else {
+ try {
+ processTransactionListenerAnnotation(listener, bean);
+ } catch (MQClientException e) {
+ log.error("Failed to process annotation " + listener, e);
+ throw new BeanCreationException("Failed to process annotation " + listener, e);
+ }
+ }
+ }
+
+ return bean;
+ }
+
+ private void processTransactionListenerAnnotation(RocketMQTransactionListener listener, Object bean)
+ throws MQClientException {
+ if (transactionHandlerRegistry == null) {
+ throw new MQClientException("Bad usage of @RocketMQTransactionListener, " +
+ "the class must work with RocketMQTemplate", null);
+ }
+ if (!RocketMQLocalTransactionListener.class.isAssignableFrom(bean.getClass())) {
+ throw new MQClientException("Bad usage of @RocketMQTransactionListener, " +
+ "the class must implement interface RocketMQLocalTransactionListener",
+ null);
+ }
+ TransactionHandler transactionHandler = new TransactionHandler();
+ transactionHandler.setBeanFactory(this.beanFactory);
+ transactionHandler.setName(listener.txProducerGroup());
+ transactionHandler.setBeanName(bean.getClass().getName());
+ transactionHandler.setListener((RocketMQLocalTransactionListener) bean);
+ transactionHandler.setCheckExecutor(listener.corePoolSize(), listener.maximumPoolSize(),
+ listener.keepAliveTime(), listener.blockingQueueSize());
+
+ transactionHandlerRegistry.registerTransactionHandler(transactionHandler);
+ }
+
+ @Override
+ public int getOrder() {
+ return LOWEST_PRECEDENCE;
+ }
+
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/config/TransactionHandler.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/config/TransactionHandler.java
new file mode 100644
index 0000000..f6f6955
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/config/TransactionHandler.java
@@ -0,0 +1,75 @@
+/*
+ * 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.spring.config;
+
+import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
+import org.springframework.beans.factory.BeanFactory;
+
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+class TransactionHandler {
+ private String name;
+ private String beanName;
+ private RocketMQLocalTransactionListener bean;
+ private BeanFactory beanFactory;
+ private ThreadPoolExecutor checkExecutor;
+
+ public String getBeanName() {
+ return beanName;
+ }
+
+ public void setBeanName(String beanName) {
+ this.beanName = beanName;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public BeanFactory getBeanFactory() {
+ return beanFactory;
+ }
+
+ public void setBeanFactory(BeanFactory beanFactory) {
+ this.beanFactory = beanFactory;
+ }
+
+ public void setListener(RocketMQLocalTransactionListener listener) {
+ this.bean = listener;
+ }
+
+ public RocketMQLocalTransactionListener getListener() {
+ return this.bean;
+ }
+
+ public void setCheckExecutor(int corePoolSize, int maxPoolSize, long keepAliveTime, int blockingQueueSize) {
+ this.checkExecutor = new ThreadPoolExecutor(corePoolSize, maxPoolSize,
+ keepAliveTime, TimeUnit.MILLISECONDS,
+ new LinkedBlockingDeque<>(blockingQueueSize));
+ }
+
+ public ThreadPoolExecutor getCheckExecutor() {
+ return checkExecutor;
+ }
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/config/TransactionHandlerRegistry.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/config/TransactionHandlerRegistry.java
new file mode 100644
index 0000000..73a1244
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/config/TransactionHandlerRegistry.java
@@ -0,0 +1,50 @@
+/*
+ * 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.spring.config;
+
+import io.netty.util.internal.ConcurrentSet;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.springframework.beans.factory.DisposableBean;
+
+import java.util.Set;
+
+public class TransactionHandlerRegistry implements DisposableBean {
+ private RocketMQTemplate rocketMQTemplate;
+
+ private final Set<String> listenerContainers = new ConcurrentSet<>();
+
+ public TransactionHandlerRegistry(RocketMQTemplate template) {
+ this.rocketMQTemplate = template;
+ }
+
+ @Override
+ public void destroy() throws Exception {
+ listenerContainers.clear();
+ }
+
+ public void registerTransactionHandler(TransactionHandler handler) throws MQClientException {
+ if (listenerContainers.contains(handler.getName()))
+ throw new MQClientException(-1,
+ String.format("The transaction name [%s] has been defined in TransactionListener [%s]", handler.getName(),
+ handler.getBeanName()));
+ listenerContainers.add(handler.getName());
+
+ rocketMQTemplate.createAndStartTransactionMQProducer(handler.getName(), handler.getListener(), handler.getCheckExecutor());
+ }
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQListener.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQListener.java
new file mode 100644
index 0000000..bd6ceeb
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQListener.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.rocketmq.spring.core;
+
+public interface RocketMQListener<T> {
+ void onMessage(T message);
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQLocalTransactionListener.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQLocalTransactionListener.java
new file mode 100644
index 0000000..890aecc
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQLocalTransactionListener.java
@@ -0,0 +1,25 @@
+/*
+ * 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.spring.core;
+
+import org.springframework.messaging.Message;
+
+public interface RocketMQLocalTransactionListener {
+ RocketMQLocalTransactionState executeLocalTransaction(final Message msg, final Object arg);
+
+ RocketMQLocalTransactionState checkLocalTransaction(final Message msg);
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQLocalTransactionState.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQLocalTransactionState.java
new file mode 100644
index 0000000..91f0bec
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQLocalTransactionState.java
@@ -0,0 +1,24 @@
+/*
+ * 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.spring.core;
+
+public enum RocketMQLocalTransactionState {
+ COMMIT,
+ ROLLBACK,
+ UNKNOWN
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQPushConsumerLifecycleListener.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQPushConsumerLifecycleListener.java
new file mode 100644
index 0000000..3be8cdc
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQPushConsumerLifecycleListener.java
@@ -0,0 +1,24 @@
+/*
+ * 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.spring.core;
+
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.spring.support.RocketMQConsumerLifecycleListener;
+
+public interface RocketMQPushConsumerLifecycleListener extends RocketMQConsumerLifecycleListener<DefaultMQPushConsumer> {
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQTemplate.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQTemplate.java
new file mode 100644
index 0000000..f8c7c64
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/core/RocketMQTemplate.java
@@ -0,0 +1,609 @@
+/*
+ * 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.spring.core;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.MessageQueueSelector;
+import org.apache.rocketmq.client.producer.SendCallback;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.client.producer.TransactionMQProducer;
+import org.apache.rocketmq.client.producer.TransactionSendResult;
+import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash;
+import org.apache.rocketmq.spring.config.RocketMQConfigUtils;
+import org.apache.rocketmq.spring.support.RocketMQUtil;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.core.AbstractMessageSendingTemplate;
+import org.springframework.messaging.core.MessagePostProcessor;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.Assert;
+import org.springframework.util.MimeTypeUtils;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@SuppressWarnings({"WeakerAccess", "unused"})
+public class RocketMQTemplate extends AbstractMessageSendingTemplate<String> implements InitializingBean, DisposableBean {
+ private static final Logger log = LoggerFactory.getLogger(RocketMQTemplate.class);
+
+ private DefaultMQProducer producer;
+
+ private ObjectMapper objectMapper;
+
+ private String charset = "UTF-8";
+
+ private MessageQueueSelector messageQueueSelector = new SelectMessageQueueByHash();
+
+ private final Map<String, TransactionMQProducer> cache = new ConcurrentHashMap<>(); //only put TransactionMQProducer by now!!!
+
+ public DefaultMQProducer getProducer() {
+ return producer;
+ }
+
+ public void setProducer(DefaultMQProducer producer) {
+ this.producer = producer;
+ }
+
+ public ObjectMapper getObjectMapper() {
+ return objectMapper;
+ }
+
+ public void setObjectMapper(ObjectMapper objectMapper) {
+ this.objectMapper = objectMapper;
+ }
+
+ public String getCharset() {
+ return charset;
+ }
+
+ public void setCharset(String charset) {
+ this.charset = charset;
+ }
+
+ public MessageQueueSelector getMessageQueueSelector() {
+ return messageQueueSelector;
+ }
+
+ public void setMessageQueueSelector(MessageQueueSelector messageQueueSelector) {
+ this.messageQueueSelector = messageQueueSelector;
+ }
+
+ /**
+ * <p> Send message in synchronous mode. This method returns only when the sending procedure totally completes.
+ * Reliable synchronous transmission is used in extensive scenes, such as important notification messages, SMS
+ * notification, SMS marketing system, etc.. </p>
+ * <p>
+ * <strong>Warn:</strong> this method has internal retry-mechanism, that is, internal implementation will retry
+ * {@link DefaultMQProducer#getRetryTimesWhenSendFailed} times before claiming failure. As a result, multiple
+ * messages may potentially delivered to broker(s). It's up to the application developers to resolve potential
+ * duplication issue.
+ *
+ * @param destination formats: `topicName:tags`
+ * @param message {@link org.springframework.messaging.Message}
+ * @return {@link SendResult}
+ */
+ public SendResult syncSend(String destination, Message<?> message) {
+ return syncSend(destination, message, producer.getSendMsgTimeout());
+ }
+
+ /**
+ * Same to {@link #syncSend(String, Message)} with send timeout specified in addition.
+ *
+ * @param destination formats: `topicName:tags`
+ * @param message {@link org.springframework.messaging.Message}
+ * @param timeout send timeout with millis
+ * @return {@link SendResult}
+ */
+ public SendResult syncSend(String destination, Message<?> message, long timeout) {
+ return syncSend(destination, message, timeout, 0);
+ }
+
+ /**
+ * Same to {@link #syncSend(String, Message)} with send timeout specified in addition.
+ *
+ * @param destination formats: `topicName:tags`
+ * @param message {@link org.springframework.messaging.Message}
+ * @param timeout send timeout with millis
+ * @param delayLevel level for the delay message
+ * @return {@link SendResult}
+ */
+ public SendResult syncSend(String destination, Message<?> message, long timeout, int delayLevel) {
+ if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
+ log.error("syncSend failed. destination:{}, message is null ", destination);
+ throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
+ }
+
+ try {
+ long now = System.currentTimeMillis();
+ org.apache.rocketmq.common.message.Message rocketMsg = RocketMQUtil.convertToRocketMessage(objectMapper,
+ charset, destination, message);
+ if (delayLevel > 0) {
+ rocketMsg.setDelayTimeLevel(delayLevel);
+ }
+ SendResult sendResult = producer.send(rocketMsg, timeout);
+ long costTime = System.currentTimeMillis() - now;
+ log.debug("send message cost: {} ms, msgId:{}", costTime, sendResult.getMsgId());
+ return sendResult;
+ } catch (Exception e) {
+ log.error("syncSend failed. destination:{}, message:{} ", destination, message);
+ throw new MessagingException(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Same to {@link #syncSend(String, Message)}.
+ *
+ * @param destination formats: `topicName:tags`
+ * @param payload the Object to use as payload
+ * @return {@link SendResult}
+ */
+ public SendResult syncSend(String destination, Object payload) {
+ return syncSend(destination, payload, producer.getSendMsgTimeout());
+ }
+
+ /**
+ * Same to {@link #syncSend(String, Object)} with send timeout specified in addition.
+ *
+ * @param destination formats: `topicName:tags`
+ * @param payload the Object to use as payload
+ * @param timeout send timeout with millis
+ * @return {@link SendResult}
+ */
+ public SendResult syncSend(String destination, Object payload, long timeout) {
+ Message<?> message = this.doConvert(payload, null, null);
+ return syncSend(destination, message, timeout);
+ }
+
+ /**
+ * Same to {@link #syncSend(String, Message)} with send orderly with hashKey by specified.
+ *
+ * @param destination formats: `topicName:tags`
+ * @param message {@link org.springframework.messaging.Message}
+ * @param hashKey use this key to select queue. for example: orderId, productId ...
+ * @return {@link SendResult}
+ */
+ public SendResult syncSendOrderly(String destination, Message<?> message, String hashKey) {
+ return syncSendOrderly(destination, message, hashKey, producer.getSendMsgTimeout());
+ }
+
+ /**
+ * Same to {@link #syncSendOrderly(String, Message, String)} with send timeout specified in addition.
+ *
+ * @param destination formats: `topicName:tags`
+ * @param message {@link org.springframework.messaging.Message}
+ * @param hashKey use this key to select queue. for example: orderId, productId ...
+ * @param timeout send timeout with millis
+ * @return {@link SendResult}
+ */
+ public SendResult syncSendOrderly(String destination, Message<?> message, String hashKey, long timeout) {
+ if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
+ log.error("syncSendOrderly failed. destination:{}, message is null ", destination);
+ throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
+ }
+
+ try {
+ long now = System.currentTimeMillis();
+ org.apache.rocketmq.common.message.Message rocketMsg = RocketMQUtil.convertToRocketMessage(objectMapper,
+ charset, destination, message);
+ SendResult sendResult = producer.send(rocketMsg, messageQueueSelector, hashKey, timeout);
+ long costTime = System.currentTimeMillis() - now;
+ log.debug("send message cost: {} ms, msgId:{}", costTime, sendResult.getMsgId());
+ return sendResult;
+ } catch (Exception e) {
+ log.error("syncSendOrderly failed. destination:{}, message:{} ", destination, message);
+ throw new MessagingException(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Same to {@link #syncSend(String, Object)} with send orderly with hashKey by specified.
+ *
+ * @param destination formats: `topicName:tags`
+ * @param payload the Object to use as payload
+ * @param hashKey use this key to select queue. for example: orderId, productId ...
+ * @return {@link SendResult}
+ */
+ public SendResult syncSendOrderly(String destination, Object payload, String hashKey) {
+ return syncSendOrderly(destination, payload, hashKey, producer.getSendMsgTimeout());
+ }
+
+ /**
+ * Same to {@link #syncSendOrderly(String, Object, String)} with send timeout specified in addition.
+ *
+ * @param destination formats: `topicName:tags`
+ * @param payload the Object to use as payload
+ * @param hashKey use this key to select queue. for example: orderId, productId ...
+ * @param timeout send timeout with millis
+ * @return {@link SendResult}
+ */
+ public SendResult syncSendOrderly(String destination, Object payload, String hashKey, long timeout) {
+ Message<?> message = this.doConvert(payload, null, null);
+ return syncSendOrderly(destination, message, hashKey, producer.getSendMsgTimeout());
+ }
+
+ /**
+ * Same to {@link #asyncSend(String, Message, SendCallback)} with send timeout specified in addition.
+ *
+ * @param destination formats: `topicName:tags`
+ * @param message {@link org.springframework.messaging.Message}
+ * @param sendCallback {@link SendCallback}
+ * @param timeout send timeout with millis
+ */
+ public void asyncSend(String destination, Message<?> message, SendCallback sendCallback, long timeout) {
+ if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
+ log.error("asyncSend failed. destination:{}, message is null ", destination);
+ throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
+ }
+
+ try {
+ org.apache.rocketmq.common.message.Message rocketMsg = RocketMQUtil.convertToRocketMessage(objectMapper,
+ charset, destination, message);
+ producer.send(rocketMsg, sendCallback, timeout);
+ } catch (Exception e) {
+ log.info("asyncSend failed. destination:{}, message:{} ", destination, message);
+ throw new MessagingException(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * <p> Send message to broker asynchronously. asynchronous transmission is generally used in response time sensitive
+ * business scenarios. </p>
+ * <p>
+ * This method returns immediately. On sending completion, <code>sendCallback</code> will be executed.
+ * <p>
+ * Similar to {@link #syncSend(String, Object)}, internal implementation would potentially retry up to {@link
+ * DefaultMQProducer#getRetryTimesWhenSendAsyncFailed} times before claiming sending failure, which may yield
+ * message duplication and application developers are the one to resolve this potential issue.
+ *
+ * @param destination formats: `topicName:tags`
+ * @param message {@link org.springframework.messaging.Message}
+ * @param sendCallback {@link SendCallback}
+ */
+ public void asyncSend(String destination, Message<?> message, SendCallback sendCallback) {
+ asyncSend(destination, message, sendCallback, producer.getSendMsgTimeout());
+ }
+
+ /**
+ * Same to {@link #asyncSend(String, Object, SendCallback)} with send timeout specified in addition.
+ *
+ * @param destination formats: `topicName:tags`
+ * @param payload the Object to use as payload
+ * @param sendCallback {@link SendCallback}
+ * @param timeout send timeout with millis
+ */
+ public void asyncSend(String destination, Object payload, SendCallback sendCallback, long timeout) {
+ Message<?> message = this.doConvert(payload, null, null);
+ asyncSend(destination, message, sendCallback, timeout);
+ }
+
+ /**
+ * Same to {@link #asyncSend(String, Message, SendCallback)}.
+ *
+ * @param destination formats: `topicName:tags`
+ * @param payload the Object to use as payload
+ * @param sendCallback {@link SendCallback}
+ */
+ public void asyncSend(String destination, Object payload, SendCallback sendCallback) {
+ asyncSend(destination, payload, sendCallback, producer.getSendMsgTimeout());
+ }
+
+ /**
+ * Same to {@link #asyncSendOrderly(String, Message, String, SendCallback)} with send timeout specified in
+ * addition.
+ *
+ * @param destination formats: `topicName:tags`
+ * @param message {@link org.springframework.messaging.Message}
+ * @param hashKey use this key to select queue. for example: orderId, productId ...
+ * @param sendCallback {@link SendCallback}
+ * @param timeout send timeout with millis
+ */
+ public void asyncSendOrderly(String destination, Message<?> message, String hashKey, SendCallback sendCallback,
+ long timeout) {
+ if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
+ log.error("asyncSendOrderly failed. destination:{}, message is null ", destination);
+ throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
+ }
+
+ try {
+ org.apache.rocketmq.common.message.Message rocketMsg = RocketMQUtil.convertToRocketMessage(objectMapper,
+ charset, destination, message);
+ producer.send(rocketMsg, messageQueueSelector, hashKey, sendCallback, timeout);
+ } catch (Exception e) {
+ log.error("asyncSendOrderly failed. destination:{}, message:{} ", destination, message);
+ throw new MessagingException(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Same to {@link #asyncSend(String, Message, SendCallback)} with send orderly with hashKey by specified.
+ *
+ * @param destination formats: `topicName:tags`
+ * @param message {@link org.springframework.messaging.Message}
+ * @param hashKey use this key to select queue. for example: orderId, productId ...
+ * @param sendCallback {@link SendCallback}
+ */
+ public void asyncSendOrderly(String destination, Message<?> message, String hashKey, SendCallback sendCallback) {
+ asyncSendOrderly(destination, message, hashKey, sendCallback, producer.getSendMsgTimeout());
+ }
+
+ /**
+ * Same to {@link #asyncSendOrderly(String, Message, String, SendCallback)}.
+ *
+ * @param destination formats: `topicName:tags`
+ * @param payload the Object to use as payload
+ * @param hashKey use this key to select queue. for example: orderId, productId ...
+ * @param sendCallback {@link SendCallback}
+ */
+ public void asyncSendOrderly(String destination, Object payload, String hashKey, SendCallback sendCallback) {
+ asyncSendOrderly(destination, payload, hashKey, sendCallback, producer.getSendMsgTimeout());
+ }
+
+ /**
+ * Same to {@link #asyncSendOrderly(String, Object, String, SendCallback)} with send timeout specified in addition.
+ *
+ * @param destination formats: `topicName:tags`
+ * @param payload the Object to use as payload
+ * @param hashKey use this key to select queue. for example: orderId, productId ...
+ * @param sendCallback {@link SendCallback}
+ * @param timeout send timeout with millis
+ */
+ public void asyncSendOrderly(String destination, Object payload, String hashKey, SendCallback sendCallback,
+ long timeout) {
+ Message<?> message = this.doConvert(payload, null, null);
+ asyncSendOrderly(destination, message, hashKey, sendCallback, timeout);
+ }
+
+ /**
+ * Similar to <a href="https://en.wikipedia.org/wiki/User_Datagram_Protocol">UDP</a>, this method won't wait for
+ * acknowledgement from broker before return. Obviously, it has maximums throughput yet potentials of message loss.
+ * <p>
+ * One-way transmission is used for cases requiring moderate reliability, such as log collection.
+ *
+ * @param destination formats: `topicName:tags`
+ * @param message {@link org.springframework.messaging.Message}
+ */
+ public void sendOneWay(String destination, Message<?> message) {
+ if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
+ log.error("sendOneWay failed. destination:{}, message is null ", destination);
+ throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
+ }
+
+ try {
+ org.apache.rocketmq.common.message.Message rocketMsg = RocketMQUtil.convertToRocketMessage(objectMapper,
+ charset, destination, message);
+ producer.sendOneway(rocketMsg);
+ } catch (Exception e) {
+ log.error("sendOneWay failed. destination:{}, message:{} ", destination, message);
+ throw new MessagingException(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Same to {@link #sendOneWay(String, Message)}
+ *
+ * @param destination formats: `topicName:tags`
+ * @param payload the Object to use as payload
+ */
+ public void sendOneWay(String destination, Object payload) {
+ Message<?> message = this.doConvert(payload, null, null);
+ sendOneWay(destination, message);
+ }
+
+ /**
+ * Same to {@link #sendOneWay(String, Message)} with send orderly with hashKey by specified.
+ *
+ * @param destination formats: `topicName:tags`
+ * @param message {@link org.springframework.messaging.Message}
+ * @param hashKey use this key to select queue. for example: orderId, productId ...
+ */
+ public void sendOneWayOrderly(String destination, Message<?> message, String hashKey) {
+ if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
+ log.error("sendOneWayOrderly failed. destination:{}, message is null ", destination);
+ throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
+ }
+
+ try {
+ org.apache.rocketmq.common.message.Message rocketMsg = RocketMQUtil.convertToRocketMessage(objectMapper,
+ charset, destination, message);
+ producer.sendOneway(rocketMsg, messageQueueSelector, hashKey);
+ } catch (Exception e) {
+ log.error("sendOneWayOrderly failed. destination:{}, message:{}", destination, message);
+ throw new MessagingException(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Same to {@link #sendOneWayOrderly(String, Message, String)}
+ *
+ * @param destination formats: `topicName:tags`
+ * @param payload the Object to use as payload
+ */
+ public void sendOneWayOrderly(String destination, Object payload, String hashKey) {
+ Message<?> message = this.doConvert(payload, null, null);
+ sendOneWayOrderly(destination, message, hashKey);
+ }
+
+ public void afterPropertiesSet() throws Exception {
+ Assert.notNull(producer, "Property 'producer' is required");
+ producer.start();
+ }
+
+ protected void doSend(String destination, Message<?> message) {
+ SendResult sendResult = syncSend(destination, message);
+ log.debug("send message to `{}` finished. result:{}", destination, sendResult);
+ }
+
+
+
+ @Override
+ protected Message<?> doConvert(Object payload, Map<String, Object> headers, MessagePostProcessor postProcessor) {
+ String content;
+ if (payload instanceof String) {
+ content = (String) payload;
+ } else {
+ // If payload not as string, use objectMapper change it.
+ try {
+ content = objectMapper.writeValueAsString(payload);
+ } catch (JsonProcessingException e) {
+ log.error("convert payload to String failed. payload:{}", payload);
+ throw new RuntimeException("convert to payload to String failed.", e);
+ }
+ }
+
+ MessageBuilder<?> builder = MessageBuilder.withPayload(content);
+ if (headers != null) {
+ builder.copyHeaders(headers);
+ }
+ builder.setHeaderIfAbsent(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN);
+
+ Message<?> message = builder.build();
+ if (postProcessor != null) {
+ message = postProcessor.postProcessMessage(message);
+ }
+ return message;
+ }
+
+ @Override
+ public void destroy() {
+ if (Objects.nonNull(producer)) {
+ producer.shutdown();
+ }
+
+ for (Map.Entry<String, TransactionMQProducer> kv : cache.entrySet()) {
+ if (Objects.nonNull(kv.getValue())) {
+ kv.getValue().shutdown();
+ }
+ }
+ cache.clear();
+ }
+
+ private String getTxProducerGroupName(String name) {
+ return name == null ? RocketMQConfigUtils.ROCKETMQ_TRANSACTION_DEFAULT_GLOBAL_NAME : name;
+ }
+
+ private TransactionMQProducer stageMQProducer(String name) throws MessagingException {
+ name = getTxProducerGroupName(name);
+
+ TransactionMQProducer cachedProducer = cache.get(name);
+ if (cachedProducer == null) {
+ throw new MessagingException(
+ String.format("Can not found MQProducer '%s' in cache! please define @RocketMQLocalTransactionListener class or invoke createOrGetStartedTransactionMQProducer() to create it firstly", name));
+ }
+
+ return cachedProducer;
+ }
+
+ /**
+ * Send Spring Message in Transaction
+ *
+ * @param txProducerGroup the validate txProducerGroup name, set null if using the default name
+ * @param destination destination formats: `topicName:tags`
+ * @param message message {@link org.springframework.messaging.Message}
+ * @param arg ext arg
+ * @return TransactionSendResult
+ * @throws MessagingException
+ */
+ public TransactionSendResult sendMessageInTransaction(final String txProducerGroup, final String destination, final Message<?> message, final Object arg) throws MessagingException {
+ try {
+ TransactionMQProducer txProducer = this.stageMQProducer(txProducerGroup);
+ org.apache.rocketmq.common.message.Message rocketMsg = RocketMQUtil.convertToRocketMessage(objectMapper,
+ charset, destination, message);
+ return txProducer.sendMessageInTransaction(rocketMsg, arg);
+ } catch (MQClientException e) {
+ throw RocketMQUtil.convert(e);
+ }
+ }
+
+ /**
+ * Remove a TransactionMQProducer from cache by manual.
+ * <p>Note: RocketMQTemplate can release all cached producers when bean destroying, it is not recommended to directly
+ * use this method by user.
+ *
+ * @param txProducerGroup
+ * @throws MessagingException
+ */
+ public void removeTransactionMQProducer(String txProducerGroup) throws MessagingException {
+ txProducerGroup = getTxProducerGroupName(txProducerGroup);
+ if (cache.containsKey(txProducerGroup)) {
+ DefaultMQProducer cachedProducer = cache.get(txProducerGroup);
+ cachedProducer.shutdown();
+ cache.remove(txProducerGroup);
+ }
+ }
+
+ /**
+ * Create and start a transaction MQProducer, this new producer is cached in memory.
+ * <p>Note: This method is invoked internally when processing {@code @RocketMQLocalTransactionListener}, it is not
+ * recommended to directly use this method by user.
+ *
+ * @param txProducerGroup Producer (group) name, unique for each producer
+ * @param transactionListener TransactoinListener impl class
+ * @param executorService Nullable.
+ * @return true if producer is created and started; false if the named producer already exists in cache.
+ * @throws MessagingException
+ */
+ public boolean createAndStartTransactionMQProducer(String txProducerGroup, RocketMQLocalTransactionListener transactionListener,
+ ExecutorService executorService) throws MessagingException {
+ txProducerGroup = getTxProducerGroupName(txProducerGroup);
+ if (cache.containsKey(txProducerGroup)) {
+ log.info(String.format("get TransactionMQProducer '%s' from cache", txProducerGroup));
+ return false;
+ }
+
+ TransactionMQProducer txProducer = createTransactionMQProducer(txProducerGroup, transactionListener, executorService);
+ try {
+ txProducer.start();
+ cache.put(txProducerGroup, txProducer);
+ } catch (MQClientException e) {
+ throw RocketMQUtil.convert(e);
+ }
+
+ return true;
+ }
+
+ private TransactionMQProducer createTransactionMQProducer(String name, RocketMQLocalTransactionListener transactionListener,
+ ExecutorService executorService) {
+ Assert.notNull(producer, "Property 'producer' is required");
+ Assert.notNull(transactionListener, "Parameter 'transactionListener' is required");
+ TransactionMQProducer txProducer = new TransactionMQProducer(name);
+ txProducer.setTransactionListener(RocketMQUtil.convert(transactionListener));
+
+ txProducer.setNamesrvAddr(producer.getNamesrvAddr());
+ if (executorService != null) {
+ txProducer.setExecutorService(executorService);
+ }
+
+ txProducer.setSendMsgTimeout(producer.getSendMsgTimeout());
+ txProducer.setRetryTimesWhenSendFailed(producer.getRetryTimesWhenSendFailed());
+ txProducer.setRetryTimesWhenSendAsyncFailed(producer.getRetryTimesWhenSendAsyncFailed());
+ txProducer.setMaxMessageSize(producer.getMaxMessageSize());
+ txProducer.setCompressMsgBodyOverHowmuch(producer.getCompressMsgBodyOverHowmuch());
+ txProducer.setRetryAnotherBrokerWhenNotStoreOK(producer.isRetryAnotherBrokerWhenNotStoreOK());
+
+ return txProducer;
+ }
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainer.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainer.java
new file mode 100644
index 0000000..1993035
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/DefaultRocketMQListenerContainer.java
@@ -0,0 +1,423 @@
+/*
+ * 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.spring.support;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.MessageSelector;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.spring.annotation.ConsumeMode;
+import org.apache.rocketmq.spring.annotation.MessageModel;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.annotation.SelectorType;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.aop.support.AopUtils;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.context.SmartLifecycle;
+import org.springframework.util.Assert;
+
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.nio.charset.Charset;
+import java.util.List;
+import java.util.Objects;
+
+@SuppressWarnings("WeakerAccess")
+public class DefaultRocketMQListenerContainer implements InitializingBean, RocketMQListenerContainer, SmartLifecycle {
+ private final static Logger log = LoggerFactory.getLogger(DefaultRocketMQListenerContainer.class);
+
+ private long suspendCurrentQueueTimeMillis = 1000;
+
+ /**
+ * Message consume retry strategy<br> -1,no retry,put into DLQ directly<br> 0,broker control retry frequency<br>
+ * >0,client control retry frequency.
+ */
+ private int delayLevelWhenNextConsume = 0;
+
+ private String nameServer;
+
+ private String consumerGroup;
+
+ private String topic;
+
+ private int consumeThreadMax = 64;
+
+ private String charset = "UTF-8";
+
+ private ObjectMapper objectMapper = new ObjectMapper();
+
+ private RocketMQListener rocketMQListener;
+
+ private RocketMQMessageListener rocketMQMessageListener;
+
+ private DefaultMQPushConsumer consumer;
+
+ private Class messageType;
+
+ private boolean running;
+
+ // The following properties came from @RocketMQMessageListener.
+ private ConsumeMode consumeMode;
+ private SelectorType selectorType;
+ private String selectorExpression;
+ private MessageModel messageModel;
+
+ public long getSuspendCurrentQueueTimeMillis() {
+ return suspendCurrentQueueTimeMillis;
+ }
+
+ public void setSuspendCurrentQueueTimeMillis(long suspendCurrentQueueTimeMillis) {
+ this.suspendCurrentQueueTimeMillis = suspendCurrentQueueTimeMillis;
+ }
+
+ public int getDelayLevelWhenNextConsume() {
+ return delayLevelWhenNextConsume;
+ }
+
+ public void setDelayLevelWhenNextConsume(int delayLevelWhenNextConsume) {
+ this.delayLevelWhenNextConsume = delayLevelWhenNextConsume;
+ }
+
+ public String getNameServer() {
+ return nameServer;
+ }
+
+ public void setNameServer(String nameServer) {
+ this.nameServer = nameServer;
+ }
+
+ public String getConsumerGroup() {
+ return consumerGroup;
+ }
+
+ public void setConsumerGroup(String consumerGroup) {
+ this.consumerGroup = consumerGroup;
+ }
+
+ public String getTopic() {
+ return topic;
+ }
+
+ public void setTopic(String topic) {
+ this.topic = topic;
+ }
+
+ public int getConsumeThreadMax() {
+ return consumeThreadMax;
+ }
+
+ public String getCharset() {
+ return charset;
+ }
+
+ public void setCharset(String charset) {
+ this.charset = charset;
+ }
+
+ public ObjectMapper getObjectMapper() {
+ return objectMapper;
+ }
+
+ public void setObjectMapper(ObjectMapper objectMapper) {
+ this.objectMapper = objectMapper;
+ }
+
+
+ public RocketMQListener getRocketMQListener() {
+ return rocketMQListener;
+ }
+
+ public void setRocketMQListener(RocketMQListener rocketMQListener) {
+ this.rocketMQListener = rocketMQListener;
+ }
+
+ public RocketMQMessageListener getRocketMQMessageListener() {
+ return rocketMQMessageListener;
+ }
+
+ public void setRocketMQMessageListener(RocketMQMessageListener anno) {
+ this.rocketMQMessageListener = anno;
+
+ this.consumeMode = anno.consumeMode();
+ this.consumeThreadMax = anno.consumeThreadMax();
+ this.messageModel = anno.messageModel();
+ this.selectorExpression = anno.selectorExpression();
+ this.selectorType = anno.selectorType();
+ }
+
+ public ConsumeMode getConsumeMode() {
+ return consumeMode;
+ }
+
+ public SelectorType getSelectorType() {
+ return selectorType;
+ }
+
+ public String getSelectorExpression() {
+ return selectorExpression;
+ }
+
+ public MessageModel getMessageModel() {
+ return messageModel;
+ }
+
+ public DefaultMQPushConsumer getConsumer() {
+ return consumer;
+ }
+
+ public void setConsumer(DefaultMQPushConsumer consumer) {
+ this.consumer = consumer;
+ }
+
+ public void setupMessageListener(RocketMQListener rocketMQListener) {
+ this.rocketMQListener = rocketMQListener;
+ }
+
+ @Override
+ public void destroy() {
+ this.setRunning(false);
+ if (Objects.nonNull(consumer)) {
+ consumer.shutdown();
+ }
+ log.info("container destroyed, {}", this.toString());
+ }
+
+ @Override
+ public boolean isAutoStartup() {
+ return true;
+ }
+
+ @Override
+ public void stop(Runnable callback) {
+ stop();
+ callback.run();
+ }
+
+ @Override
+ public void start() {
+ if (this.isRunning()) {
+ throw new IllegalStateException("container already running. " + this.toString());
+ }
+
+ try {
+ consumer.start();
+ } catch (MQClientException e) {
+ throw new IllegalStateException("Failed to start RocketMQ push consumer", e);
+ }
+ this.setRunning(true);
+
+ log.info("running container: {}", this.toString());
+ }
+
+ @Override
+ public void stop() {
+ if (this.isRunning()) {
+ if (Objects.nonNull(consumer)) {
+ consumer.shutdown();
+ }
+ setRunning(false);
+ }
+ }
+
+ @Override
+ public boolean isRunning() {
+ return running;
+ }
+
+ private void setRunning(boolean running) {
+ this.running = running;
+ }
+
+ @Override
+ public int getPhase() {
+ // Returning Integer.MAX_VALUE only suggests that
+ // we will be the first bean to shutdown and last bean to start
+ return Integer.MAX_VALUE;
+ }
+
+
+ @Override
+ public void afterPropertiesSet() throws Exception {
+ initRocketMQPushConsumer();
+
+ this.messageType = getMessageType();
+ log.debug("RocketMQ messageType: {}", messageType.getName());
+ }
+
+ @Override
+ public String toString() {
+ return "DefaultRocketMQListenerContainer{" +
+ "consumerGroup='" + consumerGroup + '\'' +
+ ", nameServer='" + nameServer + '\'' +
+ ", topic='" + topic + '\'' +
+ ", consumeMode=" + consumeMode +
+ ", selectorType=" + selectorType +
+ ", selectorExpression='" + selectorExpression + '\'' +
+ ", messageModel=" + messageModel +
+ '}';
+ }
+
+ public class DefaultMessageListenerConcurrently implements MessageListenerConcurrently {
+
+ @SuppressWarnings("unchecked")
+ public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
+ for (MessageExt messageExt : msgs) {
+ log.debug("received msg: {}", messageExt);
+ try {
+ long now = System.currentTimeMillis();
+ rocketMQListener.onMessage(doConvertMessage(messageExt));
+ long costTime = System.currentTimeMillis() - now;
+ log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
+ } catch (Exception e) {
+ log.warn("consume message failed. messageExt:{}", messageExt, e);
+ context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);
+ return ConsumeConcurrentlyStatus.RECONSUME_LATER;
+ }
+ }
+
+ return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+ }
+ }
+
+ public class DefaultMessageListenerOrderly implements MessageListenerOrderly {
+
+ @SuppressWarnings("unchecked")
+ public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
+ for (MessageExt messageExt : msgs) {
+ log.debug("received msg: {}", messageExt);
+ try {
+ long now = System.currentTimeMillis();
+ rocketMQListener.onMessage(doConvertMessage(messageExt));
+ long costTime = System.currentTimeMillis() - now;
+ log.info("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
+ } catch (Exception e) {
+ log.warn("consume message failed. messageExt:{}", messageExt, e);
+ context.setSuspendCurrentQueueTimeMillis(suspendCurrentQueueTimeMillis);
+ return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
+ }
+ }
+
+ return ConsumeOrderlyStatus.SUCCESS;
+ }
+ }
+
+
+ @SuppressWarnings("unchecked")
+ private Object doConvertMessage(MessageExt messageExt) {
+ if (Objects.equals(messageType, MessageExt.class)) {
+ return messageExt;
+ } else {
+ String str = new String(messageExt.getBody(), Charset.forName(charset));
+ if (Objects.equals(messageType, String.class)) {
+ return str;
+ } else {
+ // If msgType not string, use objectMapper change it.
+ try {
+ return objectMapper.readValue(str, messageType);
+ } catch (Exception e) {
+ log.info("convert failed. str:{}, msgType:{}", str, messageType);
+ throw new RuntimeException("cannot convert message to " + messageType, e);
+ }
+ }
+ }
+ }
+
+ private Class getMessageType() {
+ Type[] interfaces = AopUtils.getTargetClass(rocketMQListener).getGenericInterfaces();
+ if (Objects.nonNull(interfaces)) {
+ for (Type type : interfaces) {
+ if (type instanceof ParameterizedType) {
+ ParameterizedType parameterizedType = (ParameterizedType) type;
+ if (Objects.equals(parameterizedType.getRawType(), RocketMQListener.class)) {
+ Type[] actualTypeArguments = parameterizedType.getActualTypeArguments();
+ if (Objects.nonNull(actualTypeArguments) && actualTypeArguments.length > 0) {
+ return (Class) actualTypeArguments[0];
+ } else {
+ return Object.class;
+ }
+ }
+ }
+ }
+
+ return Object.class;
+ } else {
+ return Object.class;
+ }
+ }
+
+ private void initRocketMQPushConsumer() throws MQClientException {
+ Assert.notNull(rocketMQListener, "Property 'rocketMQListener' is required");
+ Assert.notNull(consumerGroup, "Property 'consumerGroup' is required");
+ Assert.notNull(nameServer, "Property 'nameServer' is required");
+ Assert.notNull(topic, "Property 'topic' is required");
+
+ consumer = new DefaultMQPushConsumer(consumerGroup);
+ consumer.setNamesrvAddr(nameServer);
+ consumer.setConsumeThreadMax(consumeThreadMax);
+ if (consumeThreadMax < consumer.getConsumeThreadMin()) {
+ consumer.setConsumeThreadMin(consumeThreadMax);
+ }
+
+ switch (messageModel) {
+ case BROADCASTING:
+ consumer.setMessageModel(org.apache.rocketmq.common.protocol.heartbeat.MessageModel.BROADCASTING);
+ break;
+ case CLUSTERING:
+ consumer.setMessageModel(org.apache.rocketmq.common.protocol.heartbeat.MessageModel.CLUSTERING);
+ break;
+ }
+
+ switch (selectorType) {
+ case TAG:
+ consumer.subscribe(topic, selectorExpression);
+ break;
+ case SQL92:
+ consumer.subscribe(topic, MessageSelector.bySql(selectorExpression));
+ break;
+ default:
+ throw new IllegalArgumentException("Property 'selectorType' was wrong.");
+ }
+
+ switch (consumeMode) {
+ case ORDERLY:
+ consumer.setMessageListener(new DefaultMessageListenerOrderly());
+ break;
+ case CONCURRENTLY:
+ consumer.setMessageListener(new DefaultMessageListenerConcurrently());
+ break;
+ default:
+ throw new IllegalArgumentException("Property 'consumeMode' was wrong.");
+ }
+
+ if (rocketMQListener instanceof RocketMQPushConsumerLifecycleListener) {
+ ((RocketMQPushConsumerLifecycleListener) rocketMQListener).prepareStart(consumer);
+ }
+
+ }
+
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQConsumerLifecycleListener.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQConsumerLifecycleListener.java
new file mode 100644
index 0000000..8cc9b80
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQConsumerLifecycleListener.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.rocketmq.spring.support;
+
+public interface RocketMQConsumerLifecycleListener<T> {
+ void prepareStart(final T consumer);
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQHeaders.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQHeaders.java
new file mode 100644
index 0000000..e21aeab
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQHeaders.java
@@ -0,0 +1,34 @@
+/*
+ * 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.spring.support;
+
+/**
+ * Represents the RocketMQ message protocol that is used during the data exchange.
+ */
+public class RocketMQHeaders {
+ public static final String KEYS = "KEYS";
+ public static final String TAGS = "TAGS";
+ public static final String TOPIC = "TOPIC";
+ public static final String MESSAGE_ID = "MESSAGE_ID";
+ public static final String BORN_TIMESTAMP = "BORN_TIMESTAMP";
+ public static final String BORN_HOST = "BORN_HOST";
+ public static final String FLAG = "FLAG";
+ public static final String QUEUE_ID = "QUEUE_ID";
+ public static final String SYS_FLAG = "SYS_FLAG";
+ public static final String TRANSACTION_ID = "TRANSACTION_ID";
+ public static final String PROPERTIES = "PROPERTIES";
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQListenerContainer.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQListenerContainer.java
new file mode 100644
index 0000000..ee52de8
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQListenerContainer.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 org.apache.rocketmq.spring.support;
+
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.beans.factory.DisposableBean;
+
+public interface RocketMQListenerContainer extends DisposableBean {
+
+ /**
+ * Setup the message listener to use. Throws an {@link IllegalArgumentException} if that message listener type is
+ * not supported.
+ */
+ void setupMessageListener(RocketMQListener<?> messageListener);
+}
diff --git a/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQUtil.java b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQUtil.java
new file mode 100644
index 0000000..7905be5
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/java/org/apache/rocketmq/spring/support/RocketMQUtil.java
@@ -0,0 +1,168 @@
+/*
+ * 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.spring.support;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.LocalTransactionState;
+import org.apache.rocketmq.client.producer.TransactionListener;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
+import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.nio.charset.Charset;
+import java.util.Objects;
+
+public class RocketMQUtil {
+ private final static Logger log = LoggerFactory.getLogger(RocketMQUtil.class);
+
+ public static TransactionListener convert(RocketMQLocalTransactionListener listener) {
+ return new TransactionListener() {
+ @Override
+ public LocalTransactionState executeLocalTransaction(Message message, Object obj) {
+ RocketMQLocalTransactionState state = listener.executeLocalTransaction(convertToSpringMessage(message), obj);
+ return convertLocalTransactionState(state);
+ }
+
+ @Override
+ public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
+ RocketMQLocalTransactionState state = listener.checkLocalTransaction(convertToSpringMessage(messageExt));
+ return convertLocalTransactionState(state);
+ }
+ };
+ }
+
+ private static LocalTransactionState convertLocalTransactionState(RocketMQLocalTransactionState state) {
+ switch (state) {
+ case UNKNOWN:
+ return LocalTransactionState.UNKNOW;
+ case COMMIT:
+ return LocalTransactionState.COMMIT_MESSAGE;
+ case ROLLBACK:
+ return LocalTransactionState.ROLLBACK_MESSAGE;
+ }
+
+ // Never happen
+ log.warn("Failed to covert enum type RocketMQLocalTransactionState.%s", state);
+ return LocalTransactionState.UNKNOW;
+ }
+
+ public static MessagingException convert(MQClientException e) {
+ return new MessagingException(e.getErrorMessage(), e);
+ }
+
+ public static org.springframework.messaging.Message convertToSpringMessage(
+ org.apache.rocketmq.common.message.MessageExt message) {
+ org.springframework.messaging.Message retMessage =
+ MessageBuilder.withPayload(message.getBody()).
+ setHeader(RocketMQHeaders.KEYS, message.getKeys()).
+ setHeader(RocketMQHeaders.TAGS, message.getTags()).
+ setHeader(RocketMQHeaders.TOPIC, message.getTopic()).
+ setHeader(RocketMQHeaders.MESSAGE_ID, message.getMsgId()).
+ setHeader(RocketMQHeaders.BORN_TIMESTAMP, message.getBornTimestamp()).
+ setHeader(RocketMQHeaders.BORN_HOST, message.getBornHostString()).
+ setHeader(RocketMQHeaders.FLAG, message.getFlag()).
+ setHeader(RocketMQHeaders.QUEUE_ID, message.getQueueId()).
+ setHeader(RocketMQHeaders.SYS_FLAG, message.getSysFlag()).
+ setHeader(RocketMQHeaders.TRANSACTION_ID, message.getTransactionId()).
+ setHeader(RocketMQHeaders.PROPERTIES, message.getProperties()).
+ build();
+
+ return retMessage;
+ }
+
+ public static org.springframework.messaging.Message convertToSpringMessage(
+ org.apache.rocketmq.common.message.Message message) {
+ org.springframework.messaging.Message retMessage =
+ MessageBuilder.withPayload(message.getBody()).
+ setHeader(RocketMQHeaders.KEYS, message.getKeys()).
+ setHeader(RocketMQHeaders.TAGS, message.getTags()).
+ setHeader(RocketMQHeaders.TOPIC, message.getTopic()).
+ setHeader(RocketMQHeaders.FLAG, message.getFlag()).
+ setHeader(RocketMQHeaders.TRANSACTION_ID, message.getTransactionId()).
+ setHeader(RocketMQHeaders.PROPERTIES, message.getProperties()).
+ build();
+
+ return retMessage;
+ }
+
+ public static org.apache.rocketmq.common.message.Message convertToRocketMessage(
+ ObjectMapper objectMapper, String charset,
+ String destination, org.springframework.messaging.Message<?> message) {
+ Object payloadObj = message.getPayload();
+ byte[] payloads;
+
+ if (payloadObj instanceof String) {
+ payloads = ((String) payloadObj).getBytes(Charset.forName(charset));
+ } else {
+ try {
+ String jsonObj = objectMapper.writeValueAsString(payloadObj);
+ payloads = jsonObj.getBytes(Charset.forName(charset));
+ } catch (Exception e) {
+ throw new RuntimeException("convert to RocketMQ message failed.", e);
+ }
+ }
+
+ String[] tempArr = destination.split(":", 2);
+ String topic = tempArr[0];
+ String tags = "";
+ if (tempArr.length > 1) {
+ tags = tempArr[1];
+ }
+
+ org.apache.rocketmq.common.message.Message rocketMsg = new org.apache.rocketmq.common.message.Message(topic, tags, payloads);
+
+ MessageHeaders headers = message.getHeaders();
+ if (Objects.nonNull(headers) && !headers.isEmpty()) {
+ Object keys = headers.get(RocketMQHeaders.KEYS);
+ if (!StringUtils.isEmpty(keys)) { // if headers has 'KEYS', set rocketMQ message key
+ rocketMsg.setKeys(keys.toString());
+ }
+
+ Object flagObj = headers.getOrDefault("FLAG", "0");
+ int flag = 0;
+ try {
+ flag = Integer.parseInt(flagObj.toString());
+ } catch (NumberFormatException e) {
+ // Ignore it
+ log.info("flag must be integer, flagObj:{}", flagObj);
+ }
+ rocketMsg.setFlag(flag);
+
+ Object waitStoreMsgOkObj = headers.getOrDefault("WAIT_STORE_MSG_OK", "true");
+ boolean waitStoreMsgOK = Boolean.TRUE.equals(waitStoreMsgOkObj);
+ rocketMsg.setWaitStoreMsgOK(waitStoreMsgOK);
+
+ headers.entrySet().stream()
+ .filter(entry -> !Objects.equals(entry.getKey(), RocketMQHeaders.KEYS)
+ && !Objects.equals(entry.getKey(), "FLAG")
+ && !Objects.equals(entry.getKey(), "WAIT_STORE_MSG_OK")) // exclude "KEYS", "FLAG", "WAIT_STORE_MSG_OK"
+ .forEach(entry -> {
+ rocketMsg.putUserProperty("USERS_" + entry.getKey(), String.valueOf(entry.getValue())); // add other properties with prefix "USERS_"
+ });
+
+ }
+
+ return rocketMsg;
+ }
+}
diff --git a/rocketmq-spring-boot/src/main/resources/META-INF/spring.factories b/rocketmq-spring-boot/src/main/resources/META-INF/spring.factories
new file mode 100644
index 0000000..92ff818
--- /dev/null
+++ b/rocketmq-spring-boot/src/main/resources/META-INF/spring.factories
@@ -0,0 +1,2 @@
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+org.apache.rocketmq.spring.config.RocketMQAutoConfiguration
\ No newline at end of file
diff --git a/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/config/RocketMQAutoConfigurationTest.java b/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/config/RocketMQAutoConfigurationTest.java
new file mode 100644
index 0000000..a841a74
--- /dev/null
+++ b/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/config/RocketMQAutoConfigurationTest.java
@@ -0,0 +1,92 @@
+/*
+ * 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.spring.config;
+
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer;
+import org.junit.Test;
+import org.springframework.beans.factory.NoSuchBeanDefinitionException;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class RocketMQAutoConfigurationTest {
+ private ApplicationContextRunner runner = new ApplicationContextRunner()
+ .withConfiguration(AutoConfigurations.of(RocketMQAutoConfiguration.class));
+
+
+ @Test(expected = NoSuchBeanDefinitionException.class)
+ public void testRocketMQAutoConfigurationNotCreatedByDefault() {
+ runner.run(context -> context.getBean(RocketMQAutoConfiguration.class));
+ }
+
+
+ @Test(expected = NoSuchBeanDefinitionException.class)
+ public void testDefaultMQProducerNotCreatedByDefault() {
+ runner.run(context -> context.getBean(DefaultMQProducer.class));
+ }
+
+
+ @Test
+ public void testDefaultMQProducer() {
+ runner.withPropertyValues("spring.rocketmq.nameServer=127.0.0.1:9876",
+ "spring.rocketmq.producer.group=spring_rocketmq").
+ run((context) -> {
+ assertThat(context).hasSingleBean(DefaultMQProducer.class);
+ assertThat(context).hasSingleBean(RocketMQProperties.class);
+ });
+
+ }
+
+ @Test
+ public void testRocketMQListenerContainer() {
+ runner.withPropertyValues("spring.rocketmq.nameServer=127.0.0.1:9876").
+ withUserConfiguration(TestConfig.class).
+ run((context) -> {
+ // No producer on consume side
+ assertThat(context).doesNotHaveBean(DefaultMQProducer.class);
+ // Auto-create consume container if existing Bean annotated with @RocketMQMessageListener
+ assertThat(context).hasSingleBean(DefaultRocketMQListenerContainer.class);
+ });
+
+ }
+
+ @Configuration
+ static class TestConfig {
+
+ @Bean
+ public Object consumeListener() {
+ return new MyMessageListener();
+ }
+ }
+
+ @RocketMQMessageListener(consumerGroup = "abc", topic = "test")
+ static class MyMessageListener implements RocketMQListener {
+
+ @Override
+ public void onMessage(Object message) {
+
+ }
+ }
+}
+
diff --git a/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/support/RocketMQUtilTest.java b/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/support/RocketMQUtilTest.java
new file mode 100644
index 0000000..8690b93
--- /dev/null
+++ b/rocketmq-spring-boot/src/test/java/org/apache/rocketmq/spring/support/RocketMQUtilTest.java
@@ -0,0 +1,37 @@
+/*
+ * 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.spring.support;
+
+
+import org.junit.Test;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+
+import static org.junit.Assert.assertTrue;
+
+public class RocketMQUtilTest {
+ @Test
+ public void testMessageBuilder() {
+ Message msg = MessageBuilder.withPayload("test").
+ setHeader("A", "test1").
+ setHeader("B", "test2").
+ build();
+ System.out.printf("header size=%d %s %n", msg.getHeaders().size(), msg.getHeaders().toString());
+ assertTrue(msg.getHeaders().get("A") != null);
+ assertTrue(msg.getHeaders().get("B") != null);
+ }
+}
\ No newline at end of file
diff --git a/style/copyright/Apache.xml b/style/copyright/Apache.xml
new file mode 100644
index 0000000..e3e3dec
--- /dev/null
+++ b/style/copyright/Apache.xml
@@ -0,0 +1,23 @@
+<!--
+ 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.
+ -->
+
+<component name="CopyrightManager">
+ <copyright>
+ <option name="myName" value="Apache" />
+ <option name="notice" value="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.a [...]
+ </copyright>
+</component>
\ No newline at end of file
diff --git a/style/copyright/profiles_settings.xml b/style/copyright/profiles_settings.xml
new file mode 100644
index 0000000..747c7e2
--- /dev/null
+++ b/style/copyright/profiles_settings.xml
@@ -0,0 +1,64 @@
+<!--
+ 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.
+ -->
+
+<component name="CopyrightManager">
+ <settings default="Apache">
+ <module2copyright>
+ <element module="All" copyright="Apache"/>
+ </module2copyright>
+ <LanguageOptions name="GSP">
+ <option name="fileTypeOverride" value="3"/>
+ <option name="prefixLines" value="false"/>
+ </LanguageOptions>
+ <LanguageOptions name="HTML">
+ <option name="fileTypeOverride" value="3"/>
+ <option name="prefixLines" value="false"/>
+ </LanguageOptions>
+ <LanguageOptions name="JAVA">
+ <option name="fileTypeOverride" value="3" />
+ <option name="addBlankAfter" value="false" />
+ </LanguageOptions>
+ <LanguageOptions name="JSP">
+ <option name="fileTypeOverride" value="3"/>
+ <option name="prefixLines" value="false"/>
+ </LanguageOptions>
+ <LanguageOptions name="JSPX">
+ <option name="fileTypeOverride" value="3"/>
+ <option name="prefixLines" value="false"/>
+ </LanguageOptions>
+ <LanguageOptions name="MXML">
+ <option name="fileTypeOverride" value="3"/>
+ <option name="prefixLines" value="false"/>
+ </LanguageOptions>
+ <LanguageOptions name="Properties">
+ <option name="fileTypeOverride" value="3"/>
+ <option name="block" value="false"/>
+ </LanguageOptions>
+ <LanguageOptions name="SPI">
+ <option name="fileTypeOverride" value="3"/>
+ <option name="block" value="false"/>
+ </LanguageOptions>
+ <LanguageOptions name="XML">
+ <option name="fileTypeOverride" value="3"/>
+ <option name="prefixLines" value="false"/>
+ </LanguageOptions>
+ <LanguageOptions name="__TEMPLATE__">
+ <option name="separateBefore" value="true"/>
+ <option name="lenBefore" value="1"/>
+ </LanguageOptions>
+ </settings>
+</component>
\ No newline at end of file
diff --git a/style/rmq_checkstyle.xml b/style/rmq_checkstyle.xml
new file mode 100644
index 0000000..2e9658f
--- /dev/null
+++ b/style/rmq_checkstyle.xml
@@ -0,0 +1,135 @@
+<?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.
+ -->
+
+<!DOCTYPE module PUBLIC
+ "-//Puppy Crawl//DTD Check Configuration 1.3//EN"
+ "http://www.puppycrawl.com/dtds/configuration_1_3.dtd">
+<!--Refer http://checkstyle.sourceforge.net/reports/google-java-style.html#s2.2-file-encoding -->
+<module name="Checker">
+
+ <property name="localeLanguage" value="en"/>
+
+ <!--To configure the check to report on the first instance in each file-->
+ <module name="FileTabCharacter"/>
+
+ <!-- header -->
+ <module name="RegexpHeader">
+ <property name="header" value="/\*\nLicensed to the Apache Software Foundation*"/>
+ <property name="fileExtensions" value="java" />
+ </module>
+
+ <module name="RegexpSingleline">
+ <property name="format" value="System\.out\.println"/>
+ <property name="message" value="Prohibit invoking System.out.println in source code !"/>
+ </module>
+
+ <module name="RegexpSingleline">
+ <property name="format" value="//FIXME"/>
+ <property name="message" value="Recommended fix FIXME task !"/>
+ </module>
+
+ <module name="RegexpSingleline">
+ <property name="format" value="//TODO"/>
+ <property name="message" value="Recommended fix TODO task !"/>
+ </module>
+
+ <module name="RegexpSingleline">
+ <property name="format" value="@alibaba"/>
+ <property name="message" value="Recommended remove @alibaba keyword!"/>
+ </module>
+ <module name="RegexpSingleline">
+ <property name="format" value="@taobao"/>
+ <property name="message" value="Recommended remove @taobao keyword!"/>
+ </module>
+ <module name="RegexpSingleline">
+ <property name="format" value="@author"/>
+ <property name="message" value="Recommended remove @author tag in javadoc!"/>
+ </module>
+
+ <module name="RegexpSingleline">
+ <property name="format"
+ value=".*[\u3400-\u4DB5\u4E00-\u9FA5\u9FA6-\u9FBB\uF900-\uFA2D\uFA30-\uFA6A\uFA70-\uFAD9\uFF00-\uFFEF\u2E80-\u2EFF\u3000-\u303F\u31C0-\u31EF]+.*"/>
+ <property name="message" value="Not allow chinese character !"/>
+ </module>
+
+ <module name="FileLength">
+ <property name="max" value="3000"/>
+ </module>
+
+ <module name="TreeWalker">
+
+ <module name="UnusedImports">
+ <property name="processJavadoc" value="true"/>
+ </module>
+ <module name="RedundantImport"/>
+
+ <!--<module name="IllegalImport" />-->
+
+ <!--Checks that classes that override equals() also override hashCode()-->
+ <module name="EqualsHashCode"/>
+ <!--Checks for over-complicated boolean expressions. Currently finds code like if (topic == true), topic || true, !false, etc.-->
+ <module name="SimplifyBooleanExpression"/>
+ <module name="OneStatementPerLine"/>
+ <module name="UnnecessaryParentheses"/>
+ <!--Checks for over-complicated boolean return statements. For example the following code-->
+ <module name="SimplifyBooleanReturn"/>
+
+ <!--Check that the default is after all the cases in producerGroup switch statement-->
+ <module name="DefaultComesLast"/>
+ <!--Detects empty statements (standalone ";" semicolon)-->
+ <module name="EmptyStatement"/>
+ <!--Checks that long constants are defined with an upper ell-->
+ <module name="UpperEll"/>
+ <module name="ConstantName">
+ <property name="format" value="(^[A-Z][A-Z0-9]*(_[A-Z0-9]+)*$)|(^log$)"/>
+ </module>
+ <!--Checks that local, non-final variable names conform to producerGroup format specified by the format property-->
+ <module name="LocalVariableName"/>
+ <!--Validates identifiers for local, final variables, including catch parameters-->
+ <module name="LocalFinalVariableName"/>
+ <!--Validates identifiers for non-static fields-->
+ <module name="MemberName"/>
+ <!--Validates identifiers for class type parameters-->
+ <module name="ClassTypeParameterName">
+ <property name="format" value="^[A-Z0-9]*$"/>
+ </module>
+ <!--Validates identifiers for method type parameters-->
+ <module name="MethodTypeParameterName">
+ <property name="format" value="^[A-Z0-9]*$"/>
+ </module>
+ <module name="PackageName"/>
+ <module name="ParameterName"/>
+ <module name="StaticVariableName"/>
+ <module name="TypeName"/>
+ <!--Checks that there are no import statements that use the * notation-->
+ <module name="AvoidStarImport"/>
+
+ <!--whitespace-->
+ <module name="GenericWhitespace"/>
+ <module name="NoWhitespaceBefore"/>
+ <module name="NoWhitespaceAfter"/>
+ <module name="WhitespaceAround">
+ <property name="allowEmptyConstructors" value="true"/>
+ <property name="allowEmptyMethods" value="true"/>
+ </module>
+ <module name="Indentation"/>
+ <module name="MethodParamPad"/>
+ <module name="ParenPad"/>
+ <module name="TypecastParenPad"/>
+ </module>
+</module>
diff --git a/style/rmq_codeStyle.xml b/style/rmq_codeStyle.xml
new file mode 100644
index 0000000..9db075e
--- /dev/null
+++ b/style/rmq_codeStyle.xml
@@ -0,0 +1,157 @@
+<!--
+ 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.
+ -->
+
+<code_scheme name="rocketmq">
+ <option name="USE_SAME_INDENTS" value="true"/>
+ <option name="IGNORE_SAME_INDENTS_FOR_LANGUAGES" value="true"/>
+ <option name="OTHER_INDENT_OPTIONS">
+ <value>
+ <option name="INDENT_SIZE" value="4"/>
+ <option name="CONTINUATION_INDENT_SIZE" value="4"/>
+ <option name="TAB_SIZE" value="4"/>
+ <option name="USE_TAB_CHARACTER" value="false"/>
+ <option name="SMART_TABS" value="false"/>
+ <option name="LABEL_INDENT_SIZE" value="0"/>
+ <option name="LABEL_INDENT_ABSOLUTE" value="false"/>
+ <option name="USE_RELATIVE_INDENTS" value="false"/>
+ </value>
+ </option>
+ <option name="PREFER_LONGER_NAMES" value="false"/>
+ <option name="CLASS_COUNT_TO_USE_IMPORT_ON_DEMAND" value="1000"/>
+ <option name="NAMES_COUNT_TO_USE_IMPORT_ON_DEMAND" value="1000"/>
+ <option name="PACKAGES_TO_USE_IMPORT_ON_DEMAND">
+ <value/>
+ </option>
+ <option name="IMPORT_LAYOUT_TABLE">
+ <value>
+ <package name="" withSubpackages="true" static="false"/>
+ <emptyLine/>
+ <package name="" withSubpackages="true" static="true"/>
+ </value>
+ </option>
+ <option name="JD_ALIGN_PARAM_COMMENTS" value="false"/>
+ <option name="JD_ALIGN_EXCEPTION_COMMENTS" value="false"/>
+ <option name="JD_P_AT_EMPTY_LINES" value="false"/>
+ <option name="JD_KEEP_INVALID_TAGS" value="false"/>
+ <option name="JD_DO_NOT_WRAP_ONE_LINE_COMMENTS" value="true"/>
+ <option name="KEEP_CONTROL_STATEMENT_IN_ONE_LINE" value="false"/>
+ <option name="KEEP_BLANK_LINES_IN_DECLARATIONS" value="1"/>
+ <option name="KEEP_BLANK_LINES_IN_CODE" value="1"/>
+ <option name="KEEP_BLANK_LINES_BEFORE_RBRACE" value="1"/>
+ <option name="ELSE_ON_NEW_LINE" value="true"/>
+ <option name="WHILE_ON_NEW_LINE" value="true"/>
+ <option name="CATCH_ON_NEW_LINE" value="true"/>
+ <option name="FINALLY_ON_NEW_LINE" value="true"/>
+ <option name="ALIGN_MULTILINE_PARAMETERS" value="false"/>
+ <option name="ALIGN_MULTILINE_FOR" value="false"/>
+ <option name="SPACE_AFTER_TYPE_CAST" value="false"/>
+ <option name="SPACE_BEFORE_ARRAY_INITIALIZER_LBRACE" value="true"/>
+ <option name="METHOD_PARAMETERS_WRAP" value="1"/>
+ <option name="ARRAY_INITIALIZER_LBRACE_ON_NEXT_LINE" value="true"/>
+ <option name="LABELED_STATEMENT_WRAP" value="1"/>
+ <option name="WRAP_COMMENTS" value="true"/>
+ <option name="METHOD_ANNOTATION_WRAP" value="1"/>
+ <option name="CLASS_ANNOTATION_WRAP" value="1"/>
+ <option name="FIELD_ANNOTATION_WRAP" value="1"/>
+ <JavaCodeStyleSettings>
+ <option name="CLASS_NAMES_IN_JAVADOC" value="3"/>
+ </JavaCodeStyleSettings>
+ <XML>
+ <option name="XML_LEGACY_SETTINGS_IMPORTED" value="true"/>
+ </XML>
+ <ADDITIONAL_INDENT_OPTIONS fileType="haml">
+ <option name="INDENT_SIZE" value="2"/>
+ </ADDITIONAL_INDENT_OPTIONS>
+ <codeStyleSettings language="Groovy">
+ <option name="KEEP_CONTROL_STATEMENT_IN_ONE_LINE" value="false"/>
+ <option name="KEEP_BLANK_LINES_IN_DECLARATIONS" value="1"/>
+ <option name="KEEP_BLANK_LINES_IN_CODE" value="1"/>
+ <option name="KEEP_BLANK_LINES_BEFORE_RBRACE" value="1"/>
+ <option name="ELSE_ON_NEW_LINE" value="true"/>
+ <option name="CATCH_ON_NEW_LINE" value="true"/>
+ <option name="FINALLY_ON_NEW_LINE" value="true"/>
+ <option name="ALIGN_MULTILINE_PARAMETERS" value="false"/>
+ <option name="ALIGN_MULTILINE_FOR" value="false"/>
+ <option name="SPACE_AFTER_TYPE_CAST" value="false"/>
+ <option name="METHOD_PARAMETERS_WRAP" value="1"/>
+ <option name="METHOD_ANNOTATION_WRAP" value="1"/>
+ <option name="CLASS_ANNOTATION_WRAP" value="1"/>
+ <option name="FIELD_ANNOTATION_WRAP" value="1"/>
+ <option name="PARENT_SETTINGS_INSTALLED" value="true"/>
+ <indentOptions>
+ <option name="CONTINUATION_INDENT_SIZE" value="4"/>
+ </indentOptions>
+ </codeStyleSettings>
+ <codeStyleSettings language="HOCON">
+ <option name="KEEP_BLANK_LINES_BEFORE_RBRACE" value="1"/>
+ <option name="PARENT_SETTINGS_INSTALLED" value="true"/>
+ </codeStyleSettings>
+ <codeStyleSettings language="JAVA">
+ <option name="KEEP_CONTROL_STATEMENT_IN_ONE_LINE" value="false"/>
+ <option name="KEEP_BLANK_LINES_IN_DECLARATIONS" value="1"/>
+ <option name="KEEP_BLANK_LINES_IN_CODE" value="1"/>
+ <option name="KEEP_BLANK_LINES_BEFORE_RBRACE" value="1"/>
+ <option name="ELSE_ON_NEW_LINE" value="true"/>
+ <option name="WHILE_ON_NEW_LINE" value="true"/>
+ <option name="CATCH_ON_NEW_LINE" value="true"/>
+ <option name="FINALLY_ON_NEW_LINE" value="true"/>
+ <option name="ALIGN_MULTILINE_PARAMETERS" value="false"/>
+ <option name="ALIGN_MULTILINE_FOR" value="false"/>
+ <option name="SPACE_AFTER_TYPE_CAST" value="false"/>
+ <option name="SPACE_BEFORE_ARRAY_INITIALIZER_LBRACE" value="true"/>
+ <option name="METHOD_PARAMETERS_WRAP" value="1"/>
+ <option name="ARRAY_INITIALIZER_LBRACE_ON_NEXT_LINE" value="true"/>
+ <option name="LABELED_STATEMENT_WRAP" value="1"/>
+ <option name="METHOD_ANNOTATION_WRAP" value="1"/>
+ <option name="CLASS_ANNOTATION_WRAP" value="1"/>
+ <option name="FIELD_ANNOTATION_WRAP" value="1"/>
+ <option name="PARENT_SETTINGS_INSTALLED" value="true"/>
+ <indentOptions>
+ <option name="CONTINUATION_INDENT_SIZE" value="4"/>
+ </indentOptions>
+ </codeStyleSettings>
+ <codeStyleSettings language="JSON">
+ <option name="KEEP_BLANK_LINES_IN_CODE" value="1"/>
+ <option name="PARENT_SETTINGS_INSTALLED" value="true"/>
+ </codeStyleSettings>
+ <codeStyleSettings language="Scala">
+ <option name="KEEP_BLANK_LINES_IN_DECLARATIONS" value="1"/>
+ <option name="KEEP_BLANK_LINES_IN_CODE" value="1"/>
+ <option name="KEEP_BLANK_LINES_BEFORE_RBRACE" value="1"/>
+ <option name="ELSE_ON_NEW_LINE" value="true"/>
+ <option name="WHILE_ON_NEW_LINE" value="true"/>
+ <option name="CATCH_ON_NEW_LINE" value="true"/>
+ <option name="FINALLY_ON_NEW_LINE" value="true"/>
+ <option name="ALIGN_MULTILINE_PARAMETERS" value="false"/>
+ <option name="ALIGN_MULTILINE_FOR" value="false"/>
+ <option name="METHOD_PARAMETERS_WRAP" value="1"/>
+ <option name="METHOD_ANNOTATION_WRAP" value="1"/>
+ <option name="CLASS_ANNOTATION_WRAP" value="1"/>
+ <option name="FIELD_ANNOTATION_WRAP" value="1"/>
+ <option name="PARENT_SETTINGS_INSTALLED" value="true"/>
+ <indentOptions>
+ <option name="INDENT_SIZE" value="4"/>
+ <option name="CONTINUATION_INDENT_SIZE" value="4"/>
+ <option name="TAB_SIZE" value="4"/>
+ </indentOptions>
+ </codeStyleSettings>
+ <codeStyleSettings language="XML">
+ <indentOptions>
+ <option name="CONTINUATION_INDENT_SIZE" value="4"/>
+ </indentOptions>
+ </codeStyleSettings>
+</code_scheme>
\ No newline at end of file