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

import java.util.List;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang.StringUtils;
import org.apache.pulsar.client.api.*;

import com.alibaba.fastjson2.JSON;
import com.alibaba.otter.canal.common.utils.MQUtil;
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.pulsarmq.config.PulsarMQConstants;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import com.google.common.collect.Lists;

/**
 * Pulsar consumer SPI 实现
 *
 * @author rewerma @ 2020-02-01
 * @version 1.0.0
 */
@SPI("pulsarmq")
public class CanalPulsarMQConsumer implements CanalMsgConsumer {

    /**
     * 连接pulsar客户端
     */
    private PulsarClient              pulsarClient;
    private Consumer<byte[]>          pulsarMQConsumer;
    /**
     * 是否为扁平消息
     */
    private boolean                   flatMessage            = false;
    /**
     * 主题名称
     */
    private String                    topic;
    /**
     * 单线程控制
     */
    private volatile Messages<byte[]> lastGetBatchMessage;
    /**
     * 环境连接URL
     */
    private String                    serviceUrl;
    /**
     * 角色认证token
     */
    private String                    roleToken;

    /**
     * listener name
     */
    private String                    listenerName;

    /**
     * 订阅客户端名称
     */
    private String                    subscriptName;
    /**
     * 每次批量获取数据的最大条目数，默认30
     */
    private int                       batchSize              = 30;
    /**
     * 与{@code batchSize}一起决定批量获取的数据大小 当：
     * <p>
     * 1. {@code batchSize} 条消息未消费时<br/>
     * 2. 距上一次批量消费时间达到{@code batchTimeoutSeconds}秒时
     * </p>
     * 任一条件满足，即执行批量消费
     */
    private int                       getBatchTimeoutSeconds = 30;
    /**
     * 批量处理消息时，一次批量处理的超时时间
     * <p>
     * 该时间应该根据{@code batchSize}和{@code batchTimeoutSeconds}合理设置
     * </p>
     */
    private long                      batchProcessTimeout    = 60 * 1000;
    /**
     * 消费失败后的重试秒数，默认60秒
     */
    private int                       redeliveryDelaySeconds = 60;
    /**
     * 当客户端接收到消息，30秒还没有返回ack给服务端时，ack超时，会重新消费该消息
     */
    private int                       ackTimeoutSeconds      = 30;
    /**
     * 是否开启消息失败重试功能，默认开启
     */
    private boolean                   isRetry                = true;
    /**
     * <p>
     * true重试(-RETRY)和死信队列(-DLQ)后缀为大写，有些地方创建的为小写，需确保正确
     * </p>
     */
    private boolean                   isRetryDLQUpperCase    = false;
    /**
     * 最大重试次数
     */
    private int                       maxRedeliveryCount     = 128;

