guzijian / wvp-GB28181-pro · Commits

Commit 08c2fa45f7f5e6626f83270363a0d16f7d752d4c

Authored by 648540858
Committed by GitHub
2 parents e3be7963 bea5b40b

Merge pull request #771 from mrjackwang/wvp-28181-2.0 

修复历史录像下载问题，查询历史录像问题
Inline Side-by-side
Showing 7 changed files with 110 additions and 37 deletions
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java
  View file @08c2fa4
... ... @@ -122,7 +122,7 @@ public interface ISIPCommander {
122 122 */
123 123 void downloadStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,
124 124 String startTime, String endTime, int downloadSpeed, InviteStreamCallback inviteStreamCallback, InviteStreamCallback hookEvent,
125   - SipSubscribe.Event errorEvent) throws InvalidArgumentException, SipException, ParseException;
  125 + SipSubscribe.Event errorEvent,SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException;
126 126  
127 127 /**
128 128 * 视频流停止
... ... @@ -221,7 +221,6 @@ public interface ISIPCommander {
221 221 *
222 222 * @param device 视频设备
223 223 * @param channelId 通道id，非通道则是设备本身
224   - * @param frontCmd 上级平台的指令，如果存在则直接下发
225 224 * @param enabled 看守位使能：1 = 开启，0 = 关闭
226 225 * @param resetTime 自动归位时间间隔，开启看守位时使用，单位:秒(s)
227 226 * @param presetIndex 调用预置位编号，开启看守位时使用，取值范围0~255
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
  View file @08c2fa4
... ... @@ -29,7 +29,6 @@ import org.springframework.beans.factory.annotation.Autowired;
29 29 import org.springframework.context.annotation.DependsOn;
30 30 import org.springframework.stereotype.Component;
31 31 import org.springframework.util.ObjectUtils;
32   -import org.springframework.util.StringUtils;
33 32  
34 33 import javax.sip.InvalidArgumentException;
35 34 import javax.sip.ResponseEvent;
... ... @@ -471,8 +470,9 @@ public class SIPCommander implements ISIPCommander {
471 470 */
472 471 @Override
473 472 public void downloadStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,
474   - String startTime, String endTime, int downloadSpeed, InviteStreamCallback inviteStreamCallback, InviteStreamCallback hookEvent,
475   - SipSubscribe.Event errorEvent) throws InvalidArgumentException, SipException, ParseException {
  473 + String startTime, String endTime, int downloadSpeed,
  474 + InviteStreamCallback inviteStreamCallback, InviteStreamCallback hookEvent,
  475 + SipSubscribe.Event errorEvent,SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException {
476 476  
477 477 logger.info("{} 分配的ZLM为: {} [{}:{}]", ssrcInfo.getStream(), mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort());
478 478 String sdpIp;
... ... @@ -541,11 +541,14 @@ public class SIPCommander implements ISIPCommander {
541 541 content.append("a=downloadspeed:" + downloadSpeed + "\r\n");
542 542  
543 543 content.append("y=" + ssrcInfo.getSsrc() + "\r\n");//ssrc
544   -
  544 + logger.debug("此时请求下载信令的ssrc===>{}",ssrcInfo.getSsrc());
545 545 HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, null, mediaServerItem.getId());
546 546 // 添加订阅
  547 + CallIdHeader newCallIdHeader = sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()), device.getTransport());
  548 + String callId=newCallIdHeader.getCallId();
547 549 subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json) -> {
548   - hookEvent.call(new InviteStreamInfo(mediaServerItem, json,sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()).getCallId(), "rtp", ssrcInfo.getStream()));
  550 + logger.debug("sipc 添加订阅===callId {}",callId);
  551 + hookEvent.call(new InviteStreamInfo(mediaServerItem, json,callId, "rtp", ssrcInfo.getStream()));
