Merge remote-tracking branch 'gitee.com/wvp-pro-record' into wvp-28181-2.0

648540858

2022-03-04 ee746e53cda8e35ede1d966b582160caeb1ca562

Merge remote-tracking branch 'gitee.com/wvp-pro-record' into wvp-28181-2.0

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

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

@@ -4,11 +4,6 @@
public class Device {

    /**
     * Id
     */
    private int id;

    /**
     * 设备Id
     */
    private String deviceId;
@@ -119,13 +114,7 @@
     */
    private int subscribeCycleForCatalog ;

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public String getDeviceId() {
        return deviceId;
@@ -294,6 +283,4 @@
    public void setSubscribeCycleForCatalog(int subscribeCycleForCatalog) {
        this.subscribeCycleForCatalog = subscribeCycleForCatalog;
    }


}

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

@@ -8,10 +8,6 @@

public class MobilePosition {
    /**
     * Id
     */
    private int id;
    /**
     * 设备Id
     */
    private String deviceId;
@@ -76,13 +72,6 @@
     */
    private String cnLat;

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public String getDeviceId() {
        return deviceId;

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

@@ -19,7 +19,9 @@
    private String name;

    

    private String filePath;

	


    private String fileSize;



    private String address;

    

    private String startTime;

@@ -104,6 +106,14 @@
        this.recorderId = recorderId;

    }



    public String getFileSize() {

        return fileSize;

    }



    public void setFileSize(String fileSize) {

        this.fileSize = fileSize;

    }



    @Override

    public int compareTo(@NotNull RecordItem recordItem) {

        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");


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

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

import javax.sdp.SessionDescription;

public class SDPInfo {
    private byte[] source;
    private SessionDescription sdpSource;
    private String sessionName;
    private Long startTime;
    private Long stopTime;
    private String username;
    private String address;
    private String ssrc;
}

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

@@ -71,6 +71,16 @@
     */
    private String mediaServerId;

    /**
     *  invite的callId
     */
    private String CallId;

    /**
     * 是否是play， false是playback
     */
    private boolean isPlay;

    public String getIp() {
        return ip;
    }
@@ -174,4 +184,20 @@
    public void setMediaServerId(String mediaServerId) {
        this.mediaServerId = mediaServerId;
    }

    public String getCallId() {
        return CallId;
    }

    public void setCallId(String callId) {
        CallId = callId;
    }

    public boolean isPlay() {
        return isPlay;
    }

    public void setPlay(boolean play) {
        isPlay = play;
    }
}

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

@@ -1,12 +1,11 @@
package com.genersoft.iot.vmp.gb28181.event;



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

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

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

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

import com.genersoft.iot.vmp.gb28181.event.offline.OfflineEvent;

import com.genersoft.iot.vmp.gb28181.event.platformKeepaliveExpire.PlatformKeepaliveExpireEvent;

import com.genersoft.iot.vmp.gb28181.event.platformNotRegister.PlatformCycleRegisterEvent;

import com.genersoft.iot.vmp.gb28181.event.platformNotRegister.PlatformNotRegisterEvent;

import com.genersoft.iot.vmp.gb28181.event.record.RecordEndEvent;

import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;

import com.genersoft.iot.vmp.media.zlm.event.ZLMOfflineEvent;

import com.genersoft.iot.vmp.media.zlm.event.ZLMOnlineEvent;

@@ -15,7 +14,6 @@
import org.springframework.scheduling.annotation.Async;

import org.springframework.stereotype.Component;



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

import com.genersoft.iot.vmp.gb28181.event.alarm.AlarmEvent;

import com.genersoft.iot.vmp.gb28181.event.online.OnlineEvent;



@@ -145,4 +143,11 @@
        gbStreamList.add(gbStream);

        catalogEventPublishForStream(platformId, gbStreamList, type);

    }



    public void recordEndEventPush(RecordInfo recordInfo) {

        RecordEndEvent outEvent = new RecordEndEvent(this);

        outEvent.setRecordInfo(recordInfo);

        applicationEventPublisher.publishEvent(outEvent);

    }



}


 src/main/java/com/genersoft/iot/vmp/gb28181/event/offline/KeepaliveTimeoutListenerForPlatform.java

@@ -76,10 +76,7 @@
            eventResult.callId = callid;
            eventResult.msg = "注册超时";
            eventResult.type = "register timeout";
            if (sipSubscribe.getErrorSubscribe(callid) != null) {
                sipSubscribe.getErrorSubscribe(callid).response(eventResult);
            }

            sipSubscribe.getErrorSubscribe(callid).response(eventResult);
        }

    }

 src/main/java/com/genersoft/iot/vmp/gb28181/event/record/RecordEndEventListener.java

@@ -1,5 +1,6 @@
package com.genersoft.iot.vmp.gb28181.event.record;

import com.genersoft.iot.vmp.gb28181.bean.RecordInfo;
import com.genersoft.iot.vmp.gb28181.bean.RecordItem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -23,12 +24,8 @@

    private static Map<String, SseEmitter> sseEmitters = new Hashtable<>();

    public void addSseEmitters(String browserId, SseEmitter sseEmitter) {
        sseEmitters.put(browserId, sseEmitter);
    }

    public interface RecordEndEventHandler{
        void  handler(List<RecordItem> recordItems);
        void  handler(RecordInfo recordInfo);
    }

    private Map<String, RecordEndEventHandler> handlerMap = new HashMap<>();
@@ -38,6 +35,15 @@
            logger.debug("录像查询完成事件触发，deviceId：{}, channelId: {}, 录像数量{}条", event.getRecordInfo().getDeviceId(),
                    event.getRecordInfo().getChannelId(), event.getRecordInfo().getRecordList().size() );
        }
        if (handlerMap.size() > 0) {
            for (RecordEndEventHandler recordEndEventHandler : handlerMap.values()) {
                recordEndEventHandler.handler(event.getRecordInfo());
            }
        }

    }

    public void addEndEventHandler(String device, String channelId, RecordEndEventHandler recordEndEventHandler) {
        handlerMap.put(device + channelId, recordEndEventHandler);
    }
}

 src/main/java/com/genersoft/iot/vmp/gb28181/session/SsrcConfig.java

@@ -81,7 +81,6 @@
            isUsed.remove(sn);
            notUsed.add(sn);
        }catch (NullPointerException e){
            System.out.printf("11111");
        }
    }


 src/main/java/com/genersoft/iot/vmp/gb28181/task/GPSSubscribeTask.java