    @Override
    public void init(Properties properties, String topic, String groupId) {
        this.topic = topic;
        String flatMessageStr = properties.getProperty(CanalConstants.CANAL_MQ_FLAT_MESSAGE);
        if (StringUtils.isNotEmpty(flatMessageStr)) {
            this.flatMessage = Boolean.parseBoolean(flatMessageStr);
        }
        this.serviceUrl = properties.getProperty(PulsarMQConstants.PULSARMQ_SERVER_URL);
        this.roleToken = properties.getProperty(PulsarMQConstants.PULSARMQ_ROLE_TOKEN);
        this.listenerName = properties.getProperty(PulsarMQConstants.PULSARMQ_LISTENER_NAME);
        this.subscriptName = properties.getProperty(PulsarMQConstants.PULSARMQ_SUBSCRIPT_NAME);
        // 采用groupId作为subscriptName，避免所有的都是同一个订阅者名称
        if (StringUtils.isEmpty(this.subscriptName)) {
            this.subscriptName = groupId;
        }

        if (StringUtils.isEmpty(this.subscriptName)) {
            throw new RuntimeException("Pulsar Consumer subscriptName required");
        }
        String batchSizeStr = properties.getProperty(CanalConstants.CANAL_MQ_CANAL_BATCH_SIZE);
        if (StringUtils.isNotEmpty(batchSizeStr)) {
            this.batchSize = Integer.parseInt(batchSizeStr);
        }
        String getBatchTimeoutSecondsStr = properties.getProperty(PulsarMQConstants.PULSARMQ_GET_BATCH_TIMEOUT_SECONDS);
        if (StringUtils.isNotEmpty(getBatchTimeoutSecondsStr)) {
            this.getBatchTimeoutSeconds = Integer.parseInt(getBatchTimeoutSecondsStr);
        }
        String batchProcessTimeoutStr = properties.getProperty(PulsarMQConstants.PULSARMQ_BATCH_PROCESS_TIMEOUT);
        if (StringUtils.isNotEmpty(batchProcessTimeoutStr)) {
            this.batchProcessTimeout = Integer.parseInt(batchProcessTimeoutStr);
        }
        String redeliveryDelaySecondsStr = properties.getProperty(PulsarMQConstants.PULSARMQ_REDELIVERY_DELAY_SECONDS);
        if (StringUtils.isNotEmpty(redeliveryDelaySecondsStr)) {
            this.redeliveryDelaySeconds = Integer.parseInt(redeliveryDelaySecondsStr);
        }
        String ackTimeoutSecondsStr = properties.getProperty(PulsarMQConstants.PULSARMQ_ACK_TIMEOUT_SECONDS);
        if (StringUtils.isNotEmpty(ackTimeoutSecondsStr)) {
            this.ackTimeoutSeconds = Integer.parseInt(ackTimeoutSecondsStr);
        }
        String isRetryStr = properties.getProperty(PulsarMQConstants.PULSARMQ_IS_RETRY);
        if (StringUtils.isNotEmpty(isRetryStr)) {
            this.isRetry = Boolean.parseBoolean(isRetryStr);
        }
        String isRetryDLQUpperCaseStr = properties.getProperty(PulsarMQConstants.PULSARMQ_IS_RETRY_DLQ_UPPERCASE);
        if (StringUtils.isNotEmpty(isRetryDLQUpperCaseStr)) {
            this.isRetryDLQUpperCase = Boolean.parseBoolean(isRetryDLQUpperCaseStr);
        }
        String maxRedeliveryCountStr = properties.getProperty(PulsarMQConstants.PULSARMQ_MAX_REDELIVERY_COUNT);
        if (StringUtils.isNotEmpty(maxRedeliveryCountStr)) {
            this.maxRedeliveryCount = Integer.parseInt(maxRedeliveryCountStr);
        }
    }

    @Override
    public void connect() {
        if (isConsumerActive()) {
            return;
        }
        // 连接创建客户端
        try {
            // AuthenticationDataProvider
            ClientBuilder builder = PulsarClient.builder().serviceUrl(serviceUrl);
            if (StringUtils.isNotEmpty(roleToken)) {
                builder.authentication(AuthenticationFactory.token(roleToken));
            }
            if (StringUtils.isNotEmpty(listenerName)) {
                builder.authentication(AuthenticationFactory.token(listenerName));
            }
            pulsarClient = builder.build();
        } catch (PulsarClientException e) {
            throw new RuntimeException(e);
        }
        ConsumerBuilder<byte[]> builder = pulsarClient.newConsumer();
        if (MQUtil.isPatternTopic(this.topic)) {
            // 正则只支持一个
            builder.topicsPattern(this.topic);
        } else {// 多个topic
            builder.topic(this.topic);
        }
        // 为保证消息的有序性，仅支持单消费实例模式
        // 灾备模式，一个分区只能有一个消费者，如果当前消费者不可用，自动切换到其他消费者
        builder.subscriptionType(SubscriptionType.Failover);

        builder
            // 调用consumer.negativeAcknowledge(message) （即nack）来表示消费失败的消息
            // 在指定的时间进行重新消费，默认是1分钟。
            .negativeAckRedeliveryDelay(this.redeliveryDelaySeconds, TimeUnit.SECONDS)
            .subscriptionName(this.subscriptName);
        if (this.isRetry) {
            DeadLetterPolicy.DeadLetterPolicyBuilder dlqBuilder = DeadLetterPolicy.builder()
                // 最大重试次数
                .maxRedeliverCount(this.maxRedeliveryCount);
            // 指定重试队列，不是多个或通配符topic才能判断重试队列
            if (!MQUtil.isPatternTag(this.topic)) {
                String retryTopic = this.topic + (this.isRetryDLQUpperCase ? "-RETRY" : "-retry");
                dlqBuilder.retryLetterTopic(retryTopic);
                String dlqTopic = this.topic + (this.isRetryDLQUpperCase ? "-DLQ" : "-dlq");
                dlqBuilder.deadLetterTopic(dlqTopic);
            }

            // 默认关闭，如果需要重试则开启
            builder.enableRetry(true).deadLetterPolicy(dlqBuilder.build());
        }

        // ack超时
        builder.ackTimeout(this.ackTimeoutSeconds, TimeUnit.SECONDS);

        // pulsar批量获取消息设置
        builder.batchReceivePolicy(new BatchReceivePolicy.Builder().maxNumMessages(this.batchSize)
            .timeout(this.getBatchTimeoutSeconds, TimeUnit.SECONDS)
            .build());

        try {
            this.pulsarMQConsumer = builder.subscribe();
        } catch (PulsarClientException e) {
            throw new CanalClientException("Subscript pulsar consumer error", e);
        }
    }

