优化点播, 级联点播级联录像。级联列表显示订阅状态

648540858

2022-03-14 354a39961ad26949f597e4c434b0cd470b7f78ee

优化点播, 级联点播级联录像。级联列表显示订阅状态

 src/main/java/com/genersoft/iot/vmp/conf/SipPlatformRunner.java

@@ -60,12 +60,9 @@

            // 取消订阅
            sipCommanderForPlatform.unregister(parentPlatform, null, (eventResult)->{
                ParentPlatform platform = storager.queryParentPlatByServerGBId(parentPlatform.getServerGBId());
                sipCommanderForPlatform.register(platform, null, null);
                // 发送平台未注册消息
                publisher.platformNotRegisterEventPublish(parentPlatform.getServerGBId());
            });

            // 发送平台未注册消息
            publisher.platformNotRegisterEventPublish(parentPlatform.getServerGBId());
        }
    }
}

 src/main/java/com/genersoft/iot/vmp/gb28181/bean/InviteStreamCallback.java

New file
@@ -0,0 +1,5 @@
package com.genersoft.iot.vmp.gb28181.bean;

public interface InviteStreamCallback {
    void call(InviteStreamInfo inviteStreamInfo);
}

 src/main/java/com/genersoft/iot/vmp/gb28181/bean/InviteStreamInfo.java

New file
@@ -0,0 +1,61 @@
package com.genersoft.iot.vmp.gb28181.bean;

import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;

public class InviteStreamInfo {

    public InviteStreamInfo(MediaServerItem mediaServerItem, JSONObject response, String callId, String app, String stream) {
        this.mediaServerItem = mediaServerItem;
        this.response = response;
        this.callId = callId;
        this.app = app;
        this.stream = stream;
    }

    private MediaServerItem mediaServerItem;
    private JSONObject response;
    private String callId;
    private String app;
    private String stream;

    public MediaServerItem getMediaServerItem() {
        return mediaServerItem;
    }

    public void setMediaServerItem(MediaServerItem mediaServerItem) {
        this.mediaServerItem = mediaServerItem;
    }

    public JSONObject getResponse() {
        return response;
    }

    public void setResponse(JSONObject response) {
        this.response = response;
    }

    public String getCallId() {
        return callId;
    }

    public void setCallId(String callId) {
        this.callId = callId;
    }

    public String getApp() {
        return app;
    }

    public void setApp(String app) {
        this.app = app;
    }

    public String getStream() {
        return stream;
    }

    public void setStream(String stream) {
        this.stream = stream;
    }
}

 src/main/java/com/genersoft/iot/vmp/gb28181/bean/ParentPlatform.java

@@ -114,6 +114,21 @@
     */
    private String catalogId;

    /**
     * 已被订阅目录信息
     */
    private boolean catalogSubscribe;

    /**
     * 已被订阅报警信息
     */
    private boolean alarmSubscribe;

    /**
     * 已被订阅GPS信息
     */
    private boolean gpsSubscribe;

    public Integer getId() {
        return id;
    }
@@ -290,4 +305,28 @@
    public void setCatalogId(String catalogId) {
        this.catalogId = catalogId;
    }

    public boolean isCatalogSubscribe() {
        return catalogSubscribe;
    }

    public void setCatalogSubscribe(boolean catalogSubscribe) {
        this.catalogSubscribe = catalogSubscribe;
    }

    public boolean isAlarmSubscribe() {
        return alarmSubscribe;
    }

    public void setAlarmSubscribe(boolean alarmSubscribe) {
        this.alarmSubscribe = alarmSubscribe;
    }

    public boolean isGpsSubscribe() {
        return gpsSubscribe;
    }

    public void setGpsSubscribe(boolean gpsSubscribe) {
        this.gpsSubscribe = gpsSubscribe;
    }
}

 src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeInfo.java

@@ -21,6 +21,8 @@
        this.eventType = eventHeader.getEventType();
        this.transaction = evt.getServerTransaction();
        this.dialog = evt.getDialog();
        CallIdHeader callIdHeader = (CallIdHeader)evt.getRequest().getHeader(CallIdHeader.NAME);
        this.callId = callIdHeader.getCallId();
    }

    private String id;

 src/main/java/com/genersoft/iot/vmp/gb28181/event/SipSubscribe.java

@@ -49,10 +49,10 @@
                errorTimeSubscribes.remove(key);
            }
        }
        logger.info("okTimeSubscribes.size:{}",okTimeSubscribes.size());
        logger.info("okSubscribes.size:{}",okSubscribes.size());
        logger.info("errorTimeSubscribes.size:{}",errorTimeSubscribes.size());
        logger.info("errorSubscribes.size:{}",errorSubscribes.size());
        logger.debug("okTimeSubscribes.size:{}",okTimeSubscribes.size());
        logger.debug("okSubscribes.size:{}",okSubscribes.size());
        logger.debug("errorTimeSubscribes.size:{}",errorTimeSubscribes.size());
        logger.debug("errorSubscribes.size:{}",errorSubscribes.size());
    }

    public interface Event {

 src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java

@@ -117,8 +117,6 @@
                        List<ParentPlatform> parentPlatforms = parentPlatformMap.get(gbId);
                        if (parentPlatforms != null && parentPlatforms.size() > 0) {
                            for (ParentPlatform platform : parentPlatforms) {
                                String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetup.getServerId() +  "_Catalog_" + platform.getServerGBId();
//                                SubscribeInfo subscribeInfo = redisCatchStorage.getSubscribe(key);
                                SubscribeInfo subscribeInfo = subscribeHolder.getCatalogSubscribe(platform.getServerGBId());
                                if (subscribeInfo == null) continue;
                                logger.info("[Catalog事件: {}]平台：{}，影响通道{}", event.getType(), platform.getServerGBId(), gbId);

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java

@@ -95,14 +95,14 @@
        logger.debug("\n收到响应：\n{}", responseEvent.getResponse());
        int status = response.getStatusCode();

        if (((status >= 200) && (status < 300)) || status == 401) { // Success!
        if (((status >= 200) && (status < 300)) || status == Response.UNAUTHORIZED) { // Success!
            CSeqHeader cseqHeader = (CSeqHeader) responseEvent.getResponse().getHeader(CSeqHeader.NAME);
            String method = cseqHeader.getMethod();
            ISIPResponseProcessor sipRequestProcessor = responseProcessorMap.get(method);
            if (sipRequestProcessor != null) {
                sipRequestProcessor.process(responseEvent);
            }
            if (responseEvent.getResponse() != null && sipSubscribe.getOkSubscribesSize() > 0 ) {
            if (status != Response.UNAUTHORIZED && responseEvent.getResponse() != null && sipSubscribe.getOkSubscribesSize() > 0 ) {
                CallIdHeader callIdHeader = (CallIdHeader)responseEvent.getResponse().getHeader(CallIdHeader.NAME);
                if (callIdHeader != null) {
                    SipSubscribe.Event subscribe = sipSubscribe.getOkSubscribe(callIdHeader.getCallId());

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java

@@ -2,6 +2,7 @@


import com.genersoft.iot.vmp.common.StreamInfo;

import com.genersoft.iot.vmp.gb28181.bean.Device;

import com.genersoft.iot.vmp.gb28181.bean.InviteStreamCallback;

import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;

import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;

import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;

@@ -103,7 +104,7 @@
     * @param startTime 开始时间,格式要求：yyyy-MM-dd HH:mm:ss

     * @param endTime 结束时间,格式要求：yyyy-MM-dd HH:mm:ss

     */

    void playbackStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInf, Device device, String channelId, String startTime, String endTime, ZLMHttpHookSubscribe.Event event, SipSubscribe.Event errorEvent);

    void playbackStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInf, Device device, String channelId, String startTime, String endTime,InviteStreamCallback inviteStreamCallback, InviteStreamCallback event, SipSubscribe.Event errorEvent);



    /**

     * 请求历史媒体下载

@@ -114,13 +115,13 @@
     * @param endTime 结束时间,格式要求：yyyy-MM-dd HH:mm:ss

     * @param downloadSpeed 下载倍速参数

     */ 

    void downloadStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, String startTime, String endTime, String downloadSpeed, ZLMHttpHookSubscribe.Event event, SipSubscribe.Event errorEvent);

    void downloadStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, String startTime, String endTime, String downloadSpeed, InviteStreamCallback event, SipSubscribe.Event errorEvent);



    /**

     * 视频流停止

     */

    void streamByeCmd(String deviceId, String channelId, String stream, SipSubscribe.Event okEvent);

    void streamByeCmd(String deviceId, String channelId, String stream);

    void streamByeCmd(String deviceId, String channelId, String stream, String callId, SipSubscribe.Event okEvent);

    void streamByeCmd(String deviceId, String channelId, String stream, String callId);



    /**

     * 回放暂停


 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java

@@ -6,6 +6,8 @@
import com.genersoft.iot.vmp.conf.SipConfig;

import com.genersoft.iot.vmp.conf.UserSetup;

import com.genersoft.iot.vmp.gb28181.bean.Device;

import com.genersoft.iot.vmp.gb28181.bean.InviteStreamCallback;

import com.genersoft.iot.vmp.gb28181.bean.InviteStreamInfo;

import com.genersoft.iot.vmp.gb28181.bean.SsrcTransaction;

import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;

import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;

@@ -445,26 +447,12 @@
     * @param endTime 结束时间,格式要求：yyyy-MM-dd HH:mm:ss

     */ 

    @Override

    public void playbackStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, String startTime, String endTime, ZLMHttpHookSubscribe.Event event

            , SipSubscribe.Event errorEvent) {

    public void playbackStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,

                                  String startTime, String endTime, InviteStreamCallback inviteStreamCallback, InviteStreamCallback hookEvent,

                                  SipSubscribe.Event errorEvent) {

        try {



            logger.info("{} 分配的ZLM为: {} [{}:{}]", ssrcInfo.getStream(), mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort());



            // 添加订阅

            JSONObject subscribeKey = new JSONObject();

            subscribeKey.put("app", "rtp");

            subscribeKey.put("stream", ssrcInfo.getStream());

            subscribeKey.put("regist", true);

            subscribeKey.put("schema", "rtmp");

            subscribeKey.put("mediaServerId", mediaServerItem.getId());

            logger.debug("录像回放添加订阅，订阅内容：" + subscribeKey.toString());

            subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,

                    (MediaServerItem mediaServerItemInUse, JSONObject json)->{

                if (event != null) {

                    event.response(mediaServerItemInUse, json);

                }

            });



            StringBuffer content = new StringBuffer(200);

            content.append("v=0\r\n");

@@ -530,6 +518,21 @@
            CallIdHeader callIdHeader = device.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()

                    : udpSipProvider.getNewCallId();



            // 添加订阅

            JSONObject subscribeKey = new JSONObject();

            subscribeKey.put("app", "rtp");

            subscribeKey.put("stream", ssrcInfo.getStream());

            subscribeKey.put("regist", true);

            subscribeKey.put("schema", "rtmp");

            subscribeKey.put("mediaServerId", mediaServerItem.getId());

            logger.debug("录像回放添加订阅，订阅内容：" + subscribeKey);

            subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,

                    (MediaServerItem mediaServerItemInUse, JSONObject json)->{

                        if (hookEvent != null) {

                            InviteStreamInfo inviteStreamInfo = new InviteStreamInfo(mediaServerItemInUse, json, callIdHeader.getCallId(), "rtp", ssrcInfo.getStream());

                            hookEvent.call(inviteStreamInfo);

                        }

                    });

            Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), null, "fromplybck" + tm, null, callIdHeader, ssrcInfo.getSsrc());



            transmitRequest(device, request, errorEvent, okEvent -> {

@@ -537,6 +540,9 @@
                streamSession.put(device.getDeviceId(), channelId, callIdHeader.getCallId(), ssrcInfo.getStream(), ssrcInfo.getSsrc(), mediaServerItem.getId(), responseEvent.getClientTransaction());

                streamSession.put(device.getDeviceId(), channelId, callIdHeader.getCallId(), okEvent.dialog);

            });

            if (inviteStreamCallback != null) {

                inviteStreamCallback.call(new InviteStreamInfo(mediaServerItem, null, callIdHeader.getCallId(), "rtp", ssrcInfo.getStream()));

            }

        } catch ( SipException | ParseException | InvalidArgumentException e) {

            e.printStackTrace();

        }

