From c3ce2bc5d069ba57309567aee4ae418bc0cf75ed Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期五, 17 三月 2023 13:22:05 +0800 Subject: [PATCH] 优化异常处理以及级联设备状态查询 --- src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java | 489 ++++++++++++++++++++++++++++++++---------------------- 1 files changed, 288 insertions(+), 201 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 7c39446..1a6358b 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,20 +1,21 @@ package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl; -import com.alibaba.fastjson.JSONObject; +import com.alibaba.fastjson2.JSONObject; 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.SsrcConfig; import com.genersoft.iot.vmp.gb28181.session.VideoStreamSessionManager; import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver; -import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander; +import com.genersoft.iot.vmp.gb28181.transmit.SIPSender; 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.ZLMMediaListManager; import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; +import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe; import com.genersoft.iot.vmp.media.zlm.dto.*; import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.IPlayService; @@ -38,12 +39,14 @@ import org.springframework.stereotype.Component; import javax.sdp.*; -import javax.sip.*; +import javax.sip.InvalidArgumentException; +import javax.sip.RequestEvent; +import javax.sip.SipException; import javax.sip.header.CallIdHeader; -import javax.sip.message.Request; import javax.sip.message.Response; import java.text.ParseException; import java.time.Instant; +import java.util.Random; import java.util.Vector; /** @@ -81,7 +84,7 @@ private IPlayService playService; @Autowired - private ISIPCommander commander; + private SIPSender sipSender; @Autowired private ZLMRTPServerFactory zlmrtpServerFactory; @@ -124,15 +127,18 @@ public void process(RequestEvent evt) { // Invite Request娑堟伅瀹炵幇锛屾娑堟伅涓�鑸负绾ц仈娑堟伅锛屼笂绾х粰涓嬬骇鍙戦�佽姹傝棰戞寚浠� try { - Request request = evt.getRequest(); + SIPRequest request = (SIPRequest)evt.getRequest(); String channelId = SipUtils.getChannelIdFromRequest(request); String requesterId = SipUtils.getUserIdFromFromHeader(request); CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME); - ServerTransaction serverTransaction = getServerTransaction(evt); if (requesterId == null || channelId == null) { logger.info("鏃犳硶浠嶧romHeader鐨凙ddress涓幏鍙栧埌骞冲彴id锛岃繑鍥�400"); // 鍙傛暟涓嶅叏锛� 鍙�400锛岃姹傞敊璇� - responseAck(serverTransaction, Response.BAD_REQUEST); + try { + responseAck(request, Response.BAD_REQUEST); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[鍛戒护鍙戦�佸け璐 invite BAD_REQUEST: {}", e.getMessage()); + } return; } @@ -140,7 +146,8 @@ // 鏌ヨ璇锋眰鏄惁鏉ヨ嚜涓婄骇骞冲彴\璁惧 ParentPlatform platform = storager.queryParentPlatByServerGBId(requesterId); if (platform == null) { - inviteFromDeviceHandle(serverTransaction, requesterId); + inviteFromDeviceHandle(request, requesterId); + } else { // 鏌ヨ骞冲彴涓嬫槸鍚︽湁璇ラ�氶亾 DeviceChannel channel = storager.queryChannelInParentPlatform(requesterId, channelId); @@ -152,13 +159,12 @@ StreamProxyItem proxyByAppAndStream =null; // 涓嶆槸閫氶亾鍙兘鏄洿鎾祦 if (channel != null && gbStream == null) { -// if (channel.getStatus() == 0) { -// logger.info("閫氶亾绂荤嚎锛岃繑鍥�400"); -// responseAck(serverTransaction, Response.BAD_REQUEST, "channel [" + channel.getChannelId() + "] offline"); -// return; -// } // 閫氶亾瀛樺湪锛屽彂100锛孴RYING - responseAck(serverTransaction, Response.TRYING); + try { + responseAck(request, Response.TRYING); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[鍛戒护鍙戦�佸け璐 invite TRYING: {}", e.getMessage()); + } } else if (channel == null && gbStream != null) { String mediaServerId = gbStream.getMediaServerId(); @@ -166,13 +172,21 @@ if (mediaServerItem == null) { if ("proxy".equals(gbStream.getStreamType())) { logger.info("[ app={}, stream={} ]鎵句笉鍒皕lm {}锛岃繑鍥�410", gbStream.getApp(), gbStream.getStream(), mediaServerId); - responseAck(serverTransaction, Response.GONE); + 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={} ]鎵句笉鍒皕lm {}锛岃繑鍥�410", gbStream.getApp(), gbStream.getStream(), mediaServerId); - responseAck(serverTransaction, Response.GONE); + try { + responseAck(request, Response.GONE); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[鍛戒护鍙戦�佸け璐 invite GONE: {}", e.getMessage()); + } return; } } @@ -181,25 +195,47 @@ streamPushItem = streamPushService.getPush(gbStream.getApp(), gbStream.getStream()); if (streamPushItem == null) { logger.info("[ app={}, stream={} ]鎵句笉鍒皕lm {}锛岃繑鍥�410", gbStream.getApp(), gbStream.getStream(), mediaServerId); - responseAck(serverTransaction, Response.GONE); + 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={} ]鎵句笉鍒皕lm {}锛岃繑鍥�410", gbStream.getApp(), gbStream.getStream(), mediaServerId); - responseAck(serverTransaction, Response.GONE); + try { + responseAck(request, Response.GONE); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[鍛戒护鍙戦�佸け璐 invite GONE: {}", e.getMessage()); + } return; } } } - responseAck(serverTransaction, Response.CALL_IS_BEING_FORWARDED); // 閫氶亾瀛樺湪锛屽彂181锛屽懠鍙浆鎺ヤ腑 + 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) { - responseAck(serverTransaction, Response.BAD_REQUEST, "catalog channel can not play"); // 鐩綍涓嶆敮鎸佺偣鎾� + 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"); - responseAck(serverTransaction, Response.NOT_FOUND); // 閫氶亾涓嶅瓨鍦紝鍙�404锛岃祫婧愪笉瀛樺湪 + logger.info("閫氶亾涓嶅瓨鍦紝杩斿洖404: {}", channelId); + try { + // 閫氶亾涓嶅瓨鍦紝鍙�404锛岃祫婧愪笉瀛樺湪 + responseAck(request, Response.NOT_FOUND); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[鍛戒护鍙戦�佸け璐 invite 閫氶亾涓嶅瓨鍦�: {}", e.getMessage()); + } return; } // 瑙f瀽sdp娑堟伅, 浣跨敤jainsip 鑷甫鐨剆dp瑙f瀽鏂瑰紡 @@ -270,11 +306,16 @@ if (port == -1) { logger.info("涓嶆敮鎸佺殑濯掍綋鏍煎紡锛岃繑鍥�415"); // 鍥炲涓嶆敮鎸佺殑鏍煎紡 - responseAck(serverTransaction, Response.UNSUPPORTED_MEDIA_TYPE); // 涓嶆敮鎸佺殑鏍煎紡锛屽彂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(); + String addressStr = sdp.getConnection().getAddress(); logger.info("[涓婄骇鐐规挱]鐢ㄦ埛锛歿}锛� 閫氶亾锛歿}, 鍦板潃锛歿}:{}锛� ssrc锛歿}", username, channelId, addressStr, port, ssrc); Device device = null; @@ -283,25 +324,36 @@ device = storager.queryVideoDeviceByPlatformIdAndChannelId(requesterId, channelId); if (device == null) { logger.warn("鐐规挱骞冲彴{}鐨勯�氶亾{}鏃舵湭鎵惧埌璁惧淇℃伅", requesterId, channel); - responseAck(serverTransaction, Response.SERVER_INTERNAL_ERROR); + 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"); - responseAck(serverTransaction, Response.BUSY_HERE); + 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); + device.getDeviceId(), channelId, mediaTransmissionTCP, platform.isRtcp()); if (tcpActive != null) { sendRtpItem.setTcpActive(tcpActive); } if (sendRtpItem == null) { logger.warn("鏈嶅姟鍣ㄧ鍙h祫婧愪笉瓒�"); - responseAck(serverTransaction, Response.BUSY_HERE); + try { + responseAck(request, Response.BUSY_HERE); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[鍛戒护鍙戦�佸け璐 invite 鏈嶅姟鍣ㄧ鍙h祫婧愪笉瓒�: {}", e.getMessage()); + } return; } sendRtpItem.setCallId(callIdHeader.getCallId()); @@ -329,7 +381,12 @@ } else { content.append("t=0 0\r\n"); } - content.append("m=video " + sendRtpItem.getLocalPort() + " RTP/AVP 96\r\n"); + int localPort = sendRtpItem.getLocalPort(); + if (localPort == 0) { + // 闈炰弗鏍兼ā寮忕鍙d笉缁熶竴, 澧炲姞鍏煎鎬э紝淇敼涓轰竴涓笉涓�0鐨勭鍙� + localPort = new Random().nextInt(65535) + 1; + } + content.append("m=video " + localPort + " 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"); @@ -341,36 +398,31 @@ logger.info("Ack 绛夊緟瓒呮椂"); mediaServerService.releaseSsrc(mediaServerItemInUSe.getId(), sendRtpItem.getSsrc()); // 鍥炲bye - cmderFroPlatform.streamByeCmd(platform, callIdHeader.getCallId()); + try { + cmderFroPlatform.streamByeCmd(platform, callIdHeader.getCallId()); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍙戦�丅YE: {}", e.getMessage()); + } }, 60 * 1000); - responseSdpAck(serverTransaction, content.toString(), platform); + responseSdpAck(request, content.toString(), platform); - } catch (SipException e) { - e.printStackTrace(); - } catch (InvalidArgumentException e) { - e.printStackTrace(); - } catch (ParseException e) { - e.printStackTrace(); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鍥炲SdpAck", e); } }; SipSubscribe.Event errorEvent = ((event) -> { // 鏈煡閿欒銆傜洿鎺ヨ浆鍙戣澶囩偣鎾殑閿欒 - Response response = null; try { - response = getMessageFactory().createResponse(event.statusCode, evt.getRequest()); - serverTransaction.sendResponse(response); - System.out.println("鏈煡閿欒銆傜洿鎺ヨ浆鍙戣澶囩偣鎾殑閿欒"); - if (serverTransaction.getDialog() != null) { - serverTransaction.getDialog().delete(); - } - } catch (ParseException | SipException | InvalidArgumentException e) { - e.printStackTrace(); + Response response = getMessageFactory().createResponse(event.statusCode, evt.getRequest()); + sipSender.transmitRequest(request.getLocalAddress().getHostAddress(), response); + } catch (ParseException | SipException e) { + logger.error("鏈鐞嗙殑寮傚父 ", e); } }); sendRtpItem.setApp("rtp"); if ("Playback".equalsIgnoreCase(sessionName)) { sendRtpItem.setPlayType(InviteStreamType.PLAYBACK); - SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, true, true); + SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, device.isSsrcCheck(), true); sendRtpItem.setStreamId(ssrcInfo.getStream()); // 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶� redisCatchStorage.updateSendRTPSever(sendRtpItem); @@ -383,13 +435,9 @@ } redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), channelId, callIdHeader.getCallId(), null); try { - responseAck(serverTransaction, Response.REQUEST_TIMEOUT); - } catch (SipException e) { - e.printStackTrace(); - } catch (InvalidArgumentException e) { - e.printStackTrace(); - } catch (ParseException e) { - e.printStackTrace(); + responseAck(request, Response.REQUEST_TIMEOUT); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 褰曞儚鍥炴斁 鍙戦�丷EQUEST_TIMEOUT: {}", e.getMessage()); } } else { if (result.getMediaServerItem() != null) { @@ -425,10 +473,11 @@ // 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶� redisCatchStorage.updateSendRTPSever(sendRtpItem); + MediaServerItem finalMediaServerItem = mediaServerItem; 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锛� 瓒呮椂鏃跺洖澶� @@ -440,65 +489,68 @@ } } } else if (gbStream != null) { + if(ssrc.equals(ssrcDefault)) + { + SsrcConfig ssrcConfig = mediaServerItem.getSsrcConfig(); + if(ssrcConfig != null) + { + ssrc = ssrcConfig.getPlaySsrc(); + ssrcConfig.releaseSsrc(ssrc); + } + } if("push".equals(gbStream.getStreamType())) { if (streamPushItem != null && streamPushItem.isPushIng()) { // 鎺ㄦ祦鐘舵�� - pushStream(evt, serverTransaction, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive, + pushStream(evt, request, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive, mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId); } else { // 鏈帹娴� 鎷夎捣 - notifyStreamOnline(evt, serverTransaction,gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive, + notifyStreamOnline(evt, request,gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive, mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId); } }else if ("proxy".equals(gbStream.getStreamType())){ - if(null != proxyByAppAndStream &&proxyByAppAndStream.isStatus()){ - pushProxyStream(evt, serverTransaction, gbStream, platform, callIdHeader, mediaServerItem, port, tcpActive, - mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId); - }else{ - //寮�鍚唬鐞嗘媺娴� - boolean start1 = streamProxyService.start(gbStream.getApp(), gbStream.getStream()); - if(start1) { - pushProxyStream(evt, serverTransaction, gbStream, platform, callIdHeader, mediaServerItem, port, tcpActive, + if (null != proxyByAppAndStream) { + if(proxyByAppAndStream.isStatus()){ + pushProxyStream(evt, request, gbStream, platform, callIdHeader, mediaServerItem, port, tcpActive, mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId); }else{ - //澶辫触鍚庨�氱煡 - notifyStreamOnline(evt, serverTransaction,gbStream, null, platform, callIdHeader, mediaServerItem, port, tcpActive, + //寮�鍚唬鐞嗘媺娴� + notifyStreamOnline(evt, request,gbStream, null, platform, callIdHeader, mediaServerItem, port, tcpActive, mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId); } } + } } } - - } catch (SipException | InvalidArgumentException | ParseException e) { - e.printStackTrace(); - logger.warn("sdp瑙f瀽閿欒"); - e.printStackTrace(); } catch (SdpParseException e) { - e.printStackTrace(); + logger.error("sdp瑙f瀽閿欒", e); } catch (SdpException e) { - e.printStackTrace(); + logger.error("鏈鐞嗙殑寮傚父 ", e); } } /** * 瀹夋帓鎺ㄦ祦 */ - private void pushProxyStream(RequestEvent evt, ServerTransaction serverTransaction, GbStream gbStream, ParentPlatform platform, + 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) throws InvalidArgumentException, ParseException, SipException { + 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); + gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp()); if (sendRtpItem == null) { logger.warn("鏈嶅姟鍣ㄧ鍙h祫婧愪笉瓒�"); - responseAck(serverTransaction, Response.BUSY_HERE); + try { + responseAck(request, Response.BUSY_HERE); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[鍛戒护鍙戦�佸け璐 invite 鏈嶅姟鍣ㄧ鍙h祫婧愪笉瓒�: {}", e.getMessage()); + } return; } if (tcpActive != null) { @@ -508,10 +560,9 @@ // 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶� sendRtpItem.setStatus(1); sendRtpItem.setCallId(callIdHeader.getCallId()); - SIPRequest request = (SIPRequest) evt.getRequest(); sendRtpItem.setFromTag(request.getFromTag()); - SIPResponse response = sendStreamAck(mediaServerItem, serverTransaction, sendRtpItem, platform, evt); + SIPResponse response = sendStreamAck(mediaServerItem, request, sendRtpItem, platform, evt); if (response != null) { sendRtpItem.setToTag(response.getToTag()); } @@ -520,22 +571,25 @@ } } - private void pushStream(RequestEvent evt, ServerTransaction serverTransaction, GbStream gbStream, StreamPushItem streamPushItem, ParentPlatform platform, + 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) throws InvalidArgumentException, ParseException, SipException { + 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); + gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp()); if (sendRtpItem == null) { logger.warn("鏈嶅姟鍣ㄧ鍙h祫婧愪笉瓒�"); - responseAck(serverTransaction, Response.BUSY_HERE); + try { + responseAck(request, Response.BUSY_HERE); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[鍛戒护鍙戦�佸け璐 invite 鏈嶅姟鍣ㄧ鍙h祫婧愪笉瓒�: {}", e.getMessage()); + } return; } if (tcpActive != null) { @@ -546,9 +600,8 @@ sendRtpItem.setStatus(1); sendRtpItem.setCallId(callIdHeader.getCallId()); - SIPRequest request = (SIPRequest) evt.getRequest(); sendRtpItem.setFromTag(request.getFromTag()); - SIPResponse response = sendStreamAck(mediaServerItem, serverTransaction, sendRtpItem, platform, evt); + SIPResponse response = sendStreamAck(mediaServerItem, request, sendRtpItem, platform, evt); if (response != null) { sendRtpItem.setToTag(response.getToTag()); } @@ -557,31 +610,62 @@ } else { // 涓嶅湪绾� 鎷夎捣 - notifyStreamOnline(evt, serverTransaction,gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive, + notifyStreamOnline(evt, request,gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive, mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId); } } else { // 鍏朵粬骞冲彴鍐呭 - otherWvpPushStream(evt, serverTransaction, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive, + otherWvpPushStream(evt, request, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive, mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId); } } /** * 閫氱煡娴佷笂绾� */ - private void notifyStreamOnline(RequestEvent evt, ServerTransaction serverTransaction, GbStream gbStream, StreamPushItem streamPushItem, ParentPlatform platform, + 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) throws InvalidArgumentException, ParseException, SipException { + String channelId, String addressStr, String ssrc, String requesterId) { if ("proxy".equals(gbStream.getStreamType())) { // TODO 鎺у埗鍚敤浠ヤ娇璁惧涓婄嚎 logger.info("[ app={}, stream={} ]閫氶亾鏈帹娴侊紝鍚敤娴佸悗寮�濮嬫帹娴�", gbStream.getApp(), gbStream.getStream()); - responseAck(serverTransaction, Response.BAD_REQUEST, "channel [" + gbStream.getGbId() + "] offline"); + // 鐩戝惉娴佷笂绾� + 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("[涓婄骇鐐规挱]鎷夋祦浠g悊宸茬粡灏辩华锛� {}/{}", 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={} ] 绛夊緟鎷夋祦浠g悊娴佽秴鏃�", 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()) { // 骞冲彴璁剧疆涓叧闂簡鎷夎捣绂荤嚎鐨勬帹娴佸垯鐩存帴鍥炲 - responseAck(serverTransaction, Response.TEMPORARILY_UNAVAILABLE, "channel stream not pushing"); + try { + logger.info("[涓婄骇鐐规挱] 澶辫触锛屾帹娴佽澶囨湭鎺ㄦ祦锛宑hannel: {}, app: {}, stream: {}", gbStream.getGbId(), gbStream.getApp(), gbStream.getStream()); + responseAck(request, Response.TEMPORARILY_UNAVAILABLE, "channel stream not pushing"); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[鍛戒护鍙戦�佸け璐 invite 閫氶亾鏈帹娴�: {}", e.getMessage()); + } return; } // 鍙戦�乺edis娑堟伅浠ヤ娇璁惧涓婄嚎 @@ -596,13 +680,13 @@ logger.info("[ app={}, stream={} ] 绛夊緟璁惧寮�濮嬫帹娴佽秴鏃�", gbStream.getApp(), gbStream.getStream()); try { mediaListManager.removedChannelOnlineEventLister(gbStream.getApp(), gbStream.getStream()); - responseAck(serverTransaction, Response.REQUEST_TIMEOUT); // 瓒呮椂 + responseAck(request, Response.REQUEST_TIMEOUT); // 瓒呮椂 } catch (SipException e) { - e.printStackTrace(); + logger.error("鏈鐞嗙殑寮傚父 ", e); } catch (InvalidArgumentException e) { - e.printStackTrace(); + logger.error("鏈鐞嗙殑寮傚父 ", e); } catch (ParseException e) { - e.printStackTrace(); + logger.error("鏈鐞嗙殑寮傚父 ", e); } }, userSetting.getPlatformPlayTimeout()); // 娣诲姞鐩戝惉 @@ -614,18 +698,18 @@ dynamicTask.stop(callIdHeader.getCallId()); if (serverId.equals(userSetting.getServerId())) { SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, finalPort, ssrc, requesterId, - app, stream, channelId, mediaTransmissionTCP); + app, stream, channelId, mediaTransmissionTCP, platform.isRtcp()); if (sendRtpItem == null) { logger.warn("涓婄骇鐐规椂鍒涘缓sendRTPItem澶辫触锛屽彲鑳芥槸鏈嶅姟鍣ㄧ鍙h祫婧愪笉瓒�"); try { - responseAck(serverTransaction, Response.BUSY_HERE); + responseAck(request, Response.BUSY_HERE); } catch (SipException e) { - e.printStackTrace(); + logger.error("鏈鐞嗙殑寮傚父 ", e); } catch (InvalidArgumentException e) { - e.printStackTrace(); + logger.error("鏈鐞嗙殑寮傚父 ", e); } catch (ParseException e) { - e.printStackTrace(); + logger.error("鏈鐞嗙殑寮傚父 ", e); } return; } @@ -637,16 +721,15 @@ sendRtpItem.setStatus(1); sendRtpItem.setCallId(callIdHeader.getCallId()); - SIPRequest request = (SIPRequest) evt.getRequest(); sendRtpItem.setFromTag(request.getFromTag()); - SIPResponse response = sendStreamAck(mediaServerItem, serverTransaction, sendRtpItem, platform, evt); + SIPResponse response = sendStreamAck(mediaServerItem, request, sendRtpItem, platform, evt); if (response != null) { sendRtpItem.setToTag(response.getToTag()); } redisCatchStorage.updateSendRTPSever(sendRtpItem); } else { // 鍏朵粬骞冲彴鍐呭 - otherWvpPushStream(evt, serverTransaction, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive, + otherWvpPushStream(evt, request, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive, mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId); } }); @@ -657,13 +740,9 @@ dynamicTask.stop(callIdHeader.getCallId()); mediaListManager.removedChannelOnlineEventLister(gbStream.getApp(), gbStream.getStream()); try { - responseAck(serverTransaction, Response.TEMPORARILY_UNAVAILABLE, response.getMsg()); - } catch (SipException e) { - throw new RuntimeException(e); - } catch (InvalidArgumentException e) { - throw new RuntimeException(e); - } catch (ParseException e) { - throw new RuntimeException(e); + responseAck(request, Response.TEMPORARILY_UNAVAILABLE, response.getMsg()); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鐐规挱鍥炲: {}", e.getMessage()); } } }); @@ -673,7 +752,7 @@ /** * 鏉ヨ嚜鍏朵粬wvp鐨勬帹娴� */ - private void otherWvpPushStream(RequestEvent evt, ServerTransaction serverTransaction, GbStream gbStream, StreamPushItem streamPushItem, ParentPlatform platform, + 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) { @@ -681,18 +760,18 @@ // 鍙戦�乺edis娑堟伅 redisGbPlayMsgListener.sendMsg(streamPushItem.getServerId(), streamPushItem.getMediaServerId(), streamPushItem.getApp(), streamPushItem.getStream(), addressStr, port, ssrc, requesterId, - channelId, mediaTransmissionTCP, null, responseSendItemMsg -> { + channelId, mediaTransmissionTCP, platform.isRtcp(),null, responseSendItemMsg -> { SendRtpItem sendRtpItem = responseSendItemMsg.getSendRtpItem(); if (sendRtpItem == null || responseSendItemMsg.getMediaServerItem() == null) { logger.warn("鏈嶅姟鍣ㄧ鍙h祫婧愪笉瓒�"); try { - responseAck(serverTransaction, Response.BUSY_HERE); + responseAck(request, Response.BUSY_HERE); } catch (SipException e) { - e.printStackTrace(); + logger.error("鏈鐞嗙殑寮傚父 ", e); } catch (InvalidArgumentException e) { - e.printStackTrace(); + logger.error("鏈鐞嗙殑寮傚父 ", e); } catch (ParseException e) { - e.printStackTrace(); + logger.error("鏈鐞嗙殑寮傚父 ", e); } return; } @@ -705,54 +784,39 @@ sendRtpItem.setStatus(1); sendRtpItem.setCallId(callIdHeader.getCallId()); - SIPRequest request = (SIPRequest) evt.getRequest(); sendRtpItem.setFromTag(request.getFromTag()); - SIPResponse response = sendStreamAck(responseSendItemMsg.getMediaServerItem(), serverTransaction,sendRtpItem, platform, evt); + SIPResponse response = sendStreamAck(responseSendItemMsg.getMediaServerItem(), request,sendRtpItem, platform, evt); if (response != null) { sendRtpItem.setToTag(response.getToTag()); } redisCatchStorage.updateSendRTPSever(sendRtpItem); }, (wvpResult) -> { - try { - // 閿欒 - if (wvpResult.getCode() == RedisGbPlayMsgListener.ERROR_CODE_OFFLINE) { - // 绂荤嚎 - // 鏌ヨ鏄惁鍦ㄦ湰鏈轰笂绾夸簡 - StreamPushItem currentStreamPushItem = streamPushService.getPush(streamPushItem.getApp(), streamPushItem.getStream()); - if (currentStreamPushItem.isPushIng()) { - // 鍦ㄧ嚎鐘舵�� - pushStream(evt, serverTransaction, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive, - mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId); - } else { - // 涓嶅湪绾� 鎷夎捣 - notifyStreamOnline(evt, serverTransaction, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive, - mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId); - } + // 閿欒 + 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); } - } catch (InvalidArgumentException e) { - throw new RuntimeException(e); - } catch (ParseException e) { - throw new RuntimeException(e); - } catch (SipException e) { - throw new RuntimeException(e); } - - try { - responseAck(serverTransaction, Response.BUSY_HERE); - } catch (SipException e) { - e.printStackTrace(); - } catch (InvalidArgumentException e) { - e.printStackTrace(); - } catch (ParseException e) { - e.printStackTrace(); + responseAck(request, Response.BUSY_HERE); + } catch (InvalidArgumentException | ParseException | SipException e) { + logger.error("[鍛戒护鍙戦�佸け璐 鍥芥爣绾ц仈 鐐规挱鍥炲 BUSY_HERE: {}", e.getMessage()); } - return; }); } - public SIPResponse sendStreamAck(MediaServerItem mediaServerItem, ServerTransaction serverTransaction, SendRtpItem sendRtpItem, ParentPlatform platform, RequestEvent evt) { + public SIPResponse sendStreamAck(MediaServerItem mediaServerItem, SIPRequest request, SendRtpItem sendRtpItem, ParentPlatform platform, RequestEvent evt) { StringBuffer content = new StringBuffer(200); content.append("v=0\r\n"); @@ -760,7 +824,13 @@ 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"); + // 闈炰弗鏍兼ā寮忕鍙d笉缁熶竴, 澧炲姞鍏煎鎬э紝淇敼涓轰竴涓笉涓�0鐨勭鍙� + int localPort = sendRtpItem.getLocalPort(); + if(localPort == 0) + { + localPort = new Random().nextInt(65535) + 1; + } + content.append("m=video " + localPort + " RTP/AVP 96\r\n"); content.append("a=sendonly\r\n"); content.append("a=rtpmap:96 PS/90000\r\n"); if (sendRtpItem.isTcp()) { @@ -775,26 +845,29 @@ content.append("f=\r\n"); try { - return responseSdpAck(serverTransaction, content.toString(), platform); + return responseSdpAck(request, content.toString(), platform); } catch (SipException e) { - e.printStackTrace(); + logger.error("鏈鐞嗙殑寮傚父 ", e); } catch (InvalidArgumentException e) { - e.printStackTrace(); + logger.error("鏈鐞嗙殑寮傚父 ", e); } catch (ParseException e) { - e.printStackTrace(); + logger.error("鏈鐞嗙殑寮傚父 ", e); } return null; } - public void inviteFromDeviceHandle(ServerTransaction serverTransaction, String requesterId) throws InvalidArgumentException, ParseException, SipException, SdpException { + public void inviteFromDeviceHandle(SIPRequest request, String requesterId) { // 闈炰笂绾у钩鍙拌姹傦紝鏌ヨ鏄惁璁惧璇锋眰锛堥�氬父涓烘帴鏀惰闊冲箍鎾殑璁惧锛� Device device = redisCatchStorage.getDevice(requesterId); if (device != null) { logger.info("鏀跺埌璁惧" + requesterId + "鐨勮闊冲箍鎾璉nvite璇锋眰"); - responseAck(serverTransaction, Response.TRYING); - - String contentString = new String(serverTransaction.getRequest().getRawContent()); + try { + responseAck(request, Response.TRYING); + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[鍛戒护鍙戦�佸け璐 invite BAD_REQUEST: {}", e.getMessage()); + } + String contentString = new String(request.getRawContent()); // jainSip涓嶆敮鎸亂=瀛楁锛� 绉婚櫎绉婚櫎浠ヨВ鏋愩�� String substring = contentString; String ssrc = "0000000404"; @@ -807,51 +880,65 @@ if (ssrcIndex > 0) { substring = contentString.substring(0, ssrcIndex); } - SessionDescription sdp = SdpFactory.getInstance().createSessionDescription(substring); + 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 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; + 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; } - 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.getConnection().getAddress(); + logger.info("璁惧{}璇锋眰璇煶娴侊紝鍦板潃锛歿}:{}锛宻src锛歿}", username, addressStr, port, ssrc); + } catch (SdpException e) { + logger.error("[SDP瑙f瀽寮傚父]", e); } - if (port == -1) { - logger.info("涓嶆敮鎸佺殑濯掍綋鏍煎紡锛岃繑鍥�415"); - // 鍥炲涓嶆敮鎸佺殑鏍煎紡 - responseAck(serverTransaction, Response.UNSUPPORTED_MEDIA_TYPE); // 涓嶆敮鎸佺殑鏍煎紡锛屽彂415 - return; - } - String username = sdp.getOrigin().getUsername(); - String addressStr = sdp.getOrigin().getAddress(); - logger.info("璁惧{}璇锋眰璇煶娴侊紝鍦板潃锛歿}:{}锛宻src锛歿}", username, addressStr, port, ssrc); + + } else { logger.warn("鏉ヨ嚜鏃犳晥璁惧/骞冲彴鐨勮姹�"); - responseAck(serverTransaction, Response.BAD_REQUEST); + try { + responseAck(request, Response.BAD_REQUEST);; // 涓嶆敮鎸佺殑鏍煎紡锛屽彂415 + } catch (SipException | InvalidArgumentException | ParseException e) { + logger.error("[鍛戒护鍙戦�佸け璐 invite 鏉ヨ嚜鏃犳晥璁惧/骞冲彴鐨勮姹傦紝 {}", e.getMessage()); + } } } } -- Gitblit v1.8.0