package com.sonic.shark.config;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.ImmutableMap;
import com.rabbitmq.client.Channel;
import com.sonic.common.AppRuntime;
import com.sonic.common.event.*;
import com.sonic.common.utils.LogUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.annotation.RabbitListeners;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;
import org.springframework.core.task.TaskExecutor;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;

import java.io.IOException;
import java.nio.charset.Charset;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

/**
 * @author code
 */
@Slf4j
public class EventConfig {

    /** TODO: 定义 Event.BuildInScene */
    public final static String DEFAULT_SCENE = Event.BuildInScene.SONIC.getCode();
    /** TODO: DEFAULT_SCENE + "_" + appName */
    public final static String DEFAULT_MODULE = DEFAULT_SCENE + "_" + "sonic";

    @Value("${mq.exchange}")
    private String mqExchange;
    @Value("${mq.default.queue}")
    private String defaultQueue;
    @Value("${mq.default.routing-key}")
    private String defaultRoutingKey;

    @Configuration
    @Profile("!unittest")
    static class Listener {
        @Autowired
        EventConsumer eventConsumer;

        @RabbitListeners({
                @RabbitListener(queues = {"${mq.default.queue}"}, concurrency = "2"),
        })
        public void process(@Payload Message message, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) {
            try {
                LogUtils.setTraceId();
                byte[] msgBody = message.getBody();
                String contentEncoding = message.getMessageProperties().getContentEncoding();
                String bodyStr = new String(msgBody, Charset.forName(ObjectUtils.defaultIfNull(contentEncoding, "UTF-8")));
                MessageProperties pro = message.getMessageProperties();
                if (log.isDebugEnabled()) {
                    log.info("receive msg routingKey:{}->consumerQueue:{}->redelivered:{}->body:{}",
                            pro.getReceivedRoutingKey(), pro.getConsumerQueue(), pro.getRedelivered(), bodyStr);
                }
                eventConsumer.onEvent(bodyStr, ImmutableMap.of("record", JSON.toJSONString(message)));

                //消息确认，multiple：是否批量.true:将一次性ack所有小于deliveryTag的消息
                channel.basicAck(deliveryTag, false);
            } catch (Exception e) {
                log.error("===> mq handler error, message: {}", JSON.toJSONString(message), e);
                //消息拒绝，//requeue=true,表示将消息重新放入到队列中，false：表示直接从队列中删除，此时和basicAck(long deliveryTag, false)的效果一样
                try {
                    channel.basicReject(deliveryTag,true);
                } catch (IOException ioException) {
                    log.error("channel.basicReject error. message: {}, deliveryTag: {}", JSON.toJSONString(message), deliveryTag, ioException);
                }
            } finally {
                LogUtils.removeTraceId();
            }
        }

    }

    @Bean
    EventProducer eventProducer(AppRuntime appRuntime, RabbitTemplate rabbitTemplate, TaskExecutor taskExecutor) {
        BiConsumer<Event, RabbitmqEventProducer.RabbitmqMessageMeta> callback = (event, meta) -> {};
        return new RabbitmqEventProducer(rabbitTemplate, DEFAULT_MODULE, DEFAULT_SCENE, appRuntime.getAppId(),
                RabbitmqEventProducer.RabbitmqMessageMeta.build(mqExchange, defaultRoutingKey), taskExecutor, callback);
    }

    @Bean
    EventConsumer eventConsumer(AppRuntime appRuntime, EventHandlerRepository eventHandlerRepository) {
        Consumer<EventConsumer.EventWrapper> callback = (eventWrapper) -> {};
        return new DefaultEventConsumer(appRuntime.getAppName(), eventHandlerRepository, callback);
    }

    @Bean
    EventHandlerRepository eventHandlerRepository() {
        return new EventHandlerRepository((ex, logText) -> log.error("MQ,handle error {}", logText, ex));
    }

    @Bean
    public DirectExchange messageServerExchange(){
        return new DirectExchange(mqExchange);
    }

    @Bean
    public Binding bindingDefaultQueueExchange(DirectExchange exchange, Queue defaultQueue) {
        return bindingExchange(exchange, defaultQueue, defaultRoutingKey);
    }

    @Bean
    public Queue defaultQueue() {
        return new Queue(defaultQueue,true);
    }

    public Binding bindingExchange(DirectExchange exchange, Queue queueMessage, String routingKey) {
        return BindingBuilder.bind(queueMessage).to(exchange).with(routingKey);
    }

}