@@ -36,7 +36,6 @@

        SubscribeInfo subscribe = redisCatchStorage.getSubscribe(key);
        if (subscribe != null) {
            System.out.println("发送GPS消息");
            ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformId);
            if (parentPlatform == null || parentPlatform.isStatus()) {
                // TODO 暂时只处理视频流的回复,后续增加对国标设备的支持

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

@@ -141,7 +141,6 @@
     */
    @Override
    public void processTimeout(TimeoutEvent timeoutEvent) {
        System.out.println("processTimeout");
        if(timeoutProcessor != null) {
            timeoutProcessor.process(timeoutEvent);
        }
@@ -173,7 +172,6 @@

    @Override
    public void processDialogTerminated(DialogTerminatedEvent dialogTerminatedEvent) {
        System.out.println("processDialogTerminated");
        CallIdHeader callId = dialogTerminatedEvent.getDialog().getCallId();
    }


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

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

import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.RecordInfo;
import com.genersoft.iot.vmp.gb28181.bean.SubscribeInfo;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
@@ -87,4 +88,12 @@
     */
    boolean sendNotifyForCatalogOther(String type, ParentPlatform parentPlatform, List<DeviceChannel> deviceChannels, SubscribeInfo subscribeInfo, Integer index);

    /**
     * 回复recordInfo
     * @param deviceChannel 通道信息
     * @param parentPlatform 平台信息
     * @param fromTag fromTag
     * @param recordInfo 录像信息
     */
    boolean recordInfo(DeviceChannel deviceChannel, ParentPlatform parentPlatform, String fromTag, RecordInfo recordInfo);
}

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

@@ -346,8 +346,11 @@
            subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,

                    (MediaServerItem mediaServerItemInUse, JSONObject json)->{

                if (userSetup.isWaitTrack() && json.getJSONArray("tracks") == null) return;

                event.response(mediaServerItemInUse, json);

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

                if (event != null) {

                    event.response(mediaServerItemInUse, json);

                }



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

            });

            //

            StringBuffer content = new StringBuffer(200);

@@ -450,13 +453,16 @@
            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)->{

                        System.out.println(344444);

                if (userSetup.isWaitTrack() && json.getJSONArray("tracks") == null) return;

                event.response(mediaServerItemInUse, json);

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

                if (event != null) {

                    event.response(mediaServerItemInUse, json);

                }

            });



            StringBuffer content = new StringBuffer(200);