    @SuppressWarnings("unchecked")
    @Override
    public List<CommonMessage> getMessage(Long timeout, TimeUnit unit) {
        List<CommonMessage> messageList = Lists.newArrayList();
        try {
            Messages<byte[]> messages = pulsarMQConsumer.batchReceive();
            if (null == messages || messages.size() == 0) {
                return messageList;
            }
            // 保存当前消费记录，用于ack和rollback
            this.lastGetBatchMessage = messages;
            for (org.apache.pulsar.client.api.Message<byte[]> msg : messages) {
                byte[] data = msg.getData();
                if (!this.flatMessage) {
                    Message message = CanalMessageSerializerUtil.deserializer(data);
                    List<CommonMessage> list = MessageUtil.convert(message);
                    messageList.addAll(list);
                } else {
                    CommonMessage commonMessage = JSON.parseObject(data, CommonMessage.class);
                    messageList.add(commonMessage);
                }
            }
        } catch (PulsarClientException e) {
            throw new CanalClientException("Receive pulsar batch message error", e);
        }

        return messageList;
    }

    @Override
    public void rollback() {
        try {
            if (isConsumerActive() && hasLastMessages()) {
                // 回滚所有消息
                this.pulsarMQConsumer.negativeAcknowledge(this.lastGetBatchMessage);
            }
        } finally {
            this.lastGetBatchMessage = null;
        }
    }

    @Override
    public void ack() {
        try {
            if (isConsumerActive() && hasLastMessages()) {
                // 确认所有消息
                this.pulsarMQConsumer.acknowledge(this.lastGetBatchMessage);
            }
        } catch (PulsarClientException e) {
            if (isConsumerActive() && hasLastMessages()) {
                this.pulsarMQConsumer.negativeAcknowledge(this.lastGetBatchMessage);
            }
        } finally {
            this.lastGetBatchMessage = null;
        }
    }

    @Override
    public void disconnect() {
        if (null == this.pulsarMQConsumer || !this.pulsarMQConsumer.isConnected()) {
            return;
        }
        try {
            // 会导致暂停期间数据丢失
            // this.pulsarMQConsumer.unsubscribe();
            this.pulsarClient.close();
        } catch (PulsarClientException e) {
            throw new CanalClientException("Disconnect pulsar consumer error", e);
        }
    }

    /**
     * 是否消费可用
     *
     * @return true消费者可用
     */
    private boolean isConsumerActive() {
        return null != this.pulsarMQConsumer && this.pulsarMQConsumer.isConnected();
    }

    /**
     * 是否有未确认消息
     *
     * @return true有正在消费的待确认消息
     */
    private boolean hasLastMessages() {
        return null != this.lastGetBatchMessage && this.lastGetBatchMessage.size() > 0;
    }
}