549 552 subscribe.removeSubscribe(hookSubscribe);
550 553 hookSubscribe.getContent().put("regist", false);
551 554 hookSubscribe.getContent().put("schema", "rtsp");
... ... @@ -554,7 +557,7 @@ public class SIPCommander implements ISIPCommander {
554 557 (MediaServerItem mediaServerItemForEnd, JSONObject jsonForEnd) -> {
555 558 logger.info("[录像]下载结束， 发送BYE");
556 559 try {
557   - streamByeCmd(device, channelId, ssrcInfo.getStream(),sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()).getCallId());
  560 + streamByeCmd(device, channelId, ssrcInfo.getStream(),callId);
558 561 } catch (InvalidArgumentException | ParseException | SipException |
559 562 SsrcTransactionNotFoundException e) {
560 563 logger.error("[录像]下载结束， 发送BYE失败 {}", e.getMessage());
... ... @@ -562,15 +565,24 @@ public class SIPCommander implements ISIPCommander {
562 565 });
563 566 });
564 567  
565   - Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), null, SipUtils.getNewFromTag(), null,sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()), ssrcInfo.getSsrc());
  568 + Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), null, SipUtils.getNewFromTag(), null,newCallIdHeader, ssrcInfo.getSsrc());
566 569 if (inviteStreamCallback != null) {
567   - inviteStreamCallback.call(new InviteStreamInfo(mediaServerItem, null,sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()).getCallId(), "rtp", ssrcInfo.getStream()));
  570 + inviteStreamCallback.call(new InviteStreamInfo(mediaServerItem, null,callId, "rtp", ssrcInfo.getStream()));
568 571 }
569 572  
570   - sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), request, errorEvent, okEvent -> {
571   - ResponseEvent responseEvent = (ResponseEvent) okEvent.event;
  573 + sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), request, errorEvent, event -> {
  574 + ResponseEvent responseEvent = (ResponseEvent) event.event;
572 575 SIPResponse response = (SIPResponse) responseEvent.getResponse();
573   - streamSession.put(device.getDeviceId(), channelId, response.getCallIdHeader().getCallId(), ssrcInfo.getStream(), ssrcInfo.getSsrc(), mediaServerItem.getId(), response, VideoStreamSessionManager.SessionType.download);
  576 + String contentString =new String(response.getRawContent());
  577 + int ssrcIndex = contentString.indexOf("y=");
  578 + String ssrc=ssrcInfo.getSsrc();
  579 + if (ssrcIndex >= 0) {
  580 + ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12);
  581 + }
  582 + logger.debug("接收到的下载响应ssrc====>{}",ssrcInfo.getSsrc());
  583 + logger.debug("接收到的下载响应ssrc====>{}",ssrc);
  584 + streamSession.put(device.getDeviceId(), channelId, response.getCallIdHeader().getCallId(), ssrcInfo.getStream(), ssrc, mediaServerItem.getId(), response, VideoStreamSessionManager.SessionType.download);
  585 + okEvent.response(event);
574 586 });
575 587 }
576 588  
... ... @@ -802,7 +814,6 @@ public class SIPCommander implements ISIPCommander {
802 814 *
803 815 * @param device 视频设备
804 816 * @param channelId 通道id，非通道则是设备本身
805   - * @param frontCmd 上级平台的指令，如果存在则直接下发
806 817 * @param enabled 看守位使能：1 = 开启，0 = 关闭
807 818 * @param resetTime 自动归位时间间隔，开启看守位时使用，单位:秒(s)
808 819 * @param presetIndex 调用预置位编号，开启看守位时使用，取值范围0~255
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java
  View file @08c2fa4
... ... @@ -165,7 +165,10 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent
165 165 // 对数据进行排序
166 166 if(recordInfo!=null && recordInfo.getRecordList()!=null) {
167 167 Collections.sort(recordInfo.getRecordList());
  168 + }else{
  169 + recordInfo.setRecordList(new ArrayList<>());
168 170 }
  171 +
169 172 RequestMessage msg = new RequestMessage();
170 173 msg.setKey(key);
171 174 msg.setData(recordInfo);
... ...
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
  View file @08c2fa4
... ... @@ -635,23 +635,75 @@ public class PlayServiceImpl implements IPlayService {
635 635 hookCallBack.call(downloadResult);
636 636 streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
637 637 };
638   -
  638 + InviteStreamCallback hookEvent = (InviteStreamInfo inviteStreamInfo) -> {
  639 + logger.info("收到订阅消息： " + inviteStreamInfo.getCallId());
  640 + dynamicTask.stop(downLoadTimeOutTaskKey);
  641 + StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId);
  642 + streamInfo.setStartTime(startTime);
  643 + streamInfo.setEndTime(endTime);
  644 + redisCatchStorage.startDownload(streamInfo, inviteStreamInfo.getCallId());
  645 + downloadResult.setCode(ErrorCode.SUCCESS.getCode());
  646 + downloadResult.setMsg(ErrorCode.SUCCESS.getMsg());
  647 + downloadResult.setData(streamInfo);
  648 + downloadResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem());
  649 + downloadResult.setResponse(inviteStreamInfo.getResponse());
  650 + hookCallBack.call(downloadResult);
  651 + };