@@ -713,6 +719,7 @@
            if (ssrcTransaction != null) {

                MediaServerItem mediaServerItem = mediaServerService.getOne(ssrcTransaction.getMediaServerId());

                mediaServerService.releaseSsrc(mediaServerItem, ssrcTransaction.getSsrc());

                mediaServerService.closeRTPServer(deviceId, channelId, ssrcTransaction.getStream());

                streamSession.remove(deviceId, channelId, ssrcTransaction.getStream());

            }

        } catch (SipException | ParseException e) {

@@ -1203,7 +1210,6 @@
        if (type == null) {

            type = "all";

        }



        try {

            StringBuffer recordInfoXml = new StringBuffer(200);

            recordInfoXml.append("<?xml version=\"1.0\" encoding=\"GB2312\"?>\r\n");

@@ -1211,11 +1217,19 @@
            recordInfoXml.append("<CmdType>RecordInfo</CmdType>\r\n");

            recordInfoXml.append("<SN>" + sn + "</SN>\r\n");

            recordInfoXml.append("<DeviceID>" + channelId + "</DeviceID>\r\n");

            recordInfoXml.append("<StartTime>" + DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(startTime) + "</StartTime>\r\n");

            recordInfoXml.append("<EndTime>" + DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(endTime) + "</EndTime>\r\n");

            recordInfoXml.append("<Secrecy> "+ secrecy + " </Secrecy>\r\n");

            // 大华NVR要求必须增加一个值为all的文本元素节点Type

            recordInfoXml.append("<Type>" + type+"</Type>\r\n");

            if (startTime != null) {

                recordInfoXml.append("<StartTime>" + DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(startTime) + "</StartTime>\r\n");

            }

            if (endTime != null) {

                recordInfoXml.append("<EndTime>" + DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(endTime) + "</EndTime>\r\n");

            }

            if (secrecy != null) {

                recordInfoXml.append("<Secrecy> "+ secrecy + " </Secrecy>\r\n");

            }

            if (type != null) {

                // 大华NVR要求必须增加一个值为all的文本元素节点Type

                recordInfoXml.append("<Type>" + type+"</Type>\r\n");

            }

            recordInfoXml.append("</Query>\r\n");

            

            String tm = Long.toString(System.currentTimeMillis());


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

@@ -1,12 +1,10 @@
package com.genersoft.iot.vmp.gb28181.transmit.cmd.impl;

import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch;
import com.genersoft.iot.vmp.gb28181.bean.SubscribeInfo;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.SIPRequestHeaderPlarformProvider;
import com.genersoft.iot.vmp.gb28181.utils.DateUtil;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import org.slf4j.Logger;
@@ -17,6 +15,7 @@
import org.springframework.context.annotation.Lazy;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.sip.*;
import javax.sip.header.CallIdHeader;
@@ -91,7 +90,7 @@

                sipSubscribe.addErrorSubscribe(callIdHeader.getCallId(), (event)->{
                    if (event != null) {
                        logger.info("向上级平台 [ {} ] 注册发上错误： {} ",
                        logger.info("向上级平台 [ {} ] 注册发生错误： {} ",
                                parentPlatform.getServerGBId(),
                                event.msg);
                    }
@@ -496,5 +495,52 @@
        catalogXml.append("</Notify>\r\n");
        return catalogXml.toString();
    }
    @Override
    public boolean recordInfo(DeviceChannel deviceChannel, ParentPlatform parentPlatform, String fromTag, RecordInfo recordInfo) {
        if ( parentPlatform ==null) {
            return false;
        }
        try {
            StringBuffer recordXml = new StringBuffer(600);
            recordXml.append("<?xml version=\"1.0\" encoding=\"GB2312\"?>\r\n");
            recordXml.append("<Response>\r\n");
            recordXml.append("<CmdType>RecordInfo</CmdType>\r\n");
            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");
                    }
                }
                recordXml.append("</Item>\r\n");
            }

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

            // callid
            CallIdHeader callIdHeader = parentPlatform.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
                    : udpSipProvider.getNewCallId();
            Request request = headerProviderPlarformProvider.createMessageRequest(parentPlatform, recordXml.toString(), fromTag, callIdHeader);
            transmitRequest(parentPlatform, request);

        } catch (SipException | ParseException | InvalidArgumentException e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }
}

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

@@ -1,10 +1,13 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
@@ -24,6 +27,8 @@
import javax.sip.header.ToHeader;
import java.util.HashMap;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;

/**
 * SIP命令类型： ACK请求
@@ -52,6 +57,9 @@
    @Autowired
    private IMediaServerService mediaServerService;

    @Autowired
    private ZLMHttpHookSubscribe subscribe;


    /**   
     * 处理  ACK请求
@@ -60,6 +68,7 @@
     */
    @Override
    public void process(RequestEvent evt) {
        logger.info("ACK请求： {}", ((System.currentTimeMillis())));
        Dialog dialog = evt.getDialog();
        if (dialog == null) return;
        if (dialog.getState()== DialogState.CONFIRMED) {
@@ -69,16 +78,17 @@
            String is_Udp = sendRtpItem.isTcp() ? "0" : "1";
            String deviceId = sendRtpItem.getDeviceId();
            StreamInfo streamInfo = null;
            if (deviceId == null) {
            if (sendRtpItem.isPlay()) {
                streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
            }else {
                streamInfo = redisCatchStorage.queryPlaybackByDevice(deviceId, channelId);
            }
            System.out.println(JSON.toJSON(streamInfo));
            if (streamInfo == null) {
                streamInfo = new StreamInfo();
                streamInfo.setApp(sendRtpItem.getApp());
                streamInfo.setStream(sendRtpItem.getStreamId());
            }else {
                streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
                sendRtpItem.setStreamId(streamInfo.getStream());
                streamInfo.setApp("rtp");
            }

            redisCatchStorage.updateSendRTPSever(sendRtpItem);
            logger.info(platformGbId);
            logger.info(channelId);
@@ -90,34 +100,42 @@
            param.put("dst_url",sendRtpItem.getIp());
            param.put("dst_port", sendRtpItem.getPort());
            param.put("is_udp", is_Udp);
            //param.put ("src_port", sendRtpItem.getLocalPort());
            // 设备推流查询，成功后才能转推
            boolean rtpPushed = false;
            long startTime = System.currentTimeMillis();
            while (!rtpPushed) {
                try {
                    if (System.currentTimeMillis() - startTime < 30 * 1000) {
                        MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
                        if (zlmrtpServerFactory.isStreamReady(mediaInfo, streamInfo.getApp(), streamInfo.getStream())) {
                            rtpPushed = true;
                            logger.info("已获取设备推流[{}/{}]，开始向上级推流[{}:{}]",
                                    streamInfo.getApp() ,streamInfo.getStream(), sendRtpItem.getIp(), sendRtpItem.getPort());
                            zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
                        } else {
                            logger.info("等待设备推流[{}/{}].......",
                                    streamInfo.getApp() ,streamInfo.getStream());
                            Thread.sleep(1000);
                            continue;
                        }
                    } else {
                        rtpPushed = true;
                        logger.info("设备推流[{}/{}]超时，终止向上级推流",
                                streamInfo.getApp() ,streamInfo.getStream());
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
            zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
//            if (zlmrtpServerFactory.isStreamReady(mediaInfo, streamInfo.getApp(), streamInfo.getStreamId())) {
//                logger.info("已获取设备推流[{}/{}]，开始向上级推流[{}:{}]",
//                        streamInfo.getApp() ,streamInfo.getStreamId(), sendRtpItem.getIp(), sendRtpItem.getPort());
//                zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
//            } else {
//                // 对hook进行订阅
//                logger.info("等待设备推流[{}/{}].......",
//                        streamInfo.getApp(), streamInfo.getStreamId());
//                Timer timer = new Timer();
//                timer.schedule(new TimerTask() {
//                    @Override
//                    public void run() {
//                        logger.info("设备推流[{}/{}]超时，终止向上级推流",
//                                finalStreamInfo.getApp() , finalStreamInfo.getStreamId());
//
//                    }
//                }, 30*1000L);
//                // 添加订阅
//                JSONObject subscribeKey = new JSONObject();
//                subscribeKey.put("app", "rtp");
//                subscribeKey.put("stream", streamInfo.getStreamId());
//                subscribeKey.put("mediaServerId", streamInfo.getMediaServerId());
//                subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_publish, subscribeKey,
//                        (MediaServerItem mediaServerItemInUse, JSONObject json) -> {
//                            logger.info("已获取设备推流[{}/{}]，开始向上级推流[{}:{}]",
//                                    finalStreamInfo.getApp(), finalStreamInfo.getStreamId(), sendRtpItem.getIp(), sendRtpItem.getPort());
//                            timer.cancel();
//                            zlmrtpServerFactory.startSendRtpStream(mediaInfo, param);
//                            subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey);
//                        });
//            }


        }
    }
}

 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.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.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
@@ -87,22 +88,34 @@
                    MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
                    zlmrtpServerFactory.stopSendRtpStream(mediaInfo, param);
                    redisCatchStorage.deleteSendRTPServer(platformGbId, channelId);
                    if (zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId) == 0) {
                    int totalReaderCount = zlmrtpServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId);
                    if (totalReaderCount == 0) {
                        logger.info(streamId + "无其它观看者，通知设备停止推流");
                        cmder.streamByeCmd(sendRtpItem.getDeviceId(), channelId, streamId);
                    }else if (totalReaderCount == -1){
                        logger.warn(streamId + " 查找其它观看者失败");
                    }
                }
                // 可能是设备主动停止
                Device device = storager.queryVideoDeviceByChannelId(platformGbId);
                if (device != null) {
                if (device != null) {
                    StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(device.getDeviceId(), channelId);
                    if (sendRtpItem != null) {
                        if (sendRtpItem.isPlay()) {
                            if (streamInfo != null) {
                                redisCatchStorage.stopPlay(streamInfo);
                            }
                        }else {
                            if (streamInfo != null) {
                                redisCatchStorage.stopPlayback(streamInfo);
                            }
                        }

                    if (streamInfo != null) {
                        redisCatchStorage.stopPlay(streamInfo);
                        storager.stopPlay(device.getDeviceId(), channelId);
                        mediaServerService.closeRTPServer(device.getDeviceId(), channelId, streamInfo.getStream());
                    }
                    storager.stopPlay(device.getDeviceId(), channelId);
                    mediaServerService.closeRTPServer(device, channelId, streamInfo.getStream());
                }

            }
        } catch (SipException e) {
            e.printStackTrace();

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

@@ -1,18 +1,29 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IPlayService;
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.PlayResult;
import gov.nist.javax.sdp.TimeDescriptionImpl;
import gov.nist.javax.sdp.fields.TimeField;
import gov.nist.javax.sip.address.AddressImpl;
import gov.nist.javax.sip.address.SipUri;
import org.slf4j.Logger;
@@ -27,10 +38,13 @@
import javax.sip.ServerTransaction;
import javax.sip.SipException;
import javax.sip.address.SipURI;
import javax.sip.header.CallIdHeader;
import javax.sip.header.FromHeader;
import javax.sip.message.Request;
import javax.sip.message.Response;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.Vector;

@@ -61,6 +75,9 @@
    private IPlayService playService;

    @Autowired
    private ISIPCommander commander;

    @Autowired
    private ZLMRTPServerFactory zlmrtpServerFactory;

    @Autowired
@@ -68,6 +85,7 @@

    @Autowired
    private SIPProcessorObserver sipProcessorObserver;


    @Override
    public void afterPropertiesSet() throws Exception {
@@ -88,22 +106,19 @@
            Request request = evt.getRequest();
            SipURI sipURI = (SipURI) request.getRequestURI();
            String channelId = sipURI.getUser();
            String requesterId = null;

            FromHeader fromHeader = (FromHeader)request.getHeader(FromHeader.NAME);
            AddressImpl address = (AddressImpl) fromHeader.getAddress();
            SipUri uri = (SipUri) address.getURI();
            requesterId = uri.getUser();

            String requesterId = SipUtils.getUserIdFromFromHeader(request);
            CallIdHeader callIdHeader = (CallIdHeader)request.getHeader(CallIdHeader.NAME);
            if (requesterId == null || channelId == null) {
                logger.info("无法从FromHeader的Address中获取到平台id，返回400");
                responseAck(evt, Response.BAD_REQUEST); // 参数不全， 发400，请求错误
                return;
            }

            // 查询请求方是否上级平台
            // 查询请求是否来自上级平台\设备
            ParentPlatform platform = storager.queryParentPlatByServerGBId(requesterId);
            if (platform != null) {
            if (platform == null) {
                inviteFromDeviceHandle(evt, requesterId);
            }else {
                // 查询平台下是否有该通道
                DeviceChannel channel = storager.queryChannelInParentPlatform(requesterId, channelId);
                GbStream gbStream = storager.queryStreamInParentPlatform(requesterId, channelId);
@@ -122,7 +137,7 @@
                    mediaServerItem = mediaServerService.getOne(mediaServerId);
                    if (mediaServerItem == null) {
                        logger.info("[ app={}, stream={} ]找不到zlm {}，返回410",gbStream.getApp(), gbStream.getStream(), mediaServerId);
                        responseAck(evt, Response.GONE, "media server not found");
                        responseAck(evt, Response.GONE);
                        return;
                    }
                    Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream());
@@ -158,13 +173,26 @@
                    ssrc = ssrcDefault;
                    sdp = SdpFactory.getInstance().createSessionDescription(contentString);
                }
                String sessionName = sdp.getSessionName().getValue();

                Long startTime = null;
                Long stopTime = null;
                Date start = null;
                Date end = null;
                if (sdp.getTimeDescriptions(false) != null && sdp.getTimeDescriptions(false).size() > 0) {
                    TimeDescriptionImpl timeDescription = (TimeDescriptionImpl)(sdp.getTimeDescriptions(false).get(0));
                    TimeField startTimeFiled = (TimeField)timeDescription.getTime();
                    startTime = startTimeFiled.getStartTime();
                    stopTime = startTimeFiled.getStopTime();

                    start = new Date(startTime*1000);
                    end = new Date(stopTime*1000);
                }
                //  获取支持的格式
                Vector mediaDescriptions = sdp.getMediaDescriptions(true);
                // 查看是否支持PS 负载96
                //String ip = null;
                int port = -1;
                //boolean recvonly = false;
                boolean mediaTransmissionTCP = false;
                Boolean tcpActive = null;
                for (Object description : mediaDescriptions) {
@@ -200,7 +228,6 @@
                }
                String username = sdp.getOrigin().getUsername();
                String addressStr = sdp.getOrigin().getAddress();
                //String sessionName = sdp.getSessionName().getValue();
                logger.info("[上级点播]用户：{}， 地址：{}:{}， ssrc：{}", username, addressStr, port, ssrc);
                Device device  = null;
                // 通过 channel 和 gbStream 是否为null 值判断来源是直播流合适国标
@@ -228,23 +255,33 @@
                        responseAck(evt, Response.BUSY_HERE);
                        return;
                    }

                    sendRtpItem.setCallId(callIdHeader.getCallId());
                    sendRtpItem.setPlay("Play".equals(sessionName));
                    // 写入redis， 超时时回复
                    redisCatchStorage.updateSendRTPSever(sendRtpItem);
                    // 通知下级推流，
                    PlayResult playResult = playService.play(mediaServerItem,device.getDeviceId(), channelId, (mediaServerItemInUSe, responseJSON)->{
                        // 收到推流， 回复200OK, 等待ack
                        // if (sendRtpItem == null) return;

                    Device finalDevice = device;
                    MediaServerItem finalMediaServerItem = mediaServerItem;
                    Long finalStartTime = startTime;
                    Long finalStopTime = stopTime;
                    ZLMHttpHookSubscribe.Event hookEvent = (mediaServerItemInUSe, responseJSON)->{
                        logger.info("[上级点播]下级已经开始推流。 回复200OK(SDP)， {}/{}", sendRtpItem.getApp(), sendRtpItem.getStreamId());
                        //     * 0 等待设备推流上来
                        //     * 1 下级已经推流，等待上级平台回复ack
                        //     * 2 推流中
                        sendRtpItem.setStatus(1);
                        redisCatchStorage.updateSendRTPSever(sendRtpItem);
                        // TODO 添加对tcp的支持

                        StringBuffer content = new StringBuffer(200);
                        content.append("v=0\r\n");
                        content.append("o="+ channelId +" 0 0 IN IP4 "+mediaServerItemInUSe.getSdpIp()+"\r\n");
                        content.append("s=Play\r\n");
                        content.append("s=" + sessionName+"\r\n");
                        content.append("c=IN IP4 "+mediaServerItemInUSe.getSdpIp()+"\r\n");
                        content.append("t=0 0\r\n");
                        if ("Playback".equals(sessionName)) {
                            content.append("t=" + finalStartTime + " " + finalStopTime + "\r\n");
                        }else {
                            content.append("t=0 0\r\n");
                        }
                        content.append("m=video "+ sendRtpItem.getLocalPort()+" RTP/AVP 96\r\n");
                        content.append("a=sendonly\r\n");
                        content.append("a=rtpmap:96 PS/90000\r\n");
@@ -260,7 +297,8 @@
                        } catch (ParseException e) {
                            e.printStackTrace();
                        }
                    } ,((event) -> {
                    };
                    SipSubscribe.Event errorEvent = ((event) -> {
                        // 未知错误。直接转发设备点播的错误
                        Response response = null;
                        try {
@@ -271,11 +309,46 @@
                        } catch (ParseException | SipException | InvalidArgumentException e) {
                            e.printStackTrace();
                        }
                    }));
                    if (logger.isDebugEnabled()) {
                        logger.debug(playResult.getResult().toString());
                    });
                    if ("Playback".equals(sessionName)) {
                        sendRtpItem.setPlay(false);
                        sendRtpItem.setStreamId(ssrc);
                        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());
                                }
                                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) {
                            if (mediaServerItem.isRtpEnable()) {
                                sendRtpItem.setStreamId(String.format("%s_%s", device.getDeviceId(), channelId));
                            }
                            sendRtpItem.setPlay(false);
                            playService.play(mediaServerItem,device.getDeviceId(), channelId, hookEvent,errorEvent);
                        }else {
                            sendRtpItem.setStreamId(streamInfo.getStream());
                            hookEvent.response(mediaServerItem, null);
                        }
                    }

                }else if (gbStream != null) {
                    SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
                            gbStream.getApp(), gbStream.getStream(), channelId,
@@ -295,7 +368,6 @@

                    sendRtpItem.setStatus(1);
                    redisCatchStorage.updateSendRTPSever(sendRtpItem);
                    // TODO 添加对tcp的支持
                    StringBuffer content = new StringBuffer(200);
                    content.append("v=0\r\n");
                    content.append("o="+ channelId +" 0 0 IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
@@ -319,72 +391,6 @@
                    }
                }

            } else {
                // 非上级平台请求，查询是否设备请求（通常为接收语音广播的设备）
                Device device = redisCatchStorage.getDevice(requesterId);
                if (device != null) {
                    logger.info("收到设备" + requesterId + "的语音广播Invite请求");
                    responseAck(evt, Response.TRYING);

                    String contentString = new String(request.getRawContent());
                    // jainSip不支持y=字段， 移除移除以解析。
                    String substring = contentString;
                    String ssrc = "0000000404";
                    int ssrcIndex = contentString.indexOf("y=");
                    if (ssrcIndex > 0) {
                        substring = contentString.substring(0, ssrcIndex);
                        ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12);
                    }
                    ssrcIndex = substring.indexOf("f=");
                    if (ssrcIndex > 0) {
                        substring = contentString.substring(0, ssrcIndex);
                    }
                    SessionDescription sdp = SdpFactory.getInstance().createSessionDescription(substring);

                    //  获取支持的格式
                    Vector mediaDescriptions = sdp.getMediaDescriptions(true);
                    // 查看是否支持PS 负载96
                    int port = -1;
                    //boolean recvonly = false;
                    boolean mediaTransmissionTCP = false;
                    Boolean tcpActive = null;
                    for (int i = 0; i < mediaDescriptions.size(); i++) {
                        MediaDescription mediaDescription = (MediaDescription)mediaDescriptions.get(i);
                        Media media = mediaDescription.getMedia();

                        Vector mediaFormats = media.getMediaFormats(false);
                        if (mediaFormats.contains("8")) {
                            port = media.getMediaPort();
                            String protocol = media.getProtocol();
                            // 区分TCP发流还是udp， 当前默认udp
                            if ("TCP/RTP/AVP".equals(protocol)) {
                                String setup = mediaDescription.getAttribute("setup");
                                if (setup != null) {
                                    mediaTransmissionTCP = true;
                                    if ("active".equals(setup)) {
                                        tcpActive = true;
                                    } else if ("passive".equals(setup)) {
                                        tcpActive = false;
                                    }
                                }
                            }
                            break;
                        }
                    }
                    if (port == -1) {
                        logger.info("不支持的媒体格式，返回415");
                        // 回复不支持的格式
                        responseAck(evt, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415
                        return;
                    }
                    String username = sdp.getOrigin().getUsername();
                    String addressStr = sdp.getOrigin().getAddress();
                    logger.info("设备{}请求语音流，地址：{}:{}，ssrc：{}", username, addressStr, port, ssrc);

                } else {
                    logger.warn("来自无效设备/平台的请求");
                    responseAck(evt, Response.BAD_REQUEST);
                }
            }

        } catch (SipException | InvalidArgumentException | ParseException e) {
@@ -397,4 +403,74 @@
            e.printStackTrace();
        }
    }

    public void inviteFromDeviceHandle(RequestEvent evt, String requesterId) throws InvalidArgumentException, ParseException, SipException, SdpException {

        // 非上级平台请求，查询是否设备请求（通常为接收语音广播的设备）
        Device device = redisCatchStorage.getDevice(requesterId);
        Request request = evt.getRequest();
        if (device != null) {
            logger.info("收到设备" + requesterId + "的语音广播Invite请求");
            responseAck(evt, Response.TRYING);

            String contentString = new String(request.getRawContent());
            // jainSip不支持y=字段， 移除移除以解析。
            String substring = contentString;
            String ssrc = "0000000404";
            int ssrcIndex = contentString.indexOf("y=");
            if (ssrcIndex > 0) {
                substring = contentString.substring(0, ssrcIndex);
                ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12);
            }
            ssrcIndex = substring.indexOf("f=");
            if (ssrcIndex > 0) {
                substring = contentString.substring(0, ssrcIndex);
            }
            SessionDescription sdp = SdpFactory.getInstance().createSessionDescription(substring);

            //  获取支持的格式
            Vector mediaDescriptions = sdp.getMediaDescriptions(true);
            // 查看是否支持PS 负载96
            int port = -1;
            //boolean recvonly = false;
            boolean mediaTransmissionTCP = false;
            Boolean tcpActive = null;
            for (int i = 0; i < mediaDescriptions.size(); i++) {
                MediaDescription mediaDescription = (MediaDescription)mediaDescriptions.get(i);
                Media media = mediaDescription.getMedia();

                Vector mediaFormats = media.getMediaFormats(false);
                if (mediaFormats.contains("8")) {
                    port = media.getMediaPort();
                    String protocol = media.getProtocol();
                    // 区分TCP发流还是udp， 当前默认udp
                    if ("TCP/RTP/AVP".equals(protocol)) {
                        String setup = mediaDescription.getAttribute("setup");
                        if (setup != null) {
                            mediaTransmissionTCP = true;
                            if ("active".equals(setup)) {
                                tcpActive = true;
                            } else if ("passive".equals(setup)) {
                                tcpActive = false;
                            }
                        }
                    }
                    break;
                }
            }
            if (port == -1) {
                logger.info("不支持的媒体格式，返回415");
                // 回复不支持的格式
                responseAck(evt, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415
                return;
            }
            String username = sdp.getOrigin().getUsername();
            String addressStr = sdp.getOrigin().getAddress();
            logger.info("设备{}请求语音流，地址：{}:{}，ssrc：{}", username, addressStr, port, ssrc);

        } else {
            logger.warn("来自无效设备/平台的请求");
            responseAck(evt, Response.BAD_REQUEST);
        }
    }
}

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

