| | |
| | | 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.conf.DynamicTask; |
| | | 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.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.redisMsg.RedisGbPlayMsgListener; |
| | | import com.genersoft.iot.vmp.storager.IRedisCatchStorage; |
| | | import com.genersoft.iot.vmp.storager.IVideoManagerStorage; |
| | | import org.slf4j.Logger; |
| | | import org.slf4j.LoggerFactory; |
| | | import org.springframework.beans.factory.InitializingBean; |
| | | import org.springframework.beans.factory.annotation.Autowired; |
| | | import org.springframework.stereotype.Component; |
| | | |
| | | import javax.sip.Dialog; |
| | | import javax.sip.DialogState; |
| | | import javax.sip.InvalidArgumentException; |
| | | import javax.sip.RequestEvent; |
| | | import javax.sip.SipException; |
| | | import javax.sip.address.SipURI; |
| | | import javax.sip.header.CallIdHeader; |
| | | import javax.sip.header.FromHeader; |
| | | import javax.sip.header.HeaderAddress; |
| | | import javax.sip.header.ToHeader; |
| | | import java.util.HashMap; |
| | | import java.util.Map; |
| | | import java.util.Timer; |
| | | import java.util.TimerTask; |
| | | import java.text.ParseException; |
| | | import java.util.*; |
| | | |
| | | /** |
| | | * SIP命令类型: ACK请求 |
| | |
| | | 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; |
| | |
| | | private IRedisCatchStorage redisCatchStorage; |
| | | |
| | | @Autowired |
| | | private IVideoManagerStorage storager; |
| | | |
| | | @Autowired |
| | | private ZLMRTPServerFactory zlmrtpServerFactory; |
| | | |
| | | @Autowired |
| | | private IMediaServerService mediaServerService; |
| | | |
| | | @Autowired |
| | | private ZLMHttpHookSubscribe subscribe; |
| | | private ZlmHttpHookSubscribe subscribe; |
| | | |
| | | @Autowired |
| | | private DynamicTask dynamicTask; |
| | | |
| | | @Autowired |
| | | private ISIPCommander cmder; |
| | | |
| | | @Autowired |
| | | private ISIPCommanderForPlatform commanderForPlatform; |
| | | |
| | | @Autowired |
| | | private RedisGbPlayMsgListener redisGbPlayMsgListener; |
| | | |
| | | |
| | | /** |
| | |
| | | */ |
| | | @Override |
| | | public void process(RequestEvent evt) { |
| | | logger.debug("ACK请求: {}", ((System.currentTimeMillis()))); |
| | | Dialog dialog = evt.getDialog(); |
| | | if (dialog == null) return; |
| | | if (dialog.getState()== DialogState.CONFIRMED) { |
| | | String platformGbId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(FromHeader.NAME)).getAddress().getURI()).getUser(); |
| | | String channelId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser(); |
| | | SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(platformGbId, channelId); |
| | | String is_Udp = sendRtpItem.isTcp() ? "0" : "1"; |
| | | String deviceId = sendRtpItem.getDeviceId(); |
| | | StreamInfo streamInfo = null; |
| | | if (sendRtpItem.isPlay()) { |
| | | streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId); |
| | | CallIdHeader callIdHeader = (CallIdHeader)evt.getRequest().getHeader(CallIdHeader.NAME); |
| | | |
| | | 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(); |
| | | SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(platformGbId, channelId, null, callIdHeader.getCallId()); |
| | | if (sendRtpItem == null) { |
| | | logger.warn("[收到ACK]:未找到通道({})的推流信息", channelId); |
| | | return; |
| | | } |
| | | String is_Udp = sendRtpItem.isTcp() ? "0" : "1"; |
| | | MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); |
| | | logger.info("收到ACK,rtp/{}开始向上级推流, 目标={}:{},SSRC={}", sendRtpItem.getStreamId(), sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc()); |
| | | Map<String, Object> param = new HashMap<>(12); |
| | | param.put("vhost","__defaultVhost__"); |
| | | param.put("app",sendRtpItem.getApp()); |
| | | param.put("stream",sendRtpItem.getStreamId()); |
| | | 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()); |
| | | param.put("pt", sendRtpItem.getPt()); |
| | | param.put("use_ps", sendRtpItem.isUsePs() ? "1" : "0"); |
| | | param.put("only_audio", sendRtpItem.isOnlyAudio() ? "1" : "0"); |
| | | if (!sendRtpItem.isTcp() && parentPlatform.isRtcp()) { |
| | | // 开启rtcp保活 |
| | | param.put("udp_rtcp_timeout", "1"); |
| | | } |
| | | |
| | | 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); |
| | | } |
| | | } |
| | | private void startSendRtpStreamHand(RequestEvent evt, SendRtpItem sendRtpItem, ParentPlatform parentPlatform, |
| | | JSONObject jsonObject, Map<String, Object> param, CallIdHeader callIdHeader) { |
| | | if (jsonObject == null) { |
| | | logger.error("RTP推流失败: 请检查ZLM服务"); |
| | | } else if (jsonObject.getInteger("code") == 0) { |
| | | logger.info("调用ZLM推流接口, 结果: {}", jsonObject); |
| | | logger.info("RTP推流成功[ {}/{} ],{}->{}:{}, " ,param.get("app"), param.get("stream"), jsonObject.getString("local_port"), param.get("dst_url"), param.get("dst_port")); |
| | | } else { |
| | | logger.error("RTP推流失败: {}, 参数:{}",jsonObject.getString("msg"),JSONObject.toJSON(param)); |
| | | if (sendRtpItem.isOnlyAudio()) { |
| | | // TODO 可能是语音对讲 |
| | | }else { |
| | | streamInfo = redisCatchStorage.queryPlaybackByDevice(deviceId, channelId); |
| | | // 向上级平台 |
| | | try { |
| | | commanderForPlatform.streamByeCmd(parentPlatform, callIdHeader.getCallId()); |
| | | } catch (SipException | InvalidArgumentException | ParseException e) { |
| | | logger.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage()); |
| | | } |
| | | } |
| | | System.out.println(JSON.toJSON(streamInfo)); |
| | | if (streamInfo == null) { |
| | | streamInfo = new StreamInfo(); |
| | | streamInfo.setApp(sendRtpItem.getApp()); |
| | | streamInfo.setStreamId(sendRtpItem.getStreamId()); |
| | | } |
| | | redisCatchStorage.updateSendRTPSever(sendRtpItem); |
| | | logger.info(platformGbId); |
| | | logger.info(channelId); |
| | | Map<String, Object> param = new HashMap<>(); |
| | | param.put("vhost","__defaultVhost__"); |
| | | param.put("app",streamInfo.getApp()); |
| | | param.put("stream",streamInfo.getStreamId()); |
| | | param.put("ssrc", sendRtpItem.getSsrc()); |
| | | param.put("dst_url",sendRtpItem.getIp()); |
| | | param.put("dst_port", sendRtpItem.getPort()); |
| | | param.put("is_udp", is_Udp); |
| | | // 设备推流查询,成功后才能转推 |
| | | 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); |
| | | // }); |
| | | // } |
| | | |
| | | |
| | | } |
| | | } |
| | | } |