From 2e98b20bea4463e4465e3c19059d0744a09aec06 Mon Sep 17 00:00:00 2001
From: zhongrj <646384940@qq.com>
Date: Tue, 27 Jun 2023 14:34:41 +0800
Subject: [PATCH] gb28181版本升级-补充
---
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java | 1313 +++++++++++++++++++++++++++++++++++++++--------------------
1 files changed, 861 insertions(+), 452 deletions(-)
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
index 77b4b32..eb38ad4 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
@@ -1,34 +1,37 @@
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.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager;
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
-import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
+import com.genersoft.iot.vmp.gb28181.transmit.SIPSender;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
-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.gb28181.transmit.event.request.ISIPRequestProcessor;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
-import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
+import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
+import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
-import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
+import com.genersoft.iot.vmp.media.zlm.dto.*;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IPlayService;
+import com.genersoft.iot.vmp.service.IStreamProxyService;
+import com.genersoft.iot.vmp.service.IStreamPushService;
+import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
import com.genersoft.iot.vmp.service.bean.SSRCInfo;
+import com.genersoft.iot.vmp.service.redisMsg.RedisGbPlayMsgListener;
+import com.genersoft.iot.vmp.service.redisMsg.RedisPushStreamResponseListener;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
-import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
-import com.genersoft.iot.vmp.utils.SerializeUtils;
-import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.PlayResult;
+import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
+import com.genersoft.iot.vmp.utils.DateUtil;
import gov.nist.javax.sdp.TimeDescriptionImpl;
import gov.nist.javax.sdp.fields.TimeField;
-import gov.nist.javax.sip.address.AddressImpl;
-import gov.nist.javax.sip.address.SipUri;
+import gov.nist.javax.sip.message.SIPRequest;
+import gov.nist.javax.sip.message.SIPResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
@@ -36,19 +39,12 @@
import org.springframework.stereotype.Component;
import javax.sdp.*;
-import javax.sip.InvalidArgumentException;
-import javax.sip.RequestEvent;
-import javax.sip.ServerTransaction;
-import javax.sip.SipException;
-import javax.sip.address.SipURI;
+import javax.sip.*;
import javax.sip.header.CallIdHeader;
-import javax.sip.header.FromHeader;
import javax.sip.message.Request;
import javax.sip.message.Response;
import java.text.ParseException;
-import java.text.SimpleDateFormat;
-import java.util.Date;
-import java.util.List;
+import java.time.Instant;
import java.util.Vector;
/**
@@ -58,466 +54,879 @@
@Component
public class InviteRequestProcessor extends SIPRequestProcessorParent implements InitializingBean, ISIPRequestProcessor {
- private final static Logger logger = LoggerFactory.getLogger(InviteRequestProcessor.class);
+ private final static Logger logger = LoggerFactory.getLogger(InviteRequestProcessor.class);
- private String method = "INVITE";
+ private final String method = "INVITE";
- @Autowired
- private SIPCommanderFroPlatform cmderFroPlatform;
+ @Autowired
+ private SIPCommanderFroPlatform cmderFroPlatform;
- @Autowired
- private IVideoManagerStorager storager;
+ @Autowired
+ private IVideoManagerStorage storager;
- @Autowired
- private IRedisCatchStorage redisCatchStorage;
+ @Autowired
+ private IStreamPushService streamPushService;
+ @Autowired
+ private IStreamProxyService streamProxyService;
- @Autowired
- private DynamicTask dynamicTask;
+ @Autowired
+ private IRedisCatchStorage redisCatchStorage;
- @Autowired
- private SIPCommander cmder;
+ @Autowired
+ private DynamicTask dynamicTask;
- @Autowired
- private IPlayService playService;
+ @Autowired
+ private RedisPushStreamResponseListener redisPushStreamResponseListener;
- @Autowired
- private ISIPCommander commander;
+ @Autowired
+ private IPlayService playService;
- @Autowired
- private ZLMRTPServerFactory zlmrtpServerFactory;
+ @Autowired
+ private SIPSender sipSender;
- @Autowired
- private IMediaServerService mediaServerService;
+ @Autowired
+ private ZLMRTPServerFactory zlmrtpServerFactory;
- @Autowired
- private SIPProcessorObserver sipProcessorObserver;
+ @Autowired
+ private IMediaServerService mediaServerService;
- @Autowired
- private VideoStreamSessionManager sessionManager;
+ @Autowired
+ private ZlmHttpHookSubscribe zlmHttpHookSubscribe;
+
+ @Autowired
+ private SIPProcessorObserver sipProcessorObserver;
+
+ @Autowired
+ private VideoStreamSessionManager sessionManager;
+
+ @Autowired
+ private UserSetting userSetting;
+
+ @Autowired
+ private ZLMMediaListManager mediaListManager;
- @Override
- public void afterPropertiesSet() throws Exception {
- // 添加消息处理的订阅
- sipProcessorObserver.addRequestProcessor(method, this);
- }
-
- /**
- * 处理invite请求
- *
- * @param evt
- * 请求消息
- */
- @Override
- public void process(RequestEvent evt) {
- // Invite Request消息实现，此消息一般为级联消息，上级给下级发送请求视频指令
- try {
- Request request = evt.getRequest();
- SipURI sipURI = (SipURI) request.getRequestURI();
- String channelId = sipURI.getUser();
- String requesterId = SipUtils.getUserIdFromFromHeader(request);
- CallIdHeader callIdHeader = (CallIdHeader)request.getHeader(CallIdHeader.NAME);
- if (requesterId == null || channelId == null) {
- logger.info("无法从FromHeader的Address中获取到平台id，返回400");
- responseAck(evt, Response.BAD_REQUEST); // 参数不全， 发400，请求错误
- return;
- }
-
- // 查询请求是否来自上级平台\设备
- ParentPlatform platform = storager.queryParentPlatByServerGBId(requesterId);
- if (platform == null) {
- inviteFromDeviceHandle(evt, requesterId);
- }else {
- // 查询平台下是否有该通道
- DeviceChannel channel = storager.queryChannelInParentPlatform(requesterId, channelId);
- GbStream gbStream = storager.queryStreamInParentPlatform(requesterId, channelId);
- PlatformCatalog catalog = storager.getCatalog(channelId);
- MediaServerItem mediaServerItem = null;
- // 不是通道可能是直播流
- if (channel != null && gbStream == null ) {
- if (channel.getStatus() == 0) {
- logger.info("通道离线，返回400");
- responseAck(evt, Response.BAD_REQUEST, "channel [" + channel.getChannelId() + "] offline");
- return;
- }
- responseAck(evt, Response.CALL_IS_BEING_FORWARDED); // 通道存在，发181，呼叫转接中
- }else if(channel == null && gbStream != null){
- String mediaServerId = gbStream.getMediaServerId();
- mediaServerItem = mediaServerService.getOne(mediaServerId);
- if (mediaServerItem == null) {
- logger.info("[ app={}, stream={} ]找不到zlm {}，返回410",gbStream.getApp(), gbStream.getStream(), mediaServerId);
- responseAck(evt, Response.GONE);
- return;
- }
- Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream());
- if (!streamReady ) {
- logger.info("[ app={}, stream={} ]通道离线，返回400",gbStream.getApp(), gbStream.getStream());
- responseAck(evt, Response.BAD_REQUEST, "channel [" + gbStream.getGbId() + "] offline");
- return;
- }
- responseAck(evt, Response.CALL_IS_BEING_FORWARDED); // 通道存在，发181，呼叫转接中
- }else if (catalog != null) {
- responseAck(evt, Response.BAD_REQUEST, "catalog channel can not play"); // 目录不支持点播
- return;
- } else {
- logger.info("通道不存在，返回404");
- responseAck(evt, Response.NOT_FOUND); // 通道不存在，发404，资源不存在
- return;
- }
- // 解析sdp消息, 使用jainsip 自带的sdp解析方式
- String contentString = new String(request.getRawContent());
-
- // jainSip不支持y=字段， 移除以解析。
- int ssrcIndex = contentString.indexOf("y=");
- // 检查是否有y字段
- String ssrcDefault = "0000000000";
- String ssrc;
- SessionDescription sdp;
- if (ssrcIndex >= 0) {
- //ssrc规定长度为10字节，不取余下长度以避免后续还有“f=”字段
- ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12);
- String substring = contentString.substring(0, contentString.indexOf("y="));
- sdp = SdpFactory.getInstance().createSessionDescription(substring);
- }else {
- ssrc = ssrcDefault;
- sdp = SdpFactory.getInstance().createSessionDescription(contentString);
- }
- String sessionName = sdp.getSessionName().getValue();
-
- Long startTime = null;
- Long stopTime = null;
- Date start = null;
- Date end = null;
- if (sdp.getTimeDescriptions(false) != null && sdp.getTimeDescriptions(false).size() > 0) {
- TimeDescriptionImpl timeDescription = (TimeDescriptionImpl)(sdp.getTimeDescriptions(false).get(0));
- TimeField startTimeFiled = (TimeField)timeDescription.getTime();
- startTime = startTimeFiled.getStartTime();
- stopTime = startTimeFiled.getStopTime();
-
- start = new Date(startTime*1000);
- end = new Date(stopTime*1000);
- }
- // 获取支持的格式
- Vector mediaDescriptions = sdp.getMediaDescriptions(true);
- // 查看是否支持PS 负载96
- //String ip = null;
- int port = -1;
- boolean mediaTransmissionTCP = false;
- Boolean tcpActive = null;
- for (Object description : mediaDescriptions) {
- MediaDescription mediaDescription = (MediaDescription) description;
- Media media = mediaDescription.getMedia();
-
- Vector mediaFormats = media.getMediaFormats(false);
- if (mediaFormats.contains("96")) {
- port = media.getMediaPort();
- //String mediaType = media.getMediaType();
- String protocol = media.getProtocol();
-
- // 区分TCP发流还是udp， 当前默认udp
- if ("TCP/RTP/AVP".equals(protocol)) {
- String setup = mediaDescription.getAttribute("setup");
- if (setup != null) {
- mediaTransmissionTCP = true;
- if ("active".equals(setup)) {
- tcpActive = true;
- } else if ("passive".equals(setup)) {
- tcpActive = false;
- }
- }
- }
- break;
- }
- }
- if (port == -1) {
- logger.info("不支持的媒体格式，返回415");
- // 回复不支持的格式
- responseAck(evt, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415
- return;
- }
- String username = sdp.getOrigin().getUsername();
- String addressStr = sdp.getOrigin().getAddress();
-
- logger.info("[上级点播]用户：{}， 地址：{}:{}， ssrc：{}", username, addressStr, port, ssrc);
- Device device = null;
- // 通过 channel 和 gbStream 是否为null 值判断来源是直播流合适国标
- if (channel != null) {
- device = storager.queryVideoDeviceByPlatformIdAndChannelId(requesterId, channelId);
- if (device == null) {
- logger.warn("点播平台{}的通道{}时未找到设备信息", requesterId, channel);
- responseAck(evt, Response.SERVER_INTERNAL_ERROR);
- return;
- }
- mediaServerItem = playService.getNewMediaServerItem(device);
- if (mediaServerItem == null) {
- logger.warn("未找到可用的zlm");
- responseAck(evt, Response.BUSY_HERE);
- return;
- }
- SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
- device.getDeviceId(), channelId,
- mediaTransmissionTCP);
- if (tcpActive != null) {
- sendRtpItem.setTcpActive(tcpActive);
- }
- if (sendRtpItem == null) {
- logger.warn("服务器端口资源不足");
- responseAck(evt, Response.BUSY_HERE);
- return;
- }
- sendRtpItem.setCallId(callIdHeader.getCallId());
- sendRtpItem.setPlay("Play".equals(sessionName));
- byte[] dialogByteArray = SerializeUtils.serialize(evt.getDialog());
- sendRtpItem.setDialog(dialogByteArray);
- byte[] transactionByteArray = SerializeUtils.serialize(evt.getServerTransaction());
- sendRtpItem.setTransaction(transactionByteArray);
+ @Autowired
+ private RedisGbPlayMsgListener redisGbPlayMsgListener;
- Long finalStartTime = startTime;
- Long finalStopTime = stopTime;
- ZLMHttpHookSubscribe.Event hookEvent = (mediaServerItemInUSe, responseJSON)->{
- String app = responseJSON.getString("app");
- String stream = responseJSON.getString("stream");
- logger.info("[上级点播]下级已经开始推流。 回复200OK(SDP)， {}/{}", app, stream);
- // * 0 等待设备推流上来
- // * 1 下级已经推流，等待上级平台回复ack
- // * 2 推流中
- sendRtpItem.setStatus(1);
- redisCatchStorage.updateSendRTPSever(sendRtpItem);
+ @Override
+ public void afterPropertiesSet() throws Exception {
+ // 添加消息处理的订阅
+ sipProcessorObserver.addRequestProcessor(method, this);
+ }
- StringBuffer content = new StringBuffer(200);
- content.append("v=0\r\n");
- content.append("o="+ channelId +" 0 0 IN IP4 "+mediaServerItemInUSe.getSdpIp()+"\r\n");
- content.append("s=" + sessionName+"\r\n");
- content.append("c=IN IP4 "+mediaServerItemInUSe.getSdpIp()+"\r\n");
- if ("Playback".equals(sessionName)) {
- content.append("t=" + finalStartTime + " " + finalStopTime + "\r\n");
- }else {
- content.append("t=0 0\r\n");
- }
- content.append("m=video "+ sendRtpItem.getLocalPort()+" RTP/AVP 96\r\n");
- content.append("a=sendonly\r\n");
- content.append("a=rtpmap:96 PS/90000\r\n");
- content.append("y="+ ssrc + "\r\n");
- content.append("f=\r\n");
+ /**
+ * 处理invite请求
+ *
+ * @param evt 请求消息
+ */
+ @Override
+ public void process(RequestEvent evt) {
+ // Invite Request消息实现，此消息一般为级联消息，上级给下级发送请求视频指令
+ try {
+ SIPRequest request = (SIPRequest)evt.getRequest();
+ String channelId = SipUtils.getChannelIdFromRequest(request);
+ String requesterId = SipUtils.getUserIdFromFromHeader(request);
+ CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME);
+ if (requesterId == null || channelId == null) {
+ logger.info("无法从FromHeader的Address中获取到平台id，返回400");
+ // 参数不全， 发400，请求错误
+ try {
+ responseAck(request, Response.BAD_REQUEST);
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] invite BAD_REQUEST: {}", e.getMessage());
+ }
+ return;
+ }
- try {
- // 超时未收到Ack应该回复bye,当前等待时间为10秒
- dynamicTask.startDelay(callIdHeader.getCallId(), ()->{
- logger.info("Ack 等待超时");
- mediaServerService.releaseSsrc(mediaServerItemInUSe.getId(), ssrc);
- // 回复bye
- cmderFroPlatform.streamByeCmd(platform, callIdHeader.getCallId());
- }, 60);
- responseSdpAck(evt, content.toString(), platform);
- } catch (SipException e) {
- e.printStackTrace();
- } catch (InvalidArgumentException e) {
- e.printStackTrace();
- } catch (ParseException e) {
- e.printStackTrace();
- }
- };
- SipSubscribe.Event errorEvent = ((event) -> {
- // 未知错误。直接转发设备点播的错误
- Response response = null;
- try {
- response = getMessageFactory().createResponse(event.statusCode, evt.getRequest());
- ServerTransaction serverTransaction = getServerTransaction(evt);
- serverTransaction.sendResponse(response);
- if (serverTransaction.getDialog() != null) serverTransaction.getDialog().delete();
- } catch (ParseException | SipException | InvalidArgumentException e) {
- e.printStackTrace();
- }
- });
- sendRtpItem.setApp("rtp");
- if ("Playback".equals(sessionName)) {
- sendRtpItem.setPlay(false);
- SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, true);
- sendRtpItem.setStreamId(ssrcInfo.getStream());
- // 写入redis， 超时时回复
- redisCatchStorage.updateSendRTPSever(sendRtpItem);
- SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
- playService.playBack(mediaServerItem, ssrcInfo, device.getDeviceId(), channelId, format.format(start),
- format.format(end), null, result -> {
- if (result.getCode() != 0){
- logger.warn("录像回放失败");
- if (result.getEvent() != null) {
- errorEvent.response(result.getEvent());
- }
- redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null);
- try {
- responseAck(evt, Response.REQUEST_TIMEOUT);
- } catch (SipException e) {
- e.printStackTrace();
- } catch (InvalidArgumentException e) {
- e.printStackTrace();
- } catch (ParseException e) {
- e.printStackTrace();
- }
- }else {
- if (result.getMediaServerItem() != null) {
- hookEvent.response(result.getMediaServerItem(), result.getResponse());
- }
- }
- });
- }else {
- sendRtpItem.setPlay(true);
- SsrcTransaction playTransaction = sessionManager.getSsrcTransaction(device.getDeviceId(), channelId, "play", null);
- if (playTransaction != null) {
- Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, "rtp", playTransaction.getStream());
- if (!streamReady) {
- playTransaction = null;
- }
- }
- if (playTransaction == null) {
- String streamId = null;
- if (mediaServerItem.isRtpEnable()) {
- streamId = String.format("%s_%s", device.getDeviceId(), channelId);
- }
- SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, true);
- sendRtpItem.setStreamId(ssrcInfo.getStream());
- // 写入redis， 超时时回复
- redisCatchStorage.updateSendRTPSever(sendRtpItem);
- playService.play(mediaServerItem, ssrcInfo, device, channelId, hookEvent, errorEvent, (code, msg)->{
- redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null);
- }, null);
- }else {
- sendRtpItem.setStreamId(playTransaction.getStream());
- // 写入redis， 超时时回复
- redisCatchStorage.updateSendRTPSever(sendRtpItem);
- JSONObject jsonObject = new JSONObject();
- jsonObject.put("app", sendRtpItem.getApp());
- jsonObject.put("stream", sendRtpItem.getStreamId());
- hookEvent.response(mediaServerItem, jsonObject);
- }
- }
- }else if (gbStream != null) {
- SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
- gbStream.getApp(), gbStream.getStream(), channelId,
- mediaTransmissionTCP);
+ // 查询请求是否来自上级平台\设备
+ ParentPlatform platform = storager.queryParentPlatByServerGBId(requesterId);
+ if (platform == null) {
+ inviteFromDeviceHandle(request, requesterId);
- if (tcpActive != null) {
- sendRtpItem.setTcpActive(tcpActive);
- }
- if (sendRtpItem == null) {
- logger.warn("服务器端口资源不足");
- responseAck(evt, Response.BUSY_HERE);
- return;
- }
+ } else {
+ // 查询平台下是否有该通道
+ DeviceChannel channel = storager.queryChannelInParentPlatform(requesterId, channelId);
+ GbStream gbStream = storager.queryStreamInParentPlatform(requesterId, channelId);
+ PlatformCatalog catalog = storager.getCatalog(channelId);
- // 写入redis， 超时时回复
- sendRtpItem.setStatus(1);
- sendRtpItem.setCallId(callIdHeader.getCallId());
- byte[] dialogByteArray = SerializeUtils.serialize(evt.getDialog());
- sendRtpItem.setDialog(dialogByteArray);
- byte[] transactionByteArray = SerializeUtils.serialize(evt.getServerTransaction());
- sendRtpItem.setTransaction(transactionByteArray);
- redisCatchStorage.updateSendRTPSever(sendRtpItem);
- StringBuffer content = new StringBuffer(200);
- content.append("v=0\r\n");
- content.append("o="+ channelId +" 0 0 IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
- content.append("s=Play\r\n");
- content.append("c=IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
- content.append("t=0 0\r\n");
- content.append("m=video "+ sendRtpItem.getLocalPort()+" RTP/AVP 96\r\n");
- content.append("a=sendonly\r\n");
- content.append("a=rtpmap:96 PS/90000\r\n");
- content.append("y="+ ssrc + "\r\n");
- content.append("f=\r\n");
+ MediaServerItem mediaServerItem = null;
+ StreamPushItem streamPushItem = null;
+ StreamProxyItem proxyByAppAndStream =null;
+ // 不是通道可能是直播流
+ if (channel != null && gbStream == null) {
+// if (channel.getStatus() == 0) {
+// logger.info("通道离线，返回400");
+// responseAck(request, Response.BAD_REQUEST, "channel [" + channel.getChannelId() + "] offline");
+// return;
+// }
+ // 通道存在，发100，TRYING
+ try {
+ responseAck(request, Response.TRYING);
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] invite TRYING: {}", e.getMessage());
+ }
+ } else if (channel == null && gbStream != null) {
- try {
- responseSdpAck(evt, content.toString(), platform);
- } catch (SipException e) {
- e.printStackTrace();
- } catch (InvalidArgumentException e) {
- e.printStackTrace();
- } catch (ParseException e) {
- e.printStackTrace();
- }
- }
+ String mediaServerId = gbStream.getMediaServerId();
+ mediaServerItem = mediaServerService.getOne(mediaServerId);
+ if (mediaServerItem == null) {
+ if ("proxy".equals(gbStream.getStreamType())) {
+ logger.info("[ app={}, stream={} ]找不到zlm {}，返回410", gbStream.getApp(), gbStream.getStream(), mediaServerId);
+ try {
+ responseAck(request, Response.GONE);
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] invite GONE: {}", e.getMessage());
+ }
+ return;
+ } else {
+ streamPushItem = streamPushService.getPush(gbStream.getApp(), gbStream.getStream());
+ if (streamPushItem == null || streamPushItem.getServerId().equals(userSetting.getServerId())) {
+ logger.info("[ app={}, stream={} ]找不到zlm {}，返回410", gbStream.getApp(), gbStream.getStream(), mediaServerId);
+ try {
+ responseAck(request, Response.GONE);
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] invite GONE: {}", e.getMessage());
+ }
+ return;
+ }
+ }
+ } else {
+ if ("push".equals(gbStream.getStreamType())) {
+ streamPushItem = streamPushService.getPush(gbStream.getApp(), gbStream.getStream());
+ if (streamPushItem == null) {
+ logger.info("[ app={}, stream={} ]找不到zlm {}，返回410", gbStream.getApp(), gbStream.getStream(), mediaServerId);
+ try {
+ responseAck(request, Response.GONE);
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] invite GONE: {}", e.getMessage());
+ }
+ return;
+ }
+ }else if("proxy".equals(gbStream.getStreamType())){
+ proxyByAppAndStream = streamProxyService.getStreamProxyByAppAndStream(gbStream.getApp(), gbStream.getStream());
+ if (proxyByAppAndStream == null) {
+ logger.info("[ app={}, stream={} ]找不到zlm {}，返回410", gbStream.getApp(), gbStream.getStream(), mediaServerId);
+ try {
+ responseAck(request, Response.GONE);
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] invite GONE: {}", e.getMessage());
+ }
+ return;
+ }
+ }
+ }
+ try {
+ responseAck(request, Response.CALL_IS_BEING_FORWARDED);
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] invite CALL_IS_BEING_FORWARDED: {}", e.getMessage());
+ }
+ } else if (catalog != null) {
+ try {
+ // 目录不支持点播
+ responseAck(request, Response.BAD_REQUEST, "catalog channel can not play");
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] invite 目录不支持点播: {}", e.getMessage());
+ }
+ return;
+ } else {
+ logger.info("通道不存在，返回404");
+ try {
+ // 通道不存在，发404，资源不存在
+ responseAck(request, Response.NOT_FOUND);
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] invite 通道不存在: {}", e.getMessage());
+ }
+ return;
+ }
+ // 解析sdp消息, 使用jainsip 自带的sdp解析方式
+ String contentString = new String(request.getRawContent());
- }
+ // jainSip不支持y=字段， 移除以解析。
+ int ssrcIndex = contentString.indexOf("y=");
+ // 检查是否有y字段
+ String ssrcDefault = "0000000000";
+ String ssrc;
+ SessionDescription sdp;
+ if (ssrcIndex >= 0) {
+ //ssrc规定长度为10个字节，不取余下长度以避免后续还有“f=”字段
+ ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12);
+ String substring = contentString.substring(0, contentString.indexOf("y="));
+ sdp = SdpFactory.getInstance().createSessionDescription(substring);
+ } else {
+ ssrc = ssrcDefault;
+ sdp = SdpFactory.getInstance().createSessionDescription(contentString);
+ }
+ String sessionName = sdp.getSessionName().getValue();
- } catch (SipException | InvalidArgumentException | ParseException e) {
- e.printStackTrace();
- logger.warn("sdp解析错误");
- e.printStackTrace();
- } catch (SdpParseException e) {
- e.printStackTrace();
- } catch (SdpException e) {
- e.printStackTrace();
- }
- }
+ Long startTime = null;
+ Long stopTime = null;
+ Instant start = null;
+ Instant end = null;
+ if (sdp.getTimeDescriptions(false) != null && sdp.getTimeDescriptions(false).size() > 0) {
+ TimeDescriptionImpl timeDescription = (TimeDescriptionImpl) (sdp.getTimeDescriptions(false).get(0));
+ TimeField startTimeFiled = (TimeField) timeDescription.getTime();
+ startTime = startTimeFiled.getStartTime();
+ stopTime = startTimeFiled.getStopTime();
- public void inviteFromDeviceHandle(RequestEvent evt, String requesterId) throws InvalidArgumentException, ParseException, SipException, SdpException {
+ start = Instant.ofEpochSecond(startTime);
+ end = Instant.ofEpochSecond(stopTime);
+ }
+ // 获取支持的格式
+ Vector mediaDescriptions = sdp.getMediaDescriptions(true);
+ // 查看是否支持PS 负载96
+ //String ip = null;
+ int port = -1;
+ boolean mediaTransmissionTCP = false;
+ Boolean tcpActive = null;
+ for (Object description : mediaDescriptions) {
+ MediaDescription mediaDescription = (MediaDescription) description;
+ Media media = mediaDescription.getMedia();
- // 非上级平台请求，查询是否设备请求（通常为接收语音广播的设备）
- Device device = redisCatchStorage.getDevice(requesterId);
- Request request = evt.getRequest();
- if (device != null) {
- logger.info("收到设备" + requesterId + "的语音广播Invite请求");
- responseAck(evt, Response.TRYING);
+ Vector mediaFormats = media.getMediaFormats(false);
+ if (mediaFormats.contains("96")) {
+ port = media.getMediaPort();
+ //String mediaType = media.getMediaType();
+ String protocol = media.getProtocol();
- String contentString = new String(request.getRawContent());
- // jainSip不支持y=字段， 移除移除以解析。
- String substring = contentString;
- String ssrc = "0000000404";
- int ssrcIndex = contentString.indexOf("y=");
- if (ssrcIndex > 0) {
- substring = contentString.substring(0, ssrcIndex);
- ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12);
- }
- ssrcIndex = substring.indexOf("f=");
- if (ssrcIndex > 0) {
- substring = contentString.substring(0, ssrcIndex);
- }
- SessionDescription sdp = SdpFactory.getInstance().createSessionDescription(substring);
+ // 区分TCP发流还是udp， 当前默认udp
+ if ("TCP/RTP/AVP".equalsIgnoreCase(protocol)) {
+ String setup = mediaDescription.getAttribute("setup");
+ if (setup != null) {
+ mediaTransmissionTCP = true;
+ if ("active".equalsIgnoreCase(setup)) {
+ tcpActive = true;
+ } else if ("passive".equalsIgnoreCase(setup)) {
+ tcpActive = false;
+ }
+ }
+ }
+ break;
+ }
+ }
+ if (port == -1) {
+ logger.info("不支持的媒体格式，返回415");
+ // 回复不支持的格式
+ try {
+ // 不支持的格式，发415
+ responseAck(request, Response.UNSUPPORTED_MEDIA_TYPE);
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] invite 不支持的格式: {}", e.getMessage());
+ }
+ return;
+ }
+ String username = sdp.getOrigin().getUsername();
+ String addressStr = sdp.getOrigin().getAddress();
- // 获取支持的格式
- Vector mediaDescriptions = sdp.getMediaDescriptions(true);
- // 查看是否支持PS 负载96
- int port = -1;
- //boolean recvonly = false;
- boolean mediaTransmissionTCP = false;
- Boolean tcpActive = null;
- for (int i = 0; i < mediaDescriptions.size(); i++) {
- MediaDescription mediaDescription = (MediaDescription)mediaDescriptions.get(i);
- Media media = mediaDescription.getMedia();
+ logger.info("[上级点播]用户：{}， 通道：{}, 地址：{}:{}， ssrc：{}", username, channelId, addressStr, port, ssrc);
+ Device device = null;
+ // 通过 channel 和 gbStream 是否为null 值判断来源是直播流合适国标
+ if (channel != null) {
+ device = storager.queryVideoDeviceByPlatformIdAndChannelId(requesterId, channelId);
+ if (device == null) {
+ logger.warn("点播平台{}的通道{}时未找到设备信息", requesterId, channel);
+ try {
+ responseAck(request, Response.SERVER_INTERNAL_ERROR);
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] invite 未找到设备信息: {}", e.getMessage());
+ }
+ return;
+ }
+ mediaServerItem = playService.getNewMediaServerItem(device);
+ if (mediaServerItem == null) {
+ logger.warn("未找到可用的zlm");
+ try {
+ responseAck(request, Response.BUSY_HERE);
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] invite BUSY_HERE: {}", e.getMessage());
+ }
+ return;
+ }
+ SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
+ device.getDeviceId(), channelId,
+ mediaTransmissionTCP);
- Vector mediaFormats = media.getMediaFormats(false);
- if (mediaFormats.contains("8")) {
- port = media.getMediaPort();
- String protocol = media.getProtocol();
- // 区分TCP发流还是udp， 当前默认udp
- if ("TCP/RTP/AVP".equals(protocol)) {
- String setup = mediaDescription.getAttribute("setup");
- if (setup != null) {
- mediaTransmissionTCP = true;
- if ("active".equals(setup)) {
- tcpActive = true;
- } else if ("passive".equals(setup)) {
- tcpActive = false;
- }
- }
- }
- break;
- }
- }
- if (port == -1) {
- logger.info("不支持的媒体格式，返回415");
- // 回复不支持的格式
- responseAck(evt, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415
- return;
- }
- String username = sdp.getOrigin().getUsername();
- String addressStr = sdp.getOrigin().getAddress();
- logger.info("设备{}请求语音流，地址：{}:{}，ssrc：{}", username, addressStr, port, ssrc);
+ if (tcpActive != null) {
+ sendRtpItem.setTcpActive(tcpActive);
+ }
+ if (sendRtpItem == null) {
+ logger.warn("服务器端口资源不足");
+ try {
+ responseAck(request, Response.BUSY_HERE);
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] invite 服务器端口资源不足: {}", e.getMessage());
+ }
+ return;
+ }
+ sendRtpItem.setCallId(callIdHeader.getCallId());
+ sendRtpItem.setPlayType("Play".equalsIgnoreCase(sessionName) ? InviteStreamType.PLAY : InviteStreamType.PLAYBACK);
- } else {
- logger.warn("来自无效设备/平台的请求");
- responseAck(evt, Response.BAD_REQUEST);
- }
- }
+ Long finalStartTime = startTime;
+ Long finalStopTime = stopTime;
+ ZlmHttpHookSubscribe.Event hookEvent = (mediaServerItemInUSe, responseJSON) -> {
+ String app = responseJSON.getString("app");
+ String stream = responseJSON.getString("stream");
+ logger.info("[上级点播]下级已经开始推流。 回复200OK(SDP)， {}/{}", app, stream);
+ // * 0 等待设备推流上来
+ // * 1 下级已经推流，等待上级平台回复ack
+ // * 2 推流中
+ sendRtpItem.setStatus(1);
+ redisCatchStorage.updateSendRTPSever(sendRtpItem);
+
+ StringBuffer content = new StringBuffer(200);
+ content.append("v=0\r\n");
+ content.append("o=" + channelId + " 0 0 IN IP4 " + mediaServerItemInUSe.getSdpIp() + "\r\n");
+ content.append("s=" + sessionName + "\r\n");
+ content.append("c=IN IP4 " + mediaServerItemInUSe.getSdpIp() + "\r\n");
+ if ("Playback".equalsIgnoreCase(sessionName)) {
+ content.append("t=" + finalStartTime + " " + finalStopTime + "\r\n");
+ } else {
+ content.append("t=0 0\r\n");
+ }
+ content.append("m=video " + sendRtpItem.getLocalPort() + " RTP/AVP 96\r\n");
+ content.append("a=sendonly\r\n");
+ content.append("a=rtpmap:96 PS/90000\r\n");
+ content.append("y=" + sendRtpItem.getSsrc() + "\r\n");
+ content.append("f=\r\n");
+
+ try {
+ // 超时未收到Ack应该回复bye,当前等待时间为10秒
+ dynamicTask.startDelay(callIdHeader.getCallId(), () -> {
+ logger.info("Ack 等待超时");
+ mediaServerService.releaseSsrc(mediaServerItemInUSe.getId(), sendRtpItem.getSsrc());
+ // 回复bye
+ try {
+ cmderFroPlatform.streamByeCmd(platform, callIdHeader.getCallId());
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage());
+ }
+ }, 60 * 1000);
+ responseSdpAck(request, content.toString(), platform);
+
+ } catch (SipException e) {
+ e.printStackTrace();
+ } catch (InvalidArgumentException e) {
+ e.printStackTrace();
+ } catch (ParseException e) {
+ e.printStackTrace();
+ }
+ };
+ SipSubscribe.Event errorEvent = ((event) -> {
+ // 未知错误。直接转发设备点播的错误
+ try {
+ Response response = getMessageFactory().createResponse(event.statusCode, evt.getRequest());
+ sipSender.transmitRequest(response);
+ } catch (ParseException | SipException e) {
+ e.printStackTrace();
+ }
+ });
+ sendRtpItem.setApp("rtp");
+ if ("Playback".equalsIgnoreCase(sessionName)) {
+ sendRtpItem.setPlayType(InviteStreamType.PLAYBACK);
+ SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, device.isSsrcCheck(), true);
+ sendRtpItem.setStreamId(ssrcInfo.getStream());
+ // 写入redis， 超时时回复
+ redisCatchStorage.updateSendRTPSever(sendRtpItem);
+ playService.playBack(mediaServerItem, ssrcInfo, device.getDeviceId(), channelId, DateUtil.formatter.format(start),
+ DateUtil.formatter.format(end), null, result -> {
+ if (result.getCode() != 0) {
+ logger.warn("录像回放失败");
+ if (result.getEvent() != null) {
+ errorEvent.response(result.getEvent());
+ }
+ redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null);
+ try {
+ responseAck(request, Response.REQUEST_TIMEOUT);
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] 国标级联 录像回放 发送REQUEST_TIMEOUT: {}", e.getMessage());
+ }
+ } else {
+ if (result.getMediaServerItem() != null) {
+ hookEvent.response(result.getMediaServerItem(), result.getResponse());
+ }
+ }
+ });
+ } else {
+ sendRtpItem.setPlayType(InviteStreamType.PLAY);
+ SsrcTransaction playTransaction = sessionManager.getSsrcTransaction(device.getDeviceId(), channelId, "play", null);
+ if (playTransaction != null) {
+ Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, "rtp", playTransaction.getStream());
+ if (!streamReady) {
+ boolean hasRtpServer = mediaServerService.checkRtpServer(mediaServerItem, "rtp", playTransaction.getStream());
+ if (hasRtpServer) {
+ logger.info("[上级点播]已经开启rtpServer但是尚未收到流，开启监听流的到来");
+ HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", playTransaction.getStream(), true, "rtsp", mediaServerItem.getId());
+ zlmHttpHookSubscribe.addSubscribe(hookSubscribe, hookEvent);
+ }else {
+ playTransaction = null;
+ }
+ }
+ }
+ if (playTransaction == null) {
+ String streamId = null;
+ if (mediaServerItem.isRtpEnable()) {
+ streamId = String.format("%s_%s", device.getDeviceId(), channelId);
+ }
+ SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, null, device.isSsrcCheck(), false);
+ logger.info(JSONObject.toJSONString(ssrcInfo));
+ sendRtpItem.setStreamId(ssrcInfo.getStream());
+ sendRtpItem.setSsrc(ssrc.equals(ssrcDefault) ? ssrcInfo.getSsrc() : ssrc);
+
+ // 写入redis， 超时时回复
+ redisCatchStorage.updateSendRTPSever(sendRtpItem);
+ playService.play(mediaServerItem, ssrcInfo, device, channelId, hookEvent, errorEvent, (code, msg) -> {
+ logger.info("[上级点播]超时, 用户：{}， 通道：{}", username, channelId);
+ redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null);
+ }, null);
+ } else {
+ sendRtpItem.setStreamId(playTransaction.getStream());
+ // 写入redis， 超时时回复
+ redisCatchStorage.updateSendRTPSever(sendRtpItem);
+ JSONObject jsonObject = new JSONObject();
+ jsonObject.put("app", sendRtpItem.getApp());
+ jsonObject.put("stream", sendRtpItem.getStreamId());
+ hookEvent.response(mediaServerItem, jsonObject);
+ }
+ }
+ } else if (gbStream != null) {
+ if("push".equals(gbStream.getStreamType())) {
+ if (streamPushItem != null && streamPushItem.isPushIng()) {
+ // 推流状态
+ pushStream(evt, request, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
+ mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+ } else {
+ // 未推流 拉起
+ notifyStreamOnline(evt, request,gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
+ mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+ }
+ }else if ("proxy".equals(gbStream.getStreamType())){
+ if (null != proxyByAppAndStream) {
+ if(proxyByAppAndStream.isStatus()){
+ pushProxyStream(evt, request, gbStream, platform, callIdHeader, mediaServerItem, port, tcpActive,
+ mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+ }else{
+ //开启代理拉流
+ notifyStreamOnline(evt, request,gbStream, null, platform, callIdHeader, mediaServerItem, port, tcpActive,
+ mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+ }
+ }
+
+
+ }
+ }
+ }
+ } catch (SdpParseException e) {
+ logger.error("sdp解析错误", e);
+ } catch (SdpException e) {
+ e.printStackTrace();
+ }
+ }
+
+ /**
+ * 安排推流
+ */
+ private void pushProxyStream(RequestEvent evt, SIPRequest request, GbStream gbStream, ParentPlatform platform,
+ CallIdHeader callIdHeader, MediaServerItem mediaServerItem,
+ int port, Boolean tcpActive, boolean mediaTransmissionTCP,
+ String channelId, String addressStr, String ssrc, String requesterId) {
+ Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream());
+ if (streamReady) {
+ // 自平台内容
+ SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
+ gbStream.getApp(), gbStream.getStream(), channelId,
+ mediaTransmissionTCP);
+
+ if (sendRtpItem == null) {
+ logger.warn("服务器端口资源不足");
+ try {
+ responseAck(request, Response.BUSY_HERE);
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] invite 服务器端口资源不足: {}", e.getMessage());
+ }
+ return;
+ }
+ if (tcpActive != null) {
+ sendRtpItem.setTcpActive(tcpActive);
+ }
+ sendRtpItem.setPlayType(InviteStreamType.PUSH);
+ // 写入redis， 超时时回复
+ sendRtpItem.setStatus(1);
+ sendRtpItem.setCallId(callIdHeader.getCallId());
+ sendRtpItem.setFromTag(request.getFromTag());
+
+ SIPResponse response = sendStreamAck(mediaServerItem, request, sendRtpItem, platform, evt);
+ if (response != null) {
+ sendRtpItem.setToTag(response.getToTag());
+ }
+ redisCatchStorage.updateSendRTPSever(sendRtpItem);
+
+ }
+
+ }
+ private void pushStream(RequestEvent evt, SIPRequest request, GbStream gbStream, StreamPushItem streamPushItem, ParentPlatform platform,
+ CallIdHeader callIdHeader, MediaServerItem mediaServerItem,
+ int port, Boolean tcpActive, boolean mediaTransmissionTCP,
+ String channelId, String addressStr, String ssrc, String requesterId) {
+ // 推流
+ if (streamPushItem.isSelf()) {
+ Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream());
+ if (streamReady) {
+ // 自平台内容
+ SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
+ gbStream.getApp(), gbStream.getStream(), channelId,
+ mediaTransmissionTCP);
+
+ if (sendRtpItem == null) {
+ logger.warn("服务器端口资源不足");
+ try {
+ responseAck(request, Response.BUSY_HERE);
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] invite 服务器端口资源不足: {}", e.getMessage());
+ }
+ return;
+ }
+ if (tcpActive != null) {
+ sendRtpItem.setTcpActive(tcpActive);
+ }
+ sendRtpItem.setPlayType(InviteStreamType.PUSH);
+ // 写入redis， 超时时回复
+ sendRtpItem.setStatus(1);
+ sendRtpItem.setCallId(callIdHeader.getCallId());
+
+ sendRtpItem.setFromTag(request.getFromTag());
+ SIPResponse response = sendStreamAck(mediaServerItem, request, sendRtpItem, platform, evt);
+ if (response != null) {
+ sendRtpItem.setToTag(response.getToTag());
+ }
+
+ redisCatchStorage.updateSendRTPSever(sendRtpItem);
+
+ } else {
+ // 不在线 拉起
+ notifyStreamOnline(evt, request,gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
+ mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+ }
+
+ } else {
+ // 其他平台内容
+ otherWvpPushStream(evt, request, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
+ mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+ }
+ }
+ /**
+ * 通知流上线
+ */
+ private void notifyStreamOnline(RequestEvent evt, SIPRequest request, GbStream gbStream, StreamPushItem streamPushItem, ParentPlatform platform,
+ CallIdHeader callIdHeader, MediaServerItem mediaServerItem,
+ int port, Boolean tcpActive, boolean mediaTransmissionTCP,
+ String channelId, String addressStr, String ssrc, String requesterId) {
+ if ("proxy".equals(gbStream.getStreamType())) {
+ // TODO 控制启用以使设备上线
+ logger.info("[ app={}, stream={} ]通道未推流，启用流后开始推流", gbStream.getApp(), gbStream.getStream());
+ // 监听流上线
+ HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed(gbStream.getApp(), gbStream.getStream(), true, "rtsp", mediaServerItem.getId());
+ zlmHttpHookSubscribe.addSubscribe(hookSubscribe, (mediaServerItemInUSe, responseJSON) -> {
+ String app = responseJSON.getString("app");
+ String stream = responseJSON.getString("stream");
+ logger.info("[上级点播]拉流代理已经就绪， {}/{}", app, stream);
+ dynamicTask.stop(callIdHeader.getCallId());
+ pushProxyStream(evt, request, gbStream, platform, callIdHeader, mediaServerItem, port, tcpActive,
+ mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+ });
+ dynamicTask.startDelay(callIdHeader.getCallId(), () -> {
+ logger.info("[ app={}, stream={} ] 等待拉流代理流超时", gbStream.getApp(), gbStream.getStream());
+ zlmHttpHookSubscribe.removeSubscribe(hookSubscribe);
+ }, userSetting.getPlatformPlayTimeout());
+ boolean start = streamProxyService.start(gbStream.getApp(), gbStream.getStream());
+ if (!start) {
+ try {
+ responseAck(request, Response.BUSY_HERE, "channel [" + gbStream.getGbId() + "] offline");
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] invite 通道未推流: {}", e.getMessage());
+ }
+ zlmHttpHookSubscribe.removeSubscribe(hookSubscribe);
+ dynamicTask.stop(callIdHeader.getCallId());
+ }
+
+
+
+ } else if ("push".equals(gbStream.getStreamType())) {
+ if (!platform.isStartOfflinePush()) {
+ // 平台设置中关闭了拉起离线的推流则直接回复
+ try {
+ responseAck(request, Response.TEMPORARILY_UNAVAILABLE, "channel stream not pushing");
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] invite 通道未推流: {}", e.getMessage());
+ }
+ return;
+ }
+ // 发送redis消息以使设备上线
+ logger.info("[ app={}, stream={} ]通道未推流，发送redis信息控制设备开始推流", gbStream.getApp(), gbStream.getStream());
+
+ MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(1,
+ gbStream.getApp(), gbStream.getStream(), gbStream.getGbId(), gbStream.getPlatformId(),
+ platform.getName(), null, gbStream.getMediaServerId());
+ redisCatchStorage.sendStreamPushRequestedMsg(messageForPushChannel);
+ // 设置超时
+ dynamicTask.startDelay(callIdHeader.getCallId(), () -> {
+ logger.info("[ app={}, stream={} ] 等待设备开始推流超时", gbStream.getApp(), gbStream.getStream());
+ try {
+ mediaListManager.removedChannelOnlineEventLister(gbStream.getApp(), gbStream.getStream());
+ responseAck(request, Response.REQUEST_TIMEOUT); // 超时
+ } catch (SipException e) {
+ e.printStackTrace();
+ } catch (InvalidArgumentException e) {
+ e.printStackTrace();
+ } catch (ParseException e) {
+ e.printStackTrace();
+ }
+ }, userSetting.getPlatformPlayTimeout());
+ // 添加监听
+ int finalPort = port;
+ Boolean finalTcpActive = tcpActive;
+
+ // 添加在本机上线的通知
+ mediaListManager.addChannelOnlineEventLister(gbStream.getApp(), gbStream.getStream(), (app, stream, serverId) -> {
+ dynamicTask.stop(callIdHeader.getCallId());
+ if (serverId.equals(userSetting.getServerId())) {
+ SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, finalPort, ssrc, requesterId,
+ app, stream, channelId, mediaTransmissionTCP);
+
+ if (sendRtpItem == null) {
+ logger.warn("上级点时创建sendRTPItem失败，可能是服务器端口资源不足");
+ try {
+ responseAck(request, Response.BUSY_HERE);
+ } catch (SipException e) {
+ e.printStackTrace();
+ } catch (InvalidArgumentException e) {
+ e.printStackTrace();
+ } catch (ParseException e) {
+ e.printStackTrace();
+ }
+ return;
+ }
+ if (finalTcpActive != null) {
+ sendRtpItem.setTcpActive(finalTcpActive);
+ }
+ sendRtpItem.setPlayType(InviteStreamType.PUSH);
+ // 写入redis， 超时时回复
+ sendRtpItem.setStatus(1);
+ sendRtpItem.setCallId(callIdHeader.getCallId());
+
+ sendRtpItem.setFromTag(request.getFromTag());
+ SIPResponse response = sendStreamAck(mediaServerItem, request, sendRtpItem, platform, evt);
+ if (response != null) {
+ sendRtpItem.setToTag(response.getToTag());
+ }
+ redisCatchStorage.updateSendRTPSever(sendRtpItem);
+ } else {
+ // 其他平台内容
+ otherWvpPushStream(evt, request, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
+ mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+ }
+ });
+
+ // 添加回复的拒绝或者错误的通知
+ redisPushStreamResponseListener.addEvent(gbStream.getApp(), gbStream.getStream(), response -> {
+ if (response.getCode() != 0) {
+ dynamicTask.stop(callIdHeader.getCallId());
+ mediaListManager.removedChannelOnlineEventLister(gbStream.getApp(), gbStream.getStream());
+ try {
+ responseAck(request, Response.TEMPORARILY_UNAVAILABLE, response.getMsg());
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] 国标级联 点播回复: {}", e.getMessage());
+ }
+ }
+ });
+ }
+ }
+
+ /**
+ * 来自其他wvp的推流
+ */
+ private void otherWvpPushStream(RequestEvent evt, SIPRequest request, GbStream gbStream, StreamPushItem streamPushItem, ParentPlatform platform,
+ CallIdHeader callIdHeader, MediaServerItem mediaServerItem,
+ int port, Boolean tcpActive, boolean mediaTransmissionTCP,
+ String channelId, String addressStr, String ssrc, String requesterId) {
+ logger.info("[级联点播]直播流来自其他平台，发送redis消息");
+ // 发送redis消息
+ redisGbPlayMsgListener.sendMsg(streamPushItem.getServerId(), streamPushItem.getMediaServerId(),
+ streamPushItem.getApp(), streamPushItem.getStream(), addressStr, port, ssrc, requesterId,
+ channelId, mediaTransmissionTCP, null, responseSendItemMsg -> {
+ SendRtpItem sendRtpItem = responseSendItemMsg.getSendRtpItem();
+ if (sendRtpItem == null || responseSendItemMsg.getMediaServerItem() == null) {
+ logger.warn("服务器端口资源不足");
+ try {
+ responseAck(request, Response.BUSY_HERE);
+ } catch (SipException e) {
+ e.printStackTrace();
+ } catch (InvalidArgumentException e) {
+ e.printStackTrace();
+ } catch (ParseException e) {
+ e.printStackTrace();
+ }
+ return;
+ }
+ // 收到sendItem
+ if (tcpActive != null) {
+ sendRtpItem.setTcpActive(tcpActive);
+ }
+ sendRtpItem.setPlayType(InviteStreamType.PUSH);
+ // 写入redis， 超时时回复
+ sendRtpItem.setStatus(1);
+ sendRtpItem.setCallId(callIdHeader.getCallId());
+
+ sendRtpItem.setFromTag(request.getFromTag());
+ SIPResponse response = sendStreamAck(responseSendItemMsg.getMediaServerItem(), request,sendRtpItem, platform, evt);
+ if (response != null) {
+ sendRtpItem.setToTag(response.getToTag());
+ }
+ redisCatchStorage.updateSendRTPSever(sendRtpItem);
+ }, (wvpResult) -> {
+
+ // 错误
+ if (wvpResult.getCode() == RedisGbPlayMsgListener.ERROR_CODE_OFFLINE) {
+ // 离线
+ // 查询是否在本机上线了
+ StreamPushItem currentStreamPushItem = streamPushService.getPush(streamPushItem.getApp(), streamPushItem.getStream());
+ if (currentStreamPushItem.isPushIng()) {
+ // 在线状态
+ pushStream(evt, request, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
+ mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+
+ } else {
+ // 不在线 拉起
+ notifyStreamOnline(evt, request, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
+ mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+ }
+ }
+ try {
+ responseAck(request, Response.BUSY_HERE);
+ } catch (InvalidArgumentException | ParseException | SipException e) {
+ logger.error("[命令发送失败] 国标级联 点播回复 BUSY_HERE: {}", e.getMessage());
+ }
+ });
+ }
+
+ public SIPResponse sendStreamAck(MediaServerItem mediaServerItem, SIPRequest request, SendRtpItem sendRtpItem, ParentPlatform platform, RequestEvent evt) {
+
+ StringBuffer content = new StringBuffer(200);
+ content.append("v=0\r\n");
+ content.append("o=" + sendRtpItem.getChannelId() + " 0 0 IN IP4 " + mediaServerItem.getSdpIp() + "\r\n");
+ content.append("s=Play\r\n");
+ content.append("c=IN IP4 " + mediaServerItem.getSdpIp() + "\r\n");
+ content.append("t=0 0\r\n");
+ content.append("m=video " + sendRtpItem.getLocalPort() + " RTP/AVP 96\r\n");
+ content.append("a=sendonly\r\n");
+ content.append("a=rtpmap:96 PS/90000\r\n");
+ if (sendRtpItem.isTcp()) {
+ content.append("a=connection:new\r\n");
+ if (!sendRtpItem.isTcpActive()) {
+ content.append("a=setup:active\r\n");
+ } else {
+ content.append("a=setup:passive\r\n");
+ }
+ }
+ content.append("y=" + sendRtpItem.getSsrc() + "\r\n");
+ content.append("f=\r\n");
+
+ try {
+ return responseSdpAck(request, content.toString(), platform);
+ } catch (SipException e) {
+ e.printStackTrace();
+ } catch (InvalidArgumentException e) {
+ e.printStackTrace();
+ } catch (ParseException e) {
+ e.printStackTrace();
+ }
+ return null;
+ }
+
+ public void inviteFromDeviceHandle(SIPRequest request, String requesterId) {
+
+ // 非上级平台请求，查询是否设备请求（通常为接收语音广播的设备）
+ Device device = redisCatchStorage.getDevice(requesterId);
+ if (device != null) {
+ logger.info("收到设备" + requesterId + "的语音广播Invite请求");
+ try {
+ responseAck(request, Response.TRYING);
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] invite BAD_REQUEST: {}", e.getMessage());
+ }
+ String contentString = new String(request.getRawContent());
+ // jainSip不支持y=字段， 移除移除以解析。
+ String substring = contentString;
+ String ssrc = "0000000404";
+ int ssrcIndex = contentString.indexOf("y=");
+ if (ssrcIndex > 0) {
+ substring = contentString.substring(0, ssrcIndex);
+ ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12);
+ }
+ ssrcIndex = substring.indexOf("f=");
+ if (ssrcIndex > 0) {
+ substring = contentString.substring(0, ssrcIndex);
+ }
+ SessionDescription sdp = null;
+ try {
+ sdp = SdpFactory.getInstance().createSessionDescription(substring);
+ // 获取支持的格式
+ Vector mediaDescriptions = sdp.getMediaDescriptions(true);
+ // 查看是否支持PS 负载96
+ int port = -1;
+ //boolean recvonly = false;
+ boolean mediaTransmissionTCP = false;
+ Boolean tcpActive = null;
+ for (int i = 0; i < mediaDescriptions.size(); i++) {
+ MediaDescription mediaDescription = (MediaDescription) mediaDescriptions.get(i);
+ Media media = mediaDescription.getMedia();
+
+ Vector mediaFormats = media.getMediaFormats(false);
+ if (mediaFormats.contains("8")) {
+ port = media.getMediaPort();
+ String protocol = media.getProtocol();
+ // 区分TCP发流还是udp， 当前默认udp
+ if ("TCP/RTP/AVP".equals(protocol)) {
+ String setup = mediaDescription.getAttribute("setup");
+ if (setup != null) {
+ mediaTransmissionTCP = true;
+ if ("active".equals(setup)) {
+ tcpActive = true;
+ } else if ("passive".equals(setup)) {
+ tcpActive = false;
+ }
+ }
+ }
+ break;
+ }
+ }
+ if (port == -1) {
+ logger.info("不支持的媒体格式，返回415");
+ // 回复不支持的格式
+ try {
+ responseAck(request, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] invite 不支持的媒体格式，返回415， {}", e.getMessage());
+ }
+ return;
+ }
+ String username = sdp.getOrigin().getUsername();
+ String addressStr = sdp.getOrigin().getAddress();
+ logger.info("设备{}请求语音流，地址：{}:{}，ssrc：{}", username, addressStr, port, ssrc);
+ } catch (SdpException e) {
+ logger.error("[SDP解析异常]", e);
+ }
+
+
+
+ } else {
+ logger.warn("来自无效设备/平台的请求");
+ try {
+ responseAck(request, Response.BAD_REQUEST);; // 不支持的格式，发415
+ } catch (SipException | InvalidArgumentException | ParseException e) {
+ logger.error("[命令发送失败] invite 来自无效设备/平台的请求， {}", e.getMessage());
+ }
+ }
+ }
}
--
Gitblit v1.9.3