@@ -552,23 +558,10 @@
     * @param downloadSpeed 下载倍速参数

     */ 

    @Override

    public void downloadStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, String startTime, String endTime, String downloadSpeed, ZLMHttpHookSubscribe.Event event

    public void downloadStreamCmd(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId, String startTime, String endTime, String downloadSpeed, InviteStreamCallback event

            , SipSubscribe.Event errorEvent) {

        try {

            logger.info("{} 分配的ZLM为: {} [{}:{}]", ssrcInfo.getStream(), mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort());



            // 添加订阅

            JSONObject subscribeKey = new JSONObject();

            subscribeKey.put("app", "rtp");

            subscribeKey.put("stream", ssrcInfo.getStream());

            subscribeKey.put("regist", true);

            subscribeKey.put("mediaServerId", mediaServerItem.getId());

            logger.debug("录像回放添加订阅，订阅内容：" + subscribeKey.toString());

            subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,

                    (MediaServerItem mediaServerItemInUse, JSONObject json)->{

                event.response(mediaServerItemInUse, json);

                subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey);

            });



            StringBuffer content = new StringBuffer(200);

            content.append("v=0\r\n");

@@ -637,6 +630,19 @@
            CallIdHeader callIdHeader = device.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()

                    : udpSipProvider.getNewCallId();



            // 添加订阅

            JSONObject subscribeKey = new JSONObject();

            subscribeKey.put("app", "rtp");

            subscribeKey.put("stream", ssrcInfo.getStream());

            subscribeKey.put("regist", true);

            subscribeKey.put("mediaServerId", mediaServerItem.getId());

            logger.debug("录像回放添加订阅，订阅内容：" + subscribeKey.toString());

            subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,

                    (MediaServerItem mediaServerItemInUse, JSONObject json)->{

                        event.call(new InviteStreamInfo(mediaServerItem, json, callIdHeader.getCallId(), "rtp", ssrcInfo.getStream()));

                        subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey);

                    });



            Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), null, "fromplybck" + tm, null, callIdHeader, ssrcInfo.getSsrc());



            ClientTransaction transaction = transmitRequest(device, request, errorEvent);

