优化历史录像下载，目前已测试大华国标级联下载

wangyimeng
1 parent 8acd64ea
Showing 4 changed files with 49 additions and 30 deletions
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java
src/main/java/com/genersoft/iot/vmp/utils/redis/RedisUtil.java
@@ -541,11 +541,14 @@ public class SIPCommander implements ISIPCommander {
         content.append("a=downloadspeed:" + downloadSpeed + "\r\n");
         content.append("y=" + ssrcInfo.getSsrc() + "\r\n");//ssrc
-        
+        logger.debug("此时请求下载信令的ssrc===>{}",ssrcInfo.getSsrc());
         HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, null, mediaServerItem.getId());
         // 添加订阅
+        CallIdHeader newCallIdHeader = sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()), device.getTransport());
+        String callId=newCallIdHeader.getCallId();
         subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json) -> {
-            hookEvent.call(new InviteStreamInfo(mediaServerItem, json,sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()).getCallId(), "rtp", ssrcInfo.getStream()));
+            logger.debug("sipc 添加订阅===callId {}",callId);
+            hookEvent.call(new InviteStreamInfo(mediaServerItem, json,callId, "rtp", ssrcInfo.getStream()));
             subscribe.removeSubscribe(hookSubscribe);
             hookSubscribe.getContent().put("regist", false);
             hookSubscribe.getContent().put("schema", "rtsp");
@@ -554,7 +557,7 @@ public class SIPCommander implements ISIPCommander {
                     (MediaServerItem mediaServerItemForEnd, JSONObject jsonForEnd) -> {
                         logger.info("[录像]下载结束， 发送BYE");
                         try {
-                            streamByeCmd(device, channelId, ssrcInfo.getStream(),sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()).getCallId());
+                            streamByeCmd(device, channelId, ssrcInfo.getStream(),callId);
                         } catch (InvalidArgumentException | ParseException | SipException |
                                  SsrcTransactionNotFoundException e) {
                             logger.error("[录像]下载结束， 发送BYE失败 {}", e.getMessage());
@@ -562,15 +565,23 @@ public class SIPCommander implements ISIPCommander {
                     });
         });
-        Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), null, SipUtils.getNewFromTag(), null,sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()), ssrcInfo.getSsrc());
+        Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), null, SipUtils.getNewFromTag(), null,newCallIdHeader, ssrcInfo.getSsrc());
         if (inviteStreamCallback != null) {
-            inviteStreamCallback.call(new InviteStreamInfo(mediaServerItem, null,sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()).getCallId(), "rtp", ssrcInfo.getStream()));
+            inviteStreamCallback.call(new InviteStreamInfo(mediaServerItem, null,callId, "rtp", ssrcInfo.getStream()));
         }
         sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), request, errorEvent, event -> {
             ResponseEvent responseEvent = (ResponseEvent) event.event;
             SIPResponse response = (SIPResponse) responseEvent.getResponse();
-            streamSession.put(device.getDeviceId(), channelId, response.getCallIdHeader().getCallId(), ssrcInfo.getStream(), ssrcInfo.getSsrc(), mediaServerItem.getId(), response, VideoStreamSessionManager.SessionType.download);
+            String contentString =new String(response.getRawContent());
+            int ssrcIndex = contentString.indexOf("y=");
+            String ssrc=ssrcInfo.getSsrc();
+            if (ssrcIndex >= 0) {
+                ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12);
+            }
+            logger.debug("接收到的下载响应ssrc====>{}",ssrcInfo.getSsrc());
+            logger.debug("接收到的下载响应ssrc====>{}",ssrc);
+            streamSession.put(device.getDeviceId(), channelId, response.getCallIdHeader().getCallId(), ssrcInfo.getStream(), ssrc, mediaServerItem.getId(), response, VideoStreamSessionManager.SessionType.download);
             okEvent.response(event);
         });
     }
@@ -635,23 +635,23 @@ public class PlayServiceImpl implements IPlayService {
             hookCallBack.call(downloadResult);
             streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
         };