@@ -3,6 +3,7 @@
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.record.RecordEndEventListener;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
@@ -47,6 +48,9 @@
    private SIPCommander commander;

    @Autowired
    private RecordEndEventListener recordEndEventListener;

    @Autowired
    private SipConfig config;

    @Autowired
@@ -65,49 +69,89 @@
    @Override
    public void handForPlatform(RequestEvent evt, ParentPlatform parentPlatform, Element rootElement) {

        String key = DeferredResultHolder.CALLBACK_CMD_CATALOG + parentPlatform.getServerGBId();
        FromHeader fromHeader = (FromHeader) evt.getRequest().getHeader(FromHeader.NAME);
        try {
            // 回复200 OK
            responseAck(evt, Response.OK);
            Element snElement = rootElement.element("SN");
            int sn = Integer.parseInt(snElement.getText());
            Element deviceIDElement = rootElement.element("DeviceID");
            String channelId = deviceIDElement.getText();
            Element startTimeElement = rootElement.element("StartTime");
            String startTime = startTimeElement.getText();
            Element endTimeElement = rootElement.element("EndTime");
            String endTime = endTimeElement.getText();
            Element secrecyElement = rootElement.element("Secrecy");
            int secrecy = Integer.parseInt(secrecyElement.getText());
            Element typeElement = rootElement.element("Type");
            String type = typeElement.getText();
            // 确认是直播还是国标， 国标直接请求下级，直播请求录像管理服务
            List<ChannelSourceInfo> channelSources = storager.getChannelSource(parentPlatform.getServerGBId(), channelId);
            if (channelSources.get(0).getCount() > 0) { // 国标
                // 向国标设备请求录像数据
                Device device = storager.queryVideoDeviceByPlatformIdAndChannelId(parentPlatform.getServerGBId(), channelId);
                commander.recordInfoQuery(device, channelId, DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(startTime),
                        DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(endTime), sn, secrecy, type, (eventResult -> {
                            // 查询成功

                        }),(eventResult -> {
                            // 查询失败

                        }));

            }else if (channelSources.get(0).getCount() > 0) { // 直播流
                // TODO
            }else { // 错误的请求

            }
        } catch (SipException e) {
            e.printStackTrace();
        } catch (InvalidArgumentException e) {
            e.printStackTrace();
        } catch (ParseException e) {
            e.printStackTrace();
        Element snElement = rootElement.element("SN");
        int sn = Integer.parseInt(snElement.getText());
        Element deviceIDElement = rootElement.element("DeviceID");
        String channelId = deviceIDElement.getText();
        Element startTimeElement = rootElement.element("StartTime");
        String startTime = null;
        if (startTimeElement != null) {
            startTime = startTimeElement.getText();
        }
        Element endTimeElement = rootElement.element("EndTime");
        String endTime = null;
        if (endTimeElement != null) {
            endTime = endTimeElement.getText();
        }
        Element secrecyElement = rootElement.element("Secrecy");
        int secrecy = 0;
        if (secrecyElement != null) {
            secrecy = Integer.parseInt(secrecyElement.getText());
        }
        String type = "all";
        Element typeElement = rootElement.element("Type");
        if (typeElement != null) {
            type =  typeElement.getText();
        }
        // 确认是直播还是国标， 国标直接请求下级，直播请求录像管理服务
        List<ChannelSourceInfo> channelSources = storager.getChannelSource(parentPlatform.getServerGBId(), channelId);

        if (channelSources.get(0).getCount() > 0) { // 国标
            // 向国标设备请求录像数据
            Device device = storager.queryVideoDeviceByPlatformIdAndChannelId(parentPlatform.getServerGBId(), channelId);
            DeviceChannel deviceChannel = storager.queryChannelInParentPlatform(parentPlatform.getServerGBId(), channelId);
            // 接收录像数据
            recordEndEventListener.addEndEventHandler(deviceChannel.getDeviceId(), channelId, (recordInfo)->{
                cmderFroPlatform.recordInfo(deviceChannel, parentPlatform, fromHeader.getTag(), recordInfo);
            });
            commander.recordInfoQuery(device, channelId, DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(startTime),
                    DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(endTime), sn, secrecy, type, (eventResult -> {
                        // 回复200 OK
                        try {
                            responseAck(evt, Response.OK);
                        } catch (SipException e) {
                            e.printStackTrace();
                        } catch (InvalidArgumentException e) {
                            e.printStackTrace();
                        } catch (ParseException e) {
                            e.printStackTrace();
                        }
                    }),(eventResult -> {
                        // 查询失败
                        try {
                            responseAck(evt, eventResult.statusCode, eventResult.msg);
                        } catch (SipException e) {
                            e.printStackTrace();
                        } catch (InvalidArgumentException e) {
                            e.printStackTrace();
                        } catch (ParseException e) {
                            e.printStackTrace();
                        }
                    }));

        }else if (channelSources.get(1).getCount() > 0) { // 直播流
            // TODO
            try {
                responseAck(evt, Response.NOT_IMPLEMENTED); // 回复未实现
            } catch (SipException e) {
                e.printStackTrace();
            } catch (InvalidArgumentException e) {
                e.printStackTrace();
            } catch (ParseException e) {
                e.printStackTrace();
            }
        }else { // 错误的请求
            try {
                responseAck(evt, Response.BAD_REQUEST);
            } catch (SipException e) {
                e.printStackTrace();
            } catch (InvalidArgumentException e) {
                e.printStackTrace();
            } catch (ParseException e) {
                e.printStackTrace();
            }
        }
    }
}

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

