From b445128fb5bdc24f6703e7a40e9fd9f40919baf2 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期三, 03 十一月 2021 10:30:42 +0800 Subject: [PATCH] 修复订阅清理请示任务的执行时长 --- src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/InviteRequestProcessor.java | 519 +++++++++++++++++++++++++++++++++++++++++++++++++++------ 1 files changed, 460 insertions(+), 59 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/InviteRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/InviteRequestProcessor.java index d3d94ee..06ae3d8 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/InviteRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/InviteRequestProcessor.java @@ -1,77 +1,478 @@ package com.genersoft.iot.vmp.gb28181.transmit.request.impl; -import javax.sip.RequestEvent; -import javax.sip.ServerTransaction; +import javax.sdp.*; +import javax.sip.*; +import javax.sip.address.Address; +import javax.sip.address.SipURI; +import javax.sip.header.*; +import javax.sip.message.Request; +import javax.sip.message.Response; -import org.springframework.stereotype.Component; +import com.genersoft.iot.vmp.gb28181.bean.*; +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.request.SIPRequestAbstractProcessor; +import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory; +import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; +import com.genersoft.iot.vmp.service.IMediaServerService; +import com.genersoft.iot.vmp.storager.IRedisCatchStorage; +import com.genersoft.iot.vmp.storager.IVideoManagerStorager; +import com.genersoft.iot.vmp.vmanager.gb28181.play.bean.PlayResult; +import com.genersoft.iot.vmp.service.IPlayService; +import gov.nist.javax.sip.address.AddressImpl; +import gov.nist.javax.sip.address.SipUri; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; -import com.genersoft.iot.vmp.gb28181.SipLayer; -import com.genersoft.iot.vmp.gb28181.transmit.request.ISIPRequestProcessor; +import java.text.ParseException; +import java.util.Vector; /** * @Description:澶勭悊INVITE璇锋眰 - * @author: swwheihei - * @date: 2020骞�5鏈�3鏃� 涓嬪崍4:43:52 + * @author: panll + * @date: 2021骞�1鏈�14鏃� */ -@Component -public class InviteRequestProcessor implements ISIPRequestProcessor { +@SuppressWarnings("rawtypes") +public class InviteRequestProcessor extends SIPRequestAbstractProcessor { + + private final static Logger logger = LoggerFactory.getLogger(MessageRequestProcessor.class); + + private SIPCommanderFroPlatform cmderFroPlatform; + + private IVideoManagerStorager storager; + + private IRedisCatchStorage redisCatchStorage; + + private SIPCommander cmder; + + private IPlayService playService; + + private ZLMRTPServerFactory zlmrtpServerFactory; + + private IMediaServerService mediaServerService; + + public ZLMRTPServerFactory getZlmrtpServerFactory() { + return zlmrtpServerFactory; + } + + public void setZlmrtpServerFactory(ZLMRTPServerFactory zlmrtpServerFactory) { + this.zlmrtpServerFactory = zlmrtpServerFactory; + } /** * 澶勭悊invite璇锋眰 * - * @param request + * @param evt * 璇锋眰娑堟伅 */ @Override - public void process(RequestEvent evt, SipLayer layer) { - // TODO Auto-generated method stub -// Request request = requestEvent.getRequest(); -// -// try { -// // 鍙戦��100 Trying -// ServerTransaction serverTransaction = getServerTransaction(requestEvent); -// // 鏌ヨ鐩爣鍦板潃 -// URI reqUri = request.getRequestURI(); -// URI contactURI = currUser.get(reqUri); -// -// System.out.println("processInvite rqStr=" + reqUri + " contact=" + contactURI); -// -// // 鏍规嵁Request uri鏉ヨ矾鐢憋紝鍚庣画鐨勫搷搴旀秷鎭�氳繃VIA鏉ヨ矾鐢� -// Request cliReq = messageFactory.createRequest(request.toString()); -// cliReq.setRequestURI(contactURI); -// -// HeaderFactory headerFactory = SipFactory.getInstance().createHeaderFactory(); -// Via callerVia = (Via) request.getHeader(Via.NAME); -// Via via = (Via) headerFactory.createViaHeader(SIPMain.ip, SIPMain.port, "UDP", -// callerVia.getBranch() + "sipphone"); -// -// // FIXME 闇�瑕佹祴璇曟槸鍚﹁兘澶熼�氳繃璁剧疆VIA澶村煙鏉ヤ慨鏀筕IA澶村煙鍊� -// cliReq.removeHeader(Via.NAME); -// cliReq.addHeader(via); -// -// // 鏇存柊contact鐨勫湴鍧� -// ContactHeader contactHeader = headerFactory.createContactHeader(); -// Address address = SipFactory.getInstance().createAddressFactory() -// .createAddress("sip:sipsoft@" + SIPMain.ip + ":" + SIPMain.port); -// contactHeader.setAddress(address); -// contactHeader.setExpires(3600); -// cliReq.setHeader(contactHeader); -// -// clientTransactionId = sipProvider.getNewClientTransaction(cliReq); -// clientTransactionId.sendRequest(); -// -// System.out.println("processInvite clientTransactionId=" + clientTransactionId.toString()); -// -// System.out.println("send invite to callee: " + cliReq); -// } catch (TransactionUnavailableException e1) { -// e1.printStackTrace(); -// } catch (SipException e) { -// e.printStackTrace(); -// } catch (ParseException e) { -// e.printStackTrace(); -// } catch (Exception e) { -// e.printStackTrace(); -// } + public void process(RequestEvent evt) { + // Invite Request娑堟伅瀹炵幇锛屾娑堟伅涓�鑸负绾ц仈娑堟伅锛屼笂绾х粰涓嬬骇鍙戦�佽姹傝棰戞寚浠� + try { + Request request = evt.getRequest(); + SipURI sipURI = (SipURI) request.getRequestURI(); + String channelId = sipURI.getUser(); + String requesterId = null; + + FromHeader fromHeader = (FromHeader)request.getHeader(FromHeader.NAME); + AddressImpl address = (AddressImpl) fromHeader.getAddress(); + SipUri uri = (SipUri) address.getURI(); + requesterId = uri.getUser(); + + if (requesterId == null || channelId == null) { + logger.info("鏃犳硶浠嶧romHeader鐨凙ddress涓幏鍙栧埌骞冲彴id锛岃繑鍥�400"); + responseAck(evt, Response.BAD_REQUEST); // 鍙傛暟涓嶅叏锛� 鍙�400锛岃姹傞敊璇� + return; + } + + // 鏌ヨ璇锋眰鏂规槸鍚︿笂绾у钩鍙� + ParentPlatform platform = storager.queryParentPlatByServerGBId(requesterId); + if (platform != null) { + // 鏌ヨ骞冲彴涓嬫槸鍚︽湁璇ラ�氶亾 + DeviceChannel channel = storager.queryChannelInParentPlatform(requesterId, channelId); + GbStream gbStream = storager.queryStreamInParentPlatform(requesterId, 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={} ]鎵句笉鍒皕lm {}锛岃繑鍥�410",gbStream.getApp(), gbStream.getStream(), mediaServerId); + responseAck(evt, Response.GONE, "media server not found"); + 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 { + logger.info("閫氶亾涓嶅瓨鍦紝杩斿洖404"); + responseAck(evt, Response.NOT_FOUND); // 閫氶亾涓嶅瓨鍦紝鍙�404锛岃祫婧愪笉瀛樺湪 + return; + } + // 瑙f瀽sdp娑堟伅, 浣跨敤jainsip 鑷甫鐨剆dp瑙f瀽鏂瑰紡 + String contentString = new String(request.getRawContent()); + + // jainSip涓嶆敮鎸亂=瀛楁锛� 绉婚櫎绉婚櫎浠ヨВ鏋愩�� + int ssrcIndex = contentString.indexOf("y="); + //ssrc瑙勫畾闀垮害涓�10瀛楄妭锛屼笉鍙栦綑涓嬮暱搴︿互閬垮厤鍚庣画杩樻湁鈥渇=鈥濆瓧娈� + String ssrc = contentString.substring(ssrcIndex + 2, ssrcIndex + 12); + String substring = contentString.substring(0, contentString.indexOf("y=")); + SessionDescription sdp = SdpFactory.getInstance().createSessionDescription(substring); + + // 鑾峰彇鏀寔鐨勬牸寮� + Vector mediaDescriptions = sdp.getMediaDescriptions(true); + // 鏌ョ湅鏄惁鏀寔PS 璐熻浇96 + //String ip = null; + int port = -1; + //boolean recvonly = false; + 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(); + //String sessionName = sdp.getSessionName().getValue(); + logger.info("[涓婄骇鐐规挱]鐢ㄦ埛锛歿}锛� 鍦板潃锛歿}:{}锛� ssrc锛歿}", username, addressStr, port, ssrc); + Device device = null; + // 閫氳繃 channel 鍜� gbStream 鏄惁涓簄ull 鍊煎垽鏂潵婧愭槸鐩存挱娴佸悎閫傚浗鏍� + 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("鏈嶅姟鍣ㄧ鍙h祫婧愪笉瓒�"); + responseAck(evt, Response.BUSY_HERE); + return; + } + + // 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶� + redisCatchStorage.updateSendRTPSever(sendRtpItem); + // 閫氱煡涓嬬骇鎺ㄦ祦锛� + PlayResult playResult = playService.play(mediaServerItem,device.getDeviceId(), channelId, (mediaServerItemInUSe, responseJSON)->{ + // 鏀跺埌鎺ㄦ祦锛� 鍥炲200OK, 绛夊緟ack + // if (sendRtpItem == null) return; + sendRtpItem.setStatus(1); + redisCatchStorage.updateSendRTPSever(sendRtpItem); + // TODO 娣诲姞瀵箃cp鐨勬敮鎸� + + 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=Play\r\n"); + content.append("c=IN IP4 "+mediaServerItemInUSe.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"); + + try { + responseAck(evt, content.toString()); + } catch (SipException e) { + e.printStackTrace(); + } catch (InvalidArgumentException e) { + e.printStackTrace(); + } catch (ParseException e) { + e.printStackTrace(); + } + } ,((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(); + } + })); + if (logger.isDebugEnabled()) { + logger.debug(playResult.getResult().toString()); + } + + }else if (gbStream != null) { + SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId, + gbStream.getApp(), gbStream.getStream(), channelId, + mediaTransmissionTCP); + + if (tcpActive != null) { + sendRtpItem.setTcpActive(tcpActive); + } + if (sendRtpItem == null) { + logger.warn("鏈嶅姟鍣ㄧ鍙h祫婧愪笉瓒�"); + responseAck(evt, Response.BUSY_HERE); + return; + } + + // 鍐欏叆redis锛� 瓒呮椂鏃跺洖澶� + redisCatchStorage.updateSendRTPSever(sendRtpItem); + + sendRtpItem.setStatus(1); + redisCatchStorage.updateSendRTPSever(sendRtpItem); + // TODO 娣诲姞瀵箃cp鐨勬敮鎸� + 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"); + + try { + responseAck(evt, content.toString()); + } catch (SipException e) { + e.printStackTrace(); + } catch (InvalidArgumentException e) { + e.printStackTrace(); + } catch (ParseException e) { + e.printStackTrace(); + } + } + + } else { + // 闈炰笂绾у钩鍙拌姹傦紝鏌ヨ鏄惁璁惧璇锋眰锛堥�氬父涓烘帴鏀惰闊冲箍鎾殑璁惧锛� + Device device = storager.queryVideoDevice(requesterId); + if (device != null) { + logger.info("鏀跺埌璁惧" + requesterId + "鐨勮闊冲箍鎾璉nvite璇锋眰"); + responseAck(evt, Response.TRYING); + + String contentString = new String(request.getRawContent()); + // jainSip涓嶆敮鎸亂=瀛楁锛� 绉婚櫎绉婚櫎浠ヨВ鏋愩�� + 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); + + // 鑾峰彇鏀寔鐨勬牸寮� + 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"); + // 鍥炲涓嶆敮鎸佺殑鏍煎紡 + responseAck(evt, 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(evt, Response.BAD_REQUEST); + } + } + + } catch (SipException | InvalidArgumentException | ParseException e) { + e.printStackTrace(); + logger.warn("sdp瑙f瀽閿欒"); + e.printStackTrace(); + } catch (SdpParseException e) { + e.printStackTrace(); + } catch (SdpException e) { + e.printStackTrace(); + } } + + /*** + * 鍥炲鐘舵�佺爜 + * 100 trying + * 200 OK + * 400 + * 404 + * @param evt + * @throws SipException + * @throws InvalidArgumentException + * @throws ParseException + */ + private void responseAck(RequestEvent evt, int statusCode) throws SipException, InvalidArgumentException, ParseException { + Response response = getMessageFactory().createResponse(statusCode, evt.getRequest()); + ServerTransaction serverTransaction = getServerTransaction(evt); + serverTransaction.sendResponse(response); + if (statusCode >= 200) { + if (serverTransaction.getDialog() != null) serverTransaction.getDialog().delete(); + } + } + + private void responseAck(RequestEvent evt, int statusCode, String msg) throws SipException, InvalidArgumentException, ParseException { + Response response = getMessageFactory().createResponse(statusCode, evt.getRequest()); + response.setReasonPhrase(msg); + ServerTransaction serverTransaction = getServerTransaction(evt); + serverTransaction.sendResponse(response); + if (statusCode >= 200) { + if (serverTransaction.getDialog() != null) serverTransaction.getDialog().delete(); + } + } + + /** + * 鍥炲甯dp鐨�200 + * @param evt + * @param sdp + * @throws SipException + * @throws InvalidArgumentException + * @throws ParseException + */ + private void responseAck(RequestEvent evt, String sdp) throws SipException, InvalidArgumentException, ParseException { + Response response = getMessageFactory().createResponse(Response.OK, evt.getRequest()); + SipFactory sipFactory = SipFactory.getInstance(); + ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("APPLICATION", "SDP"); + response.setContent(sdp, contentTypeHeader); + + SipURI sipURI = (SipURI)evt.getRequest().getRequestURI(); + + Address concatAddress = sipFactory.createAddressFactory().createAddress( + sipFactory.createAddressFactory().createSipURI(sipURI.getUser(), sipURI.getHost()+":"+sipURI.getPort() + )); + response.addHeader(sipFactory.createHeaderFactory().createContactHeader(concatAddress)); + getServerTransaction(evt).sendResponse(response); + } + + + + + + + public SIPCommanderFroPlatform getCmderFroPlatform() { + return cmderFroPlatform; + } + + public void setCmderFroPlatform(SIPCommanderFroPlatform cmderFroPlatform) { + this.cmderFroPlatform = cmderFroPlatform; + } + + public IVideoManagerStorager getStorager() { + return storager; + } + + public void setStorager(IVideoManagerStorager storager) { + this.storager = storager; + } + + public SIPCommander getCmder() { + return cmder; + } + + public void setCmder(SIPCommander cmder) { + this.cmder = cmder; + } + + public IPlayService getPlayService() { + return playService; + } + + public void setPlayService(IPlayService playService) { + this.playService = playService; + } + + public IRedisCatchStorage getRedisCatchStorage() { + return redisCatchStorage; + } + + public void setRedisCatchStorage(IRedisCatchStorage redisCatchStorage) { + this.redisCatchStorage = redisCatchStorage; + } + + public IMediaServerService getMediaServerService() { + return mediaServerService; + } + + public void setMediaServerService(IMediaServerService mediaServerService) { + this.mediaServerService = mediaServerService; + } } -- Gitblit v1.8.0