package com.liquidnet.service.consumer.base.config;

import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.consumer.base.receiver.*;
import com.liquidnet.service.consumer.base.receiver.dragon.*;
import com.liquidnet.service.consumer.base.receiver.kylin.ConsumerKylinOrderRefundExpressRdsReceiver;
import com.liquidnet.service.consumer.base.receiver.sweet.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;

import java.util.ArrayList;
import java.util.List;

/**
 * 公共的SQL队列消息消费器初始化配置
 *
 * @author zhanggb
 * Created by IntelliJ IDEA at 2022/6/2
 */
@Configuration
public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
    @Autowired
    StringRedisTemplate stringRedisTemplate;
    @Autowired
    ConsumerCommonSQL0Receiver consumerCommonSQL0Receiver;
    /*------goblin------*/
    @Autowired
    ConsumerGoblinNftOrderReceiver consumerGoblinNftOrderReceiver;
    @Autowired
    ConsumerGoblinSqlStoreReceiver consumerGoblinSqlStoreReceiver;
    @Autowired
    ConsumerGoblinSqlGoodsReceiver consumerGoblinSqlGoodsReceiver;
    @Autowired
    ConsumerGoblinSqlCouponReceiver consumerGoblinSqlCouponReceiver;
    @Autowired
    ConsumerGoblinSqlArtworkGenReceiver consumerGoblinSqlArtworkGenReceiver;
    /*------adam------*/
    @Autowired
    ConsumerAdamSqlURegisterReceiver consumerAdamURegisterRdsReceiver;
    @Autowired
    ConsumerAdamSqlUCenterReceiver consumerAdamUCenterRdsReceiver;
    @Autowired
    ConsumerAdamSqlUMemberReceiver consumerAdamUMemberRdsReceiver;
    /*------kylin------*/
    @Autowired
    ConsumerKylinSqlStationReceiver consumerKylinSqlStationReceiver;
    @Autowired
    ConsumerKylinOrderRefundExpressRdsReceiver consumerKylinOrderRefundExpressRdsReceiver;
    /*------slime------*/
    @Autowired
    ConsumerSlimeSqlFieldReceiver consumerSlimeSqlFieldReceiver;
    @Autowired
    ConsumerSlimeSqlSponsorReceiver consumerSlimeSqlSponsorReceiver;
    @Autowired
    ConsumerSlimeSqlTicketUpdateReceiver consumerSlimeSqlTicketUpdateReceiver;
    @Autowired
    ConsumerSlimeSqlPerformanceUpdateReceiver consumerSlimeSqlPerformanceUpdateReceiver;
    @Autowired
    ConsumerSlimeSqlStep2Receiver consumerSlimeSqlStep2Receiver;
    @Autowired
    ConsumerSlimeSqlLineReceiver consumerSlimeSqlLineReceiver;
    @Autowired
    ConsumerSlimeSqlPerformanceInsertReceiver consumerSlimeSqlPerformanceInsertReceiver;
    /*------galaxy------*/
    @Autowired
    ConsumerGalaxySqlUserInfoReceiver consumerGalaxySqlUserInfoReceiver;
    @Autowired
    ConsumerGalaxySqlSeriesInfoReceiver consumerGalaxySqlSeriesInfoReceiver;
    @Autowired
    ConsumerGalaxySqlNftOrderInfoReceiver consumerGalaxySqlNftOrderInfoReceiver;
    @Autowired
    ConsumerGalaxySqlNftTradeInfoReceiver consumerGalaxySqlNftTradeInfoReceiver;
    @Autowired
    ConsumerGalaxySqlNftOrderFailLogReceiver consumerGalaxySqlNftOrderFailLogReceiver;
    /*------sweet------*/
    @Autowired
    ConsumerSweetAnswerRdsReceiver consumerSweetAnswerRdsReceiver;
    @Autowired
    ConsumerSweetAntigenicQuestionRdsReceiver consumerAntigenicQuestionRdsReceiver;
    @Autowired
    ConsumerSweetCityVoteRdsReceiver consumerCityVoteRdsReceiver;
    @Autowired
    ConsumerSweetIntegralActivityDrawRdsReceiver consumerIntegralActivityDrawRdsReceiver;
    @Autowired
    ConsumerSweetMaoDengReceiver consumerSweetMaoDengReceiver;
    @Autowired
    ConsumerSweetPosterRdsReceiver consumerSweetPosterRdsReceiver;
    @Autowired
    ConsumerSweetRemindRdsReceiver consumerSweetRemindRdsReceiver;
    @Autowired
    ConsumerSweetSqlLuckDrawRdsReceiver consumerSweetSqlLuckDrawRdsReceiver;
    @Autowired
    ConsumerSweetSqlUserRelationMDSKRdsReceiver consumerSweetSqlUserRelationMDSKRdsReceiver;
    @Autowired
    ConsumerSweetSqlUserRelationRdsReceiver consumerSweetSqlUserRelationRdsReceiver;
    @Autowired
    ConsumerSweetUserRdsReceiver consumerSweetUserRdsReceiver;

    /*------dragon------*/
    @Autowired
    private ConsumerDragonPayReceiver consumerDragonPayReceiver;
    @Autowired
    private ConsumerDragonMchNotifyFailReceiver consumerDragonMchNotifyFailReceiver;
    @Autowired
    private ConsumerDragonMchNotifyReceiver consumerDragonMchNotifyReceiver;
    @Autowired
    private ConsumerDragonPayNotifyReceiver consumerDragonPayNotifyReceiver;
    @Autowired
    private ConsumerDragonRefundReceiver consumerDragonRefundReceiver;