-
+        InviteStreamCallback hookEvent = (InviteStreamInfo inviteStreamInfo) -> {
+            logger.info("收到订阅消息： " + inviteStreamInfo.getCallId());
+            dynamicTask.stop(downLoadTimeOutTaskKey);
+            StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId);
+            streamInfo.setStartTime(startTime);
+            streamInfo.setEndTime(endTime);
+            redisCatchStorage.startDownload(streamInfo, inviteStreamInfo.getCallId());
+            downloadResult.setCode(ErrorCode.SUCCESS.getCode());
+            downloadResult.setMsg(ErrorCode.SUCCESS.getMsg());
+            downloadResult.setData(streamInfo);
+            downloadResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem());
+            downloadResult.setResponse(inviteStreamInfo.getResponse());
+            hookCallBack.call(downloadResult);
+        };
         try {
             cmder.downloadStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, downloadSpeed, infoCallBack,
-                    inviteStreamInfo -> {
-                        logger.info("收到订阅消息： " + inviteStreamInfo.getResponse().toJSONString());
-                        dynamicTask.stop(downLoadTimeOutTaskKey);
-                        StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId);
-                        streamInfo.setStartTime(startTime);
-                        streamInfo.setEndTime(endTime);
-                        redisCatchStorage.startDownload(streamInfo, inviteStreamInfo.getCallId());
-                        downloadResult.setCode(ErrorCode.SUCCESS.getCode());
-                        downloadResult.setMsg(ErrorCode.SUCCESS.getMsg());
-                        downloadResult.setData(streamInfo);
-                        downloadResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem());
-                        downloadResult.setResponse(inviteStreamInfo.getResponse());
-                        hookCallBack.call(downloadResult);
-                    }, errorEvent, eventResult ->
+                    hookEvent, errorEvent, eventResult ->
                     {
                         if (eventResult.type == SipSubscribe.EventResultType.response) {
                             ResponseEvent responseEvent = (ResponseEvent) eventResult.event;
@@ -690,9 +690,9 @@ public class PlayServiceImpl implements IPlayService {
                                         subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject response) -> {
                                             logger.info("[ZLM HOOK] ssrc修正后收到订阅消息： " + response.toJSONString());
                                             dynamicTask.stop(downLoadTimeOutTaskKey);
-                                            // hook响应，TODO 此处待处理
-//                                            onPublishHandlerForPlayback(mediaServerItemInUse, response, device.getDeviceId(), channelId, playBackCallback);
-//                                            hookCallBack.call(new InviteStreamInfo(mediaServerItem, null, eventResult.callId, "rtp", ssrcInfo.getStream()));
+                                            // hook响应
+                                            onPublishHandlerForPlayback(mediaServerItemInUse, response, device.getDeviceId(), channelId, hookCallBack);
+                                            hookEvent.call(new InviteStreamInfo(mediaServerItem, null, eventResult.callId, "rtp", ssrcInfo.getStream()));
                                         });
                                     }
                                     // 关闭rtp server
@@ -177,12 +177,14 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
     @Override
     public boolean startDownload(StreamInfo stream, String callId) {
         boolean result;
+        String key=String.format("%S_%s_%s_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX,
+                userSetting.getServerId(), stream.getMediaServerId(), stream.getDeviceID(), stream.getChannelId(), stream.getStream(), callId);
         if (stream.getProgress() == 1) {
-            result = RedisUtil.set(String.format("%S_%s_%s_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX,
-                    userSetting.getServerId(), stream.getMediaServerId(), stream.getDeviceID(), stream.getChannelId(), stream.getStream(), callId), stream);
+            logger.debug("添加下载缓存==已完成下载=》{}",key);
+            result = RedisUtil.set(key, stream);
         }else {
-            result = RedisUtil.set(String.format("%S_%s_%s_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX,
-                    userSetting.getServerId(), stream.getMediaServerId(), stream.getDeviceID(), stream.getChannelId(), stream.getStream(), callId), stream, 60*60);
+            logger.debug("添加下载缓存==未完成下载=》{}",key);
+            result = RedisUtil.set(key, stream, 60*60);
         }
         return result;
     }
@@ -617,7 +619,7 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
                 stream,
                 callId
         );
-        List<Object> streamInfoScan = RedisUtil.scan(key);
+        List<Object> streamInfoScan = RedisUtil.scan2(key);
         if (streamInfoScan.size() > 0) {
             return (StreamInfo) RedisUtil.get((String) streamInfoScan.get(0));
         }else {
@@ -881,7 +881,13 @@ public class RedisUtil {
         return new ArrayList<>(resultKeys);
     }
-
+    public static List<Object> scan2(String query) {
+        if (redisTemplate == null) {
+            redisTemplate = SpringBeanFactory.getBean("redisTemplate");
+        }
+        Set<String> keys = redisTemplate.keys(query);
+        return new ArrayList<>(keys);
+    }
     //    ============================== 消息发送与订阅 ==============================
     public static void convertAndSend(String channel, JSONObject msg) {
         if (redisTemplate == null) {