@@ -652,15 +658,15 @@
     * 视频流停止, 不使用回调

     */

    @Override

    public void streamByeCmd(String deviceId, String channelId, String stream) {

        streamByeCmd(deviceId, channelId, stream, null);

    public void streamByeCmd(String deviceId, String channelId, String stream, String callId) {

        streamByeCmd(deviceId, channelId, stream, callId, null);

    }



    /**

     * 视频流停止

     */

    @Override

    public void streamByeCmd(String deviceId, String channelId, String stream, SipSubscribe.Event okEvent) {

    public void streamByeCmd(String deviceId, String channelId, String stream, String callId, SipSubscribe.Event okEvent) {

        try {

            SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(deviceId, channelId, null, stream);

            ClientTransaction transaction = streamSession.getTransactionByStream(deviceId, channelId, stream);

@@ -672,7 +678,15 @@
                }

                return;

            }

            SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, stream);

            SIPDialog dialog;

            if (callId != null) {

                dialog = streamSession.getDialogByCallId(deviceId, channelId, callId);

            }else {

                if (stream == null) return;

                dialog = streamSession.getDialogByStream(deviceId, channelId, stream);

            }





            if (dialog == null) {

                logger.warn("[ {} -> {}]停止视频流的时候发现对话已丢失", deviceId, channelId);

                return;


 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java

@@ -13,6 +13,7 @@
import com.genersoft.iot.vmp.utils.SerializeUtils;
import gov.nist.javax.sip.SipProviderImpl;
import gov.nist.javax.sip.SipStackImpl;
import gov.nist.javax.sip.message.MessageFactoryImpl;
import gov.nist.javax.sip.message.SIPRequest;
import gov.nist.javax.sip.stack.SIPDialog;
import org.slf4j.Logger;
@@ -77,11 +78,11 @@
    @Override
    public boolean unregister(ParentPlatform parentPlatform, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent) {
        ParentPlatformCatch parentPlatformCatch = redisCatchStorage.queryPlatformCatchInfo(parentPlatform.getServerGBId());
        parentPlatform.setExpires("0");
        if (parentPlatformCatch != null) {
            parentPlatformCatch.setParentPlatform(parentPlatform);
            redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
        }
        parentPlatform.setExpires("0");
        return register(parentPlatform, null, null, errorEvent, okEvent, false);
    }

@@ -416,11 +417,13 @@
    private void sendNotify(ParentPlatform parentPlatform, String catalogXmlContent,
                                   SubscribeInfo subscribeInfo, SipSubscribe.Event errorEvent,  SipSubscribe.Event okEvent )
            throws NoSuchFieldException, IllegalAccessException, SipException, ParseException {
        MessageFactoryImpl messageFactory = (MessageFactoryImpl) sipFactory.createMessageFactory();
         // 设置编码， 防止中文乱码
        messageFactory.setDefaultContentEncodingCharset("gb2312");
        Dialog dialog  = subscribeInfo.getDialog();
        Request notifyRequest = dialog.createRequest(Request.NOTIFY);

        if (dialog == null) return;
        SIPRequest notifyRequest = (SIPRequest)dialog.createRequest(Request.NOTIFY);
        ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("Application", "MANSCDP+xml");

        notifyRequest.setContent(catalogXmlContent, contentTypeHeader);

        SubscriptionStateHeader subscriptionState = sipFactory.createHeaderFactory()
@@ -511,7 +514,8 @@
    }

    @Override
    public boolean sendNotifyForCatalogOther(String type, ParentPlatform parentPlatform, List<DeviceChannel> deviceChannels, SubscribeInfo subscribeInfo, Integer index) {
    public boolean sendNotifyForCatalogOther(String type, ParentPlatform parentPlatform, List<DeviceChannel> deviceChannels,
                                             SubscribeInfo subscribeInfo, Integer index) {
        if (parentPlatform == null
                || deviceChannels == null
                || deviceChannels.size() == 0
@@ -579,24 +583,30 @@
            recordXml.append("<SN>" +recordInfo.getSn() + "</SN>\r\n");
            recordXml.append("<DeviceID>" + recordInfo.getDeviceId() + "</DeviceID>\r\n");
            recordXml.append("<SumNum>" + recordInfo.getSumNum() + "</SumNum>\r\n");
            recordXml.append("<RecordList Num=\"" + recordInfo.getRecordList().size()+"\">\r\n");
            for (RecordItem recordItem : recordInfo.getRecordList()) {
                recordXml.append("<Item>\r\n");
                if (deviceChannel != null) {
                    recordXml.append("<DeviceID>" + recordItem.getDeviceId() + "</DeviceID>\r\n");
                    recordXml.append("<Name>" + recordItem.getName() + "</Name>\r\n");
                    recordXml.append("<StartTime>" + DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(recordItem.getStartTime()) + "</StartTime>\r\n");
                    recordXml.append("<EndTime>" + DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(recordItem.getEndTime()) + "</EndTime>\r\n");
                    recordXml.append("<Secrecy>" + recordItem.getSecrecy() + "</Secrecy>\r\n");
                    recordXml.append("<Type>" + recordItem.getType() + "</Type>\r\n");
                    if (!StringUtils.isEmpty(recordItem.getFileSize())) {
                        recordXml.append("<FileSize>" + recordItem.getFileSize() + "</FileSize>\r\n");
                    }
                    if (!StringUtils.isEmpty(recordItem.getFilePath())) {
                        recordXml.append("<FilePath>" + recordItem.getFilePath() + "</FilePath>\r\n");
            if (recordInfo.getRecordList() == null ) {
                recordXml.append("<RecordList Num=\"0\">\r\n");
            }else {
                recordXml.append("<RecordList Num=\"" + recordInfo.getRecordList().size()+"\">\r\n");
                if (recordInfo.getRecordList().size() > 0) {
                    for (RecordItem recordItem : recordInfo.getRecordList()) {
                        recordXml.append("<Item>\r\n");
                        if (deviceChannel != null) {
                            recordXml.append("<DeviceID>" + recordItem.getDeviceId() + "</DeviceID>\r\n");
                            recordXml.append("<Name>" + recordItem.getName() + "</Name>\r\n");
                            recordXml.append("<StartTime>" + DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(recordItem.getStartTime()) + "</StartTime>\r\n");
                            recordXml.append("<EndTime>" + DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(recordItem.getEndTime()) + "</EndTime>\r\n");
                            recordXml.append("<Secrecy>" + recordItem.getSecrecy() + "</Secrecy>\r\n");
                            recordXml.append("<Type>" + recordItem.getType() + "</Type>\r\n");
                            if (!StringUtils.isEmpty(recordItem.getFileSize())) {
                                recordXml.append("<FileSize>" + recordItem.getFileSize() + "</FileSize>\r\n");
                            }
                            if (!StringUtils.isEmpty(recordItem.getFilePath())) {
                                recordXml.append("<FilePath>" + recordItem.getFilePath() + "</FilePath>\r\n");
                            }
                        }
                        recordXml.append("</Item>\r\n");
                    }
                }
                recordXml.append("</Item>\r\n");
            }

            recordXml.append("</RecordList>\r\n");

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java

@@ -27,10 +27,7 @@
import javax.sip.header.FromHeader;
import javax.sip.header.HeaderAddress;
import javax.sip.header.ToHeader;
import java.util.HashMap;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.*;

/**
 * SIP命令类型： ACK请求
@@ -84,44 +81,72 @@
            String channelId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser();
            SendRtpItem sendRtpItem =  redisCatchStorage.querySendRTPServer(platformGbId, channelId, null, callIdHeader.getCallId());
            String is_Udp = sendRtpItem.isTcp() ? "0" : "1";
            String deviceId = sendRtpItem.getDeviceId();
            StreamInfo streamInfo = null;
            if (sendRtpItem.isPlay()) {
                streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
            }else {
                streamInfo = redisCatchStorage.queryPlaybackByDevice(deviceId, channelId);
            }
            if (streamInfo == null) {
                streamInfo = new StreamInfo();
                streamInfo.setApp(sendRtpItem.getApp());
                streamInfo.setStream(sendRtpItem.getStreamId());
            }
            redisCatchStorage.updateSendRTPSever(sendRtpItem);
            MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
            logger.info("收到ACK，开始向上级推流 rtp/{}", sendRtpItem.getStreamId());
            Map<String, Object> param = new HashMap<>();
            param.put("vhost","__defaultVhost__");
            param.put("app",streamInfo.getApp());
            param.put("stream",streamInfo.getStream());
            param.put("app",sendRtpItem.getApp());
            param.put("stream",sendRtpItem.getStreamId());
            param.put("ssrc", sendRtpItem.getSsrc());
            param.put("dst_url",sendRtpItem.getIp());
            param.put("dst_port", sendRtpItem.getPort());
            param.put("is_udp", is_Udp);
            MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
            JSONObject jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
            if (jsonObject.getInteger("code") != 0) {
                logger.info("监听流以等待流上线{}/{}", streamInfo.getApp(), streamInfo.getStream());
                // 监听流上线
                // 添加订阅
                JSONObject subscribeKey = new JSONObject();
                subscribeKey.put("app", "rtp");
                subscribeKey.put("stream", streamInfo.getStream());
                subscribeKey.put("regist", true);
                subscribeKey.put("schema", "rtmp");
                subscribeKey.put("mediaServerId", sendRtpItem.getMediaServerId());
                subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
                        (MediaServerItem mediaServerItemInUse, JSONObject json)->{
                            zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
                        });
            }
            param.put("src_port", sendRtpItem.getLocalPort());
            zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);



//            if (streamInfo == null) { // 流还没上来，对方就回复ack
//                logger.info("监听流以等待流上线1 rtp/{}", sendRtpItem.getStreamId());
//                // 监听流上线
//                // 添加订阅
//                JSONObject subscribeKey = new JSONObject();
//                subscribeKey.put("app", "rtp");
//                subscribeKey.put("stream", sendRtpItem.getStreamId());
//                subscribeKey.put("regist", true);
//                subscribeKey.put("schema", "rtmp");
//                subscribeKey.put("mediaServerId", sendRtpItem.getMediaServerId());
//                subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
//                        (MediaServerItem mediaServerItemInUse, JSONObject json)->{
//                            Map<String, Object> param = new HashMap<>();
//                            param.put("vhost","__defaultVhost__");
//                            param.put("app",json.getString("app"));
//                            param.put("stream",json.getString("stream"));
//                            param.put("ssrc", sendRtpItem.getSsrc());
//                            param.put("dst_url",sendRtpItem.getIp());
//                            param.put("dst_port", sendRtpItem.getPort());
//                            param.put("is_udp", is_Udp);
//                            param.put("src_port", sendRtpItem.getLocalPort());
//                            zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
//                        });
//            }else {
//                Map<String, Object> param = new HashMap<>();
//                param.put("vhost","__defaultVhost__");
//                param.put("app",streamInfo.getApp());
//                param.put("stream",streamInfo.getStream());
//                param.put("ssrc", sendRtpItem.getSsrc());
//                param.put("dst_url",sendRtpItem.getIp());
//                param.put("dst_port", sendRtpItem.getPort());
//                param.put("is_udp", is_Udp);
//                param.put("src_port", sendRtpItem.getLocalPort());
//
//                JSONObject jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
//                if (jsonObject.getInteger("code") != 0) {
//                    logger.info("监听流以等待流上线2 {}/{}", streamInfo.getApp(), streamInfo.getStream());
//                    // 监听流上线
//                    // 添加订阅
//                    JSONObject subscribeKey = new JSONObject();
//                    subscribeKey.put("app", "rtp");
//                    subscribeKey.put("stream", streamInfo.getStream());
//                    subscribeKey.put("regist", true);
//                    subscribeKey.put("schema", "rtmp");
//                    subscribeKey.put("mediaServerId", sendRtpItem.getMediaServerId());
//                    subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
//                            (MediaServerItem mediaServerItemInUse, JSONObject json)->{
//                                zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
//                            });
//                }
//            }
        }
    }
}

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java

@@ -93,14 +93,16 @@
                    param.put("app",sendRtpItem.getApp());
                    param.put("stream",streamId);
                    param.put("ssrc",sendRtpItem.getSsrc());
                    logger.info("停止向上级推流：" + streamId);
                    logger.info("收到bye:停止向上级推流：" + streamId);
                    MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
                    zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);
                    redisCatchStorage.deleteSendRTPServer(platformGbId, channelId, callIdHeader.getCallId(), null);
                    int totalReaderCount = zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId);
                    if (totalReaderCount <= 0) {
                        logger.info(streamId + "无其它观看者，通知设备停止推流");
                        cmder.streamByeCmd(sendRtpItem.getDeviceId(), channelId, streamId);
                        logger.info("收到bye: {}无其它观看者，通知设备停止推流", streamId);
                        if (sendRtpItem.isPlay()) {
                            cmder.streamByeCmd(sendRtpItem.getDeviceId(), channelId, streamId, null);
                        }
                    }
                }
                // 可能是设备主动停止

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java

@@ -6,6 +6,7 @@
import com.genersoft.iot.vmp.conf.DynamicTask;
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.callback.RequestMessage;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
@@ -90,6 +91,9 @@

    @Autowired
    private SIPProcessorObserver sipProcessorObserver;

    @Autowired
    private VideoStreamSessionManager sessionManager;


    @Override
@@ -233,6 +237,7 @@
                }
                String username = sdp.getOrigin().getUsername();
                String addressStr = sdp.getOrigin().getAddress();

                logger.info("[上级点播]用户：{}， 地址：{}:{}， ssrc：{}", username, addressStr, port, ssrc);
                Device device  = null;
                // 通过 channel 和 gbStream 是否为null 值判断来源是直播流合适国标
@@ -266,13 +271,14 @@
                    sendRtpItem.setDialog(dialogByteArray);
                    byte[] transactionByteArray = SerializeUtils.serialize(evt.getServerTransaction());
                    sendRtpItem.setTransaction(transactionByteArray);
                    // 写入redis， 超时时回复
                    redisCatchStorage.updateSendRTPSever(sendRtpItem);


                    Long finalStartTime = startTime;
                    Long finalStopTime = stopTime;
                    ZLMHttpHookSubscribe.Event hookEvent = (mediaServerItemInUSe, responseJSON)->{
                        logger.info("[上级点播]下级已经开始推流。 回复200OK(SDP)， {}/{}", sendRtpItem.getApp(), sendRtpItem.getStreamId());
                        String app = responseJSON.getString("app");
                        String stream = responseJSON.getString("stream");
                        logger.info("[上级点播]下级已经开始推流。 回复200OK(SDP)， {}/{}", app, stream);
                        //     * 0 等待设备推流上来
                        //     * 1 下级已经推流，等待上级平台回复ack
                        //     * 2 推流中
@@ -325,46 +331,66 @@
                            e.printStackTrace();
                        }
                    });
                    sendRtpItem.setApp("rtp");
                    if ("Playback".equals(sessionName)) {
                        sendRtpItem.setPlay(false);
                        sendRtpItem.setStreamId(ssrc);
                        SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, true);
                        sendRtpItem.setStreamId(ssrcInfo.getStream());
                        // 写入redis， 超时时回复
                        redisCatchStorage.updateSendRTPSever(sendRtpItem);
                        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        playService.playBack(device.getDeviceId(), channelId, format.format(start), format.format(end),result -> {
                            if (result.getCode() != 0){
                                logger.warn("录像回放失败");
                                if (result.getEvent() != null) {
                                    errorEvent.response(result.getEvent());
                        playService.playBack(mediaServerItem, ssrcInfo, device.getDeviceId(), channelId, format.format(start),
                                format.format(end), null, result -> {
                                if (result.getCode() != 0){
                                    logger.warn("录像回放失败");
                                    if (result.getEvent() != null) {
                                        errorEvent.response(result.getEvent());
                                    }
                                    redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null);
                                    try {
                                        responseAck(evt, Response.REQUEST_TIMEOUT);
                                    } catch (SipException e) {
                                        e.printStackTrace();
                                    } catch (InvalidArgumentException e) {
                                        e.printStackTrace();
                                    } catch (ParseException e) {
                                        e.printStackTrace();
                                    }
                                }else {
                                    if (result.getMediaServerItem() != null) {
                                        hookEvent.response(result.getMediaServerItem(), result.getResponse());
                                    }
                                }
                                redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null);
                                try {
                                    responseAck(evt, Response.REQUEST_TIMEOUT);
                                } catch (SipException e) {
                                    e.printStackTrace();
                                } catch (InvalidArgumentException e) {
                                    e.printStackTrace();
                                } catch (ParseException e) {
                                    e.printStackTrace();
                                }
                            }else {
                                if (result.getMediaServerItem() != null) {
                                    hookEvent.response(result.getMediaServerItem(), result.getResponse());
                                }
                            }
                        });
                            });
                    }else {
                        sendRtpItem.setPlay(true);
                        StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(device.getDeviceId(), channelId);
                        if (streamInfo == null) {
                        SsrcTransaction playTransaction = sessionManager.getSsrcTransaction(device.getDeviceId(), channelId, "play", null);
                        if (playTransaction != null) {
                            Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, "rtp", playTransaction.getStream());
                            if (!streamReady) {
                                playTransaction = null;
                            }
                        }
                        if (playTransaction == null) {
                            SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, true);
                            if (mediaServerItem.isRtpEnable()) {
                                sendRtpItem.setStreamId(String.format("%s_%s", device.getDeviceId(), channelId));
                            }else {
                                sendRtpItem.setStreamId(ssrcInfo.getStream());
                            }
                            sendRtpItem.setPlay(false);
                            playService.play(mediaServerItem,device.getDeviceId(), channelId, hookEvent, errorEvent, ()->{
                            // 写入redis， 超时时回复
                            redisCatchStorage.updateSendRTPSever(sendRtpItem);
                            playService.play(mediaServerItem, ssrcInfo, device, channelId, hookEvent, errorEvent, (code, msg)->{
                                redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null);
                            });
                            }, null);
                        }else {
                            sendRtpItem.setStreamId(streamInfo.getStream());
                            hookEvent.response(mediaServerItem, null);
                            sendRtpItem.setStreamId(playTransaction.getStream());
                            // 写入redis， 超时时回复
                            redisCatchStorage.updateSendRTPSever(sendRtpItem);
                            JSONObject jsonObject = new JSONObject();
                            jsonObject.put("app", sendRtpItem.getApp());
                            jsonObject.put("stream", sendRtpItem.getStreamId());
                            hookEvent.response(mediaServerItem, jsonObject);
                        }
                    }
                }else if (gbStream != null) {

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/MediaStatusNotifyMessageHandler.java

@@ -18,6 +18,7 @@
import javax.sip.InvalidArgumentException;
import javax.sip.RequestEvent;
import javax.sip.SipException;
import javax.sip.header.CallIdHeader;
import javax.sip.message.Response;
import java.text.ParseException;

@@ -56,14 +57,15 @@
        } catch (ParseException e) {
            e.printStackTrace();
        }
        CallIdHeader callIdHeader = (CallIdHeader)evt.getRequest().getHeader(CallIdHeader.NAME);
        String NotifyType =getText(rootElement, "NotifyType");
        if (NotifyType.equals("121")){
            logger.info("媒体播放完毕，通知关流");
            StreamInfo streamInfo = redisCatchStorage.queryPlaybackByDevice(device.getDeviceId(), "*");
            if (streamInfo != null) {
                redisCatchStorage.stopPlayback(streamInfo);
                cmder.streamByeCmd(streamInfo.getDeviceID(), streamInfo.getChannelId(), streamInfo.getStream());
            }
            String channelId =getText(rootElement, "DeviceID");
            redisCatchStorage.stopPlayback(device.getDeviceId(), channelId, null, callIdHeader.getCallId());
            cmder.streamByeCmd(device.getDeviceId(), channelId, null, callIdHeader.getCallId());
            // TODO 如果级联播放，需要给上级发送此通知

        }
    }


 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/response/impl/RegisterResponseProcessor.java

@@ -2,6 +2,7 @@

import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch;
import com.genersoft.iot.vmp.gb28181.bean.SubscribeHolder;
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.event.response.SIPResponseProcessorAbstract;
@@ -39,6 +40,9 @@

    @Autowired
    private SIPProcessorObserver sipProcessorObserver;

    @Autowired
    private SubscribeHolder subscribeHolder;

    @Override
    public void afterPropertiesSet() throws Exception {
@@ -83,19 +87,19 @@
            // 注册/注销成功
            logger.info(String.format("%s %s成功", platformGBId, action));
            redisCatchStorage.delPlatformRegisterInfo(callId);
            parentPlatform.setStatus("注册".equals(action));
            redisCatchStorage.delPlatformCatchInfo(platformGBId);
            // 取回Expires设置，避免注销过程中被置为0
            if (!parentPlatformCatch.getParentPlatform().getExpires().equals("0")) {
                ParentPlatform parentPlatformTmp = storager.queryParentPlatByServerGBId(platformGBId);
                String expires = parentPlatformTmp.getExpires();
                parentPlatform.setExpires(expires);
                parentPlatform.setId(parentPlatformTmp.getId());
                redisCatchStorage.updatePlatformRegister(parentPlatform);
                redisCatchStorage.updatePlatformKeepalive(parentPlatform);
                parentPlatformCatch.setParentPlatform(parentPlatform);
                redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
            }
            ParentPlatform parentPlatformTmp = storager.queryParentPlatByServerGBId(platformGBId);
            parentPlatformTmp.setStatus("注册".equals(action));
            redisCatchStorage.updatePlatformRegister(parentPlatformTmp);
            redisCatchStorage.updatePlatformKeepalive(parentPlatformTmp);
            parentPlatformCatch.setParentPlatform(parentPlatformTmp);
            redisCatchStorage.updatePlatformCatchInfo(parentPlatformCatch);
            storager.updateParentPlatformStatus(platformGBId, "注册".equals(action));
            if ("注销".equals(action)) {
                subscribeHolder.removeCatalogSubscribe(platformGBId);
                subscribeHolder.removeMobilePositionSubscribe(platformGBId);
            }

        }
    }

 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java

@@ -45,12 +45,8 @@

        Map<String, Object> param = new HashMap<>();
        int result = -1;
        /**
         * 不设置推流端口端则使用随机端口
         */
        if (StringUtils.isEmpty(mediaServerItem.getSendRtpPortRange())){
            param.put("port", 0);
        }else {
        // 不设置推流端口端则使用随机端口
        if (!StringUtils.isEmpty(mediaServerItem.getSendRtpPortRange())){
            int newPort = getPortFromportRange(mediaServerItem);
            param.put("port", newPort);
        }

 src/main/java/com/genersoft/iot/vmp/service/IPlayService.java

@@ -2,10 +2,14 @@

import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.InviteStreamCallback;
import com.genersoft.iot.vmp.gb28181.bean.InviteStreamInfo;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.bean.InviteTimeOutCallback;
import com.genersoft.iot.vmp.service.bean.PlayBackCallback;
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.PlayResult;
import org.springframework.http.ResponseEntity;
import org.springframework.web.context.request.async.DeferredResult;
@@ -17,13 +21,17 @@

    void onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId, String uuid);

    void play(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,
              ZLMHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent,
              InviteTimeOutCallback timeoutCallback, String uuid);
    PlayResult play(MediaServerItem mediaServerItem, String deviceId, String channelId, ZLMHttpHookSubscribe.Event event, SipSubscribe.Event errorEvent, Runnable timeoutCallback);

    MediaServerItem getNewMediaServerItem(Device device);

    void onPublishHandlerForDownload(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, String toString);
    void onPublishHandlerForDownload(InviteStreamInfo inviteStreamInfo, String deviceId, String channelId, String toString);

    DeferredResult<ResponseEntity<String>> playBack(String deviceId, String channelId, String startTime, String endTime, PlayBackCallback errorCallBack);
    DeferredResult<ResponseEntity<String>> playBack(String deviceId, String channelId, String startTime, String endTime, InviteStreamCallback infoCallBack, PlayBackCallback hookCallBack);
    DeferredResult<ResponseEntity<String>> playBack(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo,String deviceId, String channelId, String startTime, String endTime, InviteStreamCallback infoCallBack, PlayBackCallback hookCallBack);

    void zlmServerOffline(String mediaServerId);
}

 src/main/java/com/genersoft/iot/vmp/service/bean/InviteTimeOutCallback.java

New file
@@ -0,0 +1,6 @@
package com.genersoft.iot.vmp.service.bean;

public interface InviteTimeOutCallback {

    void run(int code, String msg); // code: 0 sip超时, 1 收流超时
}

 src/main/java/com/genersoft/iot/vmp/service/bean/PlayBackResult.java

@@ -7,9 +7,9 @@
import javax.sip.RequestEvent;

public class PlayBackResult<T> {
     private int code;
     private T data;
     private MediaServerItem mediaServerItem;
    private int code;
    private T data;
    private MediaServerItem mediaServerItem;
    private JSONObject response;
    private SipSubscribe.EventResult event;


 src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java

@@ -16,6 +16,7 @@
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.InviteTimeOutCallback;
import com.genersoft.iot.vmp.service.bean.PlayBackCallback;
import com.genersoft.iot.vmp.service.bean.PlayBackResult;
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
@@ -27,6 +28,7 @@
import com.genersoft.iot.vmp.service.IMediaService;
import com.genersoft.iot.vmp.service.IPlayService;
import gov.nist.javax.sip.stack.SIPDialog;
import jdk.nashorn.internal.ir.RuntimeNode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -36,6 +38,9 @@
import org.springframework.util.ResourceUtils;
import org.springframework.web.context.request.async.DeferredResult;

import javax.sip.header.CallIdHeader;
import javax.sip.header.Header;
import javax.sip.message.Request;
import java.io.FileNotFoundException;
import java.util.*;

@@ -77,6 +82,8 @@

    @Autowired
    private UserSetup userSetup;




    @Override
@@ -141,67 +148,7 @@
                e.printStackTrace();
            }
        });
        if (streamInfo == null) {
            String streamId = null;
            if (mediaServerItem.isRtpEnable()) {
                streamId = String.format("%s_%s", device.getDeviceId(), channelId);
            }
            SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId);
            // 超时处理
            Timer timer = new Timer();
            timer.schedule(new TimerTask() {
                @Override
                public void run() {
                    logger.warn(String.format("设备点播超时，deviceId：%s ，channelId：%s", deviceId, channelId));
                    if (timeoutCallback != null) {
                        timeoutCallback.run();
                    }
                    WVPResult wvpResult = new WVPResult();
                    wvpResult.setCode(-1);
                    SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, ssrcInfo.getStream());
                    if (dialog != null) {
                        wvpResult.setMsg("收流超时，请稍候重试");
                        // 点播超时回复BYE 同时释放ssrc以及此次点播的资源
                        cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream());
                    }else {
                        wvpResult.setMsg("点播超时，请稍候重试");
                        mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
                        mediaServerService.closeRTPServer(deviceId, channelId, ssrcInfo.getStream());
                        streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
                    }

                    msg.setData(wvpResult);

                    // 回复之前所有的点播请求
                    resultHolder.invokeAllResult(msg);
                }
            }, userSetup.getPlayTimeout());
            // 发送点播消息
            cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInUse, JSONObject response) -> {
                logger.info("收到订阅消息： " + response.toJSONString());
                timer.cancel();
                onPublishHandlerForPlay(mediaServerItemInUse, response, deviceId, channelId, uuid);
                if (hookEvent != null) {
                    hookEvent.response(mediaServerItem, response);
                }
            }, (event) -> {
                timer.cancel();
                WVPResult wvpResult = new WVPResult();
                wvpResult.setCode(-1);
                // 点播返回sip错误
                mediaServerService.closeRTPServer(playResult.getDevice().getDeviceId(), channelId, ssrcInfo.getStream());
                // 释放ssrc
                mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
                streamSession.remove(deviceId, channelId, ssrcInfo.getStream());

                wvpResult.setMsg(String.format("点播失败， 错误码： %s, %s", event.statusCode, event.msg));
                msg.setData(wvpResult);
                resultHolder.invokeAllResult(msg);
                if (errorEvent != null) {
                    errorEvent.response(event);
                }
            });
        } else {
        if (streamInfo != null) {
            String streamId = streamInfo.getStream();
            if (streamId == null) {
                WVPResult wvpResult = new WVPResult();
@@ -227,67 +174,109 @@
                if (hookEvent != null) {
                    hookEvent.response(mediaServerItem, JSONObject.parseObject(JSON.toJSONString(streamInfo)));
                }
            } else {
                // TODO 点播前是否重置状态
            }else {
                redisCatchStorage.stopPlay(streamInfo);
                storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
                String streamId2 = null;
                if (mediaServerItem.isRtpEnable()) {
                    streamId2 = String.format("%s_%s", device.getDeviceId(), channelId);
                }
                SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId2);
                // 超时处理
                Timer timer = new Timer();
                timer.schedule(new TimerTask() {
                    @Override
                    public void run() {
                        logger.warn(String.format("设备点播超时，deviceId：%s ，channelId：%s", deviceId, channelId));
                        if (timeoutCallback != null) {
                            timeoutCallback.run();
                        }
                        WVPResult wvpResult = new WVPResult();
                        wvpResult.setCode(-1);
                        SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, ssrcInfo.getStream());
                        if (dialog != null) {
                            wvpResult.setMsg("收流超时，请稍候重试");
                            // 点播超时回复BYE 同时释放ssrc以及此次点播的资源
                            cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream());
                        }else {
                            wvpResult.setMsg("点播超时，请稍候重试");
                            mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
                            mediaServerService.closeRTPServer(deviceId, channelId, ssrcInfo.getStream());
                            streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
                        }

                        msg.setData(wvpResult);
                        // 回复之前所有的点播请求
                        resultHolder.invokeAllResult(msg);
                    }
                }, userSetup.getPlayTimeout());
                cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInuse, JSONObject response) -> {
                    logger.info("收到订阅消息： " + response.toJSONString());
                    onPublishHandlerForPlay(mediaServerItemInuse, response, deviceId, channelId, uuid);
                }, (event) -> {
                    mediaServerService.closeRTPServer(playResult.getDevice().getDeviceId(), channelId, ssrcInfo.getStream());
                    // 释放ssrc
                    mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
                    streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
                    WVPResult wvpResult = new WVPResult();
                    wvpResult.setCode(-1);
                    wvpResult.setMsg(String.format("点播失败， 错误码： %s, %s", event.statusCode, event.msg));
                    msg.setData(wvpResult);
                    resultHolder.invokeAllResult(msg);
                });
                streamInfo = null;
            }

        }
        if (streamInfo == null) {
            String streamId = null;
            if (mediaServerItem.isRtpEnable()) {
                streamId = String.format("%s_%s", device.getDeviceId(), channelId);
            }
            SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId);
            play(mediaServerItem, ssrcInfo, device, channelId, (mediaServerItemInUse, response)->{
                if (hookEvent != null) {
                    hookEvent.response(mediaServerItem, response);
                }
            }, event -> {
                // sip error错误
                WVPResult wvpResult = new WVPResult();
                wvpResult.setCode(-1);
                wvpResult.setMsg(String.format("点播失败， 错误码： %s, %s", event.statusCode, event.msg));
                msg.setData(wvpResult);
                resultHolder.invokeAllResult(msg);
                if (errorEvent != null) {
                    errorEvent.response(event);
                }
            }, (code, msgStr)->{
                // invite点播超时
                WVPResult wvpResult = new WVPResult();
                wvpResult.setCode(-1);
                if (code == 0) {
                    wvpResult.setMsg("点播超时，请稍候重试");
                }else if (code == 1) {
                    wvpResult.setMsg("收流超时，请稍候重试");
                }
                msg.setData(wvpResult);
                // 回复之前所有的点播请求
                resultHolder.invokeAllResult(msg);
            }, uuid);
        }
        return playResult;
    }



    @Override
    public void play(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, Device device, String channelId,
                           ZLMHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent,
                           InviteTimeOutCallback timeoutCallback, String uuid) {

        String streamId = null;
        if (mediaServerItem.isRtpEnable()) {
            streamId = String.format("%s_%s", device.getDeviceId(), channelId);
        }
        if (ssrcInfo == null) {
            ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId);
        }

        return playResult;
        // 超时处理
        Timer timer = new Timer();
        SSRCInfo finalSsrcInfo = ssrcInfo;
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                logger.warn(String.format("设备点播超时，deviceId：%s ，channelId：%s", device.getDeviceId(), channelId));

                SIPDialog dialog = streamSession.getDialogByStream(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
                if (dialog != null) {
                    timeoutCallback.run(1, "收流超时");
                    // 点播超时回复BYE 同时释放ssrc以及此次点播的资源
                    cmder.streamByeCmd(device.getDeviceId(), channelId, finalSsrcInfo.getStream(), null);
                }else {
                    timeoutCallback.run(0, "点播超时");
                    mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc());
                    mediaServerService.closeRTPServer(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
                    streamSession.remove(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
                }
            }
        }, userSetup.getPlayTimeout());

        cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInuse, JSONObject response) -> {
            logger.info("收到订阅消息： " + response.toJSONString());
            timer.cancel();
            // hook响应
            onPublishHandlerForPlay(mediaServerItemInuse, response, device.getDeviceId(), channelId, uuid);
            hookEvent.response(mediaServerItemInuse, response);
        }, (event) -> {
            timer.cancel();
            mediaServerService.closeRTPServer(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
            // 释放ssrc
            mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc());
            streamSession.remove(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
            errorEvent.response(event);
        });
    }

    @Override
    public void onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, String uuid) {
        RequestMessage msg = new RequestMessage();
        msg.setId(uuid);
        if (uuid != null) {
            msg.setId(uuid);
        }
        msg.setKey(DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId);
        StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId);
        if (streamInfo != null) {
@@ -297,7 +286,6 @@
                storager.startPlay(deviceId, channelId, streamInfo.getStream());
            }
            redisCatchStorage.startPlay(streamInfo);
            msg.setData(JSON.toJSONString(streamInfo));

            WVPResult wvpResult = new WVPResult();
            wvpResult.setCode(0);
