diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java index 2e70ea75..3afce16a 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java @@ -4,7 +4,6 @@ import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.event.SipSubscribe; import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform; import com.genersoft.iot.vmp.gb28181.transmit.cmd.SIPRequestHeaderPlarformProvider; -import com.genersoft.iot.vmp.utils.DateUtil; import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.service.IMediaServerService; @@ -26,12 +25,13 @@ import org.springframework.lang.Nullable; import org.springframework.stereotype.Component; import org.springframework.util.StringUtils; + +import com.genersoft.iot.vmp.utils.DateUtil; import javax.sip.*; import javax.sip.address.SipURI; import javax.sip.header.*; import javax.sip.message.Request; import java.lang.reflect.Field; -import java.net.InetAddress; import java.text.ParseException; import java.util.ArrayList; import java.util.HashSet; 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 33bee758..c35f21b6 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 @@ -1,8 +1,6 @@ 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.*; import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager; @@ -22,7 +20,6 @@ import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import gov.nist.javax.sip.message.SIPRequest; import gov.nist.javax.sip.stack.SIPDialog; 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; @@ -37,7 +34,6 @@ 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.text.ParseException; import java.util.*; @@ -99,28 +95,28 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In @Override public void process(RequestEvent evt) { Dialog dialog = evt.getDialog(); - CallIdHeader callIdHeader = (CallIdHeader)evt.getRequest().getHeader(CallIdHeader.NAME); + CallIdHeader callIdHeader = (CallIdHeader) evt.getRequest().getHeader(CallIdHeader.NAME); if (dialog == null) { return; } - if (dialog.getState()== DialogState.CONFIRMED) { + 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(); - SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(platformGbId, null, null, callIdHeader.getCallId()); + SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(platformGbId, null, null, callIdHeader.getCallId()); String is_Udp = sendRtpItem.isTcp() ? "0" : "1"; MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); logger.info("[收到ACK]，开始使用{}向上级推流 {}/{}->{}:{}({})", sendRtpItem.isTcp() ? "TCP" : "UDP", sendRtpItem.getApp(), sendRtpItem.getStreamId(), - sendRtpItem.getIp() ,sendRtpItem.getPort(), + sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc()); Map param = new HashMap<>(); - param.put("vhost","__defaultVhost__"); - param.put("app",sendRtpItem.getApp()); - param.put("stream",sendRtpItem.getStreamId()); + param.put("vhost", "__defaultVhost__"); + param.put("app", sendRtpItem.getApp()); + param.put("stream", sendRtpItem.getStreamId()); param.put("ssrc", sendRtpItem.getSsrc()); param.put("src_port", sendRtpItem.getLocalPort()); param.put("pt", sendRtpItem.getPt()); @@ -129,9 +125,9 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In JSONObject jsonObject; if (sendRtpItem.isTcpActive()) { jsonObject = zlmrtpServerFactory.startSendRtpPassive(mediaInfo, param); - }else { + } else { param.put("is_udp", is_Udp); - param.put("dst_url",sendRtpItem.getIp()); + param.put("dst_url", sendRtpItem.getIp()); param.put("dst_port", sendRtpItem.getPort()); jsonObject = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param); } @@ -149,36 +145,37 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In String waiteStreamTimeoutTaskKey = "waite-stream-" + audioBroadcastCatch.getDeviceId() + audioBroadcastCatch.getChannelId(); dynamicTask.stop(waiteStreamTimeoutTaskKey); } - logger.info("RTP推流成功[ {}/{} ]，{}->{}:{}, " ,param.get("app"), param.get("stream"), jsonObject.getString("local_port"), param.get("dst_url"), param.get("dst_port")); + 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)); + logger.error("RTP推流失败: {}, 参数：{}", jsonObject.getString("msg"), JSONObject.toJSON(param)); if (sendRtpItem.isOnlyAudio()) { // 语音对讲 try { - cmder.streamByeCmd((SIPDialog) evt.getDialog(), (SIPRequest)evt.getRequest(), null); + cmder.streamByeCmd((SIPDialog) evt.getDialog(), (SIPRequest) evt.getRequest(), null); } catch (SipException e) { throw new RuntimeException(e); } catch (ParseException e) { throw new RuntimeException(e); } - }else { + } else { // 向上级平台 commanderForPlatform.streamByeCmd(parentPlatform, callIdHeader.getCallId()); } - 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 (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, json -> { + startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, json, param, callIdHeader); + }); + } else { + JSONObject startSendRtpStreamResult = zlmrtpServerFactory.startSendRtpStream(mediaInfo, param); + startSendRtpStreamHand(evt, sendRtpItem, parentPlatform, startSendRtpStreamResult, param, callIdHeader); + } + + } - - } } private void startSendRtpStreamHand(RequestEvent evt, SendRtpItem sendRtpItem, ParentPlatform parentPlatform, 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 095f5e29..6c04c20f 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 @@ -19,15 +19,14 @@ 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.gb28181.utils.XmlUtil; import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe; import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager; import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils; import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; import com.genersoft.iot.vmp.media.zlm.dto.MediaItem; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; -import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItemLite; +import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem; import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.IPlayService; import com.genersoft.iot.vmp.service.IStreamPushService; @@ -58,8 +57,6 @@ import javax.sip.message.Request; import javax.sip.message.Response; import java.text.ParseException; import java.time.Instant; -import java.util.Date; -import java.util.List; import java.util.Vector; /** @@ -109,9 +106,6 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements @Autowired private ZLMRESTfulUtils zlmresTfulUtils; - @Autowired - private IMediaServerService mediaServerService; - @Autowired private SIPProcessorObserver sipProcessorObserver; @@ -171,7 +165,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements // 查询请求是否来自上级平台\设备 ParentPlatform platform = storager.queryParentPlatByServerGBId(requesterId); if (platform == null) { - inviteFromDeviceHandle(evt, requesterId); + inviteFromDeviceHandle(evt, requesterId, channelId); } else { // 查询平台下是否有该通道 DeviceChannel channel = storager.queryChannelInParentPlatform(requesterId, channelId); @@ -724,10 +718,16 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements } } - public void inviteFromDeviceHandle(RequestEvent evt, String requesterId) throws InvalidArgumentException, ParseException, SipException, SdpException { + public void inviteFromDeviceHandle(RequestEvent evt, String requesterId, String channelId1) throws InvalidArgumentException, ParseException, SipException, SdpException { // 非上级平台请求，查询是否设备请求（通常为接收语音广播的设备） Device device = redisCatchStorage.getDevice(requesterId); + AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(requesterId, channelId1); + if (audioBroadcastCatch == null) { + logger.warn("来自设备的Invite请求非语音广播，已忽略"); + responseAck(evt, Response.FORBIDDEN); + return; + } Request request = evt.getRequest(); if (device != null) { logger.info("收到设备" + requesterId + "的语音广播Invite请求"); @@ -740,7 +740,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements int ssrcIndex = contentString.indexOf("y="); if (ssrcIndex > 0) { substring = contentString.substring(0, ssrcIndex); - ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12); + ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12).trim(); } ssrcIndex = substring.indexOf("f="); if (ssrcIndex > 0) { @@ -750,13 +750,13 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements // 获取支持的格式 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); + MediaDescription mediaDescription = (MediaDescription)mediaDescriptions.get(i); Media media = mediaDescription.getMedia(); Vector mediaFormats = media.getMediaFormats(false); @@ -784,233 +784,248 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements 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); + logger.info("设备{}请求语音流，地址：{}:{}，ssrc：{}", requesterId, addressStr, port, ssrc); + MediaServerItem 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(), audioBroadcastCatch.getChannelId(), + mediaTransmissionTCP); + if (sendRtpItem == null) { + logger.warn("服务器端口资源不足"); + responseAck(evt, Response.BUSY_HERE); + return; + } + sendRtpItem.setTcp(mediaTransmissionTCP); + if (tcpActive != null) { + sendRtpItem.setTcpActive(tcpActive); + } + String app = "broadcast"; + String stream = device.getDeviceId() + "_" + audioBroadcastCatch.getChannelId(); + + CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME); + sendRtpItem.setPlayType(InviteStreamType.PLAY); + sendRtpItem.setCallId(callIdHeader.getCallId()); + sendRtpItem.setPlatformId(requesterId); + sendRtpItem.setStatus(1); + sendRtpItem.setApp(app); + sendRtpItem.setStreamId(stream); + sendRtpItem.setPt(8); + sendRtpItem.setUsePs(false); + sendRtpItem.setOnlyAudio(true); + redisCatchStorage.updateSendRTPSever(sendRtpItem); + + // hook监听等待设备推流上来 + // 添加订阅 + JSONObject subscribeKey = new JSONObject(); + subscribeKey.put("app", app); + subscribeKey.put("stream", stream); + subscribeKey.put("regist", true); + subscribeKey.put("schema", "rtmp"); + subscribeKey.put("mediaServerId", mediaServerItem.getId()); + String finalSsrc = ssrc; + // 流已经存在时直接推流 + JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo(mediaServerItem, app, "rtsp", stream); + JSONArray tracks = mediaInfo.getJSONArray("tracks"); + Integer codecId = null; + if (tracks != null && tracks.size() > 0) { + for (int i = 0; i < tracks.size(); i++) { + MediaItem.MediaTrack track = JSON.toJavaObject((JSON)tracks.get(i),MediaItem.MediaTrack.class); + if (track.getCodecType() == 1) { + codecId = track.getCodecId(); + break; + } + } + } + if ((mediaInfo.getInteger("code") == 0 && mediaInfo.getBoolean("online"))) { + logger.info("发现已经在推流"); + sendRtpItem.setStatus(2); + redisCatchStorage.updateSendRTPSever(sendRtpItem); + StringBuffer content = new StringBuffer(200); + content.append("v=0\r\n"); + content.append("o="+ config.getId() +" "+ sdp.getOrigin().getSessionId() +" " + sdp.getOrigin().getSessionVersion() + " 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"); + if (codecId == null) { + if (mediaTransmissionTCP) { + content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n"); + }else { + content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n"); + } + + content.append("a=rtpmap:8 PCMA/8000\r\n"); + }else { + if (codecId == 4) { + if (mediaTransmissionTCP) { + content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 0\r\n"); + }else { + content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 0\r\n"); + } + content.append("a=rtpmap:0 PCMU/8000\r\n"); + }else { + if (mediaTransmissionTCP) { + content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n"); + }else { + content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n"); + } + content.append("a=rtpmap:8 PCMA/8000\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("a=sendonly\r\n"); + content.append("y="+ finalSsrc + "\r\n"); + content.append("f=v/////a/1/8/1\r\n"); + + ParentPlatform parentPlatform = new ParentPlatform(); + parentPlatform.setServerIP(device.getIp()); + parentPlatform.setServerPort(device.getPort()); + parentPlatform.setServerGBId(device.getDeviceId()); + try { + responseSdpAck(evt, content.toString(), parentPlatform); + Dialog dialog = evt.getDialog(); + audioBroadcastCatch.setDialog((SIPDialog) dialog); + audioBroadcastCatch.setRequest((SIPRequest) request); + audioBroadcastManager.update(audioBroadcastCatch); + } catch (SipException e) { + throw new RuntimeException(e); + } catch (InvalidArgumentException e) { + throw new RuntimeException(e); + } catch (ParseException e) { + throw new RuntimeException(e); + } + }else { + // 流不存在时监听流上线 + // 设置等待推流的超时; 默认20s + String waiteStreamTimeoutTaskKey = "waite-stream-" + device.getDeviceId() + audioBroadcastCatch.getChannelId(); + dynamicTask.startDelay(waiteStreamTimeoutTaskKey, ()->{ + logger.info("等待推流超时: {}/{}", app, stream); + subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey); + playService.stopAudioBroadcast(device.getDeviceId(), audioBroadcastCatch.getChannelId()); + // 发送bye + try { + responseAck(evt, Response.BUSY_HERE); + } catch (SipException e) { + throw new RuntimeException(e); + } catch (InvalidArgumentException e) { + throw new RuntimeException(e); + } catch (ParseException e) { + throw new RuntimeException(e); + } + }, 20*1000); + + boolean finalMediaTransmissionTCP = mediaTransmissionTCP; + subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey, + (MediaServerItem mediaServerItemInUse, JSONObject json)->{ + logger.info("收到语音对讲推流"); + MediaItem mediaItem = JSON.toJavaObject(json, MediaItem.class); + Integer audioCodecId = null; + if (mediaItem.getTracks() != null && mediaItem.getTracks().size() > 0) { + for (int i = 0; i < mediaItem.getTracks().size(); i++) { + MediaItem.MediaTrack mediaTrack = mediaItem.getTracks().get(i); + if (mediaTrack.getCodecType() == 1) { + audioCodecId = mediaTrack.getCodecId(); + break; + } + } + } + + try { + sendRtpItem.setStatus(2); + redisCatchStorage.updateSendRTPSever(sendRtpItem); + StringBuffer content = new StringBuffer(200); + content.append("v=0\r\n"); + content.append("o="+ config.getId() +" "+ sdp.getOrigin().getSessionId() +" " + sdp.getOrigin().getSessionVersion() + " 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"); + if (audioCodecId == null) { + if (finalMediaTransmissionTCP) { + content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n"); + }else { + content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n"); + } + + content.append("a=rtpmap:8 PCMA/8000\r\n"); + }else { + if (audioCodecId == 4) { + if (finalMediaTransmissionTCP) { + content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 0\r\n"); + }else { + content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 0\r\n"); + } + content.append("a=rtpmap:0 PCMU/8000\r\n"); + }else { + if (finalMediaTransmissionTCP) { + content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n"); + }else { + content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n"); + } + content.append("a=rtpmap:8 PCMA/8000\r\n"); + } + } + content.append("a=sendonly\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="+ finalSsrc + "\r\n"); + content.append("f=v/////a/1/8/1\r\n"); + + ParentPlatform parentPlatform = new ParentPlatform(); + parentPlatform.setServerIP(device.getIp()); + parentPlatform.setServerPort(device.getPort()); + parentPlatform.setServerGBId(device.getDeviceId()); + + responseSdpAck(evt, content.toString(), parentPlatform); + Dialog dialog = evt.getDialog(); + audioBroadcastCatch.setDialog((SIPDialog) dialog); + audioBroadcastCatch.setRequest((SIPRequest) request); + audioBroadcastManager.update(audioBroadcastCatch); + } catch (SipException e) { + throw new RuntimeException(e); + } catch (InvalidArgumentException e) { + throw new RuntimeException(e); + } catch (ParseException e) { + throw new RuntimeException(e); + } catch (SdpParseException e) { + throw new RuntimeException(e); + } + }); + } + String key = DeferredResultHolder.CALLBACK_CMD_BROADCAST + device.getDeviceId(); + WVPResult wvpResult = new WVPResult<>(); + wvpResult.setCode(0); + wvpResult.setMsg("success"); + AudioBroadcastResult audioBroadcastResult = new AudioBroadcastResult(); + audioBroadcastResult.setApp(app); + audioBroadcastResult.setStream(stream); + audioBroadcastResult.setMediaServerItem(new MediaServerItemLite(mediaServerItem)); + audioBroadcastResult.setCodec("G.711"); + wvpResult.setData(audioBroadcastResult); + RequestMessage requestMessage = new RequestMessage(); + requestMessage.setKey(key); + requestMessage.setData(wvpResult); + resultHolder.invokeAllResult(requestMessage); } else { logger.warn("来自无效设备/平台的请求"); responseAck(evt, Response.BAD_REQUEST); } } - CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME); - sendRtpItem.setPlayType(InviteStreamType.PLAY); - sendRtpItem.setCallId(callIdHeader.getCallId()); - sendRtpItem.setPlatformId(requesterId); - sendRtpItem.setStatus(1); - sendRtpItem.setApp(app); - sendRtpItem.setStreamId(stream); - sendRtpItem.setPt(8); - sendRtpItem.setUsePs(false); - sendRtpItem.setOnlyAudio(true); - redisCatchStorage.updateSendRTPSever(sendRtpItem); - - // hook监听等待设备推流上来 - // 添加订阅 - JSONObject subscribeKey = new JSONObject(); - subscribeKey.put("app", app); - subscribeKey.put("stream", stream); - subscribeKey.put("regist", true); - subscribeKey.put("schema", "rtmp"); - subscribeKey.put("mediaServerId", mediaServerItem.getId()); - String finalSsrc = ssrc; - // 流已经存在时直接推流 - JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo(mediaServerItem, app, "rtsp", stream); - JSONArray tracks = mediaInfo.getJSONArray("tracks"); - Integer codecId = null; - if (tracks != null && tracks.size() > 0) { - for (int i = 0; i < tracks.size(); i++) { - MediaItem.MediaTrack track = JSON.toJavaObject((JSON)tracks.get(i),MediaItem.MediaTrack.class); - if (track.getCodecType() == 1) { - codecId = track.getCodecId(); - break; - } - } - } - if ((mediaInfo.getInteger("code") == 0 && mediaInfo.getBoolean("online"))) { - logger.info("发现已经在推流"); - sendRtpItem.setStatus(2); - redisCatchStorage.updateSendRTPSever(sendRtpItem); - StringBuffer content = new StringBuffer(200); - content.append("v=0\r\n"); - content.append("o="+ config.getId() +" "+ sdp.getOrigin().getSessionId() +" " + sdp.getOrigin().getSessionVersion() + " 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"); - if (codecId == null) { - if (mediaTransmissionTCP) { - content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n"); - }else { - content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n"); - } - - content.append("a=rtpmap:8 PCMA/8000\r\n"); - }else { - if (codecId == 4) { - if (mediaTransmissionTCP) { - content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 0\r\n"); - }else { - content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 0\r\n"); - } - content.append("a=rtpmap:0 PCMU/8000\r\n"); - }else { - if (mediaTransmissionTCP) { - content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n"); - }else { - content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n"); - } - content.append("a=rtpmap:8 PCMA/8000\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("a=sendonly\r\n"); - content.append("y="+ finalSsrc + "\r\n"); - content.append("f=v/////a/1/8/1\r\n"); - - ParentPlatform parentPlatform = new ParentPlatform(); - parentPlatform.setServerIP(device.getIp()); - parentPlatform.setServerPort(device.getPort()); - parentPlatform.setServerGBId(device.getDeviceId()); - try { - responseSdpAck(evt, content.toString(), parentPlatform); - Dialog dialog = evt.getDialog(); - audioBroadcastCatch.setDialog((SIPDialog) dialog); - audioBroadcastCatch.setRequest((SIPRequest) request); - audioBroadcastManager.update(audioBroadcastCatch); - } catch (SipException e) { - throw new RuntimeException(e); - } catch (InvalidArgumentException e) { - throw new RuntimeException(e); - } catch (ParseException e) { - throw new RuntimeException(e); - } - }else { - // 流不存在时监听流上线 - // 设置等待推流的超时; 默认20s - String waiteStreamTimeoutTaskKey = "waite-stream-" + device.getDeviceId() + audioBroadcastCatch.getChannelId(); - dynamicTask.startDelay(waiteStreamTimeoutTaskKey, ()->{ - logger.info("等待推流超时: {}/{}", app, stream); - subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey); - playService.stopAudioBroadcast(device.getDeviceId(), audioBroadcastCatch.getChannelId()); - // 发送bye - try { - responseAck(evt, Response.BUSY_HERE); - } catch (SipException e) { - throw new RuntimeException(e); - } catch (InvalidArgumentException e) { - throw new RuntimeException(e); - } catch (ParseException e) { - throw new RuntimeException(e); - } - }, 20*1000); - - boolean finalMediaTransmissionTCP = mediaTransmissionTCP; - subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey, - (MediaServerItem mediaServerItemInUse, JSONObject json)->{ - logger.info("收到语音对讲推流"); - MediaItem mediaItem = JSON.toJavaObject(json, MediaItem.class); - Integer audioCodecId = null; - if (mediaItem.getTracks() != null && mediaItem.getTracks().size() > 0) { - for (int i = 0; i < mediaItem.getTracks().size(); i++) { - MediaItem.MediaTrack mediaTrack = mediaItem.getTracks().get(i); - if (mediaTrack.getCodecType() == 1) { - audioCodecId = mediaTrack.getCodecId(); - break; - } - } - } - - try { - sendRtpItem.setStatus(2); - redisCatchStorage.updateSendRTPSever(sendRtpItem); - StringBuffer content = new StringBuffer(200); - content.append("v=0\r\n"); - content.append("o="+ config.getId() +" "+ sdp.getOrigin().getSessionId() +" " + sdp.getOrigin().getSessionVersion() + " 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"); - if (audioCodecId == null) { - if (finalMediaTransmissionTCP) { - content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n"); - }else { - content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n"); - } - - content.append("a=rtpmap:8 PCMA/8000\r\n"); - }else { - if (audioCodecId == 4) { - if (finalMediaTransmissionTCP) { - content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 0\r\n"); - }else { - content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 0\r\n"); - } - content.append("a=rtpmap:0 PCMU/8000\r\n"); - }else { - if (finalMediaTransmissionTCP) { - content.append("m=audio "+ sendRtpItem.getLocalPort()+" TCP/RTP/AVP 8\r\n"); - }else { - content.append("m=audio "+ sendRtpItem.getLocalPort()+" RTP/AVP 8\r\n"); - } - content.append("a=rtpmap:8 PCMA/8000\r\n"); - } - } - content.append("a=sendonly\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="+ finalSsrc + "\r\n"); - content.append("f=v/////a/1/8/1\r\n"); - - ParentPlatform parentPlatform = new ParentPlatform(); - parentPlatform.setServerIP(device.getIp()); - parentPlatform.setServerPort(device.getPort()); - parentPlatform.setServerGBId(device.getDeviceId()); - - responseSdpAck(evt, content.toString(), parentPlatform); - Dialog dialog = evt.getDialog(); - audioBroadcastCatch.setDialog((SIPDialog) dialog); - audioBroadcastCatch.setRequest((SIPRequest) request); - audioBroadcastManager.update(audioBroadcastCatch); - } catch (SipException e) { - throw new RuntimeException(e); - } catch (InvalidArgumentException e) { - throw new RuntimeException(e); - } catch (ParseException e) { - throw new RuntimeException(e); - } catch (SdpParseException e) { - throw new RuntimeException(e); - } - }); - } - String key = DeferredResultHolder.CALLBACK_CMD_BROADCAST + device.getDeviceId(); - WVPResult wvpResult = new WVPResult<>(); - wvpResult.setCode(0); - wvpResult.setMsg("success"); - AudioBroadcastResult audioBroadcastResult = new AudioBroadcastResult(); - audioBroadcastResult.setApp(app); - audioBroadcastResult.setStream(stream); - audioBroadcastResult.setMediaServerItem(new MediaServerItemLite(mediaServerItem)); - audioBroadcastResult.setCodec("G.711"); - wvpResult.setData(audioBroadcastResult); - RequestMessage requestMessage = new RequestMessage(); - requestMessage.setKey(key); - requestMessage.setData(wvpResult); - resultHolder.invokeAllResult(requestMessage); - } else { - logger.warn("来自无效设备/平台的请求"); - responseAck(evt, Response.BAD_REQUEST); - } - } } diff --git a/src/main/java/com/genersoft/iot/vmp/service/IMediaServerService.java b/src/main/java/com/genersoft/iot/vmp/service/IMediaServerService.java index e216c654..1a9c8c78 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/IMediaServerService.java +++ b/src/main/java/com/genersoft/iot/vmp/service/IMediaServerService.java @@ -1,7 +1,6 @@ package com.genersoft.iot.vmp.service; import com.alibaba.fastjson.JSONObject; -import com.genersoft.iot.vmp.gb28181.bean.Device; import com.genersoft.iot.vmp.media.zlm.ZLMServerConfig; import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; import com.genersoft.iot.vmp.service.bean.SSRCInfo; diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java index 0c84b739..86858db7 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java @@ -6,7 +6,6 @@ import com.alibaba.fastjson.JSONObject; import com.genersoft.iot.vmp.common.VideoManagerConstants; import com.genersoft.iot.vmp.conf.SipConfig; import com.genersoft.iot.vmp.conf.UserSetting; -import com.genersoft.iot.vmp.gb28181.bean.SsrcTransaction; import com.genersoft.iot.vmp.gb28181.event.EventPublisher; import com.genersoft.iot.vmp.gb28181.session.SsrcConfig; import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager; @@ -34,11 +33,8 @@ import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.TransactionStatus; import org.springframework.util.StringUtils; -import java.text.ParseException; -import java.text.SimpleDateFormat; import java.time.LocalDateTime; import java.util.*; -import java.util.stream.Collectors; /** * 媒体服务器节点管理 diff --git a/web_src/package-lock.json b/web_src/package-lock.json index 0d4c886c..e0853459 100644 --- a/web_src/package-lock.json +++ b/web_src/package-lock.json @@ -50,7 +50,7 @@ "postcss-url": "^7.2.1", "rimraf": "^2.6.0", "semver": "^5.3.0", - "shelljs": "^0.7.6", + "shelljs": "^0.8.5", "uglifyjs-webpack-plugin": "^1.1.1", "url-loader": "^0.5.8", "vue-loader": "^13.3.0", @@ -11655,9 +11655,9 @@ } }, "node_modules/shelljs": { - "version": "0.7.8", - "resolved": "https://registry.npm.taobao.org/shelljs/download/shelljs-0.7.8.tgz", - "integrity": "sha1-3svPh0sNHl+3LhSxZKloMEjprLM=", + "version": "0.8.5", + "resolved": "https://registry.npmmirror.com/shelljs/-/shelljs-0.8.5.tgz", + "integrity": "sha512-TiwcRcrkhHvbrZbnRcFYMLl30Dfov3HKqzp5tO5b4pt6G/SezKcYhmDg15zXVBswHmctSAQKznqNW2LO5tTDow==", "dev": true, "dependencies": { "glob": "^7.0.0", @@ -11668,8 +11668,7 @@ "shjs": "bin/shjs" }, "engines": { - "iojs": "*", - "node": ">=0.11.0" + "node": ">=4" } }, "node_modules/shellwords": { @@ -24156,9 +24155,9 @@ "dev": true }, "shelljs": { - "version": "0.7.8", - "resolved": "https://registry.npm.taobao.org/shelljs/download/shelljs-0.7.8.tgz", - "integrity": "sha1-3svPh0sNHl+3LhSxZKloMEjprLM=", + "version": "0.8.5", + "resolved": "https://registry.npmmirror.com/shelljs/-/shelljs-0.8.5.tgz", + "integrity": "sha512-TiwcRcrkhHvbrZbnRcFYMLl30Dfov3HKqzp5tO5b4pt6G/SezKcYhmDg15zXVBswHmctSAQKznqNW2LO5tTDow==", "dev": true, "requires": { "glob": "^7.0.0",