wvp-pro.git - Gitblit

			@@ -10,13 +10,14 @@
			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.event.subscribe.catalog.CatalogEvent;
			import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;
			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;
			@@ -66,7 +67,13 @@
			private SIPCommander cmder;

			@Autowired
			private SIPCommanderFroPlatform commanderFroPlatform;
			private ISIPCommanderForPlatform commanderFroPlatform;

			@Autowired
			private AudioBroadcastManager audioBroadcastManager;

			@Autowired
			private ZLMServerFactory zlmServerFactory;

			@Autowired
			private IPlayService playService;
			@@ -306,7 +313,15 @@
			result.setModify_stamp(2);
			}
			}
			// 如果是talk对讲，则默认获取声音
			if (ssrcTransactionForAll.get(0).getType() == InviteSessionType.TALK) {
			result.setEnable_audio(true);
			}
			}
			} else if (param.getApp().equals("broadcast")) {
			result.setEnable_audio(true);
			} else if (param.getApp().equals("talk")) {
			result.setEnable_audio(true);
			}
			if (param.getApp().equalsIgnoreCase("rtp")) {
			String receiveKey = VideoManagerConstants.WVP_OTHER_RECEIVE_RTP_INFO + userSetting.getServerId() + "_" + param.getStream();
			@@ -335,7 +350,6 @@
			} else {
			logger.info("[ZLM HOOK] 流注销, {}->{}->{}/{}", param.getMediaServerId(), param.getSchema(), param.getApp(), param.getStream());
			}


			JSONObject json = (JSONObject) JSON.toJSON(param);
			taskExecutor.execute(() -> {
			@@ -366,15 +380,14 @@
			redisCatchStorage.updateStreamAuthorityInfo(param.getApp(), param.getStream(), streamAuthorityInfo);
			}
			}

			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());
			}
			// 设置拉流代理上线/离线

			int updateStatusResult = streamProxyService.updateStatus(param.isRegist(), param.getApp(), param.getStream());
			if (updateStatusResult > 0) {

			@@ -386,6 +399,63 @@
			inviteStreamService.removeInviteInfo(inviteInfo);
			storager.stopPlay(inviteInfo.getDeviceId(), inviteInfo.getChannelId());
			}
			} 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 ("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()) {
			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())) {
			String type = OriginType.values()[param.getOriginType()].getType();
			@@ -458,13 +528,22 @@
			if (platform != null) {
			commanderFroPlatform.streamByeCmd(platform, sendRtpItem);
			redisCatchStorage.deleteSendRTPServer(platformId, sendRtpItem.getChannelId(),
			sendRtpItem.getCallId(), sendRtpItem.getStreamId());
			sendRtpItem.getCallId(), sendRtpItem.getStream());
			} else {
			cmder.streamByeCmd(device, sendRtpItem.getChannelId(), param.getStream(), sendRtpItem.getCallId());
			if (sendRtpItem.getPlayType().equals(InviteStreamType.BROADCAST)
			\|\| sendRtpItem.getPlayType().equals(InviteStreamType.TALK)) {
			AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
			if (audioBroadcastCatch != null) {
			// 来自上级平台的停止对讲
			logger.info("[停止对讲] 来自上级，平台：{}, 通道：{}", sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
			audioBroadcastManager.del(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
			}
			}
			}
			} catch (SipException \| InvalidArgumentException \| ParseException \|
			SsrcTransactionNotFoundException e) {
			logger.error("[命令发送失败] 国标级联发送BYE: {}", e.getMessage());
			logger.error("[命令发送失败] 发送BYE: {}", e.getMessage());
			}
			}
			}
			@@ -472,7 +551,6 @@
			}
			}
			});

			return HookResult.SUCCESS();
			}

			@@ -512,10 +590,10 @@
			logger.error("[命令发送失败] 国标级联发送BYE: {}", e.getMessage());
			}
			redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),
			sendRtpItem.getCallId(), sendRtpItem.getStreamId());
			sendRtpItem.getCallId(), sendRtpItem.getStream());
			if (InviteStreamType.PUSH == sendRtpItem.getPlayType()) {
			MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,
			sendRtpItem.getApp(), sendRtpItem.getStreamId(), sendRtpItem.getChannelId(),
			sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getChannelId(),
			sendRtpItem.getPlatformId(), parentPlatform.getName(), userSetting.getServerId(), sendRtpItem.getMediaServerId());
			messageForPushChannel.setPlatFormIndex(parentPlatform.getId());
			redisCatchStorage.sendPlatformStopPlayMsg(messageForPushChannel);
			@@ -548,6 +626,13 @@
			storager.stopPlay(inviteInfo.getDeviceId(), inviteInfo.getChannelId());
			return ret;
			}
			SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(null, null, param.getStream(), null);
			if (sendRtpItem != null && "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 {
			// 非国标流推流/拉流代理
			// 拉流代理
			@@ -676,7 +761,7 @@

			if (!exist) {
			SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaInfo, param.getStream(), null,
			device.isSsrcCheck(), true, 0, false, device.getStreamModeForParam());
			device.isSsrcCheck(), true, 0, false, false, device.getStreamModeForParam());
			playService.playBack(mediaInfo, ssrcInfo, deviceId, channelId, startTime, endTime, (code, message, data) -> {
			msg.setData(new HookResult(code, message));
			resultHolder.invokeResult(msg);
			@@ -749,7 +834,7 @@
			logger.error("[命令发送失败] 国标级联发送BYE: {}", e.getMessage());
			}
			redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), sendRtpItem.getChannelId(),
			sendRtpItem.getCallId(), sendRtpItem.getStreamId());
			sendRtpItem.getCallId(), sendRtpItem.getStream());
			}
			}
			});
			@@ -762,7 +847,8 @@
			*/
			@ResponseBody
			@PostMapping(value = "/on_rtp_server_timeout", produces = "application/json;charset=UTF-8")
			public HookResult onRtpServerTimeout(HttpServletRequest request, @RequestBody OnRtpServerTimeoutHookParam param) {
			public HookResult onRtpServerTimeout(HttpServletRequest request, @RequestBody OnRtpServerTimeoutHookParam
			param) {
			logger.info("[ZLM HOOK] rtpServer收流超时：{}->{}({})", param.getMediaServerId(), param.getStream_id(), param.getSsrc());

			taskExecutor.execute(() -> {