package com.qianwen.smartman.common.config;
|
|
import java.util.Arrays;
|
import java.util.HashMap;
|
import java.util.Map;
|
import java.util.Random;
|
|
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
|
import org.slf4j.Logger;
|
import org.slf4j.LoggerFactory;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.beans.factory.annotation.Value;
|
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.integration.mqtt.support.MqttHeaders;
|
import org.springframework.messaging.MessageChannel;
|
import org.springframework.messaging.MessageHandler;
|
|
import com.alibaba.fastjson.JSONObject;
|
import com.qianwen.core.websocket.distribute.MessageDO;
|
import com.qianwen.core.websocket.distribute.RedisMessageDistributor;
|
import com.qianwen.smartman.common.cache.cps.WorkstationCache;
|
import com.qianwen.smartman.common.websocket.realtime.RealTimeDaraResponseJsonWebSocketMessage;
|
import com.qianwen.smartman.modules.smis.message.dto.TelemetryDataResponseDTO;
|
import com.qianwen.smartman.modules.mdc.dto.ProcessParameterVO;
|
import com.qianwen.smartman.modules.mdc.entity.WorkstationCollectData;
|
import com.qianwen.smartman.modules.mdc.mapper.SuperProcessParameterMapper;
|
import com.qianwen.smartman.modules.mdc.service.RealTimeDataService;
|
|
import cn.hutool.core.date.DateTime;
|
import cn.hutool.core.util.ObjectUtil;
|
import cn.hutool.json.JSONUtil;
|
|
@Configuration
|
public class MdcMqttConfig {
|
private Logger logger = LoggerFactory.getLogger(this.getClass());
|
@Value("${mqtt.host}")
|
private String mqttHost;
|
|
@Value("${mqtt.username:}")
|
private String mqttUserName;
|
|
@Value("${mqtt.password:}")
|
private String mqttPassword;
|
@Value("${mqtt.isRecevieMsg:false}")
|
private boolean isRecevieMsg;
|
@Autowired
|
private RedisMessageDistributor messageDistributor;
|
|
//@Autowired
|
//private SuperProcessParameterMapper processParamMapper;
|
|
@Autowired
|
private RealTimeDataService realTimeDataService;
|
public static final String DEFAULT_TOPIC = "mdc";
|
/**
|
* 反馈创建的topic（mdc中），本应用接收并处理
|
*/
|
//public static final String FEEDBACK_TOPIC = "mdc/feedback";
|
|
public static final String WOCKSTATION_CREATE_TOPIC = "mdc/workstation-create";
|
|
public static final String WOCKSTATION_REALTIMEDATA_TOPIC = "mdc/realtimedata";
|
@Bean
|
public MqttPahoClientFactory mqttClientFactory() {
|
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
|
MqttConnectOptions options = new MqttConnectOptions();
|
options.setServerURIs(new String[] { mqttHost});//"tcp://82.156.1.83:1884"
|
|
if(ObjectUtil.isNotEmpty(mqttUserName)) {
|
options.setUserName(mqttUserName);
|
options.setPassword(mqttPassword.toCharArray());
|
}
|
factory.setConnectionOptions(options);
|
return factory;
|
}
|
|
@Bean
|
public MessageChannel mqttInputChannel() {
|
return new DirectChannel();
|
}
|
|
/**
|
*
|
* @return
|
*/
|
|
@Bean
|
public MessageProducer inbound() {
|
java.util.Random r = new java.util.Random();
|
|
String clientId = "spring-boot-mdc-mqtt-client-inbound"+r.nextInt(1000);
|
|
MqttPahoMessageDrivenChannelAdapter adapter;
|
if(!this.isRecevieMsg) {
|
adapter = new MqttPahoMessageDrivenChannelAdapter(clientId,
|
mqttClientFactory(),DEFAULT_TOPIC);
|
}else {
|
adapter = new MqttPahoMessageDrivenChannelAdapter(clientId,
|
mqttClientFactory(),WOCKSTATION_REALTIMEDATA_TOPIC);//最后一个参数允许多个topic参数
|
}
|
|
adapter.setCompletionTimeout(5000);
|
adapter.setConverter(new DefaultPahoMessageConverter());
|
adapter.setQos(1);
|
adapter.setOutputChannel(mqttInputChannel());
|
return adapter;
|
}
|
|
|
/**
|
* 入站消息处理
|
* @return
|
*/
|
@Bean
|
@ServiceActivator(inputChannel = "mqttInputChannel")
|
public MessageHandler handler() {
|
|
return message -> {
|
String topic = String.valueOf(message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC));
|
if(WOCKSTATION_REALTIMEDATA_TOPIC.equals(topic) ) {
|
//实时数据来了，数据从collect发送来
|
String payload = (String)message.getPayload();
|
logger.info("接收到mqtt消息readtime,data={}",payload);//消息体如何定义，数据点名称，值，时间，用json对象传过来
|
//TelemetryDataMessage result = TelemetryDataUtils.handleTelemetryDataMessage(telemetryDataMessage, workStationItem);
|
JSONObject payLoadJson = JSONObject.parseObject(payload);
|
long workstationId = payLoadJson.getLong("workstationId");
|
String name = payLoadJson.getString("name");//key？
|
long time = payLoadJson.getLong("time");
|
String v = payLoadJson.getString("value");
|
|
//将数据加入缓存
|
TelemetryDataResponseDTO telemetryDataResponseDTO = new TelemetryDataResponseDTO(v, time);
|
WorkstationCache.setWorkstationRealTime(workstationId, name, telemetryDataResponseDTO);
|
|
//发送websocket消息
|
RealTimeDaraResponseJsonWebSocketMessage jsonWebSocketMessage = new RealTimeDaraResponseJsonWebSocketMessage();
|
jsonWebSocketMessage.setId(""+workstationId);
|
|
|
//从缓存（或直接查询）加载最新的数据，发送websocket
|
Map<String, Object> map = WorkstationCache.getWorkstationRealTime(workstationId+"");
|
|
realTimeDataService.addPreTimeInDeviceStatus(workstationId, map);
|
|
jsonWebSocketMessage.setData(map);
|
MessageDO messageDO = new MessageDO();
|
messageDO.setNeedBroadcast(false);
|
String msgtxt =JSONUtil.toJsonStr(jsonWebSocketMessage);
|
logger.info("send_to_page_websockettxt={}",msgtxt);
|
messageDO.setMessageText(msgtxt);
|
|
messageDistributor.distribute(messageDO);
|
}
|
};
|
}
|
|
@Bean
|
public MessageChannel mqttOutboundChannel() {
|
return new DirectChannel();
|
}
|
|
@Bean
|
@ServiceActivator(inputChannel = "mqttOutboundChannel")
|
public MessageHandler mqttOutbound() {
|
Random numList = new Random();
|
|
MqttPahoMessageHandler messageHandler =
|
new MqttPahoMessageHandler("spring-boot-mqtt-client-outbound-mdc"+numList.nextInt(50), mqttClientFactory());
|
messageHandler.setAsync(true);
|
messageHandler.setDefaultTopic(DEFAULT_TOPIC);
|
return messageHandler;
|
}
|
}
|
