//package com.liquidnet.service.consumer.nft.receiver;
//
//import com.liquidnet.common.cache.redis.util.RedisUtil;
//import com.liquidnet.commons.lang.util.CollectionUtil;
//import com.liquidnet.commons.lang.util.JsonUtils;
//import com.liquidnet.service.base.constant.MQConst;
//import com.liquidnet.service.goblin.dto.GoblinQueueBizMongoDto;
//import com.mongodb.BasicDBObject;
//import com.mongodb.client.result.UpdateResult;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.data.mongodb.core.MongoTemplate;
//import org.springframework.data.mongodb.core.query.Criteria;
//import org.springframework.data.mongodb.core.query.Query;
//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;
//
///**
// * 公共的业务Mongo数据处理队列消息监听器，具体业务消费逻辑通过`consumerMessageHandler`实现
// *
// * @author zhanggb
// * Created by IntelliJ IDEA at 2022/4/22
// */
//@Slf4j
//public abstract class AbstractMdbRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
//    @Autowired
//    public StringRedisTemplate stringRedisTemplate;
//    @Autowired
//    private MongoTemplate mongoTemplate;
//    @Autowired
//    private RedisUtil redisUtil;
//
//    private static final BasicDBObject BASIC_DB_OBJECT = new BasicDBObject();
//
//    @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(MQConst.QUEUE_MESSAGE_KEY));
//        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:{}", redisStreamKey, result, message.getValue(), e);
//        }
//        try {
//            stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId());
//        } catch (Exception e) {
//            log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
//        }
//    }
//
//    private boolean consumerMessageHandler(String msg) {
//        boolean aBoolean = true;
//        try {
//            GoblinQueueBizMongoDto queueBizMongoDto = JsonUtils.fromJson(msg, GoblinQueueBizMongoDto.class);
//            if (null != queueBizMongoDto) {
//                String collectName = queueBizMongoDto.getCollect(), columnName = queueBizMongoDto.getColumn();
//                String prefix = queueBizMongoDto.getPrefix(), bizId = queueBizMongoDto.getBizId();
//
//                Object o;
//                switch (queueBizMongoDto.getOpType()) {
//                    case 1:// insert
//                        o = redisUtil.get(prefix.concat(bizId));
//                        if (null != o) {
//                            mongoTemplate.insert(o, collectName);
//                        }
//                        break;
//                    case 2:// update
//                        o = redisUtil.get(prefix.concat(bizId));
//                        if (null != o) {
//                            BasicDBObject basicDBObject = (BasicDBObject) AbstractMdbRedisReceiver.BASIC_DB_OBJECT.clone();
//                            UpdateResult updateResult = mongoTemplate.getCollection(collectName).updateOne(
//                                    Query.query(Criteria.where(columnName).is(bizId)).getQueryObject(),
//                                    basicDBObject.append("$set", mongoTemplate.getConverter().convertToMongoType(o))
//                            );
//                        }
//                        break;
//                    default:
//                        log.error("CONSUMER MSG ERR_HANDLE[未知的操作类型,{}]", msg);
//                        break;
//                }
//            }
//        } catch (Exception e) {
//            log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e);
//            aBoolean = false;
//        } finally {
//            if (!aBoolean) {
//                HashMap<String, String> map = CollectionUtil.mapStringString();
//                map.put(MQConst.QUEUE_MESSAGE_KEY, msg);
//                stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
//            }
//        }
//        return aBoolean;
//    }
//
//    protected abstract String getRedisStreamKey();
//
//    protected abstract String getRedisStreamGroup();
//}