package com.alibaba.otter.canal.connector.rocketmq.consumer;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang.StringUtils;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson2.JSON;
import com.alibaba.otter.canal.connector.core.config.CanalConstants;
import com.alibaba.otter.canal.connector.core.consumer.CommonMessage;
import com.alibaba.otter.canal.connector.core.spi.CanalMsgConsumer;
import com.alibaba.otter.canal.connector.core.spi.SPI;
import com.alibaba.otter.canal.connector.core.util.CanalMessageSerializerUtil;
import com.alibaba.otter.canal.connector.core.util.MessageUtil;
import com.alibaba.otter.canal.connector.rocketmq.config.RocketMQConstants;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;

/**
 * RocketMQ consumer SPI 实现
 *
 * @author rewerma 2020-02-01
 * @version 1.0.0
 */
@SPI("rocketmq")
public class CanalRocketMQConsumer implements CanalMsgConsumer {

    private static final Logger                                logger               = LoggerFactory.getLogger(CanalRocketMQConsumer.class);
    private static final String                                CLOUD_ACCESS_CHANNEL = "cloud";

    private String                                             nameServer;
    private String                                             topic;
    private String                                             groupName;
    private DefaultMQPushConsumer                              rocketMQConsumer;
    private BlockingQueue<ConsumerBatchMessage<CommonMessage>> messageBlockingQueue;
    private int                                                batchSize            = -1;
    private long                                               batchProcessTimeout  = 60 * 1000;
    private boolean                                            flatMessage;
    private volatile ConsumerBatchMessage<CommonMessage>       lastGetBatchMessage  = null;
    private String                                             accessKey;
    private String                                             secretKey;
    private String                                             customizedTraceTopic;
    private boolean                                            enableMessageTrace   = false;
    private String                                             accessChannel;
    private String                                             namespace;
    private String                                             filter               = "*";

    @Override
    public void init(Properties properties, String topic, String groupName) {
        this.topic = topic;
        this.groupName = groupName;
        this.flatMessage = (Boolean) properties.get(CanalConstants.CANAL_MQ_FLAT_MESSAGE);
        this.messageBlockingQueue = new LinkedBlockingQueue<>(1024);
        this.accessKey = properties.getProperty(CanalConstants.CANAL_ALIYUN_ACCESS_KEY);
        this.secretKey = properties.getProperty(CanalConstants.CANAL_ALIYUN_SECRET_KEY);
        String enableMessageTrace = properties.getProperty(RocketMQConstants.ROCKETMQ_ENABLE_MESSAGE_TRACE);
        if (StringUtils.isNotEmpty(enableMessageTrace)) {
            this.enableMessageTrace = Boolean.parseBoolean(enableMessageTrace);
        }
        this.customizedTraceTopic = properties.getProperty(RocketMQConstants.ROCKETMQ_CUSTOMIZED_TRACE_TOPIC);
        this.accessChannel = properties.getProperty(RocketMQConstants.ROCKETMQ_ACCESS_CHANNEL);
        this.namespace = properties.getProperty(RocketMQConstants.ROCKETMQ_NAMESPACE);
        this.nameServer = properties.getProperty(RocketMQConstants.ROCKETMQ_NAMESRV_ADDR);
        String batchSize = properties.getProperty(RocketMQConstants.ROCKETMQ_BATCH_SIZE);
        if (StringUtils.isNotEmpty(batchSize)) {
            this.batchSize = Integer.parseInt(batchSize);
        }
        String subscribeFilter = properties.getProperty(RocketMQConstants.ROCKETMQ_SUBSCRIBE_FILTER);
        if (StringUtils.isNotEmpty(subscribeFilter)) {
            this.filter = subscribeFilter;
        }
    }

