Merge remote-tracking branch 'origin/master'

pom.xml

@@ -67,6 +67,21 @@
             <version>RELEASE</version>
             <scope>compile</scope>
         </dependency>
+
+        <!-- mqtt支持 start -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-integration</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-stream</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-mqtt</artifactId>
+        </dependency>
+        <!-- mqtt支持 end -->
     </dependencies>
 
     <build>

src/main/java/com/sjls/nstthh/NstthhApplication.java

@@ -5,6 +5,8 @@ import com.sjls.nstthh.usb.NRJavaSerialService;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 
+import java.util.Arrays;
+
 
 @SpringBootApplication
 
@@ -17,14 +19,14 @@ public class NstthhApplication {
 //        for (String s: NRSerialPort.getAvailableSerialPorts()){
 //            System.out.println("Availible port = " + s);
 //        }
-
-        try {
+        //System.err.println("来了没有");
+        /*try {
             NRJavaSerialService nRJavaSerialService = new NRJavaSerialService();
             nRJavaSerialService.connect("/dev/ttyUSB0");// /dev/ttyUSB0
 
         } catch (Exception e) {
             e.printStackTrace();
-        }
+        }*/
 
     }
 

src/main/java/com/sjls/nstthh/controller/HardwareAdmin.java

@@ -1,8 +1,11 @@
 package com.sjls.nstthh.controller;
 
+import com.sjls.nstthh.mqtt.MsgWriter;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 
+import javax.annotation.Resource;
+
 /**
  * smp admin
  */
@@ -10,10 +13,18 @@ import org.springframework.web.bind.annotation.RestController;
 @RestController
 public class HardwareAdmin {
 
+    @Resource
+    private MsgWriter msgWriter;
+
     @RequestMapping("/hello")
     public String index() {
         return "Hello World";
     }
 
+    @RequestMapping(value = "/testmqtt")
+    public void test_mqtt(String msg) {
+        msgWriter.sendToMqtt(msg);
+    }
+
 }
 

src/main/java/com/sjls/nstthh/mqtt/MqttSenderConfig.java

@@ -0,0 +1,164 @@
+package com.sjls.nstthh.mqtt;
+
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.annotation.IntegrationComponentScan;
+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;
+
+@Configuration
+@IntegrationComponentScan
+@Slf4j
+public class MqttSenderConfig {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(MqttSenderConfig.class);
+
+    @Value("${spring.mqtt.username}")
+    private String username;
+
+    @Value("${spring.mqtt.password}")
+    private String password;
+
+    @Value("${spring.mqtt.url}")
+    private String hostUrl;
+
+    @Value("${spring.mqtt.publishclientid}")
+    private String publishclientid;
+
+    @Value("${spring.mqtt.subsribeclientid}")
+    private String subsribeclientid;
+
+    @Value("${spring.mqtt.default_topic}")
+    private String defaultTopic;
+
+    /**
+     * 订阅的bean名称
+     */
+    public static final String CHANNEL_NAME_IN = "mqttInboundChannel";
+    /**
+     * 发布的bean名称
+     */
+    public static final String CHANNEL_NAME_OUT = "mqttOutboundChannel";
+
+//    @Autowired
+//    private MqttCallbackHandler mqttCallbackHandler;
+
+    /**
+     * MQTT连接器选项
+     */
+    @Bean
+    public <hostUrl> MqttConnectOptions getMqttConnectOptions() {
+        MqttConnectOptions options = new MqttConnectOptions();
+        // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，
+        // 这里设置为true表示每次连接到服务器都以新的身份连接
+        options.setCleanSession(false);
+        // 设置连接的用户名
+        options.setUserName(username);
+        // 设置连接的密码
+        options.setPassword(password.toCharArray());
+        options.setServerURIs(new String[]{hostUrl});
+        // 设置超时时间 单位为秒
+        options.setConnectionTimeout(10);
+        // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线，但这个方法并没有重连的机制
+        options.setKeepAliveInterval(20);
+        // 设置“遗嘱”消息的话题，若客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息。
+        //options.setWill("willTopic", WILL_DATA, 2, false);
+        return options;
+    }
+
+
+    /**
+     * MQTT客户端
+     */
+    @Bean
+    public MqttPahoClientFactory mqttClientFactory() {
+        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
+        factory.setConnectionOptions(getMqttConnectOptions());
+        return factory;
+    }
+
+    /**
+     * MQTT信息通道（生产者）
+     */
+    @Bean(name = CHANNEL_NAME_OUT)
+    public MessageChannel mqttOutboundChannel() {
+        return new DirectChannel();
+    }
+
+    /**
+     * MQTT消息处理器（生产者）
+     */
+    @Bean
+    @ServiceActivator(inputChannel = CHANNEL_NAME_OUT)
+    public MessageHandler mqttOutbound() {
+        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(
+                publishclientid, mqttClientFactory());
+        messageHandler.setAsync(true);
+        //if (null == topic) topic = defaultTopic;
+        messageHandler.setDefaultTopic(defaultTopic);
+        return messageHandler;
+    }
+
+    /**
+     * MQTT消息订阅绑定（消费者）
+     */
+    @Bean
+    public MessageProducer inbound() {
+        // 可以同时消费（订阅）多个Topic
+        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
+                subsribeclientid, mqttClientFactory(), new String[]{defaultTopic});
+        adapter.setCompletionTimeout(5000);
+        adapter.setConverter(new DefaultPahoMessageConverter());
+        adapter.setQos(2);
+        // 设置订阅通道
+        adapter.setOutputChannel(mqttInboundChannel());
+        return adapter;
+    }
+
+    /**
+     * MQTT信息通道（消费者）
+     */
+    @Bean(name = CHANNEL_NAME_IN)
+    public MessageChannel mqttInboundChannel() {
+        return new DirectChannel();
+    }
+
+    /**
+     * MQTT消息处理器（消费者）
+     */
+    @Bean
+    @ServiceActivator(inputChannel = CHANNEL_NAME_IN)
+    public MessageHandler handler() {
+        return new MessageHandler() {
+            @Override
+            public void handleMessage(Message<?> message)  {
+                try {
+                    String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
+                    String qos = message.getHeaders().get("mqtt_receivedQos").toString();
+                    String payload = message.getPayload().toString();
+                    //mqttCallbackHandler.handle(topic,payload);
+                    LOGGER.info("主题："+topic);
+                    LOGGER.info("内容："+payload);
+                    LOGGER.info("级别："+qos);
+                    //message.get
+                } catch (Exception e) {
+                    LOGGER.error(e.getMessage(), e);
+                }
+            }
+        };
+    }
+}

src/main/java/com/sjls/nstthh/mqtt/MsgWriter.java

@@ -0,0 +1,16 @@
+package com.sjls.nstthh.mqtt;
+
+import org.springframework.integration.annotation.MessagingGateway;
+import org.springframework.integration.mqtt.support.MqttHeaders;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.stereotype.Component;
+
+@Component
+@MessagingGateway(defaultRequestChannel = MqttSenderConfig.CHANNEL_NAME_OUT)
+public interface MsgWriter {
+
+    void sendToMqtt(String data);
+    void sendToMqtt(String payload,@Header(MqttHeaders.TOPIC) String topic);
+    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);
+
+}

src/main/resources/application.yml

@@ -23,4 +23,20 @@ spring:
   h2:
     console:
       path: /h2-console
-      enabled: true
+      enabled: true
+
+  mqtt:
+    #MQTT-服务端账号
+    username: admin
+    #MQTT-服务端密码
+    password: public
+    #MQTT-服务端地址
+    url: tcp://192.168.50.169:1883
+    #MQTT-发布者clientid
+    publishclientid: client_publish
+    #MQTT-接收者clientid
+    subsribeclientid: client_subsribe
+    #MQTT-默认主题
+    default_topic: test
+
+