合并开源主线

648540858

2023-04-04 b4048fbe80dba8e7756ae557a15ab60b4f80a44b

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

@@ -7,11 +7,12 @@
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;

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

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

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

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

import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;

import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;

import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;

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.media.zlm.dto.HookType;

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

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

@@ -56,8 +57,14 @@
    @Autowired

    private SIPCommander cmder;



    @Autowired

    private SIPCommanderFroPlatform commanderFroPlatform;

   @Autowired

   private ISIPCommanderForPlatform commanderFroPlatform;



   @Autowired

   private AudioBroadcastManager audioBroadcastManager;



   @Autowired

   private ZLMRTPServerFactory zlmrtpServerFactory;



    @Autowired

    private IPlayService playService;

@@ -112,7 +119,7 @@
     * 服务器定时上报时间，上报间隔可配置，默认10s上报一次

     */

    @ResponseBody

    


    @PostMapping(value = "/on_server_keepalive", produces = "application/json;charset=UTF-8")

    public HookResult onServerKeepalive(@RequestBody OnServerKeepaliveHookParam param) {



@@ -136,7 +143,7 @@
     * 播放器鉴权事件，rtsp/rtmp/http-flv/ws-flv/hls的播放都将触发此鉴权事件。

     */

    @ResponseBody

    


    @PostMapping(value = "/on_play", produces = "application/json;charset=UTF-8")

    public HookResult onPlay(@RequestBody OnPlayHookParam param) {

        if (logger.isDebugEnabled()) {

@@ -244,6 +251,7 @@
            String channelId = ssrcTransactionForAll.get(0).getChannelId();

            DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);

            if (deviceChannel != null) {



                result.setEnable_audio(deviceChannel.isHasAudio());

            }

            // 如果是录像下载就设置视频间隔十秒

@@ -252,6 +260,11 @@
                result.setEnable_audio(true);

                result.setEnable_mp4(true);

            }

            // 如果是talk对讲，则默认获取声音

            if (ssrcTransactionForAll.get(0).getType() == VideoStreamSessionManager.SessionType.talk) {

                result.setEnable_audio(true);

            }



        }

        if (mediaInfo.getRecordAssistPort() > 0 && userSetting.getRecordPath() == null) {

            logger.info("推流时发现尚未设置录像路径，从assist服务中读取");

@@ -288,12 +301,15 @@
            logger.info("[ZLM HOOK] 流注销, {}->{}->{}/{}", param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());

        }





      JSONObject ret = new JSONObject();

      ret.put("code", 0);

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

        MediaServerItem mediaInfo = mediaServerService.getOne(param.getMediaServerId());

        JSONObject json = (JSONObject) JSON.toJSON(param);

        taskExecutor.execute(() -> {

            ZlmHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(HookType.on_stream_changed, json);

            if (subscribe != null) {

                MediaServerItem mediaInfo = mediaServerService.getOne(param.getMediaServerId());



                if (mediaInfo != null) {

                    subscribe.response(mediaInfo, json);

                }

@@ -321,59 +337,119 @@
                redisCatchStorage.removeStreamAuthorityInfo(param.getApp(), param.getStream());

            }



            if ("rtsp".equals(param.getSchema())) {

                // 更新流媒体负载信息

                if (param.isRegist()) {

                    mediaServerService.addCount(param.getMediaServerId());

                } else {

                    mediaServerService.removeCount(param.getMediaServerId());

                }

                // 设置拉流代理上线/离线

                streamProxyService.updateStatus(param.isRegist(), param.getApp(), param.getStream());



                if ("rtp".equals(param.getApp()) && !param.isRegist()) {

                    StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(param.getStream());

                    if (streamInfo != null) {

                        redisCatchStorage.stopPlay(streamInfo);

                        storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());

                    } else {

                        streamInfo = redisCatchStorage.queryPlayback(null, null, param.getStream(), null);

                        if (streamInfo != null) {

                            redisCatchStorage.stopPlayback(streamInfo.getDeviceID(), streamInfo.getChannelId(),

                                    streamInfo.getStream(), null);

      if ("rtsp".equals(param.getSchema())){

         logger.info("流变化：注册->{}, app->{}, stream->{}", param.isRegist(), param.getApp(), param.getStream());

         if (param.isRegist()) {

            mediaServerService.addCount(param.getMediaServerId());

         }else {

            mediaServerService.removeCount(param.getMediaServerId());

         }

         if (param.getOriginType() == OriginType.PULL.ordinal()

               || param.getOriginType() == OriginType.FFMPEG_PULL.ordinal()) {

            // 设置拉流代理上线/离线

            streamProxyService.updateStatus(param.isRegist(), param.getApp(), param.getStream());

         }

         if ("rtp".equals(param.getApp()) && !param.isRegist() ) {

            StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(param.getStream());

            if (streamInfo!=null){

               redisCatchStorage.stopPlay(streamInfo);

               storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());

            }else{

               streamInfo = redisCatchStorage.queryPlayback(null, null, param.getStream(), null);

               if (streamInfo != null) {

                  redisCatchStorage.stopPlayback(streamInfo.getDeviceID(), streamInfo.getChannelId(),

                        streamInfo.getStream(), null);

               }

            }

         }else if ("broadcast".equals(param.getApp())){

            // 语音对讲推流  stream需要满足格式deviceId_channelId

                if (param.getStream().indexOf("_") > 0) {

                    String[] streamArray = param.getStream().split("_");

                    if (streamArray.length == 2) {

                        String deviceId = streamArray[0];

                        String channelId = streamArray[1];

                        Device device = deviceService.getDevice(deviceId);

                        if (device != null) {

                            if (param.isRegist()) {

                                if (audioBroadcastManager.exit(deviceId, channelId)) {

                                    playService.stopAudioBroadcast(deviceId, channelId);

                                }

                                // 开启语音对讲通道

                                try {

                                    playService.audioBroadcastCmd(device, channelId, mediaInfo, param.getApp(), param.getStream(), 60, false, (msg)->{

                                        logger.info("[语音对讲] 通道建立成功, device: {}, channel: {}", deviceId, channelId);

                                    });

                                } catch (InvalidArgumentException | ParseException | SipException e) {

                                    logger.error("[命令发送失败] 语音对讲: {}", e.getMessage());

                                }

                            }else {

                                // 流注销

                                playService.stopAudioBroadcast(deviceId, channelId);

                            }

                        } else{

                            logger.info("[语音对讲] 未找到设备：{}", deviceId);

                        }

                    }

                } else {

                    if (!"rtp".equals(param.getApp())) {

                        String type = OriginType.values()[param.getOriginType()].getType();

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



                        if (mediaServerItem != null) {

                }

         }else if ("talk".equals(param.getApp())){

            // 语音对讲推流  stream需要满足格式deviceId_channelId

                if (param.getStream().indexOf("_") > 0) {

                    String[] streamArray = param.getStream().split("_");

                    if (streamArray.length == 2) {

                        String deviceId = streamArray[0];

                        String channelId = streamArray[1];

                        Device device = deviceService.getDevice(deviceId);

                        if (device != null) {

                            if (param.isRegist()) {

                                StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());

                                String callId = null;

                                if (streamAuthorityInfo != null) {

                                    callId = streamAuthorityInfo.getCallId();

                                if (audioBroadcastManager.exit(deviceId, channelId)) {

                                    playService.stopAudioBroadcast(deviceId, channelId);

                                }

                                StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaServerItem,

                                        param.getApp(), param.getStream(), tracks, callId);

                                param.setStreamInfo(new StreamContent(streamInfoByAppAndStream));

                                redisCatchStorage.addStream(mediaServerItem, type, param.getApp(), param.getStream(), param);

                                if (param.getOriginType() == OriginType.RTSP_PUSH.ordinal()

                                        || param.getOriginType() == OriginType.RTMP_PUSH.ordinal()

                                        || param.getOriginType() == OriginType.RTC_PUSH.ordinal()) {

                                    param.setSeverId(userSetting.getServerId());

                                    zlmMediaListManager.addPush(param);

                                }

                            } else {

                                // 兼容流注销时类型从redis记录获取

                                OnStreamChangedHookParam onStreamChangedHookParam = redisCatchStorage.getStreamInfo(param.getApp(), param.getStream(), param.getMediaServerId());

                                if (onStreamChangedHookParam != null) {

                                    type = OriginType.values()[onStreamChangedHookParam.getOriginType()].getType();

                                    redisCatchStorage.removeStream(mediaServerItem.getId(), type, param.getApp(), param.getStream());

                                }

                                GbStream gbStream = storager.getGbStream(param.getApp(), param.getStream());

                                if (gbStream != null) {

                                // 开启语音对讲通道

                                playService.talkCmd(device, channelId, mediaInfo, param.getStream(), (msg)->{

                                    logger.info("[语音对讲] 通道建立成功, device: {}, channel: {}", deviceId, channelId);

                                });

                            }else {

                                // 流注销

                                playService.stopTalk(device, channelId, param.isRegist());

                            }

                        } else{

                            logger.info("[语音对讲] 未找到设备：{}", deviceId);

                        }

                    }

                }



         }else{

            if (!"rtp".equals(param.getApp())){

               String type = OriginType.values()[param.getOriginType()].getType();

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



                  if (mediaServerItem != null){

                     if (param.isRegist()) {

                        StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(param.getApp(), param.getStream());

                        String callId = null;

                        if (streamAuthorityInfo != null) {

                           callId = streamAuthorityInfo.getCallId();

                        }

                        StreamInfo streamInfoByAppAndStream = mediaService.getStreamInfoByAppAndStream(mediaServerItem,

                              param.getApp(), param.getStream(), param.getTracks(), callId);

                        param.setStreamInfo(new StreamContent(streamInfoByAppAndStream));

                        redisCatchStorage.addStream(mediaServerItem, type, param.getApp(), param.getStream(), param);

                        if (param.getOriginType() == OriginType.RTSP_PUSH.ordinal()

                              || param.getOriginType() == OriginType.RTMP_PUSH.ordinal()

                              || param.getOriginType() == OriginType.RTC_PUSH.ordinal() ) {

                           param.setSeverId(userSetting.getServerId());

                           zlmMediaListManager.addPush(param);

                        }

                     }else {

                        // 兼容流注销时类型从redis记录获取

                        OnStreamChangedHookParam onStreamChangedHookParam = redisCatchStorage.getStreamInfo(

                              param.getApp(), param.getStream(), param.getMediaServerId());

                        if (onStreamChangedHookParam != null) {

                           type = OriginType.values()[onStreamChangedHookParam.getOriginType()].getType();

                           redisCatchStorage.removeStream(mediaServerItem.getId(), type, param.getApp(), param.getStream());

                        }

                        GbStream gbStream = storager.getGbStream(param.getApp(), param.getStream());

                        if (gbStream != null) {

//                           eventPublisher.catalogEventPublishForStream(null, gbStream, CatalogEvent.OFF);

                                }

                                zlmMediaListManager.removeMedia(param.getApp(), param.getStream());

@@ -400,16 +476,21 @@
                                ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId);

                                Device device = deviceService.getDevice(platformId);



                                try {



                                    if (platform != null) {

                                        commanderFroPlatform.streamByeCmd(platform, sendRtpItem);

                                        try {

                                            commanderFroPlatform.streamByeCmd(platform, sendRtpItem);

                                        } catch (SipException | InvalidArgumentException | ParseException e) {

                                            logger.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage());

                                        }

                                    } else {

                                        cmder.streamByeCmd(device, sendRtpItem.getChannelId(), param.getStream(), sendRtpItem.getCallId());

                                        try {

                                            cmder.streamByeCmd(device, sendRtpItem.getChannelId(), param.getStream(), sendRtpItem.getCallId());

                                        } catch (SipException | InvalidArgumentException | ParseException |

                                                 SsrcTransactionNotFoundException e) {

                                            logger.error("[命令发送失败] 发送BYE: {}", e.getMessage());

                                        }

                                    }

                                } catch (SipException | InvalidArgumentException | ParseException |

                                         SsrcTransactionNotFoundException e) {

                                    logger.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage());

                                }

                            }

                        }

                    }

@@ -427,42 +508,44 @@
    @PostMapping(value = "/on_stream_none_reader", produces = "application/json;charset=UTF-8")

    public JSONObject onStreamNoneReader(@RequestBody OnStreamNoneReaderHookParam param) {



        logger.info("[ZLM HOOK]流无人观看：{]->{}->{}/{}" + param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());

        JSONObject ret = new JSONObject();

        ret.put("code", 0);

        // 国标类型的流

        if ("rtp".equals(param.getApp())) {

            ret.put("close", userSetting.getStreamOnDemand());

            // 国标流， 点播/录像回放/录像下载

            StreamInfo streamInfoForPlayCatch = redisCatchStorage.queryPlayByStreamId(param.getStream());

            // 点播

            if (streamInfoForPlayCatch != null) {

                // 收到无人观看说明流也没有在往上级推送

                if (redisCatchStorage.isChannelSendingRTP(streamInfoForPlayCatch.getChannelId())) {

                    List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId(streamInfoForPlayCatch.getChannelId());

                    if (sendRtpItems.size() > 0) {

                        for (SendRtpItem sendRtpItem : sendRtpItems) {

                            ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());

                            try {

                                commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());

                            } catch (SipException | InvalidArgumentException | ParseException e) {

                                logger.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage());

                            }

                            redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),

                                    sendRtpItem.getCallId(), sendRtpItem.getStreamId());

                        }

                    }

                }

                Device device = deviceService.getDevice(streamInfoForPlayCatch.getDeviceID());

                if (device != null) {

                    try {

                        cmder.streamByeCmd(device, streamInfoForPlayCatch.getChannelId(),

                                streamInfoForPlayCatch.getStream(), null);

                    } catch (InvalidArgumentException | ParseException | SipException |

                             SsrcTransactionNotFoundException e) {

                        logger.error("[无人观看]点播， 发送BYE失败 {}", e.getMessage());

                    }

                }

      logger.info("[ZLM HOOK]流无人观看：{]->{}->{}/{}" + param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());

      JSONObject ret = new JSONObject();

      ret.put("code", 0);

      // 国标类型的流

      if ("rtp".equals(param.getApp())){

         ret.put("close", userSetting.getStreamOnDemand());

         // 国标流， 点播/录像回放/录像下载

         StreamInfo streamInfoForPlayCatch = redisCatchStorage.queryPlayByStreamId(param.getStream());

         // 点播

         if (streamInfoForPlayCatch != null) {

            // 收到无人观看说明流也没有在往上级推送

            if (redisCatchStorage.isChannelSendingRTP(streamInfoForPlayCatch.getChannelId())) {

               List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId(streamInfoForPlayCatch.getChannelId());

               if (sendRtpItems.size() > 0) {

                  for (SendRtpItem sendRtpItem : sendRtpItems) {

                     ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());

                     if (parentPlatform == null) {

                        continue;

                     }

                     try {

                        commanderFroPlatform.streamByeCmd(parentPlatform, sendRtpItem.getCallId());

                     } catch (SipException | InvalidArgumentException | ParseException e) {

                        logger.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage());

                     }

                     redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),

                           sendRtpItem.getCallId(), sendRtpItem.getStream());

                  }

               }

            }

            Device device = deviceService.getDevice(streamInfoForPlayCatch.getDeviceID());

            if (device != null) {

               try {

                  cmder.streamByeCmd(device, streamInfoForPlayCatch.getChannelId(),

                        streamInfoForPlayCatch.getStream(), null);

               } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) {

                  logger.error("[无人观看]点播， 发送BYE失败 {}", e.getMessage());

               }

            }



                redisCatchStorage.stopPlay(streamInfoForPlayCatch);

                storager.stopPlay(streamInfoForPlayCatch.getDeviceID(), streamInfoForPlayCatch.getChannelId());

@@ -479,8 +562,7 @@
                        try {

                            cmder.streamByeCmd(device, streamInfoForPlayBackCatch.getChannelId(),

                                    streamInfoForPlayBackCatch.getStream(), null);

                        } catch (InvalidArgumentException | ParseException | SipException |

                                 SsrcTransactionNotFoundException e) {

                        } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) {

                            logger.error("[无人观看]回放， 发送BYE失败 {}", e.getMessage());

                        }

                    }

@@ -496,6 +578,13 @@
                ret.put("close", false);

                return ret;

            }

            SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, null, param.getStream(), null);

            if ("talk".equals(sendRtpItem.getApp())){

                ret.put("close", false);

                return ret;

            }

        }else if ("talk".equals(param.getApp()) || "broadcast".equals(param.getApp())){

            ret.put("close", false);

        } else {

            // 非国标流 推流/拉流代理

            // 拉流代理

@@ -657,7 +746,7 @@
                        logger.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage());

                    }

                    redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),

                            sendRtpItem.getCallId(), sendRtpItem.getStreamId());

                            sendRtpItem.getCallId(), sendRtpItem.getStream());

                }

            }

        });