package com.liquidnet.servce.consumer.stone.service.receiver;

import com.liquidnet.commons.lang.util.JsonUtils;
import com.liquidnet.servce.consumer.stone.service.IBaseDao;
import com.liquidnet.service.base.SqlMapping;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;

import java.util.HashMap;

/**
 * @author AnJiabin <anjiabin@zhengzai.tv>
 * @version V1.0
 * @class: AbstractRedisReceiver
 * @Package com.liquidnet.service.consumer.dragon.receiver
 * @Copyright: LightNet @ Copyright (c) 2021
 * @date 2021/7/22 20:28
 */
@Slf4j
public abstract class AbstractRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
    @Autowired
    private IBaseDao baseDao;
    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        log.info("接受到来自redis PAY key:{} 的消息",getRedisStreamKey());
        log.info("message id " + message.getId());
        log.info("stream " + message.getStream());
        log.info("body " + message.getValue());
        boolean result = this.consumerSqlDaoHandler(message.getValue().get("message"));

//        if(result){
            // 消费成功确认，消息删除和消息确认是一个事务
            log.info("consumer success delete message messageId:{} ",message.getId());
            try {
//                stringRedisTemplate.multi();
                stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
                stringRedisTemplate.opsForStream().delete(this.getRedisStreamKey(), message.getId());
//                stringRedisTemplate.exec();
            } catch (Exception e) {
                e.printStackTrace();
                log.error("delete redis queue message error messageId:{} errMsg:{}",message.getId(),e.getMessage());
            }finally {
                try {
                    stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
                    stringRedisTemplate.opsForStream().delete(this.getRedisStreamKey(), message.getId());
                } catch (Exception e) {
                    log.error("error: {}",e);
                }
            }
//        }
    }

    private boolean consumerSqlDaoHandler(String msg) {
        try {
            SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(msg, SqlMapping.SqlMessage.class);
            log.debug("CONSUMER SQL ==> Preparing:{}", JsonUtils.toJson(sqlMessage.getSqls()));
            log.debug("CONSUMER SQL ==> Parameters:{}", JsonUtils.toJson(sqlMessage.getArgs()));
            Boolean rstBatchSqls = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
            log.debug("CONSUMER SQL result of execution:{}", rstBatchSqls);
            if (rstBatchSqls) {
                return true;
            }else{
                sendMySqlRedis(msg);
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("CONSUMER SQL Exception error:{}", e);
        }
        return false;
    }

    /**
     * 给 REDIS 队列发送消息 数据库相关
     *
     * @param msg 接收到的内容
     * @return
     */
    private boolean sendMySqlRedis(String msg) {
        try {
            HashMap<String, String> map = new HashMap<>();
            map.put("message", msg);
            MapRecord<String, String, String> record = StreamRecords.mapBacked(map).withStreamKey(getRedisStreamKey());
            stringRedisTemplate.opsForStream().add(record);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    protected abstract String getRedisStreamKey();

    protected abstract String getRedisStreamGroup();
}
