package com.sonic.common.event;

import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;

import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.core.task.TaskExecutor;
import org.springframework.util.CollectionUtils;

import com.google.common.base.Strings;

import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.ToString;

/**
 * 发送消息
 */
public class RabbitmqEventProducer extends AbstractEventProducer<RabbitmqEventProducer.RabbitmqMessageMeta> {

    private RabbitTemplate rabbitTemplate;

    private String defaultModule;
    private String defaultScene;
    private String appId;
    /**
     * 一个扩展点.让外部可以在发送消息后做特殊逻辑处理. 比如发送消息统计, 消息元数据监控
     */
    private BiConsumer<Event, RabbitmqMessageMeta> sendCallback;

    /**
     * 推荐使用 带callback的构造方法 可以通过callback集成apiStatsClient对消息进行治理<br>
     * 见：{@link RabbitmqEventProducer#RabbitmqEventProducer(org.springframework.amqp.rabbit.core.RabbitTemplate, String, String, String, RabbitmqMessageMeta, TaskExecutor, BiConsumer)}
     */
    @Deprecated
    public RabbitmqEventProducer(RabbitTemplate rabbitTemplate, String defaultModule, String defaultScene,
                                 String appId, RabbitmqMessageMeta defaultMeta, TaskExecutor taskExecutor) {
        this(rabbitTemplate, defaultModule, defaultScene, appId, defaultMeta, taskExecutor, null);
    }

    public RabbitmqEventProducer(RabbitTemplate rabbitTemplate, String defaultModule, String defaultScene,
                                 String appId, RabbitmqMessageMeta defaultMeta, TaskExecutor taskExecutor,
                                 BiConsumer<Event, RabbitmqMessageMeta> sendCallback) {
        super(defaultMeta, taskExecutor);
        this.rabbitTemplate = rabbitTemplate;
        this.defaultModule = defaultModule;
        this.defaultScene = defaultScene;
        this.appId = appId;
        this.sendCallback = sendCallback;
    }

    @Override
    public BiConsumer<Event, RabbitmqMessageMeta> getSender() {
        return (event, meta) -> {
            // correct scene and module if it's empty
            if (Strings.isNullOrEmpty(event.getEventScene())) {
                event.setEventScene(defaultScene);
            }
            if (Strings.isNullOrEmpty(event.getEventModule())) {
                event.setEventModule(defaultModule);
            }
            if (Strings.isNullOrEmpty(event.getEventId())) {
                // init eventId -> appId_uuid
                event.setEventId(String.format("%s_%s", appId, UUID.randomUUID().toString().replace("-", "")));
            }

            event.setSchemaHash(event.buildSchemaHash(meta.getRoutingKey()));
            event.check();

            final MessageProperties properties = new MessageProperties();
            properties.setContentType(MessageProperties.CONTENT_TYPE_JSON);
            // 设置5天有效期.避免废弃的队列一直接受数据.
            properties.setExpiration("" + TimeUnit.DAYS.toMicros(5));
            if (!CollectionUtils.isEmpty(meta.getHeaders())) {
                meta.getHeaders().entrySet()
                        .forEach(entry -> properties.getHeaders().put(entry.getKey(), entry.getValue()));
            }
            // XXX: must make sure the transaction is commit before send user message!!!

            if(event.getMessagePostProcessor() != null) {
                rabbitTemplate.convertAndSend(meta.getExchange(), meta.getRoutingKey(), event.toJsonString(), event.getMessagePostProcessor());
            } else {
                rabbitTemplate.convertAndSend(meta.getExchange(), meta.getRoutingKey(), event.toJsonString());
            }
            if (sendCallback != null) {
                sendCallback.accept(event, meta);
            }
        };
    }

    @NoArgsConstructor
    @Getter
    @ToString
    public static class RabbitmqMessageMeta {
        private String exchange;
        private String routingKey;
        private Map<String, byte[]> headers;

        public static RabbitmqMessageMeta build(String exchange, String routingKey, Map<String, byte[]> headers) {
            RabbitmqMessageMeta meta = new RabbitmqMessageMeta();
            meta.exchange = exchange;
            meta.routingKey = routingKey;
            meta.headers = Optional.ofNullable(headers).orElseGet(HashMap::new);
            return meta;
        }

        public static RabbitmqMessageMeta build(String exchange, String routingKey) {
            return build(exchange, routingKey, null);
        }

        public static RabbitmqMessageMeta build(String topic) {
            return build(topic, null, null);
        }

        public RabbitmqMessageMeta header(String key, byte[] value) {
            headers.put(key, value);
            return this;
        }
    }

}
