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

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.text.CharSequenceUtil;
import com.bonus.common.houqin.mq.MQProperties;
import com.bonus.common.houqin.mq.MQTemplate;
import com.bonus.common.houqin.mq.rabbit.mqtt.MqttConfiguration;
import com.bonus.common.houqin.mq.rabbit.tx.MessageManagerConfiguration;
import com.bonus.common.houqin.mq.rabbit.tx.PreparedCorrelationData;
import com.bonus.common.houqin.mq.rabbit.tx.RabbitTxConfiguration;
import com.bonus.common.houqin.mq.tx.TxMessageManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Declarable;
import org.springframework.amqp.core.Declarables;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.config.ContainerCustomizer;
import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.support.ValueExpression;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.amqp.DirectRabbitListenerContainerFactoryConfigurer;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.EvaluationException;
import org.springframework.expression.TypedValue;
import org.springframework.expression.common.ExpressionUtils;

import java.sql.SQLException;
import java.util.Collection;
import java.util.Objects;
import java.util.stream.Collectors;

//@AutoConfiguration(
//   after = {DataSourceAutoConfiguration.class}
//)
@ConditionalOnProperty(
   prefix = "spring.mq",
   value = {"type"},
   havingValue = "rabbitmq"
)
@Import({RabbitListenerConfiguration.class, RabbitTxConfiguration.class, MqttConfiguration.class, MessageManagerConfiguration.class})
public class RabbitConfiguration {
   private static final Logger log = LoggerFactory.getLogger(RabbitConfiguration.class);
   private final TxMessageManager txMessageManager;
   private final MQProperties mqProperties;

   public RabbitConfiguration(org.springframework.amqp.rabbit.core.RabbitTemplate rabbitTemplate, TxMessageManager txMessageManager, MQProperties mqProperties) {
      this.mqProperties = mqProperties;
      this.txMessageManager = txMessageManager;
      rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
         if (correlationData != null) {
            boolean prepared = false;
            if (correlationData instanceof PreparedCorrelationData) {
               prepared = ((PreparedCorrelationData)correlationData).isPrepared();
            }

            if (prepared) {
               if (ack) {
                  try {
                     txMessageManager.afterSendSuccess(correlationData.getId());
                  } catch (SQLException e) {
                     throw new RuntimeException(e);
                  }
               } else {
                  try {
                     txMessageManager.afterSendFailed(correlationData.getId(), cause);
                  } catch (SQLException e) {
                     throw new RuntimeException(e);
                  }
               }
            }
         }
      });
      rabbitTemplate.setMandatoryExpression(new ValueExpression<Boolean>(true) {
         private final TypedValue fv = new TypedValue(false);

         public <T> T getValue(EvaluationContext context, Object rootObject, Class<T> desiredResultType) throws EvaluationException {
            Message message = (Message)rootObject;
            return message.getMessageProperties().getHeader("__not_mandatory__") != null ? ExpressionUtils.convertTypedValue(context, this.fv, desiredResultType) : super.getValue(context, rootObject, desiredResultType);
         }
      });
      rabbitTemplate.setReturnsCallback((returned) -> {
         log.warn("Message returned: {}", returned);
      });
   }

   @Bean(
      name = {"prefetch1ListenerContainerFactory"}
   )
   @ConditionalOnMissingBean(
      name = {"prefetch1ListenerContainerFactory"}
   )
   @ConditionalOnProperty(
      prefix = "spring.mq.rabbitmq.listener",
      name = {"type"},
      havingValue = "simple",
      matchIfMissing = true
   )
   SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory, ObjectProvider<ContainerCustomizer<SimpleMessageListenerContainer>> simpleContainerCustomizer) {
      SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
      configurer.configure(factory, connectionFactory);
      factory.setPrefetchCount(1);
      Objects.requireNonNull(factory);
      simpleContainerCustomizer.ifUnique(factory::setContainerCustomizer);
      return factory;
   }

   @Bean(
      name = {"prefetch1ListenerContainerFactory"}
   )
   @ConditionalOnMissingBean(
      name = {"prefetch1ListenerContainerFactory"}
   )
   @ConditionalOnProperty(
      prefix = "spring.mqrabbitmq.listener",
      name = {"type"},
      havingValue = "direct"
   )
   DirectRabbitListenerContainerFactory directRabbitListenerContainerFactory(DirectRabbitListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory, ObjectProvider<ContainerCustomizer<DirectMessageListenerContainer>> directContainerCustomizer) {
      DirectRabbitListenerContainerFactory factory = new DirectRabbitListenerContainerFactory();
      configurer.configure(factory, connectionFactory);
      factory.setPrefetchCount(1);
      Objects.requireNonNull(factory);
      directContainerCustomizer.ifUnique(factory::setContainerCustomizer);
      return factory;
   }

   @Bean
   public Declarables declarables(RabbitPropertiesExt rabbitPropertiesExt) {
      rabbitPropertiesExt.getQueue().add(new RabbitPropertiesExt.QueueProperties(rabbitPropertiesExt.getRetryQueue(), this.mqProperties.getNamespace()));
      Collection<Declarable> queues = (Collection)rabbitPropertiesExt.getQueue().stream().map(RabbitPropertiesExt.QueueProperties::buildQueue).collect(Collectors.toSet());
      Collection<Declarable> queuesMore = (Collection)rabbitPropertiesExt.getQueueMore().stream().map(RabbitPropertiesExt.QueueProperties::buildQueue).collect(Collectors.toSet());
      if (CharSequenceUtil.isNotBlank(rabbitPropertiesExt.getDelayedExchange())) {
         rabbitPropertiesExt.getExchange().add(new RabbitPropertiesExt.ExchangeProperties(rabbitPropertiesExt.getDelayedExchange()));
      }

      Collection<Declarable> exchanges = (Collection)rabbitPropertiesExt.getExchange().stream().map(RabbitPropertiesExt.ExchangeProperties::buildExchange).collect(Collectors.toSet());
      Collection<Declarable> exchangesMore = (Collection)rabbitPropertiesExt.getExchangeMore().stream().map(RabbitPropertiesExt.ExchangeProperties::buildExchange).collect(Collectors.toSet());
      rabbitPropertiesExt.getBinding().add(new RabbitPropertiesExt.BindingProperties(rabbitPropertiesExt.getDelayedExchange(), rabbitPropertiesExt.getRetryQueue(), rabbitPropertiesExt.getRetryQueue()));
      Collection<Declarable> bindings = (Collection)rabbitPropertiesExt.getBinding().stream().map(RabbitPropertiesExt.BindingProperties::buildBinding).collect(Collectors.toSet());
      Collection<Declarable> bindingsMore = (Collection)rabbitPropertiesExt.getBindingMore().stream().map(RabbitPropertiesExt.BindingProperties::buildBinding).collect(Collectors.toSet());
      return new Declarables(CollUtil.union(queues, queuesMore, new Collection[]{exchanges, exchangesMore, bindings, bindingsMore}));
   }

   @Bean
   public MQTemplate rabbitMQProperties(org.springframework.amqp.rabbit.core.RabbitTemplate rabbitTemplate, MessageConverter messageConverter) {
      return new RabbitTemplate(rabbitTemplate, this.mqProperties.getRabbitmq(), this.mqProperties, this.txMessageManager, messageConverter);
   }
}
