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.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;
    /*------slime------*/
    @Autowired
    ConsumerSlimeSqlFieldReceiver consumerSlimeSqlFieldReceiver;
    @Autowired
    ConsumerSlimeSqlSponsorReceiver consumerSlimeSqlSponsorReceiver;
    @Autowired
    ConsumerSlimeSqlTicketUpdateReceiver consumerSlimeSqlTicketUpdateReceiver;
    @Autowired
    ConsumerSlimeSqlPerformanceUpdateReceiver consumerSlimeSqlPerformanceUpdateReceiver;
    @Autowired
    ConsumerSlimeSqlStep2Receiver consumerSlimeSqlStep2Receiver;
    @Autowired
    ConsumerSlimeSqlLineReceiver consumerSlimeSqlLineReceiver;
    @Autowired
    ConsumerSlimeSqlPerformanceInsertReceiver consumerSlimeSqlPerformanceInsertReceiver;
    /*------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;

    /*------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// NFT订单
    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;
    }

    /*------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;
    }
}