//    @Bean
//    public List<Subscription> subscriptionSQL0(RedisConnectionFactory factory) {
//        List<Subscription> subscriptionList = new ArrayList<>();
//        MQConst.CommonQueue stream = MQConst.CommonQueue.SQL0;
//        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
//        for (int i = 0; i < 5; i++) {
//            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
//            subscriptionList.add(listenerContainer.receiveAutoAck(
//                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
//                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerCommonSQL0Receiver
//            ));
//            listenerContainer.start();
//        }
//        return subscriptionList;
//    }

    /*------sweet------*/

    @Bean
    public List<Subscription> subscriptionSweetSqlWechatUser(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.SweetQueue stream = MQConst.SweetQueue.SWEET_WECHAT_USER_INSERT;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 3; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerSweetUserRdsReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionSweetSqlArtistsRelation(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.SweetQueue stream = MQConst.SweetQueue.ARTISTS_RELATION;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 3; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerSweetSqlUserRelationRdsReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionSweetSqlUserMdskRelation(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.SweetQueue stream = MQConst.SweetQueue.ARTISTS_RELATION_MDSK;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 3; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerSweetSqlUserRelationMDSKRdsReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionSweetSqlLuckDraw(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.SweetQueue stream = MQConst.SweetQueue.LUCK_DRAW;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 3; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerSweetSqlLuckDrawRdsReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionSweetSqlRemind(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.SweetQueue stream = MQConst.SweetQueue.SWEET_REMIND_INSERT;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 3; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerSweetRemindRdsReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionSweetSqlPoster(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.SweetQueue stream = MQConst.SweetQueue.SWEET_POSTER;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 3; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerSweetPosterRdsReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionSweetSqlMaoDeng(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.SweetQueue stream = MQConst.SweetQueue.SWEET_WECHAT_MAODENG_TEXT_INSERT;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 3; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerSweetMaoDengReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionSweetSqlIntegralActivityDraw(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.SweetQueue stream = MQConst.SweetQueue.SWEET_INTEGRAL_ACTIVITY_DRAW;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 3; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerIntegralActivityDrawRdsReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionSweetSqlCityVote(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.SweetQueue stream = MQConst.SweetQueue.SWEET_CITY_VOTE;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 3; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerCityVoteRdsReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionSweetSqlAntigenicQuestion(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.SweetQueue stream = MQConst.SweetQueue.SWEET_ANTIGENIC_QUESTION;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 3; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerAntigenicQuestionRdsReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionSweetSqlAnswer(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.SweetQueue stream = MQConst.SweetQueue.SWEET_ANSWER_INSERT;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 3; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerSweetAnswerRdsReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    /*------goblin------*/

    @Bean
    public List<Subscription> subscriptionGoblinSqlNftOrder(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.GoblinQueue stream = MQConst.GoblinQueue.GOBLIN_NFT_ORDER;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 20; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerGoblinNftOrderReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionGoblinSqlStore(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.GoblinQueue stream = MQConst.GoblinQueue.SQL_STORE;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 1; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerGoblinSqlStoreReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionGoblinSqlGoods(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.GoblinQueue stream = MQConst.GoblinQueue.SQL_GOODS;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 1; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerGoblinSqlGoodsReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionGoblinSqlStoreCoupon(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.GoblinQueue stream = MQConst.GoblinQueue.SQL_COUPON;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 1; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerGoblinSqlCouponReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionGoblinSqlArtworkGen(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.GoblinQueue stream = MQConst.GoblinQueue.SQL_ARTWORK_GEN;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 5; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerGoblinSqlArtworkGenReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    /*------adam------*/

    @Bean
    public List<Subscription> subscriptionAdamSqlURegister(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.AdamQueue stream = MQConst.AdamQueue.SQL_UREGISTER;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 10; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerAdamURegisterRdsReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionAdamSqlUCenter(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.AdamQueue stream = MQConst.AdamQueue.SQL_UCENTER;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 10; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerAdamUCenterRdsReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionAdamSqlUMember(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.AdamQueue stream = MQConst.AdamQueue.SQL_UMEMBER;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 10; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerAdamUMemberRdsReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    /*------kylin------*/

    @Bean
    public List<Subscription> subscriptionKylinSqlStation(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.KylinQueue stream = MQConst.KylinQueue.SQL_STATION;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 10; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerKylinSqlStationReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionKylinSqlOrderRefundExpress(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.KylinQueue stream = MQConst.KylinQueue.SQL_ORDER_REFUND_EXPRESS;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 10; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerKylinOrderRefundExpressRdsReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    /*------slime------*/

    @Bean
    public List<Subscription> subscriptionSlimeSqlField(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.SlimeQueue stream = MQConst.SlimeQueue.SQL_SLIME_FIELD;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 1; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerSlimeSqlFieldReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionSlimeSqlSponsor(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.SlimeQueue stream = MQConst.SlimeQueue.SQL_SLIME_SPONSOR;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 1; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerSlimeSqlSponsorReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionSlimeSqlTicketUpdate(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.SlimeQueue stream = MQConst.SlimeQueue.SQL_SLIME_TICKET_UPDATE;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 1; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerSlimeSqlTicketUpdateReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionSlimeSqlPerformanceUpdate(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.SlimeQueue stream = MQConst.SlimeQueue.SQL_SLIME_PERFORMANCE_UPDATE;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 1; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerSlimeSqlPerformanceUpdateReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionSlimeSqlStep2(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.SlimeQueue stream = MQConst.SlimeQueue.SQL_SLIME_STEP2;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 1; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerSlimeSqlStep2Receiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionSlimeSqlLine(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.SlimeQueue stream = MQConst.SlimeQueue.SQL_SLIME_LINE;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 1; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerSlimeSqlLineReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionSlimeSqlPerformanceInsert(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.SlimeQueue stream = MQConst.SlimeQueue.SQL_SLIME_PERFORMANCE_INSERT;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 1; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerSlimeSqlPerformanceInsertReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionGalaxySqlUserInfo(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.GalaxyQueue stream = MQConst.GalaxyQueue.SQL_USER_INFO;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 1; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerGalaxySqlUserInfoReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionGalaxySqlSeriesInfo(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.GalaxyQueue stream = MQConst.GalaxyQueue.SQL_SERIES_INFO;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 1; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerGalaxySqlSeriesInfoReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionGalaxySqlNftOrderInfo(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.GalaxyQueue stream = MQConst.GalaxyQueue.SQL_NFT_ORDER_INFO;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 1; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerGalaxySqlNftOrderInfoReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionGalaxySqlNftTradeInfo(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.GalaxyQueue stream = MQConst.GalaxyQueue.SQL_NFT_TRADE_INFO;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 1; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerGalaxySqlNftTradeInfoReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean
    public List<Subscription> subscriptionGalaxySqlNftOrderFailLog(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.GalaxyQueue stream = MQConst.GalaxyQueue.SQL_NFT_ORDER_FAIL_LOG;

        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 1; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerGalaxySqlNftOrderFailLogReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }

    @Bean// dragon-支付
    public List<Subscription> subscriptionDragonPay(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.DragonQueue stream = MQConst.DragonQueue.DRAGON_PAY_KEY;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 20; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerDragonPayReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }
    @Bean// dragon-商户通知失败
    public List<Subscription> subscriptionDragonMchNotifyFail(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.DragonQueue stream = MQConst.DragonQueue.PAY_MCH_NOTIFY_ERROR_KEY;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 5; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerDragonMchNotifyFailReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }
    @Bean// dragon-商户通知
    public List<Subscription> subscriptionDragonMchNotify(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.DragonQueue stream = MQConst.DragonQueue.PAY_MCH_NOTIFY_KEY;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 5; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerDragonMchNotifyReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }
    @Bean// dragon-支付通知
    public List<Subscription> subscriptionDragonPayNotify(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.DragonQueue stream = MQConst.DragonQueue.DRAGON_PAY_NOTIFY_KEY;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 5; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerDragonPayNotifyReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }
    @Bean// dragon-退款
    public List<Subscription> subscriptionDragonRefund(RedisConnectionFactory factory) {
        List<Subscription> subscriptionList = new ArrayList<>();
        MQConst.DragonQueue stream = MQConst.DragonQueue.DRAGON_REFUND_KEY;
        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
        for (int i = 0; i < 5; i++) {
            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
            subscriptionList.add(listenerContainer.receiveAutoAck(
                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerDragonRefundReceiver
            ));
            listenerContainer.start();
        }
        return subscriptionList;
    }
}
