forked from uav/ard-uav-server
ard-sample/src/main/java/com/ard/sample/manage/service/impl/DockOnlineServiceImpl.java
@@ -1,33 +1,170 @@ package com.ard.sample.manage.service.impl; import com.ard.common.core.domain.model.LoginUser; import com.ard.common.core.redis.RedisConst; import com.ard.common.core.redis.RedisOpsUtils; import com.ard.common.utils.SecurityUtils; import com.ard.sample.common.model.ResponseResult; 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.manage.model.dto.DeviceDTO; import com.ard.sample.manage.model.dto.LiveDTO; import com.ard.sample.manage.model.dto.LiveTypeDTO; import com.ard.sample.manage.service.IDeviceService; import com.ard.sample.manage.service.ILiveStreamService; import com.fasterxml.jackson.databind.ObjectMapper; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import javax.annotation.Resource; import java.net.URL; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; @Slf4j @Component public class DockOnlineServiceImpl { @Autowired private IDeviceService _deviceService; @Autowired private DockWebSocketHandler _dockWebSocketHandler; @Autowired protected ObjectMapper _objectMapper; @Autowired private LiveStreamHandler _liveStreamHandler; @Autowired private ILiveStreamService _liveStreamService; /** * 定时任务，每5秒取一次机场设备的在线状态，如已离线更新状态 * 定时任务，每2秒取一次机场设备的在线状态，如已离线更新状态 * */ @Scheduled(cron = "0/5 * * * * ?") @Scheduled(cron = "0/2 * * * * ?") public void cronOnlineTask() { List<DeviceDTO> devicesList = _deviceService.getDevicesTopoForWeb(); LoginUser loginUser = SecurityUtils.getLoginUser(); for (DeviceDTO device : devicesList) { String key = "online:" + device.getDeviceSn(); //String key = "online:" + device.getDeviceSn(); //System.out.println("在线状态："+RedisOpsUtils.checkExist(key)); // 更新机场设备状态："3"-"设备已离线"， if(!RedisOpsUtils.checkExist(key)){ if(!_deviceService.checkDeviceOnline(device.getDeviceSn())){ _deviceService.updateWorkType(device.getDeviceSn(), 3L); RedisOpsUtils.set(RedisConst.OSD_STATUS_PREFIX + device.getDeviceSn(), 3); RedisOpsUtils.del(RedisConst.WEBSOCKET_VIDEO_ADDRESS_PREFIX + device.getDeviceSn()); RedisOpsUtils.del(RedisConst.COVER_STATE_PREFIX + device.getDeviceSn()); } else { StartLiveStreaming(device.getDeviceSn()); } if(!_deviceService.checkDeviceOnline(device.getChildDeviceSn())){ _deviceService.updateWorkType(device.getChildDeviceSn(), 6L); RedisOpsUtils.del(RedisConst.WEBSOCKET_VIDEO_ADDRESS_PREFIX + device.getChildDeviceSn()); RedisOpsUtils.del(RedisConst.COVER_STATE_PREFIX + device.getChildDeviceSn()); } else{ StartLiveStreaming(device.getChildDeviceSn()); } } } /** * 定时任务，每1秒读一次redis的机场设备的状态，并发送到websocket客户端 * */ //@Scheduled(cron = "0/1 * * * * ?") @Scheduled(initialDelay = 2, fixedRate = 1, timeUnit = TimeUnit.SECONDS) public void sendDeviceStatus(){ // 测试 机场oss读取文件 // URL url = _ossService.getObjectUrl("kmz","/2025/11/25/新建航点航线(1)_20251125161322A001.kmz"); // log.info("机场任务航线文件url地址：{}", url.toString()); // 发送设备状态到websocket客户端 List<DeviceDTO> devicesList = _deviceService.getDevicesTopoForWeb(); for (DeviceDTO device : devicesList) { String dockStatus = ""; String droneStatus = ""; // 判断redis中是否存在设备状态数据 if(RedisOpsUtils.checkExist(RedisConst.OSD_STATUS_PREFIX + device.getDeviceSn())){ dockStatus = String.valueOf(RedisOpsUtils.get(RedisConst.OSD_STATUS_PREFIX + device.getDeviceSn())); } if(RedisOpsUtils.checkExist(RedisConst.OSD_STATUS_PREFIX + device.getChildDeviceSn())){ droneStatus = String.valueOf(RedisOpsUtils.get(RedisConst.OSD_STATUS_PREFIX + device.getChildDeviceSn())); } String json = getJSONString(MessageType.STATE.getType(), device.getDeviceSn(), dockStatus, droneStatus); // 发送给websocket客户端 _dockWebSocketHandler.sendMessageToSubscribers(device.getDeviceSn(), MessageType.STATE.getType(), json, ""); // log.info("发送设备状态{}到websocket客户端：{}", MessageType.STATE.getType(), json); } } public String getJSONString(String messageType, String dockSn, String dockStatus, String droneStatus){ Map<String, Object> json = new HashMap<>(); json.put("code", 200); json.put("gateway", dockSn); json.put("type", messageType); json.put("message", "success"); Map<String, String> data = new HashMap<>(); data.put("dockstate", dockStatus); data.put("dronestate", droneStatus); json.put("data", data); // 将Map转换为JSON字符串 String jsonString = ""; try{ jsonString = _objectMapper.writeValueAsString(json); } catch (Exception e){ e.printStackTrace(); } return jsonString; } public void StartLiveStreaming(String sn) { //调用开始直播 ResponseResult rlive = null; String resultLive = ""; try { // 获取直播参数 LiveTypeDTO param = _liveStreamHandler.getLiveParam(sn); // 如果存在直播则读取直播地址返回 if (RedisOpsUtils.checkExist(RedisConst.WEBSOCKET_VIDEO_ADDRESS_PREFIX + sn)) { resultLive = String.valueOf(RedisOpsUtils.get(RedisConst.WEBSOCKET_VIDEO_ADDRESS_PREFIX + sn)); // log.info("SN: {}-redis中存在相机的播放地址信息，直接返回地址！", sn); } else { // log.info("SN: {}-redis中未找到相机的播放地址信息！停止直播后再开启直播", sn); _liveStreamHandler.StopLiveStreaming(sn); // 如果不存在直播，则开始直播，并返回直播地址 rlive = _liveStreamService.liveStart(param); if (rlive.getCode() == 0) { // 直播成功后，在redis中存一个标记，将videoid保留，留给停止直播使用 log.info("SN: {}-开启直播成功！", sn); rlive.setGateway(sn); rlive.setType(MessageType.LIVESTREAM.getType()); resultLive = _objectMapper.writeValueAsString(rlive); RedisOpsUtils.set(RedisConst.WEBSOCKET_VIDEO_ADDRESS_PREFIX + sn, resultLive); LiveDTO live = (LiveDTO) rlive.getData(); Boolean isWeb = _deviceService.updateWebrtcUrl(sn,live.getWebrtcUrl()); // if (isWeb) { // log.info("SN: {}-更新webrtc地址成功！", sn); // } } } } catch (Exception e) { log.info("订时任务推流-SN: {}-开启直播失败！原因：{}", sn, e.getMessage()); e.printStackTrace(); } } }
