package cn.iocoder.yudao.module.iot.core.messagebus.config;
|
|
import cn.iocoder.yudao.framework.mq.redis.core.RedisMQTemplate;
|
import cn.iocoder.yudao.framework.mq.redis.core.job.RedisPendingMessageResendJob;
|
import cn.iocoder.yudao.framework.mq.redis.core.job.RedisStreamMessageCleanupJob;
|
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;
|
import lombok.extern.slf4j.Slf4j;
|
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;
|
|
import static cn.iocoder.yudao.framework.common.util.collection.CollectionUtils.convertList;
|
|
/**
|
* IoT 消息总线自动配置
|
*
|
* @author 芋道源码
|
*/
|
@AutoConfiguration
|
@EnableConfigurationProperties(IotMessageBusProperties.class)
|
@Slf4j
|
public class IotMessageBusAutoConfiguration {
|
|
@Bean
|
public IotDeviceMessageProducer deviceMessageProducer(IotMessageBus messageBus) {
|
return new IotDeviceMessageProducer(messageBus);
|
}
|
|
// ==================== Local 实现 ====================
|
|
@Configuration
|
@ConditionalOnProperty(prefix = "yudao.iot.message-bus", name = "type", havingValue = "local", matchIfMissing = true)
|
public static class IotLocalMessageBusConfiguration {
|
|
@Bean
|
public IotLocalMessageBus iotLocalMessageBus(ApplicationContext applicationContext) {
|
log.info("[iotLocalMessageBus][创建 IoT Local 消息总线]");
|
return new IotLocalMessageBus(applicationContext);
|
}
|
|
}
|
|
// ==================== RocketMQ 实现 ====================
|
|
@Configuration
|
@ConditionalOnProperty(prefix = "yudao.iot.message-bus", name = "type", havingValue = "rocketmq")
|
@ConditionalOnClass(RocketMQTemplate.class)
|
public static class IotRocketMQMessageBusConfiguration {
|
|
@Bean
|
public IotRocketMQMessageBus iotRocketMQMessageBus(RocketMQProperties rocketMQProperties,
|
RocketMQTemplate rocketMQTemplate) {
|
log.info("[iotRocketMQMessageBus][创建 IoT RocketMQ 消息总线]");
|
return new IotRocketMQMessageBus(rocketMQProperties, rocketMQTemplate);
|
}
|
|
}
|
|
// ==================== 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 实现 ====================
|
|
/**
|
* 特殊：由于 YudaoRedisMQConsumerAutoConfiguration 关于 Redis stream 的消费是动态注册，所以这里只能拷贝相关的逻辑！！！
|
*
|
* @see cn.iocoder.yudao.framework.mq.redis.config.YudaoRedisMQConsumerAutoConfiguration
|
*/
|
@Configuration
|
@ConditionalOnProperty(prefix = "yudao.iot.message-bus", name = "type", havingValue = "redis")
|
@ConditionalOnClass(RedisTemplate.class)
|
public static class IotRedisMessageBusConfiguration {
|
|
@Bean
|
public IotRedisMessageBus iotRedisMessageBus(StringRedisTemplate redisTemplate) {
|
log.info("[iotRedisMessageBus][创建 IoT Redis 消息总线]");
|
return new IotRedisMessageBus(redisTemplate);
|
}
|
|
/**
|
* 创建 Redis Stream 重新消费的任务
|
*/
|
@Bean
|
public RedisPendingMessageResendJob iotRedisPendingMessageResendJob(IotRedisMessageBus messageBus,
|
RedisMQTemplate redisTemplate,
|
RedissonClient redissonClient) {
|
List<AbstractRedisStreamMessageListener<?>> listeners = getListeners(messageBus);
|
return new RedisPendingMessageResendJob(listeners, redisTemplate, redissonClient,
|
RedisPendingMessageResendJob.IOT_RESEND_LOCK_KEY);
|
}
|
|
/**
|
* 创建 Redis Stream 消息清理任务
|
*/
|
@Bean
|
public RedisStreamMessageCleanupJob iotRedisStreamMessageCleanupJob(IotRedisMessageBus messageBus,
|
RedisMQTemplate redisTemplate,
|
RedissonClient redissonClient) {
|
List<AbstractRedisStreamMessageListener<?>> listeners = getListeners(messageBus);
|
return new RedisStreamMessageCleanupJob(listeners, redisTemplate, redissonClient,
|
RedisStreamMessageCleanupJob.IOT_CLEANUP_LOCK_KEY);
|
}
|
|
private List<AbstractRedisStreamMessageListener<?>> getListeners(IotRedisMessageBus messageBus) {
|
return convertList(messageBus.getSubscribers(), subscriber ->
|
new AbstractRedisStreamMessageListener<>(subscriber.getTopic(), subscriber.getGroup()) {
|
|
@Override
|
public void onMessage(AbstractRedisStreamMessage message) {
|
throw new UnsupportedOperationException("不应该调用！！！");
|
}
|
});
|
}
|
|
}
|
|
// ==================== 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);
|
}
|
|
}
|
|
}
|
