feature: Integrate RocketMQ into Seata

script/client/spring/application.properties

@@ -15,6 +15,7 @@
 #
 
 seata.enabled=true
+seata.rocketmq-enabled=false
 seata.scan-packages=firstPackage,secondPackage
 seata.excludes-for-scanning=firstBeanNameForExclude,secondBeanNameForExclude
 seata.excludes-for-auto-proxying=firstClassNameForExclude,secondClassNameForExclude

script/client/spring/application.yml

@@ -1,5 +1,6 @@
 seata:
   enabled: true
+  rocketmq-enabled: false
   application-id: applicationName
   tx-service-group: default_tx_group
   enable-auto-data-source-proxy: true

seata-spring-autoconfigure/seata-spring-autoconfigure-client/src/main/java/io/seata/spring/boot/autoconfigure/properties/SeataProperties.java

@@ -32,6 +32,10 @@ public class SeataProperties {
      * whether enable auto configuration
      */
     private boolean enabled = true;
+    /**
+     * whether enable rocketmq integrate
+     */
+    private boolean rocketmqEnabled = false;
     /**
      * application id
      */
@@ -77,6 +81,14 @@ public SeataProperties setEnabled(boolean enabled) {
         return this;
     }
 
+    public boolean isRocketmqEnabled() {
+        return rocketmqEnabled;
+    }
+
+    public void setRocketmqEnabled(boolean rocketmqEnabled) {
+        this.rocketmqEnabled = rocketmqEnabled;
+    }
+
     public String getApplicationId() {
         if (applicationId == null) {
             applicationId = springCloudAlibabaConfiguration.getApplicationId();

seata-spring-autoconfigure/seata-spring-autoconfigure-client/src/main/resources/META-INF/additional-spring-configuration-metadata.json

@@ -13,6 +13,13 @@
       "sourceType": "io.seata.spring.boot.autoconfigure.properties.SeataProperties",
       "defaultValue": "AT"
     },
+    {
+      "name": "seata.rocketmq-enabled",
+      "type": "java.lang.Boolean",
+      "sourceType": "io.seata.spring.boot.autoconfigure.properties.SeataProperties",
+      "description": "Enable seata integrate rocketmq",
+      "defaultValue": false
+    },
     {
       "name": "spring.cloud.alibaba.seata.application-id",
       "type": "java.lang.String",

seata-spring-boot-starter/pom.xml

@@ -50,6 +50,13 @@
             <artifactId>spring-boot-configuration-processor</artifactId>
             <optional>true</optional>
         </dependency>
+        <!-- rocketmq -->
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-client</artifactId>
+            <version>4.9.0</version>
+            <optional>true</optional>
+        </dependency>
     </dependencies>
 
 </project>

seata-spring-boot-starter/src/main/java/io/seata/spring/boot/autoconfigure/RocketMQAutoConfiguration.java

@@ -0,0 +1,43 @@
+/*
+ *  Copyright 1999-2019 Seata.io Group.
+ *
+ *  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.
+ */
+package io.seata.spring.boot.autoconfigure;
+
+import io.seata.rm.tcc.rocketmq.RocketMQAspect;
+import io.seata.rm.tcc.rocketmq.TCCRocketMQ;
+import io.seata.rm.tcc.rocketmq.TCCRocketMQImpl;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.context.annotation.Bean;
+
+@ConditionalOnClass(DefaultMQProducer.class)
+@ConditionalOnBean(DefaultMQProducer.class)
+@ConditionalOnExpression("${seata.enabled:true} && ${seata.rocketmq-enabled:false}")
+public class RocketMQAutoConfiguration {
+
+    @Bean
+    @ConditionalOnMissingBean
+    public TCCRocketMQ tccRocketMQ() {
+        return new TCCRocketMQImpl();
+    }
+
+    @Bean
+    public RocketMQAspect rocketMQAspect(TCCRocketMQ tccRocketMQ) {
+        return new RocketMQAspect(tccRocketMQ);
+    }
+}

seata-spring-boot-starter/src/main/resources/META-INF/spring.factories

@@ -3,4 +3,5 @@ org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
 io.seata.spring.boot.autoconfigure.SeataDataSourceAutoConfiguration,\
 io.seata.spring.boot.autoconfigure.SeataAutoConfiguration,\
 io.seata.spring.boot.autoconfigure.HttpAutoConfiguration,\
-io.seata.spring.boot.autoconfigure.SeataSagaAutoConfiguration
+io.seata.spring.boot.autoconfigure.SeataSagaAutoConfiguration,\
+io.seata.spring.boot.autoconfigure.RocketMQAutoConfiguration

spring/src/main/java/io/seata/spring/annotation/EnableRocketMQAspect.java

@@ -0,0 +1,32 @@
+/*
+ *  Copyright 1999-2019 Seata.io Group.
+ *
+ *  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.
+ */
+package io.seata.spring.annotation;
+
+import io.seata.rm.tcc.rocketmq.RocketMQAspect;
+import io.seata.rm.tcc.rocketmq.TCCRocketMQImpl;
+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;
+import org.springframework.context.annotation.Import;
+
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@Import({RocketMQAspect.class, TCCRocketMQImpl.class})
+public @interface EnableRocketMQAspect {
+}

spring/src/main/java/io/seata/spring/annotation/scannercheckers/PackageScannerChecker.java

@@ -58,7 +58,7 @@ public static void addScannablePackages(String... packages) {
 
     @Override
     public boolean check(Object bean, String beanName, @Nullable ConfigurableListableBeanFactory beanFactory) throws Exception {
-        if (SCANNABLE_PACKAGE_SET.isEmpty()) {
+        if (SCANNABLE_PACKAGE_SET.isEmpty() || bean.getClass().getName().startsWith("io.seata")) {
             // if empty, pass this checker
             return true;
         }

tcc/pom.xml

@@ -47,6 +47,18 @@
             <groupId>com.alibaba</groupId>
             <artifactId>fastjson</artifactId>
         </dependency>
+
+        <!-- tcc rocketmq  -->
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-aop</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-client</artifactId>
+            <version>4.9.0</version>
+            <optional>true</optional>
+        </dependency>
     </dependencies>
 
 

tcc/src/main/java/io/seata/rm/tcc/rocketmq/RocketMQAspect.java

@@ -0,0 +1,47 @@
+/*
+ *  Copyright 1999-2019 Seata.io Group.
+ *
+ *  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.
+ */
+package io.seata.rm.tcc.rocketmq;
+
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+
+public class RocketMQAspect implements BeanPostProcessor {
+    public static Logger LOGGER = LoggerFactory.getLogger(RocketMQAspect.class);
+
+    private final TCCRocketMQ tccRocketMQ;
+
+    public RocketMQAspect(TCCRocketMQ tccRocketMQ) {
+        this.tccRocketMQ = tccRocketMQ;
+    }
+
+    @Override
+    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
+        return bean;
+    }
+
+    @Override
+    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+        if (bean instanceof DefaultMQProducer) {
+            LOGGER.info("Generate RocketMQ Producer Proxy");
+            tccRocketMQ.setDefaultMQProducer((DefaultMQProducer) bean);
+            return new SeataMQProducer((DefaultMQProducer) bean, tccRocketMQ);
+        }
+        return bean;
+    }
+}

tcc/src/main/java/io/seata/rm/tcc/rocketmq/RocketMQUtils.java

@@ -0,0 +1,84 @@
+/*
+ *  Copyright 1999-2019 Seata.io Group.
+ *
+ *  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.
+ */
+package io.seata.rm.tcc.rocketmq;
+
+import java.net.UnknownHostException;
+import org.apache.rocketmq.client.Validators;
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.LocalTransactionState;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageAccessor;
+import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.remoting.exception.RemotingException;
+
+public class RocketMQUtils {
+
+    public static SendResult halfSend(DefaultMQProducer defaultMQProducer,
+        Message msg) throws MQClientException {
+        // ignore DelayTimeLevel parameter
+        if (msg.getDelayTimeLevel() != 0) {
+            MessageAccessor.clearProperty(msg, MessageConst.PROPERTY_DELAY_TIME_LEVEL);
+        }
+
+        Validators.checkMessage(msg, defaultMQProducer);
+
+        MessageAccessor.putProperty(msg, MessageConst.PROPERTY_TRANSACTION_PREPARED, "true");
+        MessageAccessor.putProperty(msg, MessageConst.PROPERTY_PRODUCER_GROUP, defaultMQProducer.getProducerGroup());
+        DefaultMQProducerImpl defaultMQProducerImpl = defaultMQProducer.getDefaultMQProducerImpl();
+        SendResult sendResult = null;
+        try {
+            sendResult = defaultMQProducerImpl.send(msg);
+        } catch (Exception e) {
+            throw new MQClientException("send message Exception", e);
+        }
+
+        LocalTransactionState localTransactionState = LocalTransactionState.UNKNOW;
+        switch (sendResult.getSendStatus()) {
+            case SEND_OK: {
+                if (sendResult.getTransactionId() != null) {
+                    msg.putUserProperty("__transactionId__", sendResult.getTransactionId());
+                }
+                String transactionId = msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
+                if (null != transactionId && !"".equals(transactionId)) {
+                    msg.setTransactionId(transactionId);
+                }
+            }
+            break;
+            case FLUSH_DISK_TIMEOUT:
+            case FLUSH_SLAVE_TIMEOUT:
+            case SLAVE_NOT_AVAILABLE:
+            default:
+                throw new RuntimeException("Message send fail.");
+        }
+        return sendResult;
+    }
+
+    public static void confirm(DefaultMQProducer defaultMQProducer, Message msg,
+        SendResult sendResult) throws UnknownHostException, MQBrokerException, RemotingException, InterruptedException {
+        DefaultMQProducerImpl defaultMQProducerImpl = defaultMQProducer.getDefaultMQProducerImpl();
+        defaultMQProducerImpl.endTransaction(msg, sendResult, LocalTransactionState.COMMIT_MESSAGE, null);
+    }
+
+    public static void cancel(DefaultMQProducer defaultMQProducer, Message msg,
+        SendResult sendResult) throws UnknownHostException, MQBrokerException, RemotingException, InterruptedException {
+        DefaultMQProducerImpl defaultMQProducerImpl = defaultMQProducer.getDefaultMQProducerImpl();
+        defaultMQProducerImpl.endTransaction(msg, sendResult, LocalTransactionState.ROLLBACK_MESSAGE, null);
+    }
+}