package com.bonus.common.houqin.mq.rabbit;

import com.bonus.common.houqin.mq.MQProperties;
import com.bonus.common.houqin.mq.MQTemplate;
import com.bonus.common.houqin.mq.rabbit.tx.PreparedCorrelationData;
import com.bonus.common.houqin.mq.rabbit.tx.RabbitTxHolder;
import com.bonus.common.houqin.mq.tx.TxHolder;
import com.bonus.common.houqin.mq.tx.TxMessageManager;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.converter.MessageConverter;

import java.nio.charset.StandardCharsets;
import java.sql.SQLException;
import java.util.Map;
import java.util.UUID;

public class RabbitTemplate extends MQTemplate {
   static final String NOT_MANDATORY_HEADER = "__not_mandatory__";
   private final org.springframework.amqp.rabbit.core.RabbitTemplate rabbitmqTemplate;
   private final RabbitPropertiesExt rabbitPropertiesExt;
   private final TxMessageManager txMessageManager;
   private final MessageConverter messageConverter;

   public RabbitTemplate(org.springframework.amqp.rabbit.core.RabbitTemplate rabbitmqTemplate, RabbitPropertiesExt rabbitPropertiesExt, MQProperties mqProperties, TxMessageManager txMessageManager, MessageConverter messageConverter) {
      super(mqProperties);
      this.rabbitmqTemplate = rabbitmqTemplate;
      this.rabbitPropertiesExt = rabbitPropertiesExt;
      this.txMessageManager = txMessageManager;
      this.messageConverter = messageConverter;
   }

   public void sendInternal(boolean mandatory, String routingKey, Object payload, Map<String, Object> header, String messageId, Integer delay, boolean txMessage) {
      routingKey = routingKey.replace(":", ".");
      String msgId = messageId == null ? UUID.randomUUID().toString() : messageId;
      this.rabbitmqTemplate.convertAndSend(delay != null ? this.rabbitPropertiesExt.getDelayedExchange() : this.rabbitPropertiesExt.getNormalExchange(), routingKey, payload, (msg) -> {
         msg.getMessageProperties().setMessageId(msgId);
         msg.getMessageProperties().setCorrelationId(msg.getMessageProperties().getMessageId());
         if (!mandatory) {
            msg.getMessageProperties().setHeader("__not_mandatory__", true);
         }

         if (delay != null) {
            msg.getMessageProperties().setDelay(delay);
         }

         if (header != null) {
            header.forEach((k, v) -> {
               msg.getMessageProperties().setHeader(k, v);
            });
         }

         if (msg.getMessageProperties().getHeader("X-Trace-Id") == null) {
            msg.getMessageProperties().setHeader("X-Trace-Id", this.getTraceId());
         }

         return msg;
      }, new PreparedCorrelationData(msgId, txMessage));
   }

   public void sendDelayImpl(String destination, Object payload, int delayTime) {
      this.sendInternal(false, destination, payload, (Map)null, (String)null, delayTime, false);
   }

   public void sendImpl(String destination, Object payload) {
      this.sendInternal(true, destination, payload, (Map)null, (String)null, (Integer) null, false);
   }

   public TxHolder txBeginImpl(String destination, Object payload) throws SQLException {
      return this.txBeginInternal(destination, payload);
   }

   public TxHolder mqttTxBegin(String destination, Object payload) throws SQLException {
      return this.txBeginInternal(destination.replace("/", "."), payload);
   }

   private TxHolder txBeginInternal(String destination, Object payload) throws SQLException {
      try {
         Message message = this.messageConverter.toMessage(payload, new MessageProperties());
         RabbitPropertiesExt.RabbitMessage rabbitMessage = new RabbitPropertiesExt.RabbitMessage(message.getMessageProperties().getMessageId(), new String(message.getBody(), StandardCharsets.UTF_8), this.rabbitPropertiesExt.getNormalExchange(), destination);
         this.txMessageManager.saveTxMessage(rabbitMessage);
         return new RabbitTxHolder(rabbitMessage, this);
      } catch (Throwable var5) {
         throw var5;
      }
   }

   public void txCommit(TxHolder txHolder) {
      RabbitPropertiesExt.RabbitMessage message = ((RabbitTxHolder)txHolder).getMessage();
      this.sendInternal(true, message.getRoutingKey(), message.getPayload(), (Map)null, message.getId(), (Integer) null, true);
   }

   public void txRollback(TxHolder txHolder, Throwable exception) throws SQLException {
      try {
         RabbitPropertiesExt.RabbitMessage message = ((RabbitTxHolder)txHolder).getMessage();
         this.txMessageManager.afterRollback(message.getId());
      } catch (Throwable var4) {
         throw var4;
      }
   }

   public void txUnknown(TxHolder txHolder, Throwable exception) {
   }

   public void sendMqtt(String destination, Object payload) {
      this.sendInternal(false, destination.replace("/", "."), payload, (Map)null, (String)null, (Integer) null, false);
   }
}
