package com.elab.mq.rocket.msg;

import org.apache.rocketmq.client.hook.ConsumeMessageContext;
import org.apache.rocketmq.client.hook.ConsumeMessageHook;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

/**
 * 默认消费者消费消息存储
 *
 * @author ： liukx
 * @time ： 2019/7/18 - 16:52
 */
@Component
public class DefaultConsumeMessageHook implements ConsumeMessageHook {
    private Logger logger = LoggerFactory.getLogger(DefaultSendMessageHook.class);

//    @Autowired
//    private ConsumerDataStore consumerDataStore;
//
//    @Value("${spring.application.name}")
//    private String group;
//    @Value("${spring.cloud.client.ipAddress}:${server.port}")
//    private String ip;

    @Override
    public String hookName() {
        return DefaultConsumeMessageHook.class.getName();
    }

    @Override
    public void consumeMessageBefore(ConsumeMessageContext context) {
        logger.info("-----------------------------------DefaultConsumeMessageHook---------sendMessageBefore----------------");
    }

    @Override
    public void consumeMessageAfter(ConsumeMessageContext context) {
        // 查看消费是否成功
        String status = context.getStatus();

        logger.info("-----------------------------------DefaultConsumeMessageHook---------sendMessageBefore----------------");
//        List<MessageExt> msgList = context.getMsgList();
//        for (MessageExt messageExt : msgList) {
//            RmqConsumerEntity entity = null;
//            String topic = messageExt.getTopic();
//            String keys = messageExt.getKeys();
//            entity = consumerDataStore.getMsgKey(group, keys);
//            boolean isInsert = false;
//            // 1. 构建数据
//            if (entity == null) {
//                isInsert = true;
//                entity = new RmqConsumerEntity();
//                entity.setConsuemrGroup(group);
//                entity.setConsumerIp(ip);
//                entity.setConsumerStatus(MsgConstants.MSG_OK);
//                entity.setCreated(new Date());
//                entity.setMsgKey(keys);
//                entity.setRetryCount(0);
//                entity.setTopic(topic);
//            } else {
//                // 如果存在消息记录，开始做状态比较
//                Integer consumerStatus = entity.getConsumerStatus();
//                if (MsgConstants.MSG_OK == consumerStatus) {
//                    // 已经消费成功了,不做处理
//                    return;
//                } else if (MsgConstants.MSG_WAIT == consumerStatus) {
//                    // 还在处理中 ... 这个时候需要返回一个重新消费的状态
//                    throw new MsgWaitException(" 该 key [" + keys + "] 还在消费中 ... ");
//                } else {
//                    int count = entity.getRetryCount() + 1;
//                    entity.setRetryCount(count);
//                    entity.setUpdated(new Date());
//                }
//            }
//
//            // 2. 具体执行消息
//            try {
//                processMessage(message);
//                // 3. 保存结果
//                saveData(entity, isInsert);
//            } catch (Exception e) {
//                logger.error(" 业务消息消费失败 topic : " + topic + " key : " + keys, e);
//                // 消息执行失败
//                e.printStackTrace();
//                saveExceptionData(message, entity, isInsert);
//            }
//        }

    }

//    private void saveData(RmqConsumerEntity entity, boolean isInsert) {
//        // 如果没有异常发生
//        entity.setConsumerStatus(MsgConstants.MSG_OK);
//        if (isInsert) {
//            consumerDataStore.add(entity);
//        } else {
//            consumerDataStore.udpate(entity);
//        }
//    }
//
//    private void saveExceptionData(MessageExt message, RmqConsumerEntity entity, boolean isInsert) {
//        if (isInsert) {
//            entity.setConsumerStatus(MsgConstants.MSG_FAILS);
//            consumerDataStore.add(entity);
//            logger.info(" 添加失败消息保存成功 " + message.toString());
//            throw new MsgProcessException(" 消费消息执行失败 ... ");
//        } else {
//            consumerDataStore.udpate(entity);
//            logger.info(" 修改失败消息保存成功 " + message.toString());
//            throw new MsgRecurringException(" 重试消息执行失败 ... ");
//        }
//    }
}