639 652 try {
640 653 cmder.downloadStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, downloadSpeed, infoCallBack,
641   - inviteStreamInfo -> {
642   - logger.info("收到订阅消息： " + inviteStreamInfo.getResponse().toJSONString());
643   - dynamicTask.stop(downLoadTimeOutTaskKey);
644   - StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId);
645   - streamInfo.setStartTime(startTime);
646   - streamInfo.setEndTime(endTime);
647   - redisCatchStorage.startDownload(streamInfo, inviteStreamInfo.getCallId());
648   - downloadResult.setCode(ErrorCode.SUCCESS.getCode());
649   - downloadResult.setMsg(ErrorCode.SUCCESS.getMsg());
650   - downloadResult.setData(streamInfo);
651   - downloadResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem());
652   - downloadResult.setResponse(inviteStreamInfo.getResponse());
653   - hookCallBack.call(downloadResult);
654   - }, errorEvent);
  654 + hookEvent, errorEvent, eventResult ->
  655 + {
  656 + if (eventResult.type == SipSubscribe.EventResultType.response) {
  657 + ResponseEvent responseEvent = (ResponseEvent) eventResult.event;
  658 + String contentString = new String(responseEvent.getResponse().getRawContent());
  659 + // 获取ssrc
  660 + int ssrcIndex = contentString.indexOf("y=");
  661 + // 检查是否有y字段
  662 + if (ssrcIndex >= 0) {
  663 + //ssrc规定长度为10字节，不取余下长度以避免后续还有“f=”字段 TODO 后续对不规范的非10位ssrc兼容
  664 + String ssrcInResponse = contentString.substring(ssrcIndex + 2, ssrcIndex + 12);
  665 + // 查询到ssrc不一致且开启了ssrc校验则需要针对处理
  666 + if (ssrcInfo.getSsrc().equals(ssrcInResponse)) {
  667 + return;
  668 + }
  669 + logger.info("[回放消息] 收到invite 200, 发现下级自定义了ssrc: {}", ssrcInResponse);
  670 + if (!mediaServerItem.isRtpEnable() || device.isSsrcCheck()) {
  671 + logger.info("[回放消息] SSRC修正 {}->{}", ssrcInfo.getSsrc(), ssrcInResponse);
  672 +
  673 + if (!mediaServerItem.getSsrcConfig().checkSsrc(ssrcInResponse)) {
  674 + // ssrc 不可用
  675 + // 释放ssrc
  676 + mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
  677 + streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
  678 + eventResult.msg = "下级自定义了ssrc,但是此ssrc不可用";
  679 + eventResult.statusCode = 400;
  680 + errorEvent.response(eventResult);
  681 + return;
  682 + }
  683 +
  684 + // 单端口模式streamId也有变化，需要重新设置监听
  685 + if (!mediaServerItem.isRtpEnable()) {
  686 + // 添加订阅
  687 + HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());
  688 + subscribe.removeSubscribe(hookSubscribe);
  689 + hookSubscribe.getContent().put("stream", String.format("%08x", Integer.parseInt(ssrcInResponse)).toUpperCase());
  690 + subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response) -> {
  691 + logger.info("[ZLM HOOK] ssrc修正后收到订阅消息： " + response.toJSONString());
  692 + dynamicTask.stop(downLoadTimeOutTaskKey);
  693 + // hook响应
  694 + onPublishHandlerForPlayback(mediaServerItemInUse, response, device.getDeviceId(), channelId, hookCallBack);
  695 + hookEvent.call(new InviteStreamInfo(mediaServerItem, null, eventResult.callId, "rtp", ssrcInfo.getStream()));
  696 + });
  697 + }
  698 + // 关闭rtp server
  699 + mediaServerService.closeRTPServer(mediaServerItem, ssrcInfo.getStream());
  700 + // 重新开启ssrc server
  701 + mediaServerService.openRTPServer(mediaServerItem, ssrcInfo.getStream(), ssrcInResponse, device.isSsrcCheck(), true, ssrcInfo.getPort());
  702 + }
  703 + }
  704 + }
  705 +
  706 + });