    @Override
    public void connect() {
        RPCHook rpcHook = null;
        if (null != accessKey && accessKey.length() > 0 && null != secretKey && secretKey.length() > 0) {
            SessionCredentials sessionCredentials = new SessionCredentials();
            sessionCredentials.setAccessKey(accessKey);
            sessionCredentials.setSecretKey(secretKey);
            rpcHook = new AclClientRPCHook(sessionCredentials);
        }

        rocketMQConsumer = new DefaultMQPushConsumer(groupName,
            rpcHook,
            new AllocateMessageQueueAveragely(),
            enableMessageTrace,
            customizedTraceTopic);
        rocketMQConsumer.setVipChannelEnabled(false);
        if (CLOUD_ACCESS_CHANNEL.equals(this.accessChannel)) {
            rocketMQConsumer.setAccessChannel(AccessChannel.CLOUD);
        }

        if (!StringUtils.isEmpty(this.namespace)) {
            rocketMQConsumer.setNamespace(this.namespace);
        }

        if (!StringUtils.isBlank(nameServer)) {
            rocketMQConsumer.setNamesrvAddr(nameServer);
        }
        if (batchSize != -1) {
            rocketMQConsumer.setConsumeMessageBatchMaxSize(batchSize);
        }

        try {
            if (rocketMQConsumer == null) {
                this.connect();
            }
            rocketMQConsumer.subscribe(this.topic, this.filter);
            rocketMQConsumer.registerMessageListener((MessageListenerOrderly) (messageExts, context) -> {
                context.setAutoCommit(true);
                boolean isSuccess = process(messageExts);
                if (isSuccess) {
                    return ConsumeOrderlyStatus.SUCCESS;
                } else {
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
            });
            rocketMQConsumer.start();
        } catch (MQClientException ex) {
            logger.error("Start RocketMQ consumer error", ex);
        }
    }

    private boolean process(List<MessageExt> messageExts) {
        if (logger.isDebugEnabled()) {
            logger.debug("Get Message: {}", messageExts);
        }
        List<CommonMessage> messageList = new ArrayList<>();
        for (MessageExt messageExt : messageExts) {
            byte[] data = messageExt.getBody();
            if (data != null) {
                try {
                    if (!flatMessage) {
                        Message message = CanalMessageSerializerUtil.deserializer(data);
                        messageList.addAll(MessageUtil.convert(message));
                    } else {
                        CommonMessage commonMessage = JSON.parseObject(data, CommonMessage.class);
                        messageList.add(commonMessage);
                    }
                } catch (Exception ex) {
                    logger.error("Add message error", ex);
                    throw new CanalClientException(ex);
                }
            } else {
                logger.warn("Received message data is null");
            }
        }
        ConsumerBatchMessage<CommonMessage> batchMessage = new ConsumerBatchMessage<>(messageList);
        try {
            messageBlockingQueue.put(batchMessage);
        } catch (InterruptedException e) {
            logger.error("Put message to queue error", e);
            throw new RuntimeException(e);
        }
        boolean isCompleted;
        try {
            isCompleted = batchMessage.waitFinish(batchProcessTimeout);
        } catch (InterruptedException e) {
            logger.error("Interrupted when waiting messages to be finished.", e);
            throw new RuntimeException(e);
        }
        boolean isSuccess = batchMessage.isSuccess();
        return isCompleted && isSuccess;
    }

    @Override
    public List<CommonMessage> getMessage(Long timeout, TimeUnit unit) {
        try {
            if (this.lastGetBatchMessage != null) {
                throw new CanalClientException("mq get/ack not support concurrent & async ack");
            }

            ConsumerBatchMessage<CommonMessage> batchMessage = messageBlockingQueue.poll(timeout, unit);
            if (batchMessage != null) {
                this.lastGetBatchMessage = batchMessage;
                return batchMessage.getData();
            }
        } catch (InterruptedException ex) {
            logger.warn("Get message timeout", ex);
            throw new CanalClientException("Failed to fetch the data after: " + timeout);
        }
        return null;
    }

    @Override
    public void rollback() {
        try {
            if (this.lastGetBatchMessage != null) {
                this.lastGetBatchMessage.fail();
            }
        } finally {
            this.lastGetBatchMessage = null;
        }
    }

    @Override
    public void ack() {
        try {
            if (this.lastGetBatchMessage != null) {
                this.lastGetBatchMessage.ack();
            }
        } catch (Throwable e) {
            if (this.lastGetBatchMessage != null) {
                this.lastGetBatchMessage.fail();
            }
        } finally {
            this.lastGetBatchMessage = null;
        }
    }

    @Override
    public void disconnect() {
        rocketMQConsumer.unsubscribe(topic);
        rocketMQConsumer.shutdown();
    }
}
