设备控制

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
@Component
public class ServerRpc {


    @Autowired
    private EmqClient emqClient;

    @Autowired
    private MqttProperties properties;


    @PostConstruct
    public void init(){
        //连接服务端
        emqClient.connect(properties.getUsername(),properties.getPassword());
        //订阅一个主题
        emqClient.subscribe("v1/devices/me/rpc/request/+", QosEnum.QoS1);
    }
    
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
@Component
public class MessageCallback implements MqttCallback {
    
    
    /**
     * 应用收到消息后触发的回调
     * @param topic
     * @param message
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.info("订阅者订阅到了消息,topic={},messageid={},qos={},payload={}",
                topic,
                message.getId(),
                message.getQos(),
                new String(message.getPayload()));

        // 订阅者订阅到了消息,topic=v1/devices/me/rpc/request/7,messageid=6,qos=1,payload={"method":"setValue","params":false}
        // 订阅后客户端会收到一条命令作为对相应主题的PUBLISH命令：
        // v1/devices/me/rpc/request/$request_id
        String[] buff = topic.split("/");
        String request_id = buff[buff.length-1];


        // 客户端PUBLISH下面主题进行响应：
        // v1/devices/me/rpc/response/$request_id
        emqClient.publish("v1/devices/me/rpc/response/" + request_id, "{}", QosEnum.QoS1,false);

    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
package com.iiotos.rpc;


import com.iiotos.mqtt.EmqClient;
import com.iiotos.mqtt.MqttProperties;
import com.iiotos.mqtt.QosEnum;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;


@Component
public class ClientRpc {


    @Autowired
    private EmqClient emqClient;

    @Autowired
    private MqttProperties properties;


    @PostConstruct
    public void init(){
        //连接服务端
        emqClient.connect(properties.getUsername(),properties.getPassword());
        //订阅一个主题
        emqClient.subscribe("v1/devices/me/rpc/response/+", QosEnum.QoS1);

    }


    @Scheduled(fixedRate = 3000)
    public void publish(){

        String data = getData();

        emqClient.publish("v1/devices/me/rpc/request/1",data,QosEnum.QoS1,false);
    }

    
    private String getData(){

        String data = "{\n" +
                "\t\"method\": \"getServerValue\",\n" +
                "\t\"params\": \"\"\n" +
                "}";

        return data;

    }

}