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

import com.bonus.common.houqin.mq.MQMessageConverter;
import com.bonus.common.houqin.mq.MQProperties;
import com.bonus.common.houqin.mq.MQTemplate;
import com.bonus.common.houqin.mq.mqtt.MqttProperties;
import com.bonus.common.houqin.mq.rocket.mqtt.MqttConfiguration;
import com.bonus.common.houqin.mq.rocket.tx.RocketTxConfiguration;
import com.bonus.common.houqin.mq.tx.MQTxCheckComposite;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

@AutoConfiguration(
   before = {MqttConfiguration.class}
)
@ConditionalOnProperty(
   prefix = "spring.mq",
   value = {"type"},
   havingValue = "rocketmq"
)
@Import({RocketTxConfiguration.class, MqttConfiguration.class, RocketListenerConfiguration.class})
public class RocketConfiguration {
   private static final Logger log = LoggerFactory.getLogger(RocketConfiguration.class);
   private final ApplicationContext applicationContext;
   private final MqttProperties mqttProperties;

   public RocketConfiguration(ApplicationContext applicationContext, MqttProperties mqttProperties) {
      this.applicationContext = applicationContext;
      this.mqttProperties = mqttProperties;
   }

   @Bean(
      destroyMethod = "destroy"
   )
   @ConditionalOnMissingBean(
      name = {"rocketMQTemplate"}
   )
   public RocketMQTemplate rocketMQTemplate(MQMessageConverter MQMessageConverter) {
      RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
      if (this.applicationContext.containsBean("defaultMQProducer")) {
         TransactionMQProducer producer = (TransactionMQProducer)this.applicationContext.getBean("defaultMQProducer");
         producer.setTransactionCheckListener(new MQTxCheckComposite(MQMessageConverter.getMessageConverter()));
         producer.setCheckThreadPoolMaxSize(10);
         rocketMQTemplate.setProducer(producer);
      }

      if (this.applicationContext.containsBean("defaultLitePullConsumer")) {
         rocketMQTemplate.setConsumer((DefaultLitePullConsumer)this.applicationContext.getBean("defaultLitePullConsumer"));
      }

      rocketMQTemplate.setMessageConverter(MQMessageConverter.getMessageConverter());
      return rocketMQTemplate;
   }

   @Bean({"defaultMQProducer"})
   @ConditionalOnMissingBean({DefaultMQProducer.class})
   public DefaultMQProducer defaultMQProducer(RocketMQProperties rocketMQProperties, MQProperties mqProperties) {
      RocketMQProperties.Producer producerConfig = rocketMQProperties.getProducer();
      String nameServer = rocketMQProperties.getNameServer();
      String groupName = producerConfig.getGroup();
      Assert.hasText(nameServer, "Rocketmq nameserver must not be null");
      Assert.hasText(groupName, "Rocketmq producer's group must not be null");
      String accessChannel = rocketMQProperties.getAccessChannel();
      String ak = mqProperties.getRocketmq().getAccessKey();
      String sk = mqProperties.getRocketmq().getSecretKey();
      boolean isEnableMsgTrace = rocketMQProperties.getProducer().isEnableMsgTrace();
      String customizedTraceTopic = rocketMQProperties.getProducer().getCustomizedTraceTopic();
      DefaultMQProducer producer = RocketMQUtil.createDefaultMQProducer(groupName, ak, sk, isEnableMsgTrace, customizedTraceTopic);
      producer.setNamesrvAddr(nameServer);
      if (StringUtils.hasLength(accessChannel)) {
         producer.setAccessChannel(AccessChannel.valueOf(accessChannel));
      }

      producer.setSendMsgTimeout(producerConfig.getSendMessageTimeout());
      producer.setRetryTimesWhenSendFailed(producerConfig.getRetryTimesWhenSendFailed());
      producer.setRetryTimesWhenSendAsyncFailed(producerConfig.getRetryTimesWhenSendAsyncFailed());
      producer.setMaxMessageSize(producerConfig.getMaxMessageSize());
      producer.setCompressMsgBodyOverHowmuch(producerConfig.getCompressMessageBodyThreshold());
      producer.setRetryAnotherBrokerWhenNotStoreOK(producerConfig.isRetryNextServer());
      producer.setUseTLS(producerConfig.isTlsEnable());
      if (StringUtils.hasText(producerConfig.getNamespace())) {
         producer.setNamespace(producerConfig.getNamespace());
      }

      if (StringUtils.hasText(producerConfig.getNamespaceV2())) {
         producer.setNamespaceV2(producerConfig.getNamespaceV2());
      }

      producer.setInstanceName(producerConfig.getInstanceName());
      log.info("a producer ({}) init on namesrv {}", groupName, nameServer);
      return producer;
   }

   @Bean({"defaultLitePullConsumer"})
   @ConditionalOnMissingBean({DefaultLitePullConsumer.class})
   @ConditionalOnProperty(
      prefix = "spring.mq.rocketmq",
      value = {"name-server", "pull-consumer.group", "pull-consumer.topic"}
   )
   public DefaultLitePullConsumer defaultLitePullConsumer(RocketMQProperties rocketMQProperties) throws MQClientException {
      RocketMQProperties.PullConsumer consumerConfig = rocketMQProperties.getPullConsumer();
      String nameServer = rocketMQProperties.getNameServer();
      String groupName = consumerConfig.getGroup();
      String topicName = consumerConfig.getTopic();
      Assert.hasText(nameServer, "Rocketmq nameserver must not be null");
      Assert.hasText(groupName, "Rocketmq producer's group must not be null");
      Assert.hasText(topicName, "Rocketmq pullConsumer's topic must not be null");
      String accessChannel = rocketMQProperties.getAccessChannel();
      MessageModel messageModel = MessageModel.valueOf(consumerConfig.getMessageModel());
      SelectorType selectorType = SelectorType.valueOf(consumerConfig.getSelectorType());
      String selectorExpression = consumerConfig.getSelectorExpression();
      String ak = consumerConfig.getAccessKey();
      String sk = consumerConfig.getSecretKey();
      int pullBatchSize = consumerConfig.getPullBatchSize();
      boolean useTLS = consumerConfig.isTlsEnable();
      DefaultLitePullConsumer litePullConsumer = RocketMQUtil.createDefaultLitePullConsumer(nameServer, accessChannel, groupName, topicName, messageModel, selectorType, selectorExpression, ak, sk, pullBatchSize, useTLS);
      litePullConsumer.setEnableMsgTrace(consumerConfig.isEnableMsgTrace());
      litePullConsumer.setCustomizedTraceTopic(consumerConfig.getCustomizedTraceTopic());
      if (StringUtils.hasText(consumerConfig.getNamespace())) {
         litePullConsumer.setNamespace(consumerConfig.getNamespace());
      }

      if (StringUtils.hasText(consumerConfig.getNamespaceV2())) {
         litePullConsumer.setNamespaceV2(consumerConfig.getNamespaceV2());
      }

      litePullConsumer.setInstanceName(consumerConfig.getInstanceName());
      log.info("a pull consumer({} sub {}) init on namesrv {}", new Object[]{groupName, topicName, nameServer});
      return litePullConsumer;
   }

   @Bean
   public MQTemplate mqTemplate(RocketMQTemplate rocketMQTemplate, MQMessageConverter messageConverter, MQProperties mqProperties) {
      return new RocketTemplate(rocketMQTemplate, this.mqttProperties.getEmqx().getParentTopic(), messageConverter.getMessageConverter(), mqProperties);
   }
}
