blame | 历史 | 补丁 | 提交 | 提交对比 | 忽略空白
云
11 小时以前 8f3bf7050e65fdbe55eaad74fde307c57dab960e 
 yudao-module-iot/yudao-module-iot-core/src/main/java/cn/iocoder/yudao/module/iot/core/messagebus/config/IotMessageBusAutoConfiguration.java


@@ -6,7 +6,9 @@


import cn.iocoder.yudao.framework.mq.redis.core.stream.AbstractRedisStreamMessage;


import cn.iocoder.yudao.framework.mq.redis.core.stream.AbstractRedisStreamMessageListener;


import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;


import cn.iocoder.yudao.module.iot.core.messagebus.core.kafka.IotKafkaMessageBus;


import cn.iocoder.yudao.module.iot.core.messagebus.core.local.IotLocalMessageBus;


import cn.iocoder.yudao.module.iot.core.messagebus.core.rabbitmq.IotRabbitMQMessageBus;


import cn.iocoder.yudao.module.iot.core.messagebus.core.redis.IotRedisMessageBus;


import cn.iocoder.yudao.module.iot.core.messagebus.core.rocketmq.IotRocketMQMessageBus;


import cn.iocoder.yudao.module.iot.core.mq.producer.IotDeviceMessageProducer;


@@ -14,15 +16,20 @@


import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;


import org.apache.rocketmq.spring.core.RocketMQTemplate;


import org.redisson.api.RedissonClient;


import org.springframework.amqp.rabbit.core.RabbitAdmin;


import org.springframework.amqp.rabbit.core.RabbitTemplate;


import org.springframework.boot.autoconfigure.AutoConfiguration;


import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;


import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;


import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;


import org.springframework.boot.context.properties.EnableConfigurationProperties;


import org.springframework.boot.kafka.autoconfigure.KafkaProperties;


import org.springframework.context.ApplicationContext;


import org.springframework.context.annotation.Bean;


import org.springframework.context.annotation.Configuration;


import org.springframework.data.redis.core.RedisTemplate;


import org.springframework.data.redis.core.StringRedisTemplate;


import org.springframework.kafka.core.KafkaTemplate;




import java.util.List;




@@ -73,6 +80,21 @@




    }




    // ==================== Kafka 实现 ====================




    @Configuration


    @ConditionalOnProperty(prefix = "yudao.iot.message-bus", name = "type", havingValue = "kafka")


    @ConditionalOnClass(KafkaTemplate.class)


    public static class IotKafkaMessageBusConfiguration {




        @Bean


        public IotKafkaMessageBus iotKafkaMessageBus(KafkaProperties kafkaProperties) {


            log.info("[iotKafkaMessageBus][创建 IoT Kafka 消息总线]");


            return new IotKafkaMessageBus(kafkaProperties);


        }




    }




    // ==================== Redis 实现 ====================




    /**


@@ -99,7 +121,8 @@


                                                                            RedisMQTemplate redisTemplate,


                                                                            RedissonClient redissonClient) {


            List<AbstractRedisStreamMessageListener<?>> listeners = getListeners(messageBus);


            return new RedisPendingMessageResendJob(listeners, redisTemplate, redissonClient);


            return new RedisPendingMessageResendJob(listeners, redisTemplate, redissonClient,


                    RedisPendingMessageResendJob.IOT_RESEND_LOCK_KEY);


        }




        /**


@@ -110,7 +133,8 @@


                                                                            RedisMQTemplate redisTemplate,


                                                                            RedissonClient redissonClient) {


            List<AbstractRedisStreamMessageListener<?>> listeners = getListeners(messageBus);


            return new RedisStreamMessageCleanupJob(listeners, redisTemplate, redissonClient);


            return new RedisStreamMessageCleanupJob(listeners, redisTemplate, redissonClient,


                    RedisStreamMessageCleanupJob.IOT_CLEANUP_LOCK_KEY);


        }




        private List<AbstractRedisStreamMessageListener<?>> getListeners(IotRedisMessageBus messageBus) {


@@ -126,4 +150,25 @@




    }




}


    // ==================== RabbitMQ 实现 ====================




    @Configuration


    @ConditionalOnProperty(prefix = "yudao.iot.message-bus", name = "type", havingValue = "rabbitmq")


    @ConditionalOnClass(RabbitTemplate.class)


    public static class IotRabbitMQMessageBusConfiguration {




        @Bean


        @ConditionalOnMissingBean


        public RabbitAdmin rabbitAdmin(RabbitTemplate rabbitTemplate) {


            return new RabbitAdmin(rabbitTemplate);


        }




        @Bean


        public IotRabbitMQMessageBus iotRabbitMQMessageBus(RabbitTemplate rabbitTemplate, RabbitAdmin rabbitAdmin) {


            log.info("[iotRabbitMQMessageBus][创建 IoT RabbitMQ 消息总线]");


            return new IotRabbitMQMessageBus(rabbitTemplate, rabbitAdmin);


        }




    }




}