@@ -329,9 +317,24 @@
        return mediaServerItem;
    }

    @Override
    public DeferredResult<ResponseEntity<String>> playBack(String deviceId, String channelId, String startTime,
                                                           String endTime,InviteStreamCallback inviteStreamCallback,
                                                           PlayBackCallback callback) {
        Device device = storager.queryVideoDevice(deviceId);
        if (device == null) return null;
        MediaServerItem newMediaServerItem = getNewMediaServerItem(device);
        SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, true);

        return playBack(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, inviteStreamCallback, callback);
    }

    @Override
    public DeferredResult<ResponseEntity<String>> playBack(String deviceId, String channelId, String startTime, String endTime, PlayBackCallback callback) {
    public DeferredResult<ResponseEntity<String>> playBack(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo,
                                                           String deviceId, String channelId, String startTime,
                                                           String endTime, InviteStreamCallback infoCallBack,
                                                           PlayBackCallback playBackCallback) {
        if (mediaServerItem == null || ssrcInfo == null) return null;
        String uuid = UUID.randomUUID().toString();
        String key = DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId;
        DeferredResult<ResponseEntity<String>> result = new DeferredResult<>(30000L);
@@ -341,8 +344,6 @@
            return result;
        }

        MediaServerItem newMediaServerItem = getNewMediaServerItem(device);
        SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, true);
        resultHolder.put(DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId, uuid, result);
        RequestMessage msg = new RequestMessage();
        msg.setId(uuid);
