//package com.liquidnet.service.consumer.dragon.service.processor;
//
//import com.liquidnet.commons.lang.util.JsonUtils;
//import com.liquidnet.service.base.SqlMapping;
//import com.liquidnet.service.consumer.dragon.service.IBaseDao;
//import com.rabbitmq.client.Channel;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.amqp.core.Message;
//import org.springframework.amqp.core.MessageProperties;
//import org.springframework.data.redis.connection.stream.MapRecord;
//import org.springframework.data.redis.stream.StreamListener;
//import org.springframework.stereotype.Component;
//
//import javax.annotation.Resource;
//import java.io.IOException;
//
///**
// * @author AnJiabin <anjiabin@zhengzai.tv>
// * @version V1.0
// * @Description: TODO
// * @class: ConsumerPayProcessor
// * @Package com.liquidnet.service.consumer.dragon.service.processor
// * @Copyright: LightNet @ Copyright (c) 2021
// * @date 2021/7/8 10:59
// */
//@Slf4j
//@Component
//public class ConsumerPayProcessor implements StreamListener<String, MapRecord<String, String, String>> {
//    @Resource
//    IBaseDao baseDao;
//    @Override
//    public void onMessage(MapRecord<String, String, String> message) {
//        log.info("接受到来自redis PAY 的消息");
//        System.out.println("message id "+message.getId());
//        System.out.println("stream "+message.getStream());
//        System.out.println("body "+message.getValue());
//    }
//
//
//    private void consumerSqlDaoHandler(Message msg, Channel channel) {
//        MessageProperties properties = msg.getMessageProperties();
//        String consumerQueue = properties.getConsumerQueue();
//        long deliveryTag = properties.getDeliveryTag();
//        log.info("CONSUMER SQL ==> [consumerQueue:{},deliveryTag:{}]", consumerQueue, deliveryTag);
//        SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(new String(msg.getBody()), SqlMapping.SqlMessage.class);
//        log.debug("CONSUMER SQL ==> Preparing:{}", JsonUtils.toJson(sqlMessage.getSqls()));
//        log.debug("CONSUMER SQL ==> Parameters:{}", JsonUtils.toJson(sqlMessage.getArgs()));
//        try {
//            Boolean rstBatchSqls = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
//            log.debug("CONSUMER SQL result of execution:{}", rstBatchSqls);
//            if (rstBatchSqls) {
//                channel.basicAck(deliveryTag, false);
//            } else {
//                log.warn("###CONSUMER SQL[consumerQueue:{},deliveryTag={},sqlMessage:{}]", consumerQueue, deliveryTag, JsonUtils.toJson(sqlMessage));
//                channel.basicAck(deliveryTag, false);
//            }
//        } catch (IOException e) {
//            log.error("CONSUMER SQL[consumerQueue:{},deliveryTag:{},sqlMessage:{}]", consumerQueue, deliveryTag, JsonUtils.toJson(sqlMessage), e);
//        }
//    }
//}
