package com.ard.sample.manage.service.impl;
|
|
import com.ard.sample.common.error.CommonErrorEnum;
|
import com.ard.sample.common.model.ResponseResult;
|
import com.ard.sample.component.mqtt.model.*;
|
import com.ard.sample.component.mqtt.service.IMessageSenderService;
|
import com.ard.common.core.redis.RedisConst;
|
import com.ard.common.core.redis.RedisOpsUtils;
|
import com.ard.sample.component.websocket.enums.MessageType;
|
import com.ard.sample.component.websocket.handler.DockWebSocketHandler;
|
import com.ard.sample.component.websocket.handler.LiveStreamHandler;
|
import com.ard.sample.component.websocket.service.ISendMessageService;
|
import com.ard.sample.manage.model.dto.BasicDeviceProperty;
|
import com.ard.sample.manage.model.dto.DeviceDTO;
|
import com.ard.sample.manage.model.enums.RemoteControlMethodEnum;
|
import com.ard.sample.manage.model.param.RemoteDebugParam;
|
import com.ard.sample.manage.service.IControlService;
|
import com.ard.sample.manage.service.IDeviceService;
|
import com.fasterxml.jackson.core.type.TypeReference;
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.integration.annotation.ServiceActivator;
|
import org.springframework.integration.mqtt.support.MqttHeaders;
|
import org.springframework.messaging.MessageHeaders;
|
import org.springframework.stereotype.Service;
|
import java.util.Objects;
|
import java.util.UUID;
|
|
/**
|
* @author sean
|
* @version 1.2
|
* @date 2022/7/29
|
*/
|
@Service
|
@Slf4j
|
public class ControlServiceImpl implements IControlService {
|
|
@Autowired
|
private IMessageSenderService messageSenderService;
|
|
@Autowired
|
private DockWebSocketHandler _dockWebSocketHandler;
|
|
@Autowired
|
private ISendMessageService webSocketMessageService;
|
|
@Autowired
|
private IDeviceService _deviceService;
|
|
@Autowired
|
private LiveStreamHandler _liveStreamHandler;
|
|
// @Autowired
|
// private IWebSocketManageService webSocketManageService;
|
|
@Autowired
|
private IDeviceService deviceService;
|
|
@Autowired
|
private ObjectMapper mapper;
|
|
@Override
|
public ResponseResult controlDock(String sn, String serviceIdentifier, RemoteDebugParam param) {
|
RemoteControlMethodEnum controlMethodEnum = RemoteControlMethodEnum.find(serviceIdentifier);
|
if (RemoteControlMethodEnum.UNKNOWN == controlMethodEnum) {
|
return ResponseResult.error("The " + serviceIdentifier + " method does not exist.");
|
}
|
|
|
Object data = "";
|
// Add parameter validation.
|
if (Objects.nonNull(controlMethodEnum.getClazz())) {
|
if (Objects.isNull(param)) {
|
return ResponseResult.error(CommonErrorEnum.ILLEGAL_ARGUMENT);
|
}
|
BasicDeviceProperty basicDeviceProperty = null;
|
if(serviceIdentifier.equals("sdr_workmode_switch")){
|
basicDeviceProperty= mapper.convertValue(param.getLink_workmode(), controlMethodEnum.getClazz());
|
}
|
else
|
{
|
basicDeviceProperty = mapper.convertValue(param.getAction(), controlMethodEnum.getClazz());
|
}
|
if (!basicDeviceProperty.valid()) {
|
return ResponseResult.error(CommonErrorEnum.ILLEGAL_ARGUMENT);
|
}
|
data = basicDeviceProperty;
|
}
|
|
boolean isExist = deviceService.checkDeviceOnline(sn);
|
if (!isExist) {
|
return ResponseResult.error(CommonErrorEnum.DEVICE_OFFLINE.getErrorCode(), CommonErrorEnum.DEVICE_OFFLINE.getErrorMsg());
|
}
|
DeviceDTO device = (DeviceDTO) RedisOpsUtils.get(RedisConst.DEVICE_ONLINE_PREFIX + sn);
|
// 重启机场时清redis视频直播地址，机场重启不掉线。但是流停了。
|
if(RemoteControlMethodEnum.DEVICE_REBOOT == controlMethodEnum) {
|
// 重启机场时，读取舱盖状态，开舱盖不让重启，关舱盖重启
|
Integer cover_state = (Integer)RedisOpsUtils.get(RedisConst.COVER_STATE_OSD_PREFIX + device.getDeviceSn());
|
if(cover_state != null && cover_state != 0) {
|
return ResponseResult.error(111111,"请关闭舱盖再执行重启操作！");
|
}
|
RedisOpsUtils.del(RedisConst.WEBSOCKET_VIDEO_ADDRESS_PREFIX + sn);
|
RedisOpsUtils.del(RedisConst.COVER_STATE_PREFIX + sn);
|
RedisOpsUtils.del(RedisConst.WEBSOCKET_VIDEO_ADDRESS_PREFIX + device.getChildDeviceSn());
|
}
|
if(RemoteControlMethodEnum.DRONE_CLOSE == controlMethodEnum) {
|
// 飞行器关机时drone_close，关闭飞行器视频流
|
_liveStreamHandler.StopLiveStreaming(device.getChildDeviceSn());
|
RedisOpsUtils.del(RedisConst.WEBSOCKET_VIDEO_ADDRESS_PREFIX + device.getChildDeviceSn());
|
log.info("SN: {}-飞行器关机，停止直播并删除redis直播信息！", device.getChildDeviceSn());
|
}
|
String topic = TopicConst.THING_MODEL_PRE + TopicConst.PRODUCT + sn + TopicConst.SERVICES_SUF;
|
String bid = UUID.randomUUID().toString();
|
ServiceReply serviceReplyOpt = messageSenderService.publishWithReply(
|
topic, CommonTopicResponse.builder()
|
.tid(UUID.randomUUID().toString())
|
.bid(bid)
|
.method(serviceIdentifier)
|
.timestamp(System.currentTimeMillis())
|
.data(data)
|
.build());
|
|
ServiceReply<EventsOutputReceiver> serviceReply = mapper.convertValue(
|
serviceReplyOpt, new TypeReference<ServiceReply<EventsOutputReceiver>>() {});
|
if (ResponseResult.CODE_SUCCESS != serviceReply.getResult()) {
|
String errorMsg = CommonErrorEnum.getMsgByCode(serviceReply.getResult());
|
return ResponseResult.error(serviceReply.getResult(),errorMsg);
|
// return ResponseResult.error(serviceReply.getResult(),
|
// Objects.nonNull(serviceReply.getOutput()) ? serviceReply.getOutput().getStatus() : "errorCode: "
|
// + serviceReply.getResult());
|
}
|
if (controlMethodEnum.getProgress()) {
|
RedisOpsUtils.setWithExpire(serviceIdentifier + RedisConst.DELIMITER + bid, sn,
|
RedisConst.DEVICE_ALIVE_SECOND * RedisConst.DEVICE_ALIVE_SECOND);
|
}
|
// 执行操作成功后，更新摄像头内外镜头状态
|
if(RemoteControlMethodEnum.COVER_OPEN == controlMethodEnum) {
|
// 打开盖子,切换机场舱内摄像头镜头
|
//打开，摄像头为舱内；将摄像头状态存到redis，格式为：CONVERSTATE:设备SN：1（舱外）或0（舱内）
|
_deviceService.setRedisCameraPosition(sn,0);
|
}
|
if(RemoteControlMethodEnum.COVER_CLOSE == controlMethodEnum) {
|
// 关闭盖子,切换机场舱外摄像头镜头
|
//关闭，摄像头为舱外；将摄像头状态存到redis，格式为：CONVERSTATE:设备SN：1（舱外）或0（舱内）
|
_deviceService.setRedisCameraPosition(sn,1);
|
}
|
|
return ResponseResult.success();
|
}
|
|
@Override
|
@ServiceActivator(inputChannel = ChannelName.INBOUND_EVENTS_CONTROL_PROGRESS,
|
outputChannel = ChannelName.OUTBOUND)
|
public void handleControlProgress(CommonTopicReceiver receiver, MessageHeaders headers) {
|
String key = receiver.getMethod() + RedisConst.DELIMITER + receiver.getBid();
|
if (RedisOpsUtils.getExpire(key) <= 0) {
|
return;
|
}
|
// 更新设备列表状态--关闭舱盖中...
|
String method = receiver.getMethod();
|
String deviceSn = receiver.getGateway();
|
if(method.equals("cover_close")){
|
_deviceService.updateWorkType(deviceSn, 2L);
|
RedisOpsUtils.set(RedisConst.OSD_STATUS_PREFIX + deviceSn, 2);
|
}
|
String sn = RedisOpsUtils.get(key).toString();
|
receiver.setType(MessageType.PROGRESS.getType());
|
EventsReceiver<EventsOutputReceiver> eventsReceiver = mapper.convertValue(receiver.getData(),
|
new TypeReference<EventsReceiver<EventsOutputReceiver>>(){});
|
eventsReceiver.setBid(receiver.getBid());
|
eventsReceiver.setGateway(sn);
|
eventsReceiver.setType(MessageType.PROGRESS.getType());
|
eventsReceiver.setMethod(receiver.getMethod());
|
|
log.info("SN: {}, {} ===> Control progress: {}",
|
sn, receiver.getMethod(), eventsReceiver.getOutput().getProgress().toString());
|
|
if (eventsReceiver.getResult() != ResponseResult.CODE_SUCCESS) {
|
log.error("SN: {}, {} ===> Error code: {}", sn, receiver.getMethod(), eventsReceiver.getResult());
|
}
|
|
if (eventsReceiver.getOutput().getProgress().getPercent() == 100 ||
|
EventsResultStatusEnum.find(eventsReceiver.getOutput().getStatus()).getEnd()) {
|
RedisOpsUtils.del(key);
|
}
|
|
DeviceDTO device = (DeviceDTO) RedisOpsUtils.get(RedisConst.DEVICE_ONLINE_PREFIX + sn);
|
// webSocketMessageService.sendBatch(
|
// WebSocketManager.getValueWithWorkspaceAndUserType(
|
// device.getWorkspaceId(), UserTypeEnum.WEB.getVal()),
|
// CustomWebSocketMessage.builder()
|
// .data(eventsReceiver)
|
// .timestamp(System.currentTimeMillis())
|
// .bizCode(receiver.getMethod())
|
// .build());
|
// 更新设备列表状态--飞行器开机中...
|
if(method.equals("drone_open")){
|
_deviceService.updateWorkType(device.getChildDeviceSn(), 7L);
|
RedisOpsUtils.set(RedisConst.OSD_STATUS_PREFIX + device.getChildDeviceSn(), 7);
|
}
|
// 机场重启状态--ok后删除redis状态
|
if(method.equals("device_reboot")){
|
if(eventsReceiver.getOutput().getProgress().getPercent() == 100){
|
RedisOpsUtils.del(RedisConst.WEBSOCKET_VIDEO_ADDRESS_PREFIX + sn);
|
RedisOpsUtils.del(RedisConst.COVER_STATE_PREFIX + sn);
|
RedisOpsUtils.del(RedisConst.WEBSOCKET_VIDEO_ADDRESS_PREFIX + device.getChildDeviceSn());
|
}
|
}
|
|
|
String dockProgress ="";
|
try {
|
dockProgress = mapper.writeValueAsString(eventsReceiver);
|
}
|
catch (Exception e) {
|
e.printStackTrace();
|
}
|
// 设备控制进度消息websocket发给前端
|
_dockWebSocketHandler.sendMessageToSubscribers(sn, MessageType.PROGRESS.getType(), dockProgress, "");
|
|
if (receiver.getNeedReply() != null && receiver.getNeedReply() == 1) {
|
String topic = headers.get(MqttHeaders.RECEIVED_TOPIC) + TopicConst._REPLY_SUF;
|
messageSenderService.publish(topic,
|
CommonTopicResponse.builder()
|
.tid(receiver.getTid())
|
.bid(receiver.getBid())
|
.method(receiver.getMethod())
|
.timestamp(System.currentTimeMillis())
|
.data(RequestsReply.success())
|
.build());
|
}
|
}
|
}
|