@@ -356,63 +357,62 @@
                logger.warn(String.format("设备回放超时，deviceId：%s ，channelId：%s", deviceId, channelId));
                playBackResult.setCode(-1);
                playBackResult.setData(msg);
                callback.call(playBackResult);
                playBackCallback.call(playBackResult);
                SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, ssrcInfo.getStream());
                // 点播超时回复BYE 同时释放ssrc以及此次点播的资源
                if (dialog != null) {
                    // 点播超时回复BYE 同时释放ssrc以及此次点播的资源
                    cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream());
                    cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream(), null);
                }else {
                    mediaServerService.releaseSsrc(newMediaServerItem.getId(), ssrcInfo.getSsrc());
                    mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
                    mediaServerService.closeRTPServer(deviceId, channelId, ssrcInfo.getStream());
                    streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
                }
                cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream());
                cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream(), null);
                // 回复之前所有的点播请求
                callback.call(playBackResult);
                playBackCallback.call(playBackResult);
            }
        }, userSetup.getPlayTimeout());
        cmder.playbackStreamCmd(newMediaServerItem, ssrcInfo, device, channelId, startTime, endTime, (MediaServerItem mediaServerItem, JSONObject response) -> {
            logger.info("收到订阅消息： " + response.toJSONString());
            timer.cancel();
            StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId);
            if (streamInfo == null) {
                logger.warn("设备回放API调用失败！");
                msg.setData("设备回放API调用失败！");
                playBackResult.setCode(-1);
                playBackResult.setData(msg);
                callback.call(playBackResult);
                return;
            }
            redisCatchStorage.startPlayback(streamInfo);
            msg.setData(JSON.toJSONString(streamInfo));
            playBackResult.setCode(0);
            playBackResult.setData(msg);
            playBackResult.setMediaServerItem(mediaServerItem);
            playBackResult.setResponse(response);
            callback.call(playBackResult);
        }, event -> {
            timer.cancel();
            msg.setData(String.format("回放失败， 错误码： %s, %s", event.statusCode, event.msg));
            playBackResult.setCode(-1);
            playBackResult.setData(msg);
            playBackResult.setEvent(event);
            callback.call(playBackResult);
            streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
        });
        cmder.playbackStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, infoCallBack,
                (InviteStreamInfo inviteStreamInfo) -> {
                    logger.info("收到订阅消息： " + inviteStreamInfo.getResponse().toJSONString());
                    timer.cancel();
                    StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId);
                    if (streamInfo == null) {
                        logger.warn("设备回放API调用失败！");
                        msg.setData("设备回放API调用失败！");
                        playBackResult.setCode(-1);
                        playBackResult.setData(msg);
                        playBackCallback.call(playBackResult);
                        return;
                    }
                    redisCatchStorage.startPlayback(streamInfo, inviteStreamInfo.getCallId());
                    msg.setData(JSON.toJSONString(streamInfo));
                    playBackResult.setCode(0);
                    playBackResult.setData(msg);
                    playBackResult.setMediaServerItem(inviteStreamInfo.getMediaServerItem());
                    playBackResult.setResponse(inviteStreamInfo.getResponse());
                    playBackCallback.call(playBackResult);
                }, event -> {
                    timer.cancel();
                    msg.setData(String.format("回放失败， 错误码： %s, %s", event.statusCode, event.msg));
                    playBackResult.setCode(-1);
                    playBackResult.setData(msg);
                    playBackResult.setEvent(event);
                    playBackCallback.call(playBackResult);
                    streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
                });
        return result;
    }



    @Override
    public void onPublishHandlerForDownload(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, String uuid) {
    public void onPublishHandlerForDownload(InviteStreamInfo inviteStreamInfo, String deviceId, String channelId, String uuid) {
        RequestMessage msg = new RequestMessage();
        msg.setKey(DeferredResultHolder.CALLBACK_CMD_DOWNLOAD + deviceId + channelId);
        msg.setId(uuid);
        StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId);
        StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId);
        if (streamInfo != null) {
            redisCatchStorage.startDownload(streamInfo);
            redisCatchStorage.startDownload(streamInfo, inviteStreamInfo.getCallId());
            msg.setData(JSON.toJSONString(streamInfo));
            resultHolder.invokeResult(msg);
        } else {
@@ -449,7 +449,8 @@
        if (allSsrc.size() > 0) {
            for (SsrcTransaction ssrcTransaction : allSsrc) {
                if(ssrcTransaction.getMediaServerId().equals(mediaServerId)) {
                    cmder.streamByeCmd(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId(), ssrcTransaction.getStream());
                    cmder.streamByeCmd(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId(),
                            ssrcTransaction.getStream(), null);
                }
            }
        }

 src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java

@@ -47,17 +47,15 @@

    StreamInfo queryPlayByStreamId(String steamId);

    StreamInfo queryPlaybackByStreamId(String steamId);

    StreamInfo queryPlayByDevice(String deviceId, String channelId);

    Map<String, StreamInfo> queryPlayByDeviceId(String deviceId);

    boolean startPlayback(StreamInfo stream);
    boolean startPlayback(StreamInfo stream, String callId);

    boolean stopPlayback(StreamInfo streamInfo);
    boolean stopPlayback(String deviceId, String channelId, String stream, String callId);

    StreamInfo queryPlaybackByDevice(String deviceId, String code);
    StreamInfo queryPlayback(String deviceId, String channelID, String stream, String callId);

    void updatePlatformCatchInfo(ParentPlatformCatch parentPlatformCatch);

@@ -167,9 +165,9 @@
     * 开始下载录像时存入
     * @param streamInfo
     */
    boolean startDownload(StreamInfo streamInfo);
    boolean startDownload(StreamInfo streamInfo, String callId);

    StreamInfo queryDownloadByStreamId(String streamId);
    StreamInfo queryDownload(String deviceId, String channelId, String stream, String callId);

    /**
     * 查找第三方系统留下的国标预设值

 src/main/java/com/genersoft/iot/vmp/storager/dao/PlatformChannelMapper.java

@@ -55,7 +55,7 @@
    int cleanChannelForGB(String platformId);

    @Select("SELECT dc.* FROM platform_gb_channel pgc left join device_channel dc on dc.id = pgc.deviceChannelId WHERE dc.channelId='${channelId}' and pgc.platformId='${platformId}'")
    DeviceChannel queryChannelInParentPlatform(String platformId, String channelId);
    List<DeviceChannel> queryChannelInParentPlatform(String platformId, String channelId);

    @Select(" select dc.channelId as id, dc.name as name, pgc.platformId as platformId, pgc.catalogId as parentId, 0 as childrenCount, 1 as type " +
            " from device_channel dc left join platform_gb_channel pgc on dc.id = pgc.deviceChannelId " +
@@ -67,7 +67,7 @@
            "         left join device_channel dc on dc.id = pgc.deviceChannelId\n" +
            "         left join device d on dc.deviceId = d.deviceId\n" +
            "where dc.channelId = #{channelId} and pgc.platformId=#{platformId}")
    Device queryVideoDeviceByPlatformIdAndChannelId(String platformId, String channelId);
    List<Device> queryVideoDeviceByPlatformIdAndChannelId(String platformId, String channelId);

    @Delete("<script> "+
            "DELETE FROM platform_gb_channel WHERE catalogId=#{id}"  +

 src/main/java/com/genersoft/iot/vmp/storager/dao/PlatformGbStreamMapper.java

@@ -78,7 +78,7 @@
            "left join platform_gb_stream pgs on " +
            "pp.serverGBId = pgs.platformId " +
            "left join gb_stream gs " +
            "gs.gbStreamId = pgs.gbStreamId " +
            "on gs.gbStreamId = pgs.gbStreamId " +
            "WHERE " +
            "gs.app = #{app} " +
            "AND gs.stream = #{stream}" +

 src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java

@@ -134,13 +134,6 @@
    }

    @Override
    public StreamInfo queryPlaybackByStreamId(String streamId) {
        List<Object> playLeys = redis.scan(String.format("%S_%s_%s_*", VideoManagerConstants.PLAY_BLACK_PREFIX, userSetup.getServerId(),  streamId));
        if (playLeys == null || playLeys.size() == 0) return null;
        return (StreamInfo)redis.get(playLeys.get(0).toString());
    }

    @Override
    public StreamInfo queryPlayByDevice(String deviceId, String channelId) {
        List<Object> playLeys = redis.scan(String.format("%S_%s_*_%s_%s", VideoManagerConstants.PLAYER_PREFIX,
                userSetup.getServerId(),
@@ -166,49 +159,67 @@


    @Override
    public boolean startPlayback(StreamInfo stream) {
        return redis.set(String.format("%S_%s_%s_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
                userSetup.getServerId(), stream.getStream(), stream.getDeviceID(), stream.getChannelId()), stream);
    public boolean startPlayback(StreamInfo stream, String callId) {
        return redis.set(String.format("%S_%s_%s_%s_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
                userSetup.getServerId(), stream.getDeviceID(), stream.getChannelId(), stream.getStream(), callId), stream);
    }

    @Override
    public boolean startDownload(StreamInfo streamInfo) {
        return redis.set(String.format("%S_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX, userSetup.getServerId(),
                streamInfo.getStream(), streamInfo.getDeviceID(), streamInfo.getChannelId()), streamInfo);
    public boolean startDownload(StreamInfo stream, String callId) {
        return redis.set(String.format("%S_%s_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX,
                userSetup.getServerId(), stream.getDeviceID(), stream.getChannelId(), stream.getStream(), callId), stream);
    }

    @Override
    public boolean stopPlayback(StreamInfo streamInfo) {
        if (streamInfo == null) return false;
        DeviceChannel deviceChannel = deviceChannelMapper.queryChannel(streamInfo.getDeviceID(), streamInfo.getChannelId());
    public boolean stopPlayback(String deviceId, String channelId, String stream, String callId) {
        DeviceChannel deviceChannel = deviceChannelMapper.queryChannel(deviceId, channelId);
        if (deviceChannel != null) {
            deviceChannel.setStreamId(null);
            deviceChannel.setDeviceId(streamInfo.getDeviceID());
            deviceChannel.setDeviceId(deviceId);
            deviceChannelMapper.update(deviceChannel);
        }
        return redis.del(String.format("%S_%s_%s_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
        if (deviceId == null) deviceId = "*";
        if (channelId == null) channelId = "*";
        if (stream == null) stream = "*";
        if (callId == null) callId = "*";
        String key = String.format("%S_%s_%s_%s_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
                userSetup.getServerId(),
                streamInfo.getStream(),
                streamInfo.getDeviceID(),
                streamInfo.getChannelId()));
                deviceId,
                channelId,
                stream,
                callId
        );
        List<Object> scan = redis.scan(key);
        if (scan.size() > 0) {
            for (Object keyObj : scan) {
                redis.del((String) keyObj);
            }
        }
        return true;
    }

    @Override
    public StreamInfo queryPlaybackByDevice(String deviceId, String code) {
        // String format = String.format("%S_*_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
        //         deviceId,
        //         code);
        List<Object> playLeys = redis.scan(String.format("%S_%s_*_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
    public StreamInfo queryPlayback(String deviceId, String channelId, String stream, String callId) {
        if (stream == null && callId == null) {
            return null;
        }
        if (deviceId == null) deviceId = "*";
        if (channelId == null) channelId = "*";
        if (stream == null) stream = "*";
        if (callId == null) callId = "*";
        String key = String.format("%S_%s_%s_%s_%s_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
                userSetup.getServerId(),
                deviceId,
                code));
        if (playLeys == null || playLeys.size() == 0) {
            playLeys = redis.scan(String.format("%S_%s_*_*_%s", VideoManagerConstants.PLAY_BLACK_PREFIX,
                    userSetup.getServerId(),
                    deviceId));
                channelId,
                stream,
                callId
        );
        List<Object> streamInfoScan = redis.scan(key);
        if (streamInfoScan.size() > 0) {
            return (StreamInfo) redis.get((String) streamInfoScan.get(0));
        }else {
            return null;
        }
        if (playLeys == null || playLeys.size() == 0) return null;
        return (StreamInfo)redis.get(playLeys.get(0).toString());
    }

    @Override
@@ -361,7 +372,7 @@
            }
        }

        List<Object> playBackers = redis.scan(String.format("%S_%s_*_%s_*", VideoManagerConstants.PLAY_BLACK_PREFIX,
        List<Object> playBackers = redis.scan(String.format("%S_%s_%s_*_*_*", VideoManagerConstants.PLAY_BLACK_PREFIX,
                userSetup.getServerId(),
                deviceId));
        if (playBackers.size() > 0) {
@@ -426,10 +437,27 @@
    }

    @Override
    public StreamInfo queryDownloadByStreamId(String streamId) {
        List<Object> playLeys = redis.scan(String.format("%S_%s_%s_*", VideoManagerConstants.DOWNLOAD_PREFIX, userSetup.getServerId(), streamId));
        if (playLeys == null || playLeys.size() == 0) return null;
        return (StreamInfo)redis.get(playLeys.get(0).toString());
    public StreamInfo queryDownload(String deviceId, String channelId, String stream, String callId) {
        if (stream == null && callId == null) {
            return null;
        }
        if (deviceId == null) deviceId = "*";
        if (channelId == null) channelId = "*";
        if (stream == null) stream = "*";
        if (callId == null) callId = "*";
        String key = String.format("%S_%s_%s_%s_%s_%s", VideoManagerConstants.DOWNLOAD_PREFIX,
                userSetup.getServerId(),
                deviceId,
                channelId,
                stream,
                callId
        );
        List<Object> streamInfoScan = redis.scan(key);
        if (streamInfoScan.size() > 0) {
            return (StreamInfo) redis.get((String) streamInfoScan.get(0));
        }else {
            return null;
        }
    }

    @Override

 src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStoragerImpl.java

@@ -663,8 +663,16 @@

    @Override
    public DeviceChannel queryChannelInParentPlatform(String platformId, String channelId) {
        DeviceChannel channel = platformChannelMapper.queryChannelInParentPlatform(platformId, channelId);
        return channel;
        List<DeviceChannel> channels = platformChannelMapper.queryChannelInParentPlatform(platformId, channelId);
        if (channels.size() > 1) {
            // 出现长度大于0的时候肯定是国标通道的ID重复了
            logger.warn("国标ID存在重复：{}", channelId);
        }
        if (channels.size() == 0) {
            return null;
        }else {
            return channels.get(0);
        }
    }

    @Override
@@ -681,8 +689,18 @@

    @Override
    public Device queryVideoDeviceByPlatformIdAndChannelId(String platformId, String channelId) {
        Device device = platformChannelMapper.queryVideoDeviceByPlatformIdAndChannelId(platformId, channelId);
        return device;
        List<Device> devices = platformChannelMapper.queryVideoDeviceByPlatformIdAndChannelId(platformId, channelId);
        if (devices.size() > 1) {
            // 出现长度大于0的时候肯定是国标通道的ID重复了
            logger.warn("国标ID存在重复：{}", channelId);
        }
        if (devices.size() == 0) {
            return null;
        }else {
            return devices.get(0);
        }


    }

    /**
@@ -1084,6 +1102,9 @@

    @Override
    public List<ParentPlatform> queryPlatFormListForStreamWithGBId(String app, String stream, List<String> platforms) {
        if (platforms == null || platforms.size() == 0) {
            return new ArrayList<>();
        }
        return platformGbStreamMapper.queryPlatFormListForGBWithGBId(app, stream, platforms);
    }


 src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/platform/PlatformController.java

@@ -7,6 +7,7 @@
import com.genersoft.iot.vmp.conf.UserSetup;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.PlatformCatalog;
import com.genersoft.iot.vmp.gb28181.bean.SubscribeHolder;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
@@ -48,6 +49,9 @@

    @Autowired
    private IRedisCatchStorage redisCatchStorage;

    @Autowired
    private SubscribeHolder subscribeHolder;

    @Autowired
    private ISIPCommanderForPlatform commanderForPlatform;
@@ -110,10 +114,14 @@
    })
    public PageInfo<ParentPlatform> platforms(@PathVariable int page, @PathVariable int count) {

//        if (logger.isDebugEnabled()) {
//            logger.debug("查询所有上级设备API调用");
//        }
        return storager.queryParentPlatformList(page, count);
        PageInfo<ParentPlatform> parentPlatformPageInfo = storager.queryParentPlatformList(page, count);
        if (parentPlatformPageInfo.getList().size() > 0) {
            for (ParentPlatform platform : parentPlatformPageInfo.getList()) {
                platform.setGpsSubscribe(subscribeHolder.getMobilePositionSubscribe(platform.getServerGBId()) != null);
                platform.setCatalogSubscribe(subscribeHolder.getCatalogSubscribe(platform.getServerGBId()) != null);
            }
        }
        return parentPlatformPageInfo;
    }

    /**

 src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java

@@ -120,7 +120,7 @@
            storager.stopPlay(deviceId, channelId);
            return result;
        }
        cmder.streamByeCmd(deviceId, channelId, streamInfo.getStream(), (event) -> {
        cmder.streamByeCmd(deviceId, channelId, streamInfo.getStream(), null, (event) -> {
            redisCatchStorage.stopPlay(streamInfo);
            storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
            RequestMessage msg = new RequestMessage();
@@ -174,7 +174,7 @@
    public ResponseEntity<String> playConvert(@PathVariable String streamId) {
        StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(streamId);
        if (streamInfo == null) {
            streamInfo = redisCatchStorage.queryPlaybackByStreamId(streamId);
            streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null);
        }
        if (streamInfo == null) {
            logger.warn("视频转码API调用失败！, 视频流已经停止!");

 src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/playback/DownloadController.java

@@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.vmanager.gb28181.playback;

import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.gb28181.bean.InviteStreamInfo;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
@@ -93,11 +94,6 @@
        }
        resultHolder.put(key, uuid, result);
        Device device = storager.queryVideoDevice(deviceId);
        StreamInfo streamInfo = redisCatchStorage.queryPlaybackByDevice(deviceId, channelId);
        if (streamInfo != null) {
            // 停止之前的下载
            cmder.streamByeCmd(deviceId, channelId, streamInfo.getStream());
        }

        MediaServerItem newMediaServerItem = playService.getNewMediaServerItem(device);
        if (newMediaServerItem == null) {
@@ -112,9 +108,9 @@

        SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, true);

        cmder.downloadStreamCmd(newMediaServerItem, ssrcInfo, device, channelId, startTime, endTime, downloadSpeed, (MediaServerItem mediaServerItem, JSONObject response) -> {
            logger.info("收到订阅消息： " + response.toJSONString());
            playService.onPublishHandlerForDownload(mediaServerItem, response, deviceId, channelId, uuid);
        cmder.downloadStreamCmd(newMediaServerItem, ssrcInfo, device, channelId, startTime, endTime, downloadSpeed, (InviteStreamInfo inviteStreamInfo) -> {
            logger.info("收到订阅消息： " + inviteStreamInfo.getResponse().toJSONString());
            playService.onPublishHandlerForDownload(inviteStreamInfo, deviceId, channelId, uuid);
        }, event -> {
            RequestMessage msg = new RequestMessage();
            msg.setId(uuid);
@@ -135,7 +131,7 @@
    @GetMapping("/stop/{deviceId}/{channelId}/{stream}")
    public ResponseEntity<String> playStop(@PathVariable String deviceId, @PathVariable String channelId, @PathVariable String stream) {

        cmder.streamByeCmd(deviceId, channelId, stream);
        cmder.streamByeCmd(deviceId, channelId, stream, null);

        if (logger.isDebugEnabled()) {
            logger.debug(String.format("设备历史媒体下载停止 API调用，deviceId/channelId：%s_%s", deviceId, channelId));

 src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/playback/PlaybackController.java

@@ -77,7 +77,7 @@
            logger.debug(String.format("设备回放 API调用，deviceId：%s ，channelId：%s", deviceId, channelId));
        }

        DeferredResult<ResponseEntity<String>> result = playService.playBack(deviceId, channelId, startTime, endTime, wvpResult->{
        DeferredResult<ResponseEntity<String>> result = playService.playBack(deviceId, channelId, startTime, endTime, null, wvpResult->{
            resultHolder.invokeResult(wvpResult.getData());
        });

@@ -96,7 +96,7 @@
            @PathVariable String channelId,
            @PathVariable String stream) {

        cmder.streamByeCmd(deviceId, channelId, stream);
        cmder.streamByeCmd(deviceId, channelId, stream, null);

        if (logger.isDebugEnabled()) {
            logger.debug(String.format("设备录像回放停止 API调用，deviceId/channelId：%s/%s", deviceId, channelId));
@@ -124,7 +124,7 @@
    public ResponseEntity<String> playPause(@PathVariable String streamId) {
        logger.info("playPause: "+streamId);
        JSONObject json = new JSONObject();
        StreamInfo streamInfo = redisCatchStorage.queryPlaybackByStreamId(streamId);
        StreamInfo streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null);
        if (null == streamInfo) {
            json.put("msg", "streamId不存在");
            logger.warn("streamId不存在!");
@@ -144,7 +144,7 @@
    public ResponseEntity<String> playResume(@PathVariable String streamId) {
        logger.info("playResume: "+streamId);
        JSONObject json = new JSONObject();
        StreamInfo streamInfo = redisCatchStorage.queryPlaybackByStreamId(streamId);
        StreamInfo streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null);
        if (null == streamInfo) {
            json.put("msg", "streamId不存在");
            logger.warn("streamId不存在!");
@@ -165,7 +165,7 @@
    public ResponseEntity<String> playSeek(@PathVariable String streamId, @PathVariable long seekTime) {
        logger.info("playSeek: "+streamId+", "+seekTime);
        JSONObject json = new JSONObject();
        StreamInfo streamInfo = redisCatchStorage.queryPlaybackByStreamId(streamId);
        StreamInfo streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null);
        if (null == streamInfo) {
            json.put("msg", "streamId不存在");
            logger.warn("streamId不存在!");
@@ -186,7 +186,7 @@
    public ResponseEntity<String> playSpeed(@PathVariable String streamId, @PathVariable Double speed) {
        logger.info("playSpeed: "+streamId+", "+speed);
        JSONObject json = new JSONObject();
        StreamInfo streamInfo = redisCatchStorage.queryPlaybackByStreamId(streamId);
        StreamInfo streamInfo = redisCatchStorage.queryPlayback(null, null, streamId, null);
        if (null == streamInfo) {
            json.put("msg", "streamId不存在");
            logger.warn("streamId不存在!");

 src/main/java/com/genersoft/iot/vmp/web/gb28181/ApiStreamController.java

@@ -177,7 +177,7 @@
            result.put("error","未找到流信息");
            return result;
        }
        cmder.streamByeCmd(serial, code, streamInfo.getStream());
        cmder.streamByeCmd(serial, code, streamInfo.getStream(), null);
        redisCatchStorage.stopPlay(streamInfo);
        storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
        return null;

 web_src/src/components/ParentPlatformList.vue

@@ -13,7 +13,7 @@
        </div>
        <!--设备列表-->
        <el-table :data="platformList" border style="width: 100%" :height="winHeight">
          <el-table-column prop="name" label="名称" width="240" align="center"></el-table-column>
          <el-table-column prop="name" label="名称" align="center"></el-table-column>
          <el-table-column prop="serverGBId" label="平台编号" width="180" align="center"></el-table-column>
          <el-table-column label="是否启用" width="120" align="center">
            <template slot-scope="scope">
@@ -38,9 +38,19 @@
              </div>
            </template>
          </el-table-column>
          <el-table-column prop="deviceGBId" label="设备国标编号" width="240" align="center"></el-table-column>
          <el-table-column prop="deviceGBId" label="设备国标编号" width="200" align="center"></el-table-column>
          <el-table-column prop="transport" label="信令传输模式" width="120" align="center"></el-table-column>
          <el-table-column prop="channelCount" label="通道数" align="center"></el-table-column>
          <el-table-column prop="channelCount" label="通道数" width="120" align="center"></el-table-column>
          <el-table-column label="订阅信息" width="240" align="center" fixed="right">
            <template slot-scope="scope">
              <i v-if="scope.row.alarmSubscribe" style="font-size: 1.5rem;" title="报警订阅" class="subscribe-on iconfont icon-gbaojings" ></i>
              <i v-if="!scope.row.alarmSubscribe" style="font-size: 1.5rem;" title="报警订阅" class="subscribe-off iconfont icon-gbaojings" ></i>
              <i v-if="scope.row.catalogSubscribe" title="目录订阅"  class="subscribe-on iconfont icon-gjichus" ></i>
              <i v-if="!scope.row.catalogSubscribe" title="目录订阅" class="subscribe-off iconfont icon-gjichus" ></i>
              <i v-if="scope.row.gpsSubscribe" title="位置订阅" class="subscribe-on iconfont icon-gxunjians" ></i>
              <i v-if="!scope.row.gpsSubscribe" title="位置订阅" class="subscribe-off iconfont icon-gxunjians" ></i>
            </template>
          </el-table-column>

          <el-table-column label="操作" width="300" align="center" fixed="right">
            <template slot-scope="scope">
@@ -169,3 +179,13 @@
  }
};
</script>
<style>
.subscribe-on{
  color: #409EFF;
  font-size: 1.3rem;
}
.subscribe-off{
  color: #afafb3;
  font-size: 1.3rem;
}
</style>

 web_src/src/components/dialog/chooseChannelForGb.vue

@@ -1,8 +1,8 @@
<template>
<div id="chooseChannelForGb" >
  <div style="font-size: 17px; color: #606060; white-space: nowrap; line-height: 30px; font-family: monospace;">
    <span v-if="catalogId == null">{{catalogName}}的直播流</span>
    <span v-if="catalogId != null">{{catalogName}}({{catalogId}})的直播流</span>
    <span v-if="catalogId == null">{{catalogName}}的国标通道</span>
    <span v-if="catalogId != null">{{catalogName}}({{catalogId}})的国标通道</span>
  </div>
   <div style="background-color: #FFFFFF; position: relative; padding: 0.5rem; text-align: left;font-size: 14px;">
        搜索: <el-input @input="search" style="margin-right: 1rem; width: auto;" size="mini" placeholder="关键字" prefix-icon="el-icon-search" v-model="searchSrt" clearable> </el-input>

 web_src/src/components/dialog/chooseChannelForStream.vue

@@ -1,8 +1,8 @@
<template>
<div id="chooseChannelFoStream" >
    <div style="font-size: 17px; color: #606060; white-space: nowrap; line-height: 30px; font-family: monospace;">
      <span v-if="catalogId == null">{{catalogName}}的直播流</span>
      <span v-if="catalogId != null">{{catalogName}}({{catalogId}})的直播流</span>
      <span v-if="catalogId == null">{{catalogName}}的直播通道</span>
      <span v-if="catalogId != null">{{catalogName}}({{catalogId}})的直播通道</span>
    </div>
    <div style="background-color: #FFFFFF; margin-bottom: 1rem; position: relative; padding: 0.5rem; text-align: left;font-size: 14px;">


 web_src/static/css/iconfont.css

@@ -1,8 +1,8 @@
@font-face {
  font-family: "iconfont"; /* Project id 1291092 */
  src: url('iconfont.woff2?t=1644809302709') format('woff2'),
       url('iconfont.woff?t=1644809302709') format('woff'),
       url('iconfont.ttf?t=1644809302709') format('truetype');
  src: url('iconfont.woff2?t=1647245982270') format('woff2'),
       url('iconfont.woff?t=1647245982270') format('woff'),
       url('iconfont.ttf?t=1647245982270') format('truetype');
}

.iconfont {
@@ -13,6 +13,22 @@
  -moz-osx-font-smoothing: grayscale;
}

.icon-xitongxinxi:before {
  content: "\e7d6";
}

.icon-gbaojings:before {
  content: "\e7d7";
}

.icon-gjichus:before {
  content: "\e7d8";
}

.icon-gxunjians:before {
  content: "\e7d9";
}

.icon-ziyuan:before {
  content: "\e7d5";
}

 web_src/static/css/iconfont.woff2

Binary files differ