@@ -4,6 +4,7 @@
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.RecordInfo;
import com.genersoft.iot.vmp.gb28181.bean.RecordItem;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.transmit.callback.CheckForAllRecordsThread;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
@@ -49,6 +50,9 @@
    @Autowired
    private DeferredResultHolder deferredResultHolder;

    @Autowired
    private EventPublisher eventPublisher;

    @Override
    public void afterPropertiesSet() throws Exception {
        responseMessageHandler.addHandler(cmdType, this);
@@ -77,6 +81,7 @@
            Element recordListElement = rootElement.element("RecordList");
            if (recordListElement == null || recordInfo.getSumNum() == 0) {
                logger.info("无录像数据");
                eventPublisher.recordEndEventPush(recordInfo);
                RequestMessage msg = new RequestMessage();
                msg.setKey(key);
                msg.setData(recordInfo);
@@ -99,6 +104,7 @@
                        record.setDeviceId(getText(itemRecord, "DeviceID"));
                        record.setName(getText(itemRecord, "Name"));
                        record.setFilePath(getText(itemRecord, "FilePath"));
                        record.setFileSize(getText(itemRecord, "FileSize"));
                        record.setAddress(getText(itemRecord, "Address"));
                        record.setStartTime(
                                DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(getText(itemRecord, "StartTime")));
@@ -112,7 +118,7 @@
                    }
                    recordInfo.setRecordList(recordList);
                }

                eventPublisher.recordEndEventPush(recordInfo);
                // 改用单独线程统计已获取录像文件数量，避免多包并行分别统计不完整的问题
                String cacheKey = CACHE_RECORDINFO_KEY + device.getDeviceId() + sn;
                redis.set(cacheKey + "_" + uuid, recordList, 90);

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