655 707 } catch (InvalidArgumentException | SipException | ParseException e) {
656 708 logger.error("[命令发送失败] 录像下载: {}", e.getMessage());
657 709  
... ...
src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java
  View file @08c2fa4
... ... @@ -316,10 +316,10 @@ public interface DeviceChannelMapper {
316 316 "select * " +
317 317 "from device_channel " +
318 318 "where deviceId=#{deviceId}" +
319   - " <if test='parentId != null and length != null' > and parentId = #{parentId} or left(channelId, #{parentId.length()}) = #{parentId} and length(channelId)=#{length} </if>" +
  319 + " <if test='parentId != null and length != null' > and parentId = #{parentId} or left(channelId, LENGTH(#{parentId})) = #{parentId} and length(channelId)=#{length} </if>" +
320 320 " <if test='parentId == null and length != null' > and parentId = #{parentId} or length(channelId)=#{length} </if>" +
321 321 " <if test='parentId == null and length == null' > and parentId = #{parentId} </if>" +
322   - " <if test='parentId != null and length == null' > and parentId = #{parentId} or left(channelId, #{parentId.length()}) = #{parentId} </if>" +
  322 + " <if test='parentId != null and length == null' > and parentId = #{parentId} or left(channelId, LENGTH(#{parentId})) = #{parentId} </if>" +
323 323 " </script>"})
324 324 List<DeviceChannel> getChannelsWithCivilCodeAndLength(String deviceId, String parentId, Integer length);
325 325  
... ...
src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java
  View file @08c2fa4
... ... @@ -177,12 +177,14 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
177 177 @Override
178 178 public boolean startDownload(StreamInfo stream, String callId) {
179 179 boolean result;
  180 + String key=String.format("%S_%s_%s_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX,
  181 + userSetting.getServerId(), stream.getMediaServerId(), stream.getDeviceID(), stream.getChannelId(), stream.getStream(), callId);
180 182 if (stream.getProgress() == 1) {
181   - result = RedisUtil.set(String.format("%S_%s_%s_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX,
182   - userSetting.getServerId(), stream.getMediaServerId(), stream.getDeviceID(), stream.getChannelId(), stream.getStream(), callId), stream);
  183 + logger.debug("添加下载缓存==已完成下载=》{}",key);
  184 + result = RedisUtil.set(key, stream);
183 185 }else {
184   - result = RedisUtil.set(String.format("%S_%s_%s_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX,
185   - userSetting.getServerId(), stream.getMediaServerId(), stream.getDeviceID(), stream.getChannelId(), stream.getStream(), callId), stream, 60*60);
  186 + logger.debug("添加下载缓存==未完成下载=》{}",key);
  187 + result = RedisUtil.set(key, stream, 60*60);
186 188 }
187 189 return result;
188 190 }
... ... @@ -617,7 +619,7 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
617 619 stream,
618 620 callId
619 621 );
620   - List<Object> streamInfoScan = RedisUtil.scan(key);
  622 + List<Object> streamInfoScan = RedisUtil.scan2(key);
621 623 if (streamInfoScan.size() > 0) {
622 624 return (StreamInfo) RedisUtil.get((String) streamInfoScan.get(0));
623 625 }else {
... ...
src/main/java/com/genersoft/iot/vmp/utils/redis/RedisUtil.java
  View file @08c2fa4
... ... @@ -881,7 +881,13 @@ public class RedisUtil {
881 881  
882 882 return new ArrayList<>(resultKeys);
883 883 }
884   -
  884 + public static List<Object> scan2(String query) {
  885 + if (redisTemplate == null) {
  886 + redisTemplate = SpringBeanFactory.getBean("redisTemplate");
  887 + }
  888 + Set<String> keys = redisTemplate.keys(query);
  889 + return new ArrayList<>(keys);
  890 + }
885 891 // ============================== 消息发送与订阅 ==============================
886 892 public static void convertAndSend(String channel, JSONObject msg) {
887 893 if (redisTemplate == null) {
... ...