//package com.liquidnet.servce.consumer.stone.service.receiver;
//
//import com.liquidnet.commons.lang.util.CollectionUtil;
//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) {
//        String redisStreamKey = this.getRedisStreamKey();
//        log.debug("CONSUMER MSG[streamKey:{},messageId:{},stream:{},body:{}]", redisStreamKey, message.getId(), message.getStream(), message.getValue());
//        boolean result = this.consumerMessageHandler(message.getValue().get("message"));
//        log.info("CONSUMER MSG RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId());
//
//        try {
//            stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
//        } catch (Exception e) {
//            log.error("#CONSUMER MSG EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", this.getRedisStreamKey(), result, message.getValue(), e);
//        }
//        try {
//            stringRedisTemplate.opsForStream().delete(this.getRedisStreamKey(), message.getId());
//        } catch (Exception e) {
//            log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", this.getRedisStreamKey(), result, message.getValue(), e);
//        }
//    }
//
//    private boolean consumerMessageHandler(String msg) {
//        boolean aBoolean = false;
//        try {
//            SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(msg, SqlMapping.SqlMessage.class);
//            aBoolean = null == sqlMessage || baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
//        } catch (Exception e) {
//            log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e);
//        } finally {
//            if (!aBoolean) {
//                HashMap<String, String> map = CollectionUtil.mapStringString();
//                map.put("message", msg);
//                stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
//            }
//        }
//        return aBoolean;
//    }
//
////    @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();
//}
