From 28b5cc39d0a2d9939f70b4c980a31d9b27fc1e4c Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期二, 19 七月 2022 14:27:30 +0800 Subject: [PATCH] 修复更新推流状态sql错误 --- src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java | 112 ++++++++++++++++++++++++++++--------------------------- 1 files changed, 57 insertions(+), 55 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java index 873a0a3..78b8e62 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/AckRequestProcessor.java @@ -4,15 +4,24 @@ import com.alibaba.fastjson.JSONObject; import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.conf.DynamicTask; +import com.genersoft.iot.vmp.gb28181.bean.InviteStreamType; +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; +import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform; 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; +import com.genersoft.iot.vmp.service.bean.RequestPushStreamMsg; +import com.genersoft.iot.vmp.service.impl.RedisGbPlayMsgListener; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; +import com.genersoft.iot.vmp.storager.IVideoManagerStorage; +import com.genersoft.iot.vmp.utils.SerializeUtils; +import org.ehcache.shadow.org.terracotta.offheapstore.storage.IntegerStorageEngine; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; @@ -36,7 +45,7 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements InitializingBean, ISIPRequestProcessor { private Logger logger = LoggerFactory.getLogger(AckRequestProcessor.class); - private String method = "ACK"; + private final String method = "ACK"; @Autowired private SIPProcessorObserver sipProcessorObserver; @@ -51,6 +60,9 @@ private IRedisCatchStorage redisCatchStorage; @Autowired + private IVideoManagerStorage storager; + + @Autowired private ZLMRTPServerFactory zlmrtpServerFactory; @Autowired @@ -61,6 +73,15 @@ @Autowired private DynamicTask dynamicTask; + + @Autowired + private ISIPCommander cmder; + + @Autowired + private ISIPCommanderForPlatform commanderForPlatform; + + @Autowired + private RedisGbPlayMsgListener redisGbPlayMsgListener; /** @@ -78,6 +99,7 @@ if (dialog.getState()== DialogState.CONFIRMED) { String platformGbId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(FromHeader.NAME)).getAddress().getURI()).getUser(); logger.info("ACK璇锋眰锛� platformGbId->{}", platformGbId); + ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformGbId); // 鍙栨秷璁剧疆鐨勮秴鏃朵换鍔� dynamicTask.stop(callIdHeader.getCallId()); String channelId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser(); @@ -96,62 +118,42 @@ param.put("src_port", sendRtpItem.getLocalPort()); param.put("pt", sendRtpItem.getPt()); param.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0"); - param.put("only_audio", sendRtpItem.isOnlyAudio()); - zlmrtpServerFactory.startSendRtpStream(mediaInfo, param); + param.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0"); + if (mediaInfo == null) { + RequestPushStreamMsg requestPushStreamMsg = RequestPushStreamMsg.getInstance( + sendRtpItem.getMediaServerId(), sendRtpItem.getApp(), sendRtpItem.getStreamId(), + sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc(), sendRtpItem.isTcp(), + sendRtpItem.getLocalPort(), sendRtpItem.getPt(), sendRtpItem.isUsePs(), sendRtpItem.isOnlyAudio()); + redisGbPlayMsgListener.sendMsgForStartSendRtpStream(sendRtpItem.getServerId(), requestPushStreamMsg, jsonObject->{ + startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, jsonObject, param, callIdHeader); + }); + }else { + JSONObject jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param); + startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, jsonObject, param, callIdHeader); + } - -// if (streamInfo == null) { // 娴佽繕娌′笂鏉ワ紝瀵规柟灏卞洖澶峚ck -// logger.info("鐩戝惉娴佷互绛夊緟娴佷笂绾�1 rtp/{}", sendRtpItem.getStreamId()); -// // 鐩戝惉娴佷笂绾� -// // 娣诲姞璁㈤槄 -// JSONObject subscribeKey = new JSONObject(); -// subscribeKey.put("app", "rtp"); -// subscribeKey.put("stream", sendRtpItem.getStreamId()); -// subscribeKey.put("regist", true); -// subscribeKey.put("schema", "rtmp"); -// subscribeKey.put("mediaServerId", sendRtpItem.getMediaServerId()); -// subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey, -// (MediaServerItem mediaServerItemInUse, JSONObject json)->{ -// Map<String, Object> param = new HashMap<>(); -// param.put("vhost","__defaultVhost__"); -// param.put("app",json.getString("app")); -// param.put("stream",json.getString("stream")); -// param.put("ssrc", sendRtpItem.getSsrc()); -// param.put("dst_url",sendRtpItem.getIp()); -// param.put("dst_port", sendRtpItem.getPort()); -// param.put("is_udp", is_Udp); -// param.put("src_port", sendRtpItem.getLocalPort()); -// zlmrtpServerFactory.startSendRtpStream(mediaInfo, param); -// }); -// }else { -// Map<String, Object> param = new HashMap<>(); -// param.put("vhost","__defaultVhost__"); -// param.put("app",streamInfo.getApp()); -// param.put("stream",streamInfo.getStream()); -// param.put("ssrc", sendRtpItem.getSsrc()); -// param.put("dst_url",sendRtpItem.getIp()); -// param.put("dst_port", sendRtpItem.getPort()); -// param.put("is_udp", is_Udp); -// param.put("src_port", sendRtpItem.getLocalPort()); -// -// JSONObject jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param); -// if (jsonObject.getInteger("code") != 0) { -// logger.info("鐩戝惉娴佷互绛夊緟娴佷笂绾�2 {}/{}", streamInfo.getApp(), streamInfo.getStream()); -// // 鐩戝惉娴佷笂绾� -// // 娣诲姞璁㈤槄 -// JSONObject subscribeKey = new JSONObject(); -// subscribeKey.put("app", "rtp"); -// subscribeKey.put("stream", streamInfo.getStream()); -// subscribeKey.put("regist", true); -// subscribeKey.put("schema", "rtmp"); -// subscribeKey.put("mediaServerId", sendRtpItem.getMediaServerId()); -// subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey, -// (MediaServerItem mediaServerItemInUse, JSONObject json)->{ -// zlmrtpServerFactory.startSendRtpStream(mediaInfo, param); -// }); -// } -// } + } + } + private void startSendRtpStreamHand(RequestEvent evt, SendRtpItem sendRtpItem, ParentPlatform parentPlatform, + JSONObject jsonObject, Map<String, Object> param, CallIdHeader callIdHeader) { + if (jsonObject == null) { + logger.error("RTP鎺ㄦ祦澶辫触: 璇锋鏌LM鏈嶅姟"); + } else if (jsonObject.getInteger("code") == 0) { + logger.info("RTP鎺ㄦ祦鎴愬姛[ {}/{} ]锛寋}->{}:{}, " ,param.get("app"), param.get("stream"), jsonObject.getString("local_port"), param.get("dst_url"), param.get("dst_port")); + byte[] dialogByteArray = SerializeUtils.serialize(evt.getDialog()); + sendRtpItem.setDialog(dialogByteArray); + byte[] transactionByteArray = SerializeUtils.serialize(evt.getServerTransaction()); + sendRtpItem.setTransaction(transactionByteArray); + redisCatchStorage.updateSendRTPSever(sendRtpItem); + } else { + logger.error("RTP鎺ㄦ祦澶辫触: {}, 鍙傛暟锛歿}",jsonObject.getString("msg"),JSONObject.toJSON(param)); + if (sendRtpItem.isOnlyAudio()) { + // TODO 鍙兘鏄闊冲璁� + }else { + // 鍚戜笂绾у钩鍙� + commanderForPlatform.streamByeCmd(parentPlatform, callIdHeader.getCallId()); + } } } } -- Gitblit v1.8.0