package com.lusifer.sca.message.producer.service;

import com.lusifer.sca.message.producer.domain.MyObject;
import java.util.UUID;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

@Slf4j
@Service
@RequiredArgsConstructor
public class TransactionProducerService {

  private final RocketMQTemplate rocketMQTemplate;

  /**
   * 发送事务消息
   *
   * @param topic 消息主题
   * @param msgBody 消息内容
   */
  public void sendMessageInTransaction(String topic, String msgBody) {
    // 构建事务消息，绑定唯一事务ID
    Message<String> message =
        MessageBuilder.withPayload(msgBody)
            .setHeader(RocketMQHeaders.TRANSACTION_ID, UUID.randomUUID().toString())
            .build();

    // 发送事务消息（主题:标签，消息，业务参数）
    SendResult sendResult =
        rocketMQTemplate.sendMessageInTransaction(topic + ":tagA", message, new MyObject("张三", 18));

    log.info("事务半消息发送成功，发送结果：{}", sendResult);
  }
}

package com.lusifer.sca.message.producer.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;

/**
 * 事务消息监听器
 * 核心：执行本地事务 + 事务状态回查
 */
@Slf4j
@RocketMQTransactionListener(rocketMQTemplateBeanName = "rocketMQTemplate")
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {

    /**
     * 执行本地事务（半消息发送成功后自动调用）
     * @param message 事务消息
     * @param arg 业务参数
     * @return 事务状态
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object arg) {
        String payload = message.getPayload().toString();
        log.info("开始执行本地事务，消息内容：{}，业务参数：{}", payload, arg);

        try {
            // 模拟本地事务：数据库增删改查
            boolean transactionSuccess = true;
            if (transactionSuccess) {
                log.info("本地事务执行成功，提交事务消息");
                return RocketMQLocalTransactionState.COMMIT;
            } else {
                log.info("本地事务执行失败，回滚事务消息");
                return RocketMQLocalTransactionState.ROLLBACK;
            }
        } catch (Exception e) {
            log.error("本地事务执行异常", e);
            // 异常返回未知状态，等待Broker回查
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }

    /**
     * 事务状态回查（Broker 主动回调）
     * @param message 事务消息
     * @return 事务状态
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        String payload = message.getPayload().toString();
        log.info("Broker 触发事务回查，消息内容：{}", payload);

        // 模拟查询本地事务最终状态
        boolean transactionFinalSuccess = true;
        if (transactionFinalSuccess) {
            return RocketMQLocalTransactionState.COMMIT;
        } else {
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }
}

package com.lusifer.sca.message.consumer.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;

@Slf4j
@Service
@RocketMQMessageListener(topic = "transaction-topic", consumerGroup = "transaction-consumer-group")
public class TransactionConsumer implements RocketMQListener<String> {

  /** 消费事务消息（仅本地事务提交后才会触发） */
  @Override
  public void onMessage(String message) {
    log.info("成功消费事务消息：{}", message);
  }
}

package com.lusifer.sca.message.producer.test;

import com.lusifer.sca.message.producer.service.TransactionProducerService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

@Slf4j
@SpringBootTest
public class TransactionMessageTest {

  @Resource private TransactionProducerService transactionProducerService;

  @Test
  public void testSendTransactionMessage() {
    transactionProducerService.sendMessageInTransaction(
        "transaction-topic", "Hello RocketMQ Transaction Message!");
  }
}