diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java index f4bcbb64..4c147d0c 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java @@ -143,6 +143,14 @@ public interface ISIPCommander { * 回放倍速播放 */ void playSpeedCmd(Device device, StreamInfo streamInfo, Double speed); + + /** + * 回放控制 + * @param device + * @param streamInfo + * @param content + */ + void playbackControlCmd(Device device, StreamInfo streamInfo, String content,SipSubscribe.Event errorEvent, SipSubscribe.Event okEvent); /** * 语音广播 diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java index 7b144d16..0097ce0a 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java @@ -1828,6 +1828,43 @@ public class SIPCommander implements ISIPCommander { e.printStackTrace(); } } + + @Override + public void playbackControlCmd(Device device, StreamInfo streamInfo, String content,SipSubscribe.Event errorEvent, SipSubscribe.Event okEvent) { + try { + Request request = headerProvider.createInfoRequest(device, streamInfo, content); + if (request == null) { + return; + } + logger.info(request.toString()); + ClientTransaction clientTransaction = null; + if ("TCP".equals(device.getTransport())) { + clientTransaction = tcpSipProvider.getNewClientTransaction(request); + } else if ("UDP".equals(device.getTransport())) { + clientTransaction = udpSipProvider.getNewClientTransaction(request); + } + CallIdHeader callIdHeader = (CallIdHeader)request.getHeader(CallIdHeader.NAME); + if(errorEvent != null) { + sipSubscribe.addErrorSubscribe(callIdHeader.getCallId(), (eventResult -> { + errorEvent.response(eventResult); + sipSubscribe.removeErrorSubscribe(eventResult.callId); + sipSubscribe.removeOkSubscribe(eventResult.callId); + })); + } + + if(okEvent != null) { + sipSubscribe.addOkSubscribe(callIdHeader.getCallId(), eventResult -> { + okEvent.response(eventResult); + sipSubscribe.removeOkSubscribe(eventResult.callId); + sipSubscribe.removeErrorSubscribe(eventResult.callId); + }); + } + clientTransaction.sendRequest(); + + } catch (SipException | ParseException | InvalidArgumentException e) { + e.printStackTrace(); + } + } @Override public boolean sendAlarmMessage(Device device, DeviceAlarm deviceAlarm) { diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/info/InfoRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/info/InfoRequestProcessor.java new file mode 100644 index 00000000..a1e98f9e --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/info/InfoRequestProcessor.java @@ -0,0 +1,143 @@ +package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.info; + +import com.genersoft.iot.vmp.common.StreamInfo; +import com.genersoft.iot.vmp.gb28181.bean.*; +import com.genersoft.iot.vmp.gb28181.event.SipSubscribe; +import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager; +import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver; +import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander; +import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor; +import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent; +import com.genersoft.iot.vmp.gb28181.utils.SipUtils; +import com.genersoft.iot.vmp.storager.IRedisCatchStorage; +import com.genersoft.iot.vmp.storager.IVideoManagerStorage; +import gov.nist.javax.sip.message.SIPRequest; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; +import javax.sip.InvalidArgumentException; +import javax.sip.RequestEvent; +import javax.sip.SipException; +import javax.sip.header.*; +import javax.sip.message.Response; +import java.text.ParseException; + +@Component +public class InfoRequestProcessor extends SIPRequestProcessorParent implements InitializingBean, ISIPRequestProcessor { + + private final static Logger logger = LoggerFactory.getLogger(InfoRequestProcessor.class); + + private final String method = "INFO"; + + @Autowired + private SIPProcessorObserver sipProcessorObserver; + + @Autowired + private IVideoManagerStorage storage; + + @Autowired + private SipSubscribe sipSubscribe; + + @Autowired + private IRedisCatchStorage redisCatchStorage; + + @Autowired + private IVideoManagerStorage storager; + + @Autowired + private SIPCommander cmder; + + @Autowired + private VideoStreamSessionManager sessionManager; + + @Override + public void afterPropertiesSet() throws Exception { + // 添加消息处理的订阅 + sipProcessorObserver.addRequestProcessor(method, this); + } + + @Override + public void process(RequestEvent evt) { + logger.debug("接收到消息：" + evt.getRequest()); + String deviceId = SipUtils.getUserIdFromFromHeader(evt.getRequest()); + CallIdHeader callIdHeader = (CallIdHeader)evt.getRequest().getHeader(CallIdHeader.NAME); + // 先从会话内查找 + SsrcTransaction ssrcTransaction = sessionManager.getSsrcTransaction(null, null, callIdHeader.getCallId(), null); + if (ssrcTransaction != null) { // 兼容海康 媒体通知 消息from字段不是设备ID的问题 + deviceId = ssrcTransaction.getDeviceId(); + } + // 查询设备是否存在 + Device device = redisCatchStorage.getDevice(deviceId); + // 查询上级平台是否存在 + ParentPlatform parentPlatform = storage.queryParentPlatByServerGBId(deviceId); + try { + if (device != null && parentPlatform != null) { + logger.warn("[重复]平台与设备编号重复：{}", deviceId); + SIPRequest request = (SIPRequest) evt.getRequest(); + String hostAddress = request.getRemoteAddress().getHostAddress(); + int remotePort = request.getRemotePort(); + if (device.getHostAddress().equals(hostAddress + ":" + remotePort)) { + parentPlatform = null; + }else { + device = null; + } + } + if (device == null && parentPlatform == null) { + // 不存在则回复404 + responseAck(evt, Response.NOT_FOUND, "device "+ deviceId +" not found"); + logger.warn("[设备未找到 ]： {}", deviceId); + if (sipSubscribe.getErrorSubscribe(callIdHeader.getCallId()) != null){ + SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(new DeviceNotFoundEvent(evt.getDialog())); + sipSubscribe.getErrorSubscribe(callIdHeader.getCallId()).response(eventResult); + }; + }else { + ContentTypeHeader header = (ContentTypeHeader)evt.getRequest().getHeader(ContentTypeHeader.NAME); + String contentType = header.getContentType(); + String contentSubType = header.getContentSubType(); + if ("Application".equals(contentType) && "MANSRTSP".equals(contentSubType)) { + SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, null, null, callIdHeader.getCallId()); + String streamId = sendRtpItem.getStreamId(); + StreamInfo streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null); + if (null == streamInfo) { + responseAck(evt, Response.NOT_FOUND, "stream " + streamId + " not found"); + return; + } + Device device1 = storager.queryVideoDevice(streamInfo.getDeviceID()); + cmder.playbackControlCmd(device1,streamInfo,new String(evt.getRequest().getRawContent()),eventResult -> { + // 失败的回复 + try { + responseAck(evt, eventResult.statusCode, eventResult.msg); + } catch (SipException e) { + e.printStackTrace(); + } catch (InvalidArgumentException e) { + e.printStackTrace(); + } catch (ParseException e) { + e.printStackTrace(); + } + }, eventResult -> { + // 成功的回复 + try { + responseAck(evt, eventResult.statusCode); + } catch (SipException e) { + e.printStackTrace(); + } catch (InvalidArgumentException e) { + e.printStackTrace(); + } catch (ParseException e) { + e.printStackTrace(); + } + }); + } + } + } catch (SipException e) { + logger.warn("SIP 回复错误", e); + } catch (InvalidArgumentException e) { + logger.warn("参数无效", e); + } catch (ParseException e) { + logger.warn("SIP回复时解析异常", e); + } + } + + +} diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/DeviceStatusResponseMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/DeviceStatusResponseMessageHandler.java index 1fed401a..17738a5f 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/DeviceStatusResponseMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/DeviceStatusResponseMessageHandler.java @@ -82,7 +82,7 @@ public class DeviceStatusResponseMessageHandler extends SIPRequestProcessorParen deviceService.offline(device.getDeviceId()); } RequestMessage msg = new RequestMessage(); - msg.setKey(DeferredResultHolder.CALLBACK_CMD_DEVICESTATUS + device.getDeviceId() + channelId); + msg.setKey(DeferredResultHolder.CALLBACK_CMD_DEVICESTATUS + device.getDeviceId()); msg.setData(json); deferredResultHolder.invokeAllResult(msg); } diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/device/DeviceQuery.java b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/device/DeviceQuery.java index 4c895378..5e1e40be 100644 --- a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/device/DeviceQuery.java +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/device/DeviceQuery.java @@ -342,6 +342,11 @@ public class DeviceQuery { Device device = storager.queryVideoDevice(deviceId); String uuid = UUID.randomUUID().toString(); String key = DeferredResultHolder.CALLBACK_CMD_DEVICESTATUS + deviceId; + DeferredResult> result = new DeferredResult>(2*1000L); + if(device == null) { + result.setResult(new ResponseEntity(String.format("设备%s不存在", deviceId),HttpStatus.OK)); + return result; + } cmder.deviceStatusQuery(device, event -> { RequestMessage msg = new RequestMessage(); msg.setId(uuid); @@ -349,7 +354,6 @@ public class DeviceQuery { msg.setData(String.format("获取设备状态失败，错误码： %s, %s", event.statusCode, event.msg)); resultHolder.invokeResult(msg); }); - DeferredResult> result = new DeferredResult>(2*1000L); result.onTimeout(()->{ logger.warn(String.format("获取设备状态超时")); // 释放rtpserver