合并开源主线

648540858

2023-04-04 b4048fbe80dba8e7756ae557a15ab60b4f80a44b

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

@@ -9,9 +9,9 @@
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.cmd.ISIPCommanderForPlatform;

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

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

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

     */

    @ResponseBody



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

    public HookResult onServerKeepalive(@RequestBody OnServerKeepaliveHookParam param) {



        logger.info("[ZLM HOOK] 收到zlm心跳：" + param.getMediaServerId());

//        logger.info("[ZLM HOOK] 收到zlm心跳：" + param.getMediaServerId());



        taskExecutor.execute(() -> {

            List<ZlmHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(HookType.on_server_keepalive);

@@ -142,6 +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()) {

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

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

            if (deviceChannel != null) {



                result.setEnable_audio(deviceChannel.isHasAudio());

            }

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

@@ -257,9 +260,33 @@
                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服务中读取");

            JSONObject info = assistRESTfulUtils.getInfo(mediaInfo, null);

            if (info != null && info.getInteger("code") != null && info.getInteger("code") == 0 ) {

                JSONObject dataJson = info.getJSONObject("data");

                if (dataJson != null) {

                    String recordPath = dataJson.getString("record");

                    userSetting.setRecordPath(recordPath);

                    result.setMp4_save_path(recordPath);

                    // 修改zlm中的录像路径

                    if (mediaInfo.isAutoConfig()) {

                        taskExecutor.execute(() -> {

                            mediaServerService.setZLMConfig(mediaInfo, false);

                        });

                    }

                }

            }

        }

        return result;

    }





    /**

     * rtsp/rtmp流注册或注销时触发此事件；此事件对回复不敏感。

@@ -277,19 +304,22 @@
      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);

                }

            }

            // 流消失移除redis play



            List<OnStreamChangedHookParam.MediaTrack> tracks = param.getTracks();

            // TODO 重构此处逻辑



            if (param.isRegist()) {

                // 处理流注册的鉴权信息

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

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

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

@@ -333,85 +363,60 @@
            }

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

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

            if (param.isRegist() && 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)) {

                           // 直接推流

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

                           if (sendRtpItem == null) {

                              // TODO 可能数据错误，重新开启语音通道

                           }else {

                              JSONObject jsonObject = zlmrtpServerFactory.startSendRtp(mediaInfo, sendRtpItem);

                              if (jsonObject != null && jsonObject.getInteger("code") == 0 ) {

                                 logger.info("[语音喊话] 自动推流成功, device: {}, channel: {}", device.getDeviceId(), channelId);

                              }else {

                                 logger.info("[语音喊话] 推流失败, 结果： {}", jsonObject);

                              }



                           }

                        }else {

                           // 开启语音喊话通道

                           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 {

                  logger.info("[语音喊话] 推流格式有误, 格式为： broadcast/设备编号_通道编号 ");

               }

            }

                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 ("talk".equals(param.getApp())){

            // 语音喊话推流  stream需要满足格式deviceId_channelId

            if (param.isRegist() && 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) {

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

                     if (deviceChannel != null) {

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

                           // 直接推流

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

                           if (sendRtpItem == null) {

                              // TODO 可能数据错误，重新开启语音通道

                           }else {

                              logger.info("rtp/{}开始向上级推流, 目标={}:{}，SSRC={}", sendRtpItem.getStreamId(), sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc());

                              zlmrtpServerFactory.startSendRtp(mediaInfo, sendRtpItem);

                           }

                        }else {

                           // 开启语音喊话通道

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

                           playService.talk(mediaServerItem, device, channelId, (mediaServer, jsonObject)->{

                              System.out.println("开始推流");

                           }, eventResult -> {

                              System.out.println(eventResult.msg);

                           }, ()->{

                              System.out.println("超时");

                           });

                        }



                     }

                  }

               }

            }

            // 语音对讲推流  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);

                                }

                                // 开启语音对讲通道

                                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())){

@@ -471,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());

                                }

                            }

                        }

                    }

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

                     }

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

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

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

                  }

               }

            }

@@ -552,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());

                        }

                    }

@@ -569,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 {

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

            // 拉流代理

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

                    }

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

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

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

                }

            }

        });