From 67a62aae4ed3aecf901cfd86f5c482c8ac2d5a59 Mon Sep 17 00:00:00 2001 From: panlinlin <648540858@qq.com> Date: Tue, 2 Mar 2021 18:23:06 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AF=B9=E7=BA=A7=E8=81=94=E7=82=B9=E6=92=AD?= =?UTF-8?q?=E4=BF=A1=E4=BB=A4=E8=BF=9B=E8=A1=8C=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../iot/vmp/common/VideoManagerConstants.java | 2 + .../iot/vmp/gb28181/bean/SendRtpItem.java | 150 ++++++++++ .../gb28181/transmit/SIPProcessorFactory.java | 1 + .../request/impl/AckRequestProcessor.java | 9 +- .../request/impl/InviteRequestProcessor.java | 256 +++++++----------- .../vmp/media/zlm/ZLMRTPServerFactory.java | 32 +++ .../iot/vmp/storager/IRedisCatchStorage.java | 3 + .../storager/impl/RedisCatchStorageImpl.java | 11 +- 8 files changed, 294 insertions(+), 170 deletions(-) create mode 100644 src/main/java/com/genersoft/iot/vmp/gb28181/bean/SendRtpItem.java diff --git a/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java b/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java index 832bce38..8e9e871c 100644 --- a/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java +++ b/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java @@ -30,6 +30,8 @@ public class VideoManagerConstants { public static final String PLATFORM_REGISTER_INFO_PREFIX = "VMP_platform_register_info_"; + public static final String PLATFORM_SEND_RTP_INFO_PREFIX = "VMP_platform_send_rtp_info_"; + public static final String Pattern_Topic = "VMP_keeplive_platform_"; public static final String EVENT_ONLINE_REGISTER = "1"; diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SendRtpItem.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SendRtpItem.java new file mode 100644 index 00000000..f3ec337d --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SendRtpItem.java @@ -0,0 +1,150 @@ +package com.genersoft.iot.vmp.gb28181.bean; + +public class SendRtpItem { + + /** + * 推流ip + */ + private String ip; + + /** + * 推流端口 + */ + private int port; + + /** + * 推流标识 + */ + private String ssrc; + + /** + * 平台id + */ + private String platformId; + + /** + * 通道id + */ + private String channelId; + + /** + * 推流状态 + * 0 等待设备推流上来 + * 1 等待上级平台回复ack + * 2 推流中 + */ + private int status = 0; + + /** + * 设备推流的app + */ + private String app = "rtp"; + + /** + * 设备推流的streamId + */ + private String streamId; + + /** + * 是否为tcp + */ + private boolean tcp; + + /** + * 是否为tcp主动模式 + */ + private boolean tcpActive; + + /** + * 自己推流使用的端口 + */ + private int localPort; + + public String getIp() { + return ip; + } + + public void setIp(String ip) { + this.ip = ip; + } + + public int getPort() { + return port; + } + + public void setPort(int port) { + this.port = port; + } + + public String getSsrc() { + return ssrc; + } + + public void setSsrc(String ssrc) { + this.ssrc = ssrc; + } + + public String getPlatformId() { + return platformId; + } + + public void setPlatformId(String platformId) { + this.platformId = platformId; + } + + public String getChannelId() { + return channelId; + } + + public void setChannelId(String channelId) { + this.channelId = channelId; + } + + public int getStatus() { + return status; + } + + public void setStatus(int status) { + this.status = status; + } + + public String getApp() { + return app; + } + + public void setApp(String app) { + this.app = app; + } + + public String getStreamId() { + return streamId; + } + + public void setStreamId(String streamId) { + this.streamId = streamId; + } + + public boolean isTcp() { + return tcp; + } + + public void setTcp(boolean tcp) { + this.tcp = tcp; + } + + public int getLocalPort() { + return localPort; + } + + public void setLocalPort(int localPort) { + this.localPort = localPort; + } + + public boolean isTcpActive() { + return tcpActive; + } + + public void setTcpActive(boolean tcpActive) { + this.tcpActive = tcpActive; + } +} diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorFactory.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorFactory.java index 9c8129d6..829bc7f4 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorFactory.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorFactory.java @@ -136,6 +136,7 @@ public class SIPProcessorFactory { processor.setCmderFroPlatform(cmderFroPlatform); processor.setPlayService(playService); processor.setStorager(storager); + processor.setRedisCatchStorage(redisCatchStorage); processor.setZlmrtpServerFactory(zlmrtpServerFactory); return processor; } else if (Request.REGISTER.equals(method)) { diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/AckRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/AckRequestProcessor.java index 13d630c4..b72f0139 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/AckRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/AckRequestProcessor.java @@ -1,9 +1,6 @@ package com.genersoft.iot.vmp.gb28181.transmit.request.impl; -import javax.sip.Dialog; -import javax.sip.InvalidArgumentException; -import javax.sip.RequestEvent; -import javax.sip.SipException; +import javax.sip.*; import javax.sip.message.Request; import com.genersoft.iot.vmp.gb28181.transmit.request.SIPRequestAbstractProcessor; @@ -26,7 +23,11 @@ public class AckRequestProcessor extends SIPRequestAbstractProcessor { public void process(RequestEvent evt) { Request request = evt.getRequest(); Dialog dialog = evt.getDialog(); + DialogState state = dialog.getState(); if (dialog == null) return; + if (request.getMethod().equals(Request.INVITE) && dialog.getState()== DialogState.CONFIRMED) { + // TODO 查询并开始推流 + } try { Request ackRequest = null; CSeq csReq = (CSeq) request.getHeader(CSeq.NAME); 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 7e9e626b..ac19d6f6 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 @@ -4,9 +4,12 @@ import javax.sdp.*; import javax.sip.InvalidArgumentException; import javax.sip.RequestEvent; import javax.sip.SipException; +import javax.sip.SipFactory; +import javax.sip.address.Address; import javax.sip.address.SipURI; import javax.sip.header.ContentTypeHeader; import javax.sip.header.FromHeader; +import javax.sip.header.HeaderFactory; import javax.sip.header.SubjectHeader; import javax.sip.message.Request; import javax.sip.message.Response; @@ -15,6 +18,7 @@ import com.alibaba.fastjson.JSONObject; import com.genersoft.iot.vmp.conf.MediaServerConfig; import com.genersoft.iot.vmp.gb28181.bean.Device; import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel; +import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem; 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; @@ -100,16 +104,18 @@ public class InviteRequestProcessor extends SIPRequestAbstractProcessor { platformId = uri.getUser(); if (platformId == null || channelId == null) { - response400Ack(evt); // 参数不全， 发400，请求错误 + logger.info("无法从FromHeader的Address中获取到平台id，返回404"); + responseAck(evt, Response.BAD_REQUEST); // 参数不全， 发400，请求错误 return; } // 查询平台下是否有该通道 DeviceChannel channel = storager.queryChannelInParentPlatform(platformId, channelId); if (channel == null) { - response404Ack(evt); // 通道不存在，发404，资源不存在 + logger.info("通道不存在，返回404"); + responseAck(evt, Response.NOT_FOUND); // 通道不存在，发404，资源不存在 return; }else { - response100Ack(evt); // 通道存在，发100，trying + responseAck(evt, Response.TRYING); // 通道存在，发100，trying } // 解析sdp消息, 使用jainsip 自带的sdp解析方式 String contentString = new String(request.getRawContent()); @@ -152,107 +158,79 @@ public class InviteRequestProcessor extends SIPRequestAbstractProcessor { } } } -// Vector attributes = mediaDescription.getAttributes(false); -// for (Object attributeObj : attributes) { -// Attribute attribute = (Attribute)attributeObj; -// String name = attribute.getName(); -// switch (name){ -// case "recvonly": -// recvonly = true; -// break; -// case "rtpmap": -// case "connection": -// break; -// case "setup": -// mediaTransmissionTCP = true; -// if ("active".equals(attribute.getValue())) { // tcp主动模式 -// tcpActive = true; -// }else if ("passive".equals(attribute.getValue())){ // tcp被动模式 -// tcpActive = false; -// } -// break; -// -// } -// if ("recvonly".equals(name)) { -// recvonly = true; -// } -// -// String value = attribute.getValue(); -// } break; } } if (port == -1) { + logger.info("不支持的媒体格式，返回415"); // 回复不支持的格式 - response415Ack(evt); // 不支持的格式，发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 = storager.queryVideoDeviceByPlatformIdAndChannelId(platformId, channelId); -// if (device == null) { -// logger.warn("点播平台{}的通道{}时未找到设备信息", platformId, channel); -// response500Ack(evt); -// return; -// } -// -// // 通知下级推流， -// PlayResult playResult = playService.play(device.getDeviceId(), channelId, (responseJSON)->{ -// // 收到推流， 回复200OK -// UUID uuid = UUID.randomUUID(); -// int rtpServer = zlmrtpServerFactory.createRTPServer(uuid.toString()); -// if (rtpServer == -1) { -// logger.error("为获取到可用端口"); -// return; -// }else { -// zlmrtpServerFactory.closeRTPServer(uuid.toString()); -// } -// // TODO 添加对tcp的支持 -// MediaServerConfig mediaInfo = redisCatchStorage.getMediaInfo(); -// StringBuffer content = new StringBuffer(200); -// content.append("v=0\r\n"); -// content.append("o="+"00000"+" 0 0 IN IP4 "+mediaInfo.getWanIp()+"\r\n"); -// content.append("s=Play\r\n"); -// content.append("c=IN IP4 "+mediaInfo.getWanIp()+"\r\n"); -// content.append("t=0 0\r\n"); -// content.append("m=video "+ rtpServer+" 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(); -// } -// -// // 写入redis， 超时时回复 -//// redisCatchStorage.waiteAck() -// },(event -> { -// // 未知错误。直接转发设备点播的错误 -// Response response = null; -// try { -// response = getMessageFactory().createResponse(event.getResponse().getStatusCode(), evt.getRequest()); -// getServerTransaction(evt).sendResponse(response); -// -// } catch (ParseException | SipException | InvalidArgumentException e) { -// e.printStackTrace(); -// } -// })); -// playResult.getResult(); - // 查找合适的端口推流， - // 收到ack后调用推流接口 + Device device = storager.queryVideoDeviceByPlatformIdAndChannelId(platformId, channelId); + if (device == null) { + logger.warn("点播平台{}的通道{}时未找到设备信息", platformId, channel); + responseAck(evt, Response.SERVER_INTERNAL_ERROR); + return; + } + SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(ip, port, platformId, ssrc, channelId, + mediaTransmissionTCP); + if (tcpActive != null) { + sendRtpItem.setTcpActive(tcpActive); + } + if (sendRtpItem == null) { + logger.warn("服务器端口资源不足"); + responseAck(evt, Response.BUSY_HERE); + return; + } + // 写入redis， 超时时回复 + redisCatchStorage.updateSendRTPSever(sendRtpItem); + // 通知下级推流， + PlayResult playResult = playService.play(device.getDeviceId(), channelId, (responseJSON)->{ + // 收到推流， 回复200OK, 等待ack + sendRtpItem.setStatus(1); + redisCatchStorage.updateSendRTPSever(sendRtpItem); + // TODO 添加对tcp的支持 + MediaServerConfig mediaInfo = redisCatchStorage.getMediaInfo(); + StringBuffer content = new StringBuffer(200); + content.append("v=0\r\n"); + content.append("o="+"00000"+" 0 0 IN IP4 "+mediaInfo.getWanIp()+"\r\n"); + content.append("s=Play\r\n"); + content.append("c=IN IP4 "+mediaInfo.getWanIp()+"\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.getResponse().getStatusCode(), evt.getRequest()); + getServerTransaction(evt).sendResponse(response); + + } catch (ParseException | SipException | InvalidArgumentException e) { + e.printStackTrace(); + } + })); + playResult.getResult(); } catch (SipException | InvalidArgumentException | ParseException e) { e.printStackTrace(); @@ -263,101 +241,47 @@ public class InviteRequestProcessor extends SIPRequestAbstractProcessor { } catch (SdpException e) { e.printStackTrace(); } - } /*** - * 回复100 trying + * 回复状态码 + * 100 trying + * 200 OK + * 400 + * 404 * @param evt * @throws SipException * @throws InvalidArgumentException * @throws ParseException */ - private void response100Ack(RequestEvent evt) throws SipException, InvalidArgumentException, ParseException { + private void responseAck(RequestEvent evt, int statusCode) throws SipException, InvalidArgumentException, ParseException { Response response = getMessageFactory().createResponse(Response.TRYING, evt.getRequest()); getServerTransaction(evt).sendResponse(response); } - /*** - * 回复200 OK + /** + * 回复带sdp的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()); - ContentTypeHeader contentTypeHeader = getHeaderFactory().createContentTypeHeader("APPLICATION", "SDP"); + 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); } - /*** - * 回复400 - * @param evt - * @throws SipException - * @throws InvalidArgumentException - * @throws ParseException - */ - private void response400Ack(RequestEvent evt) throws SipException, InvalidArgumentException, ParseException { - Response response = getMessageFactory().createResponse(Response.BAD_REQUEST, evt.getRequest()); - getServerTransaction(evt).sendResponse(response); - } - - /*** - * 回复404 - * @param evt - * @throws SipException - * @throws InvalidArgumentException - * @throws ParseException - */ - private void response404Ack(RequestEvent evt) throws SipException, InvalidArgumentException, ParseException { - Response response = getMessageFactory().createResponse(Response.NOT_FOUND, evt.getRequest()); - getServerTransaction(evt).sendResponse(response); - } - - /*** - * 回复415 不支持的媒体类型 - * @param evt - * @throws SipException - * @throws InvalidArgumentException - * @throws ParseException - */ - private void response415Ack(RequestEvent evt) throws SipException, InvalidArgumentException, ParseException { - Response response = getMessageFactory().createResponse(Response.UNSUPPORTED_MEDIA_TYPE, evt.getRequest()); - getServerTransaction(evt).sendResponse(response); - } - - /*** - * 回复488 - * @param evt - * @throws SipException - * @throws InvalidArgumentException - * @throws ParseException - */ - private void response488Ack(RequestEvent evt) throws SipException, InvalidArgumentException, ParseException { - Response response = getMessageFactory().createResponse(Response.NOT_ACCEPTABLE_HERE, evt.getRequest()); - getServerTransaction(evt).sendResponse(response); - } - - /*** - * 回复500 - * @param evt - * @throws SipException - * @throws InvalidArgumentException - * @throws ParseException - */ - private void response500Ack(RequestEvent evt) throws SipException, InvalidArgumentException, ParseException { - Response response = getMessageFactory().createResponse(Response.SERVER_INTERNAL_ERROR, evt.getRequest()); - getServerTransaction(evt).sendResponse(response); - } - - - - - - - @@ -394,4 +318,12 @@ public class InviteRequestProcessor extends SIPRequestAbstractProcessor { public void setPlayService(IPlayService playService) { this.playService = playService; } + + public IRedisCatchStorage getRedisCatchStorage() { + return redisCatchStorage; + } + + public void setRedisCatchStorage(IRedisCatchStorage redisCatchStorage) { + this.redisCatchStorage = redisCatchStorage; + } } diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java index f69ff0f0..a556ba7f 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java @@ -1,6 +1,8 @@ package com.genersoft.iot.vmp.media.zlm; import com.alibaba.fastjson.JSONObject; +import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem; +import com.genersoft.iot.vmp.gb28181.session.SsrcUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -92,4 +94,34 @@ public class ZLMRTPServerFactory { return currentPort++; } } + + /** + * 创建一个推流 + * @param ip 推流ip + * @param port 推流端口 + * @param ssrc 推流唯一标识 + * @param platformId 平台id + * @param channelId 通道id + * @param tcp 是否为tcp + * @return SendRtpItem + */ + public SendRtpItem createSendRtpItem(String ip, int port, String ssrc, String platformId, String channelId, boolean tcp){ + String playSsrc = SsrcUtil.getPlaySsrc(); + int localPort = createRTPServer(SsrcUtil.getPlaySsrc()); + if (localPort != -1) { + closeRTPServer(playSsrc); + }else { + logger.error("没有可用的端口"); + return null; + } + SendRtpItem sendRtpItem = new SendRtpItem(); + sendRtpItem.setIp(ip); + sendRtpItem.setPort(port); + sendRtpItem.setSsrc(ssrc); + sendRtpItem.setPlatformId(platformId); + sendRtpItem.setChannelId(channelId); + sendRtpItem.setTcp(tcp); + sendRtpItem.setLocalPort(localPort); + return sendRtpItem; + } } diff --git a/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java b/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java index 49e55786..4ca25e27 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java @@ -5,6 +5,7 @@ import com.genersoft.iot.vmp.conf.MediaServerConfig; import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform; import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch; import com.genersoft.iot.vmp.gb28181.bean.PlatformRegister; +import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem; import java.util.Map; @@ -78,4 +79,6 @@ public interface IRedisCatchStorage { String queryPlatformRegisterInfo(String callId); void delPlatformRegisterInfo(String callId); + + void updateSendRTPSever(SendRtpItem sendRtpItem); } diff --git a/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java b/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java index 083c86ca..a9abdd9e 100644 --- a/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java @@ -3,10 +3,7 @@ package com.genersoft.iot.vmp.storager.impl; import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.common.VideoManagerConstants; import com.genersoft.iot.vmp.conf.MediaServerConfig; -import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel; -import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform; -import com.genersoft.iot.vmp.gb28181.bean.ParentPlatformCatch; -import com.genersoft.iot.vmp.gb28181.bean.PlatformRegister; +import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.dao.DeviceChannelMapper; import com.genersoft.iot.vmp.utils.redis.RedisUtil; @@ -215,4 +212,10 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage { public void delPlatformRegisterInfo(String callId) { redis.del(VideoManagerConstants.PLATFORM_REGISTER_INFO_PREFIX + callId); } + + @Override + public void updateSendRTPSever(SendRtpItem sendRtpItem) { + String key = VideoManagerConstants.PLATFORM_SEND_RTP_INFO_PREFIX + sendRtpItem.getPlatformId() + "_" + sendRtpItem.getChannelId(); + redis.set(key, sendRtpItem); + } }