|
|
@@ -0,0 +1,195 @@
|
|
|
+package com.hfln.portal.infrastructure.config;
|
|
|
+
|
|
|
+import com.hfln.portal.common.constant.mqtt.topic.TopicConstants;
|
|
|
+import com.hfln.portal.infrastructure.mqtt.MqttSubHandle;
|
|
|
+import lombok.extern.slf4j.Slf4j;
|
|
|
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
|
|
|
+import org.springframework.beans.factory.annotation.Value;
|
|
|
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
|
|
+import org.springframework.context.annotation.Bean;
|
|
|
+import org.springframework.context.annotation.Configuration;
|
|
|
+import org.springframework.integration.annotation.ServiceActivator;
|
|
|
+import org.springframework.integration.channel.DirectChannel;
|
|
|
+import org.springframework.integration.core.MessageProducer;
|
|
|
+import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
|
|
|
+import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
|
|
|
+import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
|
|
|
+import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
|
|
|
+import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
|
|
|
+import org.springframework.messaging.Message;
|
|
|
+import org.springframework.messaging.MessageChannel;
|
|
|
+import org.springframework.messaging.MessageHandler;
|
|
|
+import org.springframework.messaging.MessagingException;
|
|
|
+import org.springframework.scheduling.TaskScheduler;
|
|
|
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
|
|
|
+
|
|
|
+/**
|
|
|
+ * MQTT配置类
|
|
|
+ *
|
|
|
+ * 统一管理所有MQTT相关配置，包括：
|
|
|
+ * 1. MQTT客户端工厂
|
|
|
+ * 2. Spring Integration入站和出站适配器
|
|
|
+ * 3. 消息处理器
|
|
|
+ * 4. 任务调度器
|
|
|
+ */
|
|
|
+@Configuration
|
|
|
+@Slf4j
|
|
|
+@ConditionalOnProperty(prefix = "mqtt", name = "enabled", havingValue = "true", matchIfMissing = true)
|
|
|
+public class MqttConfig {
|
|
|
+
|
|
|
+ @Value("${mqtt.broker:tcp://8.130.28.21:1883}")
|
|
|
+ private String serverUri;
|
|
|
+
|
|
|
+ @Value("${mqtt.client.id:hfln-device-service}")
|
|
|
+ private String clientId;
|
|
|
+
|
|
|
+ @Value("${mqtt.username:admin}")
|
|
|
+ private String username;
|
|
|
+
|
|
|
+ @Value("${mqtt.password:public}")
|
|
|
+ private String password;
|
|
|
+
|
|
|
+ @Value("${mqtt.connect.timeout:30}")
|
|
|
+ private int connectTimeout;
|
|
|
+
|
|
|
+ @Value("${mqtt.keep.alive.interval:60}")
|
|
|
+ private int keepAliveInterval;
|
|
|
+
|
|
|
+ @Value("${mqtt.clean.session:true}")
|
|
|
+ private boolean cleanSession;
|
|
|
+
|
|
|
+ // ===========================================
|
|
|
+ // 基础配置
|
|
|
+ // ===========================================
|
|
|
+
|
|
|
+ @Bean
|
|
|
+ public TaskScheduler taskScheduler() {
|
|
|
+ ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
|
|
|
+ scheduler.setPoolSize(10);
|
|
|
+ scheduler.setThreadNamePrefix("mqtt-task-");
|
|
|
+ scheduler.setWaitForTasksToCompleteOnShutdown(true);
|
|
|
+ scheduler.setAwaitTerminationSeconds(60);
|
|
|
+ return scheduler;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * MQTT客户端工厂
|
|
|
+ */
|
|
|
+ @Bean
|
|
|
+ public MqttPahoClientFactory mqttClientFactory() {
|
|
|
+ DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
|
|
|
+ MqttConnectOptions options = new MqttConnectOptions();
|
|
|
+
|
|
|
+ options.setServerURIs(new String[] { serverUri });
|
|
|
+
|
|
|
+ if (username != null && !username.isEmpty()) {
|
|
|
+ options.setUserName(username);
|
|
|
+ }
|
|
|
+
|
|
|
+ if (password != null && !password.isEmpty()) {
|
|
|
+ options.setPassword(password.toCharArray());
|
|
|
+ }
|
|
|
+
|
|
|
+ options.setConnectionTimeout(connectTimeout);
|
|
|
+ options.setKeepAliveInterval(keepAliveInterval);
|
|
|
+ options.setCleanSession(cleanSession);
|
|
|
+
|
|
|
+ factory.setConnectionOptions(options);
|
|
|
+
|
|
|
+ return factory;
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ // ===========================================
|
|
|
+ // 消息通道和适配器
|
|
|
+ // ===========================================
|
|
|
+
|
|
|
+ @Bean
|
|
|
+ public MessageChannel inputChannel() {
|
|
|
+ return new DirectChannel();
|
|
|
+ }
|
|
|
+
|
|
|
+ @Bean
|
|
|
+ public MessageProducer mpsInbound() {
|
|
|
+ String[] topics = {
|
|
|
+ TopicConstants.TOPIC_DAS_EVENT,
|
|
|
+ TopicConstants.TOPIC_DAS_ALARM_EVENT,
|
|
|
+ TopicConstants.TOPIC_DAS_DEV_STATUS,
|
|
|
+ TopicConstants.TOPIC_DAS_ESIST,
|
|
|
+ TopicConstants.TOPIC_DAS_REALTIME_POS
|
|
|
+ };
|
|
|
+
|
|
|
+ MqttPahoMessageDrivenChannelAdapter adapter =
|
|
|
+ new MqttPahoMessageDrivenChannelAdapter(clientId + "_das", mqttClientFactory(), topics);
|
|
|
+ adapter.setCompletionTimeout(5000);
|
|
|
+ adapter.setConverter(new DefaultPahoMessageConverter());
|
|
|
+ adapter.setQos(2, 2, 0, 0, 0);
|
|
|
+ adapter.setOutputChannel(inputChannel());
|
|
|
+ adapter.setTaskScheduler(taskScheduler());
|
|
|
+ return adapter;
|
|
|
+ }
|
|
|
+
|
|
|
+ @Bean
|
|
|
+ @ServiceActivator(inputChannel = "inputChannel")
|
|
|
+ public MessageHandler dasMqttMessageHandler(MqttSubHandle handler) {
|
|
|
+ return new MessageHandler() {
|
|
|
+ @Override
|
|
|
+ public void handleMessage(Message<?> message) throws MessagingException {
|
|
|
+ handler.handleMessage(message);
|
|
|
+ }
|
|
|
+ };
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ // ===========================================
|
|
|
+ // 出站配置和框架兼容
|
|
|
+ // ===========================================
|
|
|
+
|
|
|
+ /**
|
|
|
+ * MQTT消息输入通道（框架需要）
|
|
|
+ */
|
|
|
+ @Bean
|
|
|
+ public MessageChannel mqttInputChannel() {
|
|
|
+ return new DirectChannel();
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * MQTT消息输出通道
|
|
|
+ */
|
|
|
+ @Bean
|
|
|
+ public MessageChannel mqttOutputChannel() {
|
|
|
+ return new DirectChannel();
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * MQTT消息处理器（出站）
|
|
|
+ */
|
|
|
+ @Bean
|
|
|
+ @ServiceActivator(inputChannel = "mqttOutputChannel")
|
|
|
+ public MessageHandler outbound() {
|
|
|
+ String clientIdWithRandom = clientId + System.currentTimeMillis();
|
|
|
+ MqttPahoMessageHandler messageHandler =
|
|
|
+ new MqttPahoMessageHandler(clientIdWithRandom + "-out", mqttClientFactory());
|
|
|
+
|
|
|
+ messageHandler.setAsync(true);
|
|
|
+ messageHandler.setDefaultQos(1);
|
|
|
+
|
|
|
+ return messageHandler;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * MQTT出站处理器
|
|
|
+ */
|
|
|
+ @Bean
|
|
|
+ public MqttPahoMessageHandler mqttOutbound() {
|
|
|
+ String clientIdWithRandom = clientId + System.currentTimeMillis();
|
|
|
+ MqttPahoMessageHandler messageHandler =
|
|
|
+ new MqttPahoMessageHandler(clientIdWithRandom + "-outbound", mqttClientFactory());
|
|
|
+
|
|
|
+ messageHandler.setAsync(true);
|
|
|
+ messageHandler.setDefaultQos(1);
|
|
|
+
|
|
|
+ return messageHandler;
|
|
|
+ }
|
|
|
+}
|
