package com.fastbee.mqtt.service.impl;
|
|
import com.alibaba.fastjson2.JSON;
|
import com.alibaba.fastjson2.JSONArray;
|
import com.fastbee.common.core.mq.DeviceReportBo;
|
import com.fastbee.common.core.mq.MQSendMessageBo;
|
import com.fastbee.common.core.mq.message.DeviceData;
|
import com.fastbee.common.core.mq.message.DeviceDownMessage;
|
import com.fastbee.common.core.mq.message.InstructionsMessage;
|
import com.fastbee.common.core.mq.message.MqttBo;
|
import com.fastbee.common.core.mq.ota.OtaUpgradeBo;
|
import com.fastbee.common.core.protocol.modbus.ModbusCode;
|
import com.fastbee.common.core.thingsModel.ThingsModelSimpleItem;
|
import com.fastbee.common.enums.ServerType;
|
import com.fastbee.common.enums.TopicType;
|
import com.fastbee.common.exception.ServiceException;
|
import com.fastbee.common.utils.DateUtils;
|
import com.fastbee.common.utils.StringUtils;
|
import com.fastbee.common.utils.gateway.CRC16Utils;
|
import com.fastbee.common.utils.gateway.mq.TopicsUtils;
|
import com.fastbee.common.utils.ip.IpUtils;
|
import com.fastbee.iot.domain.Device;
|
import com.fastbee.iot.domain.FunctionLog;
|
import com.fastbee.iot.domain.Product;
|
import com.fastbee.iot.model.NtpModel;
|
import com.fastbee.iot.model.ThingsModels.PropertyDto;
|
import com.fastbee.iot.service.IDeviceService;
|
import com.fastbee.iot.service.IProductService;
|
import com.fastbee.iot.service.IThingsModelService;
|
import com.fastbee.iot.service.cache.IFirmwareCache;
|
import com.fastbee.iot.util.SnowflakeIdWorker;
|
import com.fastbee.json.JsonProtocolService;
|
import com.fastbee.mq.model.ReportDataBo;
|
import com.fastbee.mq.mqttClient.PubMqttClient;
|
import com.fastbee.mq.service.IDataHandler;
|
import com.fastbee.mq.service.IMqttMessagePublish;
|
import com.fastbee.mqtt.manager.MqttRemoteManager;
|
import com.fastbee.mqtt.model.PushMessageBo;
|
import lombok.extern.slf4j.Slf4j;
|
import org.java_websocket.protocols.IProtocol;
|
import org.springframework.stereotype.Service;
|
|
import javax.annotation.Resource;
|
import java.util.ArrayList;
|
import java.util.List;
|
import java.util.Objects;
|
import java.util.Optional;
|
|
/**
|
* 消息推送方法集合
|
*
|
* @author bill
|
*/
|
@Slf4j
|
@Service
|
public class MqttMessagePublishImpl implements IMqttMessagePublish {
|
|
@Resource
|
private IProductService productService;
|
@Resource
|
private PubMqttClient mqttClient;
|
@Resource
|
private IFirmwareCache firmwareCache;
|
@Resource
|
private TopicsUtils topicsUtils;
|
@Resource
|
private IDeviceService deviceService;
|
@Resource
|
private MqttRemoteManager remoteManager;
|
|
@Resource
|
private IDataHandler dataHandler;
|
@Resource
|
private IThingsModelService thingsModelService;
|
@Resource
|
private JsonProtocolService jsonProtocolService;
|
private SnowflakeIdWorker snowflakeIdWorker = new SnowflakeIdWorker(3);
|
|
|
@Override
|
public InstructionsMessage buildMessage(DeviceDownMessage downMessage, TopicType type) {
|
/*返回的组将数据*/
|
InstructionsMessage message = new InstructionsMessage();
|
/*根据设备编号查询产品信息*/
|
if (StringUtils.isEmpty(downMessage.getProtocolCode())) {
|
Product product = productService.getProductBySerialNumber(downMessage.getSerialNumber());
|
Optional.ofNullable(product).orElseThrow(() -> new ServiceException("产品为空"));
|
downMessage.setProtocolCode(product.getProtocolCode());
|
}
|
String serialNumber = downMessage.getSerialNumber() == null ? "" : downMessage.getSerialNumber();
|
|
/*组建Topic*/
|
String topicName = "";
|
if (downMessage.getServerType().equals(ServerType.MQTT)) {
|
topicName = topicsUtils.buildTopic(downMessage.getProductId(), serialNumber, type);
|
}
|
|
DeviceData encodeData = DeviceData.builder()
|
.downMessage(downMessage)
|
.serialNumber(serialNumber)
|
.body(downMessage.getBody())
|
.code(downMessage.getCode())
|
.topicName(topicName).build();
|
//根据协议编码后数据
|
byte[] data = jsonProtocolService.encode(encodeData, null);
|
message.setMessage(data);
|
message.setSerialNumber(serialNumber);
|
message.setTopicName(topicName);
|
|
return message;
|
}
|
|
/**
|
* 服务(指令)下发
|
*/
|
@Override
|
public void funcSend(MQSendMessageBo bo) {
|
//如果协议编号为空，则获取
|
if (StringUtils.isEmpty(bo.getProtocolCode())) {
|
Product product = productService.selectProductByProductId(bo.getProductId());
|
//bo.setType(ThingsModelType.SERVICE);
|
bo.setProtocolCode(product.getProtocolCode());
|
bo.setTransport(product.getTransport());
|
}
|
|
//处理设备影子模式
|
if (null != bo.getIsShadow() && bo.getIsShadow()){
|
List<ThingsModelSimpleItem> dataList = new ArrayList<>();
|
bo.getValue().forEach((key,value) ->{
|
ThingsModelSimpleItem item = new ThingsModelSimpleItem();
|
item.setId(key);
|
item.setValue(value+"");
|
dataList.add(item);
|
});
|
ReportDataBo dataBo = new ReportDataBo();
|
dataBo.setDataList(dataList);
|
dataBo.setProductId(bo.getProductId());
|
dataBo.setSerialNumber(bo.getSerialNumber());
|
dataBo.setRuleEngine(false);
|
dataBo.setShadow(true);
|
dataBo.setSlaveId(bo.getSlaveId());
|
dataBo.setType(bo.getType().getCode());
|
dataHandler.reportData(dataBo);
|
return;
|
}
|
|
/* 下发服务数据存储对象*/
|
FunctionLog log = new FunctionLog();
|
log.setCreateTime(DateUtils.getNowDate());
|
log.setFunValue(bo.getValue().get(bo.getIdentifier()).toString());
|
log.setMessageId(bo.getMessageId());
|
log.setSerialNumber(bo.getSerialNumber());
|
log.setIdentify(bo.getIdentifier());
|
log.setShowValue(bo.getShowValue());
|
log.setFunType(1);
|
log.setModelName(bo.getModelName());
|
//兼容子设备
|
if (null != bo.getSlaveId()) {
|
PropertyDto thingModels = thingsModelService.getSingleThingModels(bo.getProductId(), bo.getIdentifier() + "#" + bo.getSlaveId());
|
log.setSerialNumber(bo.getSerialNumber() + "_" + bo.getSlaveId());
|
bo.setCode(ModbusCode.Write06);
|
if (!Objects.isNull(thingModels.getCode())){
|
bo.setCode(ModbusCode.getInstance(Integer.parseInt(thingModels.getCode())));
|
}
|
}
|
|
ServerType serverType = ServerType.explain(bo.getTransport());
|
Optional.ofNullable(serverType).orElseThrow(() -> new ServiceException("产品的传输协议编码为空!"));
|
/*下发服务数据处理对象*/
|
DeviceDownMessage downMessage = DeviceDownMessage.builder()
|
.messageId(bo.getMessageId())
|
.body(bo.getValue())
|
.serialNumber(bo.getSerialNumber())
|
.productId(bo.getProductId())
|
.timestamp(DateUtils.getTimestamp())
|
.identifier(bo.getIdentifier())
|
.slaveId(bo.getSlaveId())
|
.code(bo.getCode() == ModbusCode.Read01 ? ModbusCode.Write05 : ModbusCode.Write06)
|
.serverType(serverType)
|
.build();
|
switch (serverType) {
|
case MQTT:
|
//组建下发服务指令
|
InstructionsMessage instruction = buildMessage(downMessage, TopicType.FUNCTION_GET);
|
mqttClient.publish(instruction.getTopicName(), instruction.getMessage(), log);
|
MqttMessagePublishImpl.log.debug("=>服务下发,topic=[{}],指令=[{}]", instruction.getTopicName(),new String(instruction.getMessage()));
|
break;
|
|
}
|
}
|
|
/**
|
* OTA升级下发
|
*
|
* @param bo
|
*/
|
@Override
|
public void upGradeOTA(OtaUpgradeBo bo) {
|
|
}
|
|
@Override
|
public void sendFunctionMessage(DeviceReportBo bo) {
|
log.warn("=>功能指令下发,sendFunctionMessage bo=[{}]", bo);
|
Device device = deviceService.selectDeviceBySerialNumber(bo.getSerialNumber());
|
Optional.ofNullable(device).orElseThrow(()->new ServiceException("服务下发的设备:["+bo.getSerialNumber()+"]不存在"));
|
|
Product product = productService.selectProductByProductId(topicsUtils.parseProductId(bo.getTopicName()));
|
ServerType serverType = ServerType.explain(product.getTransport());
|
Optional.ofNullable(serverType).orElseThrow(() -> new ServiceException("产品的传输协议编码为空!"));
|
|
switch (serverType) {
|
case GB28181:
|
break;
|
}
|
}
|
|
/**
|
* 1.发布设备状态
|
*/
|
@Override
|
public void publishStatus(Long productId, String deviceNum, int deviceStatus, int isShadow, int rssi) {
|
String message = "{\"status\":" + deviceStatus + ",\"isShadow\":" + isShadow + ",\"rssi\":" + rssi + "}";
|
String topic = topicsUtils.buildTopic(productId, deviceNum, TopicType.STATUS_POST);
|
mqttClient.publish(1, false, topic, message);
|
}
|
|
|
/**
|
* 2.发布设备信息
|
*/
|
@Override
|
public void publishInfo(Long productId, String deviceNum) {
|
String topic = topicsUtils.buildTopic(productId, deviceNum, TopicType.INFO_GET);
|
mqttClient.publish(1, false, topic, "");
|
}
|
|
/**
|
* 3.发布时钟同步信息
|
*
|
* @param bo 数据模型
|
*/
|
public void publishNtp(ReportDataBo bo) {
|
NtpModel ntpModel = JSON.parseObject(bo.getMessage(), NtpModel.class);
|
ntpModel.setServerRecvTime(System.currentTimeMillis());
|
ntpModel.setServerSendTime(System.currentTimeMillis());
|
String topic = topicsUtils.buildTopic(bo.getProductId(), bo.getSerialNumber(), TopicType.NTP_GET);
|
mqttClient.publish(1, false, topic, JSON.toJSONString(ntpModel));
|
}
|
|
/**
|
* 4.发布属性
|
* delay 延时，秒为单位
|
*/
|
@Override
|
public void publishProperty(Long productId, String deviceNum, List<ThingsModelSimpleItem> thingsList, int delay) {
|
String pre = "";
|
if (delay > 0) {
|
pre = "$delayed/" + String.valueOf(delay) + "/";
|
}
|
String topic = topicsUtils.buildTopic(productId, deviceNum, TopicType.FUNCTION_GET);
|
if (thingsList == null) {
|
mqttClient.publish(1, true, topic, "");
|
} else {
|
mqttClient.publish(1, true, topic, JSON.toJSONString(thingsList));
|
}
|
}
|
|
/**
|
* 5.发布功能
|
* delay 延时，秒为单位
|
*/
|
@Override
|
public void publishFunction(Long productId, String deviceNum, List<ThingsModelSimpleItem> thingsList, int delay) {
|
String pre = "";
|
if (delay > 0) {
|
pre = "$delayed/" + String.valueOf(delay) + "/";
|
}
|
String topic = topicsUtils.buildTopic(productId, deviceNum, TopicType.FUNCTION_GET);
|
if (thingsList == null) {
|
mqttClient.publish(1, true, topic, "");
|
} else {
|
mqttClient.publish(1, true, topic, JSON.toJSONString(thingsList));
|
}
|
|
}
|
|
/**
|
* 设备数据同步
|
*
|
* @param deviceNumber 设备编号
|
* @return 设备
|
*/
|
public Device deviceSynchronization(String deviceNumber) {
|
Device device = deviceService.selectDeviceBySerialNumber(deviceNumber);
|
// 1-未激活，2-禁用，3-在线，4-离线
|
if (device.getStatus() == 3) {
|
device.setStatus(4);
|
deviceService.updateDeviceStatus(device);
|
// 发布设备信息
|
publishInfo(device.getProductId(), device.getSerialNumber());
|
}
|
return device;
|
}
|
|
|
/**
|
* 发送模拟设备到WS
|
*/
|
public void sendSimulationWs(MqttBo send ,MqttBo receive,String topic){
|
PushMessageBo messageBo = new PushMessageBo();
|
messageBo.setTopic(topic);
|
JSONArray array = new JSONArray();
|
send.setDirection("send");
|
send.setTs(DateUtils.getNowDate());
|
receive.setTs(DateUtils.getNowDate());
|
receive.setDirection("receive");
|
array.add(send);
|
array.add(receive);
|
messageBo.setMessage(array.toJSONString());
|
remoteManager.pushCommon(messageBo);
|
}
|
|
public byte[] CRC(byte[] source) {
|
source[2] = (byte)((int) source[2] * 2);
|
byte[] result = new byte[source.length + 2];
|
byte[] crc16Byte = CRC16Utils.getCrc16Byte(source);
|
System.arraycopy(source, 0, result, 0, source.length);
|
System.arraycopy(crc16Byte, 0, result, result.length - 2, 2);
|
return result;
|
}
|
|
|
/**
|
* 搭建消息
|
*
|
* @param bo
|
* @return
|
*/
|
private DeviceDownMessage buildMessage(OtaUpgradeBo bo) {
|
String messageId = String.valueOf(snowflakeIdWorker.nextId());
|
bo.setMessageId(messageId);
|
bo.setOtaUrl("http://" + IpUtils.getHostIp()+bo.getOtaUrl());
|
return DeviceDownMessage.builder()
|
.productId(bo.getProductId())
|
.serialNumber(bo.getSerialNumber())
|
.body(JSON.toJSON(bo))
|
.timestamp(DateUtils.getTimestamp())
|
.messageId(messageId)
|
.build();
|
|
}
|
|
}
|