@@ -82,9 +82,6 @@
                requestURI.setPort(event.getRemotePort());
                reqAck.setRequestURI(requestURI);
                logger.info("向 " + event.getRemoteIpAddress() + ":" + event.getRemotePort() + "回复ack");
                SipURI sipURI = (SipURI)dialog.getRemoteParty().getURI();
                String deviceId = requestURI.getUser();
                String channelId = sipURI.getUser();

                dialog.sendAck(reqAck);


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

@@ -181,7 +181,7 @@
    @PostMapping(value = "/on_publish", produces = "application/json;charset=UTF-8")

    public ResponseEntity<String> onPublish(@RequestBody JSONObject json) {



        logger.debug("[ ZLM HOOK ]on_publish API调用，参数：" + json.toString());

        logger.info("[ ZLM HOOK ]on_publish API调用，参数：" + json.toString());

        JSONObject ret = new JSONObject();

        ret.put("code", 0);

        ret.put("msg", "success");


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

@@ -77,22 +77,23 @@
        if (eventMap == null) {
            return;
        }
        Iterator<Map.Entry<JSONObject, Event>> iterator = eventMap.entrySet().iterator();
        while (iterator.hasNext()){
            Map.Entry<JSONObject, Event> next = iterator.next();
            JSONObject key = next.getKey();
            Boolean result = null;
            for (String s : key.keySet()) {
                if (result == null) {
                    result = key.getString(s).equals(hookResponse.getString(s));
                }else {
                    if (key.getString(s) == null) continue;
                    result = result && key.getString(s).equals(hookResponse.getString(s));

        Set<Map.Entry<JSONObject, Event>> entries = eventMap.entrySet();
        if (entries.size() > 0) {
            for (Map.Entry<JSONObject, Event> entry : entries) {
                JSONObject key = entry.getKey();
                Boolean result = null;
                for (String s : key.keySet()) {
                    if (result == null) {
                        result = key.getString(s).equals(hookResponse.getString(s));
                    }else {
                        if (key.getString(s) == null) continue;
                        result = result && key.getString(s).equals(hookResponse.getString(s));
                    }
                }
            }
            if (null != result && result){
                // TODO 报错未处理
                iterator.remove();
                if (null != result && result){
                    entries.remove(entry);
                }
            }
        }
    }

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

@@ -242,9 +242,18 @@
     */
    public int totalReaderCount(MediaServerItem mediaServerItem, String app, String streamId) {
        JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo(mediaServerItem, app, "rtmp", streamId);
        Integer code = mediaInfo.getInteger("code");
        if (mediaInfo == null) {
            return 0;
        }
        if ( code < 0) {
            logger.warn("查询流({}/{})是否有其它观看者时得到： {}", app, streamId, mediaInfo.getString("msg"));
            return -1;
        }
        if ( code == 0 && ! mediaInfo.getBoolean("online")) {
            logger.warn("查询流({}/{})是否有其它观看者时得到： {}", app, streamId, mediaInfo.getString("msg"));
            return -1;
        }
        return mediaInfo.getInteger("totalReaderCount");
    }


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

@@ -48,7 +48,7 @@

    SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, boolean isPlayback);

    void closeRTPServer(Device device, String channelId, String ssrc);
    void closeRTPServer(String deviceId, String channelId, String ssrc);

    void clearRTPServer(MediaServerItem mediaServerItem);


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

@@ -1,9 +1,10 @@
package com.genersoft.iot.vmp.service.bean;

import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;

public interface PlayBackCallback {

    void call(RequestMessage msg);
    void call(PlayBackResult<RequestMessage> msg);

}

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

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

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

import javax.sip.RequestEvent;

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

    public int getCode() {
        return code;
    }

    public void setCode(int code) {
        this.code = code;
    }

    public T getData() {
        return data;
    }

    public void setData(T data) {
        this.data = data;
    }

    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 SipSubscribe.EventResult getEvent() {
        return event;
    }

    public void setEvent(SipSubscribe.EventResult event) {
        this.event = event;
    }
}

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

@@ -160,16 +160,16 @@
    }

    @Override
    public void closeRTPServer(Device device, String channelId, String stream) {
        String mediaServerId = streamSession.getMediaServerId(device.getDeviceId(), channelId, stream);
        String ssrc = streamSession.getSSRC(device.getDeviceId(), channelId, stream);
    public void closeRTPServer(String deviceId, String channelId, String stream) {
        String mediaServerId = streamSession.getMediaServerId(deviceId, channelId, stream);
        String ssrc = streamSession.getSSRC(deviceId, channelId, stream);
        MediaServerItem mediaServerItem = this.getOne(mediaServerId);
        if (mediaServerItem != null) {
            String streamId = String.format("%s_%s", device.getDeviceId(), channelId);
            String streamId = String.format("%s_%s", deviceId, channelId);
            zlmrtpServerFactory.closeRTPServer(mediaServerItem, streamId);
            releaseSsrc(mediaServerItem, ssrc);
        }
        streamSession.remove(device.getDeviceId(), channelId, stream);
        streamSession.remove(deviceId, channelId, stream);
    }

    @Override

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

@@ -17,9 +17,11 @@
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.PlayBackCallback;
import com.genersoft.iot.vmp.service.bean.PlayBackResult;
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
import com.genersoft.iot.vmp.utils.redis.RedisUtil;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.PlayResult;
import com.genersoft.iot.vmp.service.IMediaService;
@@ -52,6 +54,9 @@

    @Autowired
    private IRedisCatchStorage redisCatchStorage;

    @Autowired
    private RedisUtil redis;

    @Autowired
    private DeferredResultHolder resultHolder;
@@ -121,7 +126,6 @@
            // 点播结束时调用截图接口
            try {
                String classPath = ResourceUtils.getURL("classpath:").getPath();
                // System.out.println(classPath);
                // 兼容打包为jar的class路径
                if(classPath.contains("jar")) {
                    classPath = classPath.substring(0, classPath.lastIndexOf("."));
@@ -170,7 +174,10 @@
                WVPResult wvpResult = new WVPResult();
                wvpResult.setCode(-1);
                // 点播返回sip错误
                mediaServerService.closeRTPServer(playResult.getDevice(), channelId, ssrcInfo.getStream());
                mediaServerService.closeRTPServer(playResult.getDevice().getDeviceId(), channelId, ssrcInfo.getStream());
                // 释放ssrc
                mediaServerService.releaseSsrc(mediaServerItem, ssrcInfo.getSsrc());
                streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
                wvpResult.setMsg(String.format("点播失败， 错误码： %s, %s", event.statusCode, event.msg));
                msg.setData(wvpResult);
                resultHolder.invokeAllResult(msg);
@@ -219,7 +226,10 @@
                    logger.info("收到订阅消息： " + response.toJSONString());
                    onPublishHandlerForPlay(mediaServerItemInuse, response, deviceId, channelId, uuid);
                }, (event) -> {
                    mediaServerService.closeRTPServer(playResult.getDevice(), channelId, ssrcInfo.getStream());
                    mediaServerService.closeRTPServer(playResult.getDevice().getDeviceId(), channelId, ssrcInfo.getStream());
                    // 释放ssrc
                    mediaServerService.releaseSsrc(mediaServerItem, 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));
@@ -233,11 +243,11 @@
    }

    @Override
    public void onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject resonse, String deviceId, String channelId, String uuid) {
    public void onPublishHandlerForPlay(MediaServerItem mediaServerItem, JSONObject response, String deviceId, String channelId, String uuid) {
        RequestMessage msg = new RequestMessage();
        msg.setId(uuid);
        msg.setKey(DeferredResultHolder.CALLBACK_CMD_PLAY + deviceId + channelId);
        StreamInfo streamInfo = onPublishHandler(mediaServerItem, resonse, deviceId, channelId);
        StreamInfo streamInfo = onPublishHandler(mediaServerItem, response, deviceId, channelId);
        if (streamInfo != null) {
            DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
            if (deviceChannel != null) {
@@ -295,9 +305,12 @@
        RequestMessage msg = new RequestMessage();
        msg.setId(uuid);
        msg.setKey(key);
        PlayBackResult<RequestMessage> playBackResult = new PlayBackResult<>();
        result.onTimeout(()->{
            msg.setData("回放超时");
            callback.call(msg);
            playBackResult.setCode(-1);
            playBackResult.setData(msg);
            callback.call(playBackResult);
        });
        cmder.playbackStreamCmd(newMediaServerItem, ssrcInfo, device, channelId, startTime, endTime, (MediaServerItem mediaServerItem, JSONObject response) -> {
            logger.info("收到订阅消息： " + response.toJSONString());
@@ -305,15 +318,25 @@
            if (streamInfo == null) {
                logger.warn("设备回放API调用失败！");
                msg.setData("设备回放API调用失败！");
                callback.call(msg);
                playBackResult.setCode(-1);
                playBackResult.setData(msg);
                callback.call(playBackResult);
                return;
            }
            redisCatchStorage.startPlayback(streamInfo);
            msg.setData(JSON.toJSONString(streamInfo));
            callback.call(msg);
            playBackResult.setCode(0);
            playBackResult.setData(msg);
            playBackResult.setMediaServerItem(mediaServerItem);
            playBackResult.setResponse(response);
            callback.call(playBackResult);
        }, event -> {
            msg.setData(String.format("回放失败， 错误码： %s, %s", event.statusCode, event.msg));
            callback.call(msg);
            playBackResult.setCode(-1);
            playBackResult.setData(msg);
            playBackResult.setEvent(event);
            callback.call(playBackResult);
            streamSession.remove(device.getDeviceId(), channelId, ssrcInfo.getStream());
        });
        return result;
    }

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

@@ -88,8 +88,8 @@
            "</script>"})
    int setDefaultCatalog(String platformId, String catalogId);

    @Select("select 'channel' as name, count(pgc.platformId) count from platform_gb_channel pgc  where  pgc.platformId=#{platformId} and pgc.channelId =#{gbId} " +
    @Select("select 'channel' as name, count(pgc.platformId) count from platform_gb_channel pgc left join device_channel dc on dc.id = pgc.deviceChannelId where  pgc.platformId=#{platformId} and dc.channelId =#{gbId} " +
            "union " +
            "select 'stream' as name, count(pgs.platformId) count from platform_gb_stream pgs left join gb_stream gs on pgs.gbStreamId = gs.id where  pgs.platformId=#{platformId} and gs.gbId = #{gbId}")
            "select 'stream' as name, count(pgs.platformId) count from platform_gb_stream pgs left join gb_stream gs on pgs.gbStreamId = gs.gbStreamId where  pgs.platformId=#{platformId} and gs.gbId = #{gbId}")
    List<ChannelSourceInfo> getChannelSource(String platformId, String gbId);
}

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

