package org.jeecg.modules.iot.mqtt.config;
|
|
import com.alibaba.fastjson.JSONObject;
|
import lombok.extern.slf4j.Slf4j;
|
import org.eclipse.paho.client.mqttv3.*;
|
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
|
import org.jeecg.modules.iot.entity.MqttParameter;
|
import org.jeecg.modules.iot.mqtt.callback.PushCallback;
|
import org.jeecg.modules.iot.util.DispatchUtil;
|
import org.springframework.stereotype.Component;
|
|
import javax.annotation.Resource;
|
import java.time.LocalDateTime;
|
import java.time.format.DateTimeFormatter;
|
import java.util.Arrays;
|
import java.util.LinkedHashMap;
|
import java.util.Map;
|
|
/**
|
* mqtt客户端
|
*
|
* @author: cuikaidong
|
* @create: 2023-09-27 11:36
|
*/
|
@Slf4j
|
@Component
|
public class MqttCustomerClient {
|
|
@Resource
|
private PushCallback pushCallback;
|
|
|
private static MqttClient client;
|
|
public static MqttClient getClient() {
|
return client;
|
}
|
|
public static void setClient(MqttClient client) {
|
MqttCustomerClient.client = client;
|
}
|
|
/**
|
* 客户端连接
|
*
|
* @param host ip+端口
|
* @param clientID 客户端Id
|
* @param username 用户名
|
* @param password 密码
|
* @param timeout 超时时间
|
* @param keeplive 保留数
|
*/
|
public void connect(String host, String clientID, String username, String password, int timeout, int keeplive) {
|
MqttClient client;
|
|
try {
|
client = new MqttClient(host, clientID, new MemoryPersistence());
|
MqttConnectOptions options = new MqttConnectOptions();
|
options.setCleanSession(true);
|
options.setUserName(username);
|
options.setPassword(password.toCharArray());
|
options.setConnectionTimeout(timeout);
|
options.setKeepAliveInterval(keeplive);
|
MqttCustomerClient.setClient(client);
|
try {
|
client.setCallback(pushCallback);
|
client.connect(options);
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
}
|
|
/**
|
* 发布，默认qos为0，非持久化
|
*
|
* @param topic
|
* @param mqttParameter
|
*/
|
public void pushlish(String topic, MqttParameter mqttParameter) {
|
pushlish(0, false, topic, mqttParameter);
|
}
|
|
/**
|
* 发布
|
*
|
* @param qos 连接方式
|
* @param retained 是否保留
|
* @param topic 主题
|
* @param mqttParameter 参数对象
|
*/
|
public void pushlish(int qos, boolean retained, String topic, MqttParameter mqttParameter) {
|
topic = "IOT\\" + topic + "\\Mutually1";
|
Map<String, String> map = new LinkedHashMap<>();
|
map.put("id", mqttParameter.getId());
|
map.put("type", mqttParameter.getType());
|
LocalDateTime now = LocalDateTime.now();
|
DateTimeFormatter formatter2 = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
|
map.put("time", now.format(formatter2));
|
map.put("parameter1", mqttParameter.getParameter1());
|
map.put("parameter2", mqttParameter.getParameter2());
|
map.put("parameter3", mqttParameter.getParameter3());
|
map.put("parameter4", mqttParameter.getParameter4());
|
map.put("parameter5", mqttParameter.getParameter5());
|
Object json = JSONObject.toJSON(map);
|
String pushMessage = DispatchUtil.replaceBlank(json.toString());
|
MqttMessage message = new MqttMessage();
|
message.setQos(qos);
|
message.setRetained(retained);
|
message.setPayload(pushMessage.getBytes());
|
MqttTopic mqttTopic = MqttCustomerClient.getClient().getTopic(topic);
|
if (null == mqttTopic) {
|
log.error("topic not exist");
|
}
|
MqttDeliveryToken token;
|
try {
|
token = mqttTopic.publish(message);
|
token.waitForCompletion();
|
} catch (MqttPersistenceException e) {
|
e.printStackTrace();
|
} catch (MqttException e) {
|
e.printStackTrace();
|
// 重连
|
pushCallback.connectionLost(null);
|
}
|
}
|
|
/**
|
* 订阅某个主题，qos为0
|
*
|
* @param topic
|
*/
|
public void subscribe(String[] topic) {
|
log.info("开始订阅主题" + Arrays.toString(topic));
|
try {
|
int[] qos = new int[topic.length];
|
for (int i = 0; i < topic.length; i++) {
|
qos[i] = 2;
|
}
|
MqttCustomerClient.getClient().subscribe(topic, qos);
|
} catch (MqttException e) {
|
e.printStackTrace();
|
}
|
}
|
|
}
|
