|
|
|
@ -1,181 +1,167 @@
|
|
|
|
package com.ruoyi.mqtt.config;
|
|
|
|
package com.ruoyi.mqtt.config;
|
|
|
|
|
|
|
|
|
|
|
|
import com.ruoyi.mqtt.MqttEventHandler;
|
|
|
|
import com.ruoyi.mqtt.MqttConnectionEvent;
|
|
|
|
import com.ruoyi.mqtt.MqttFactory;
|
|
|
|
import com.ruoyi.mqtt.MqttConnectionLostEvent;
|
|
|
|
import com.ruoyi.mqtt.MqttItem;
|
|
|
|
import com.ruoyi.mqtt.MqttMessageDeliveryEvent;
|
|
|
|
import com.ruoyi.mqtt.MqttUtil;
|
|
|
|
import com.ruoyi.mqtt.MqttMessageEvent;
|
|
|
|
import com.ruoyi.mqtt.event.MqttEvent;
|
|
|
|
|
|
|
|
import com.ruoyi.mqtt.event.MqttSendEvent;
|
|
|
|
|
|
|
|
import com.ruoyi.mqtt.event.MqttSendTopicEvent;
|
|
|
|
|
|
|
|
import lombok.Getter;
|
|
|
|
|
|
|
|
import lombok.RequiredArgsConstructor;
|
|
|
|
import lombok.RequiredArgsConstructor;
|
|
|
|
import lombok.extern.slf4j.Slf4j;
|
|
|
|
import lombok.extern.slf4j.Slf4j;
|
|
|
|
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
|
|
|
|
import org.eclipse.paho.client.mqttv3.*;
|
|
|
|
import org.eclipse.paho.client.mqttv3.MqttException;
|
|
|
|
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
|
|
|
|
import org.springframework.boot.ApplicationArguments;
|
|
|
|
import org.springframework.beans.factory.DisposableBean;
|
|
|
|
import org.springframework.boot.ApplicationRunner;
|
|
|
|
import org.springframework.beans.factory.config.BeanPostProcessor;
|
|
|
|
|
|
|
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
|
|
|
|
|
|
|
|
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.context.properties.EnableConfigurationProperties;
|
|
|
|
import org.springframework.context.ApplicationContext;
|
|
|
|
import org.springframework.context.ApplicationEventPublisher;
|
|
|
|
|
|
|
|
import org.springframework.context.annotation.Bean;
|
|
|
|
import org.springframework.context.annotation.Configuration;
|
|
|
|
import org.springframework.context.annotation.Configuration;
|
|
|
|
import org.springframework.context.event.EventListener;
|
|
|
|
import org.springframework.context.event.EventListener;
|
|
|
|
import org.springframework.core.Ordered;
|
|
|
|
import org.springframework.integration.mqtt.support.MqttUtils;
|
|
|
|
import org.springframework.core.PriorityOrdered;
|
|
|
|
import org.springframework.scheduling.annotation.Async;
|
|
|
|
|
|
|
|
import org.springframework.scheduling.annotation.Scheduled;
|
|
|
|
|
|
|
|
|
|
|
|
import javax.annotation.PostConstruct;
|
|
|
|
import javax.annotation.PostConstruct;
|
|
|
|
import javax.annotation.PreDestroy;
|
|
|
|
import javax.annotation.PreDestroy;
|
|
|
|
import java.util.Collections;
|
|
|
|
import java.util.Arrays;
|
|
|
|
import java.util.Map;
|
|
|
|
import java.util.Random;
|
|
|
|
import java.util.concurrent.ConcurrentHashMap;
|
|
|
|
import java.util.concurrent.TimeUnit;
|
|
|
|
import java.util.concurrent.ScheduledExecutorService;
|
|
|
|
import java.util.concurrent.atomic.AtomicBoolean;
|
|
|
|
import java.util.concurrent.ScheduledThreadPoolExecutor;
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@Configuration
|
|
|
|
@Configuration(proxyBeanMethods = false)
|
|
|
|
|
|
|
|
@ConditionalOnProperty(prefix = MqttProperties.PREFIX, name = "enabled", havingValue = "true", matchIfMissing = false)
|
|
|
|
@EnableConfigurationProperties(MqttProperties.class)
|
|
|
|
@EnableConfigurationProperties(MqttProperties.class)
|
|
|
|
@Slf4j
|
|
|
|
@Slf4j
|
|
|
|
@RequiredArgsConstructor
|
|
|
|
@RequiredArgsConstructor
|
|
|
|
public class MqttConfig implements MqttFactory, MqttEventHandler, PriorityOrdered {
|
|
|
|
public class MqttConfig {
|
|
|
|
|
|
|
|
|
|
|
|
private final MqttProperties properties;
|
|
|
|
private final MqttProperties properties;
|
|
|
|
private final ApplicationContext act;
|
|
|
|
private final ApplicationEventPublisher publisher;
|
|
|
|
|
|
|
|
|
|
|
|
@Getter
|
|
|
|
private static MqttClient client;
|
|
|
|
private static MqttFactory mqttFactory;
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
private ScheduledExecutorService executorService;
|
|
|
|
private static MqttCallback mqttCallback;
|
|
|
|
|
|
|
|
|
|
|
|
private final Map<String, MqttItem> clients = new ConcurrentHashMap<>();
|
|
|
|
private static MqttConnectOptions options;
|
|
|
|
|
|
|
|
|
|
|
|
@PostConstruct
|
|
|
|
private static AtomicBoolean isReConnect = new AtomicBoolean(false);
|
|
|
|
public void init() throws Exception {
|
|
|
|
|
|
|
|
if (!properties.getEnabled()) {
|
|
|
|
|
|
|
|
log.info("mqtt模块未激活");
|
|
|
|
|
|
|
|
return;
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
log.info("mqtt模块启动中...");
|
|
|
|
|
|
|
|
try {
|
|
|
|
|
|
|
|
executorService = act.getBeansOfType(ScheduledExecutorService.class).values().stream().findFirst().get();
|
|
|
|
|
|
|
|
} catch (Exception e) {
|
|
|
|
|
|
|
|
executorService = new ScheduledThreadPoolExecutor(Runtime.getRuntime().availableProcessors());
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
properties.getConfigs().forEach((a, b) -> {
|
|
|
|
|
|
|
|
if (!b.getEnabled()) {
|
|
|
|
|
|
|
|
return;
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
put(a, b, this);
|
|
|
|
|
|
|
|
});
|
|
|
|
|
|
|
|
mqttFactory = this;
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public static MqttClient getClient() {
|
|
|
|
public boolean next(MqttEvent event, MqttItem item) {
|
|
|
|
return client;
|
|
|
|
log.debug("mqtt event: {} = {}", event.getConfigName(), event.getClass().getName());
|
|
|
|
|
|
|
|
act.publishEvent(event);
|
|
|
|
|
|
|
|
return true;
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
@PreDestroy
|
|
|
|
@Bean
|
|
|
|
public void destroy() {
|
|
|
|
@ConditionalOnMissingBean(MqttClient.class)
|
|
|
|
clients.forEach((a, b) -> {
|
|
|
|
public MqttClient mqttClient(MqttConnectOptions options, MqttCallback callback) throws Exception {
|
|
|
|
|
|
|
|
log.debug("mqtt:开始创建mqtt客户端");
|
|
|
|
|
|
|
|
client = new MqttClient(properties.getUrl(), properties.getClientId()+'_'+Long.toString(System.currentTimeMillis()-new Random().nextInt(),36), new MemoryPersistence());
|
|
|
|
try{
|
|
|
|
try{
|
|
|
|
b.destroy();
|
|
|
|
connect();
|
|
|
|
} catch (Exception e) {
|
|
|
|
}catch (Exception ex) {
|
|
|
|
log.error("mqtt客户端关闭失败:" + a, e);
|
|
|
|
log.error("mqtt:连接异常:"+properties.getUrl(), ex);
|
|
|
|
}
|
|
|
|
|
|
|
|
});
|
|
|
|
|
|
|
|
try {
|
|
|
|
|
|
|
|
executorService.shutdown();
|
|
|
|
|
|
|
|
} catch (Exception e) {
|
|
|
|
|
|
|
|
throw new RuntimeException(e);
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return client;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
public MqttItem get(String configName) {
|
|
|
|
@Bean
|
|
|
|
return clients.get(configName);
|
|
|
|
public MqttCallback mqttCallback() {
|
|
|
|
}
|
|
|
|
mqttCallback = new MqttCallback() {
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public void put(String configName, MqttProperties.Config config, MqttEventHandler... handlers) {
|
|
|
|
public void connectionLost(Throwable throwable) {
|
|
|
|
if (clients.containsKey(configName)) {
|
|
|
|
log.warn("mqtt:连接断开", throwable);
|
|
|
|
throw new RuntimeException("配置项已经存在");
|
|
|
|
publisher.publishEvent(new MqttConnectionLostEvent(throwable));
|
|
|
|
}
|
|
|
|
|
|
|
|
MqttItem mqttItem = new MqttItem(configName, config, executorService);
|
|
|
|
|
|
|
|
if (handlers != null && handlers.length > 0) {
|
|
|
|
|
|
|
|
Collections.addAll(mqttItem.getMessageHandlers(), handlers);
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
clients.put(configName, mqttItem);
|
|
|
|
|
|
|
|
log.debug("mqtt配置项添加:" + configName);
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
public boolean contains(String configName) {
|
|
|
|
@Override
|
|
|
|
return clients.containsKey(configName);
|
|
|
|
public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
|
|
|
|
|
|
|
|
log.trace("mqtt:接收到主题为{}的消息", s);
|
|
|
|
|
|
|
|
publisher.publishEvent(new MqttMessageEvent(s, mqttMessage));
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
public void remove(String configName) {
|
|
|
|
@Override
|
|
|
|
if (contains(configName)) {
|
|
|
|
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
|
|
|
|
get(configName).destroy();
|
|
|
|
log.trace("mqtt:确认发送到主题为{}的消息", iMqttDeliveryToken.getTopics());
|
|
|
|
clients.remove(configName);
|
|
|
|
publisher.publishEvent(new MqttMessageDeliveryEvent(iMqttDeliveryToken));
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
};
|
|
|
|
|
|
|
|
return mqttCallback;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@Bean
|
|
|
|
|
|
|
|
public MqttConnectOptions mqttConnectOptions() {
|
|
|
|
|
|
|
|
// 连接设置
|
|
|
|
|
|
|
|
options = new MqttConnectOptions();
|
|
|
|
|
|
|
|
// 是否清空session，设置false表示服务器会保留客户端的连接记录（订阅主题，qos）,客户端重连之后能获取到服务器在客户端断开连接期间推送的消息
|
|
|
|
|
|
|
|
// 设置为true表示每次连接服务器都是以新的身份
|
|
|
|
|
|
|
|
options.setCleanSession(properties.getCleanSession());
|
|
|
|
|
|
|
|
// 设置连接用户名
|
|
|
|
|
|
|
|
options.setUserName(properties.getUsername());
|
|
|
|
|
|
|
|
// 设置连接密码
|
|
|
|
|
|
|
|
options.setPassword(properties.getPassword().toCharArray());
|
|
|
|
|
|
|
|
// 设置超时时间，单位为秒
|
|
|
|
|
|
|
|
options.setConnectionTimeout(properties.getConnectionTimeout());
|
|
|
|
|
|
|
|
// 设置心跳时间 单位为秒，表示服务器每隔 1.5*20秒的时间向客户端发送心跳判断客户端是否在线
|
|
|
|
|
|
|
|
options.setKeepAliveInterval(properties.getKeepAliveInterval());
|
|
|
|
|
|
|
|
// 设置遗嘱消息的话题，若客户端和服务器之间的连接意外断开，服务器将发布客户端的遗嘱信息
|
|
|
|
|
|
|
|
options.setWill(properties.getWillTopic(), properties.getWillMessage().getBytes(), properties.getWillQos(), false);
|
|
|
|
|
|
|
|
// 设置重连
|
|
|
|
|
|
|
|
options.setAutomaticReconnect(false);
|
|
|
|
|
|
|
|
|
|
|
|
public IMqttDeliveryToken send(String configName, String topic, byte[] payload, int qos, boolean retained) throws MqttException {
|
|
|
|
return options;
|
|
|
|
MqttItem item = get(configName);
|
|
|
|
|
|
|
|
if (item == null) {
|
|
|
|
|
|
|
|
throw new RuntimeException("mqtt客户端未找到:" + configName);
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
log.debug("mqtt发送成功:configName={},topic={}", configName, topic);
|
|
|
|
|
|
|
|
return item.getClient().publish(topic, payload, qos, retained);
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
public IMqttDeliveryToken send(String configName, String sendName, byte[] payload, String... params) throws MqttException {
|
|
|
|
public void connect() throws Exception {
|
|
|
|
|
|
|
|
|
|
|
|
MqttItem item = get(configName);
|
|
|
|
log.debug("mqtt:url={},clientId={}", client.getServerURI(), client.getClientId());
|
|
|
|
if (item == null) {
|
|
|
|
// 设置回调
|
|
|
|
throw new RuntimeException("mqtt客户端未找到:" + configName);
|
|
|
|
client.setCallback(mqttCallback);
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
MqttProperties.Topic topic = item.getConfig().getSends().get(sendName);
|
|
|
|
client.connect(MqttUtils.cloneConnectOptions(options));
|
|
|
|
if (topic == null) {
|
|
|
|
if (properties.getTopic() != null && properties.getTopic().length > 0) {
|
|
|
|
throw new RuntimeException("mqtt配置的主题未找到:" + configName + " = " + sendName);
|
|
|
|
if (properties.getTopic().length != properties.getQos().length) {
|
|
|
|
|
|
|
|
throw new Exception("mqtt:订阅的主题和qos不一致");
|
|
|
|
}
|
|
|
|
}
|
|
|
|
String topicTempalte = topic.getTopic();
|
|
|
|
client.subscribe(properties.getTopic(), properties.getQos());
|
|
|
|
if (params != null) {
|
|
|
|
log.debug("mqtt:订阅了主题={}", Arrays.toString(properties.getTopic()));
|
|
|
|
for (int i = 0; i < params.length; i++) {
|
|
|
|
|
|
|
|
topicTempalte = topicTempalte.replace("{" + i + "}", params[i]);
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
log.debug("mqtt:创建客户端成功");
|
|
|
|
|
|
|
|
publisher.publishEvent(new MqttConnectionEvent());
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
log.debug("mqtt发送成功:configName={},sendName={},topic={}", configName, sendName, topicTempalte);
|
|
|
|
@Scheduled(fixedDelay = 5000)
|
|
|
|
return item.getClient().publish(topicTempalte, payload, topic.getQos(), topic.getRetained());
|
|
|
|
public void reConnect() {
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if (!client.isConnected()) {
|
|
|
|
@EventListener
|
|
|
|
try {
|
|
|
|
public void listener(MqttSendEvent event) throws MqttException {
|
|
|
|
client.disconnect();
|
|
|
|
send(event.getConfigName(), event.getSendName(), event.getPayload(), event.getParams());
|
|
|
|
} catch (Exception e) {
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
/* try {
|
|
|
|
|
|
|
|
client.close();
|
|
|
|
|
|
|
|
} catch (Exception e) {
|
|
|
|
|
|
|
|
}*/
|
|
|
|
|
|
|
|
try {
|
|
|
|
|
|
|
|
connect();
|
|
|
|
|
|
|
|
} catch (Exception ex) {
|
|
|
|
|
|
|
|
log.error("mqtt:连接异常:"+properties.getUrl(), ex);
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
@EventListener
|
|
|
|
|
|
|
|
public void listener(MqttSendTopicEvent event) throws MqttException {
|
|
|
|
|
|
|
|
send(event.getConfigName(), event.getTopic(), event.getPayload(), event.getQos(), event.isRetained());
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// @Scheduled(cron = "*/5 * * * * ?")
|
|
|
|
/**
|
|
|
|
// public void test() {
|
|
|
|
* 断开连接
|
|
|
|
// String s = Long.toString(System.currentTimeMillis(), 36);
|
|
|
|
*/
|
|
|
|
// try {
|
|
|
|
@PreDestroy
|
|
|
|
//// act.publishEvent(new MqttSendEvent("test",s.getBytes(StandardCharsets.UTF_8),s));
|
|
|
|
public void disConnect() {
|
|
|
|
// MqttUtil.send("test", s, new String[]{s});
|
|
|
|
try {
|
|
|
|
// log.info("test success:{}", s);
|
|
|
|
if (client != null) {
|
|
|
|
// } catch (Exception e) {
|
|
|
|
client.disconnect();
|
|
|
|
// log.info("test error:" + s, e);
|
|
|
|
}
|
|
|
|
// }
|
|
|
|
} catch (Exception e) {
|
|
|
|
// }
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
@Override
|
|
|
|
|
|
|
|
public int getOrder() {
|
|
|
|
|
|
|
|
return Ordered.HIGHEST_PRECEDENCE; // 最高优先级
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