@@ -75,7 +75,7 @@
    int delByCatalogId(String id);

    @Delete("<script> "+
           "DELETE FROM platform_gb_channel WHERE catalogId=#{parentId} AND platformId=#{platformId} AND channelId=#{id}"  +
           "DELETE FROM platform_gb_channel  WHERE catalogId=#{parentId} AND platformId=#{platformId} AND channelId=#{id}"  +
           "</script>")
    int delByCatalogIdAndChannelIdAndPlatformId(PlatformCatalog platformCatalog);


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

@@ -141,7 +141,6 @@

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

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

@@ -129,7 +129,6 @@
            //Response response = event.getResponse();
            msg.setData(String.format("success"));
            resultHolder.invokeAllResult(msg);
            mediaServerService.closeRTPServer(device, channelId, streamInfo.getStream());
        });

        if (deviceId != null || channelId != null) {

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

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

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

        return result;

 src/test/java/com/genersoft/iot/vmp/service/impl/DeviceAlarmServiceImplTest.java

@@ -50,14 +50,7 @@
//        System.out.println(deviceAlarmService.getAllAlarm(0, 10000, "11111111111111111111", null, "1", null,
//                null, null).getSize());

        System.out.println(deviceAlarmService.getAllAlarm(0, 10000, "11111111111111111111", null, null, null,
                "2021-01-01 00:00:00", null).getSize());

        System.out.println(deviceAlarmService.getAllAlarm(0, 10000, "11111111111111111111", null, null, null,
                null, "2021-04-01 09:00:00").getSize());

        System.out.println(deviceAlarmService.getAllAlarm(0, 10000, "11111111111111111111", null, null, null,
                "2021-02-01 01:00:00", "2021-04-01 04:00:00").getSize());
    }