dragon-redis队列迁移

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/config/ConsumerCommonSqlRedisStreamConfig.java

@@ -3,6 +3,7 @@ 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;
@@ -104,6 +105,17 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
     @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<>();
@@ -658,4 +670,80 @@ public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
         }
         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 < 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()), 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 < 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()), 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 < 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()), 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 < 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()), 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 < 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()), consumerDragonRefundReceiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
 }