临时提交

src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java

@@ -72,6 +72,8 @@ public class VideoManagerConstants {
 
 	public static final String REGISTER_EXPIRE_TASK_KEY_PREFIX = "VMP_device_register_expire_";
 	public static final String PUSH_STREAM_LIST = "VMP_PUSH_STREAM_LIST_";
+	public static final String WAITE_SEND_PUSH_STREAM = "VMP_WAITE_SEND_PUSH_STREAM:";
+	public static final String START_SEND_PUSH_STREAM = "VMP_START_SEND_PUSH_STREAM:";
 
 
 

src/main/java/com/genersoft/iot/vmp/conf/redis/RedisMsgListenConfig.java

@@ -31,9 +31,6 @@ public class RedisMsgListenConfig {
 	@Autowired
 	private RedisStreamMsgListener redisStreamMsgListener;
 
-	@Autowired
-	private RedisGbPlayMsgListener redisGbPlayMsgListener;
-
 	@Autowired
 	private RedisPushStreamStatusMsgListener redisPushStreamStatusMsgListener;
 
@@ -49,6 +46,12 @@ public class RedisMsgListenConfig {
 	@Autowired
 	private RedisPushStreamCloseResponseListener redisPushStreamCloseResponseListener;
 
+	@Autowired
+	private RedisPlatformWaitPushStreamOnlineListener redisPlatformWaitPushStreamOnlineListener;
+
+	@Autowired
+	private RedisPlatformStartSendRtpListener redisPlatformStartSendRtpListener;
+
 
 	/**
 	 * redis消息监听器容器 可以添加多个监听不同话题的redis监听器，只需要把消息监听器和相应的消息订阅处理器绑定，该消息监听器
@@ -65,12 +68,13 @@ public class RedisMsgListenConfig {
 		container.addMessageListener(redisGPSMsgListener, new PatternTopic(VideoManagerConstants.VM_MSG_GPS));
 		container.addMessageListener(redisAlarmMsgListener, new PatternTopic(VideoManagerConstants.VM_MSG_SUBSCRIBE_ALARM_RECEIVE));
 		container.addMessageListener(redisStreamMsgListener, new PatternTopic(VideoManagerConstants.WVP_MSG_STREAM_CHANGE_PREFIX + "PUSH"));
-		container.addMessageListener(redisGbPlayMsgListener, new PatternTopic(RedisGbPlayMsgListener.WVP_PUSH_STREAM_KEY));
 		container.addMessageListener(redisPushStreamStatusMsgListener, new PatternTopic(VideoManagerConstants.VM_MSG_PUSH_STREAM_STATUS_CHANGE));
 		container.addMessageListener(redisPushStreamListMsgListener, new PatternTopic(VideoManagerConstants.VM_MSG_PUSH_STREAM_LIST_CHANGE));
 		container.addMessageListener(redisPushStreamResponseListener, new PatternTopic(VideoManagerConstants.VM_MSG_STREAM_PUSH_RESPONSE));
 		container.addMessageListener(redisCloseStreamMsgListener, new PatternTopic(VideoManagerConstants.VM_MSG_STREAM_PUSH_CLOSE));
 		container.addMessageListener(redisPushStreamCloseResponseListener, new PatternTopic(VideoManagerConstants.VM_MSG_STREAM_PUSH_CLOSE_REQUESTED));
+		container.addMessageListener(redisPlatformWaitPushStreamOnlineListener, new PatternTopic(VideoManagerConstants.WAITE_SEND_PUSH_STREAM));
+		container.addMessageListener(redisPlatformStartSendRtpListener, new PatternTopic(VideoManagerConstants.START_SEND_PUSH_STREAM));
         return container;
     }
 }

src/main/java/com/genersoft/iot/vmp/gb28181/bean/SendRtpItem.java

@@ -22,6 +22,11 @@ public class SendRtpItem {
      */
     private String platformId;
 
+    /**
+     * 平台名称
+     */
+    private String platformName;
+
      /**
      * 对应设备id
      */
@@ -61,6 +66,11 @@ public class SendRtpItem {
      */
     private boolean tcpActive;
 
+    /**
+     * 自己推流使用的IP
+     */
+    private String localIp;
+
     /**
      * 自己推流使用的端口
      */
@@ -306,6 +316,22 @@ public class SendRtpItem {
         this.receiveStream = receiveStream;
     }
 
+    public String getPlatformName() {
+        return platformName;
+    }
+
+    public void setPlatformName(String platformName) {
+        this.platformName = platformName;
+    }
+
+    public String getLocalIp() {
+        return localIp;
+    }
+
+    public void setLocalIp(String localIp) {
+        this.localIp = localIp;
+    }
+
     @Override
     public String toString() {
         return "SendRtpItem{" +
@@ -313,6 +339,7 @@ public class SendRtpItem {
                 ", port=" + port +
                 ", ssrc='" + ssrc + '\'' +
                 ", platformId='" + platformId + '\'' +
+                ", platformName='" + platformName + '\'' +
                 ", deviceId='" + deviceId + '\'' +
                 ", app='" + app + '\'' +
                 ", channelId='" + channelId + '\'' +
@@ -320,6 +347,7 @@ public class SendRtpItem {
                 ", stream='" + stream + '\'' +
                 ", tcp=" + tcp +
                 ", tcpActive=" + tcpActive +
+                ", localIp=" + localIp +
                 ", localPort=" + localPort +
                 ", mediaServerId='" + mediaServerId + '\'' +
                 ", serverId='" + serverId + '\'' +

src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java

@@ -18,6 +18,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
 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.SendRtpPortManager;
 import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
 import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
 import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
@@ -28,7 +29,6 @@ import com.genersoft.iot.vmp.service.bean.ErrorCallback;
 import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
 import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
 import com.genersoft.iot.vmp.service.bean.SSRCInfo;
-import com.genersoft.iot.vmp.service.redisMsg.RedisGbPlayMsgListener;
 import com.genersoft.iot.vmp.service.redisMsg.RedisPushStreamResponseListener;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
@@ -127,12 +127,11 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
     @Autowired
     private SipConfig config;
 
-
     @Autowired
-    private RedisGbPlayMsgListener redisGbPlayMsgListener;
+    private VideoStreamSessionManager streamSession;
 
     @Autowired
-    private VideoStreamSessionManager streamSession;
+    private SendRtpPortManager sendRtpPortManager;
 
 
     @Override
@@ -577,21 +576,40 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
                     }else {
                         ssrc = gb28181Sdp.getSsrc();
                     }
+                    SendRtpItem sendRtpItem = new SendRtpItem();
+                    sendRtpItem.setTcpActive(tcpActive);
+                    sendRtpItem.setTcp(mediaTransmissionTCP);
+                    sendRtpItem.setRtcp(platform.isRtcp());
+                    sendRtpItem.setSsrc(ssrc);
+                    sendRtpItem.setPlatformName(platform.getName());
+                    sendRtpItem.setPlatformId(platform.getServerGBId());
+                    sendRtpItem.setMediaServerId(mediaServerItem.getId());
+                    sendRtpItem.setChannelId(channelId);
+                    sendRtpItem.setIp(addressStr);
+                    sendRtpItem.setPort(port);
+                    sendRtpItem.setUsePs(true);
+                    sendRtpItem.setApp(gbStream.getApp());
+                    sendRtpItem.setStream(gbStream.getStream());
+                    sendRtpItem.setCallId(callIdHeader.getCallId());
+                    sendRtpItem.setFromTag(request.getFromTag());
+                    sendRtpItem.setOnlyAudio(false);
+                    sendRtpItem.setPlayType(InviteStreamType.PUSH);
+                    sendRtpItem.setStatus(0);
 
                     if ("push".equals(gbStream.getStreamType())) {
                         if (streamPushItem != null) {
                             // 从redis查询是否正在接收这个推流
                             OnStreamChangedHookParam pushListItem = redisCatchStorage.getPushListItem(gbStream.getApp(), gbStream.getStream());
+
+                            sendRtpItem.setServerId(pushListItem.getSeverId());
                             if (pushListItem != null) {
                                 StreamPushItem transform = streamPushService.transform(pushListItem);
                                 transform.setSelf(userSetting.getServerId().equals(pushListItem.getSeverId()));
                                 // 推流状态
-                                pushStream(evt, request, gbStream, transform, platform, callIdHeader, mediaServerItem, port, tcpActive,
-                                        mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+                                pushStream(sendRtpItem, mediaServerItem, platform, request);
                             }else {
                                 // 未推流 拉起
-                                notifyStreamOnline(evt, request, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
-                                        mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+                                notifyPushStreamOnline(sendRtpItem, mediaServerItem, platform, request);
                             }
                         }
                     } else if ("proxy".equals(gbStream.getStreamType())) {
@@ -601,8 +619,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
                                         mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
                             } else {
                                 //开启代理拉流
-                                notifyStreamOnline(evt, request, gbStream, null, platform, callIdHeader, mediaServerItem, port, tcpActive,
-                                        mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+                                notifyProxyStreamOnline(sendRtpItem, mediaServerItem, platform, request);
                             }
                         }
 
@@ -659,8 +676,9 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
             sendRtpItem.setStatus(1);
             sendRtpItem.setCallId(callIdHeader.getCallId());
             sendRtpItem.setFromTag(request.getFromTag());
+            sendRtpItem.setLocalIp(mediaServerItem.getSdpIp());
 
-            SIPResponse response = sendStreamAck(mediaServerItem, request, sendRtpItem, platform, evt);
+            SIPResponse response = sendStreamAck(request, sendRtpItem, platform);
             if (response != null) {
                 sendRtpItem.setToTag(response.getToTag());
             }
@@ -670,19 +688,14 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
 
     }
 
-    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) {
+    private void pushStream(SendRtpItem sendRtpItem, MediaServerItem mediaServerItem, ParentPlatform platform, SIPRequest request) {
         // 推流
-        if (streamPushItem.isSelf()) {
-            Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream());
+        if (sendRtpItem.getServerId().equals(userSetting.getServerId())) {
+            Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream());
             if (streamReady != null && streamReady) {
                 // 自平台内容
-                SendRtpItem sendRtpItem = zlmServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
-                        gbStream.getApp(), gbStream.getStream(), channelId, mediaTransmissionTCP, platform.isRtcp());
-
-                if (sendRtpItem == null) {
+                int localPort = sendRtpPortManager.getNextPort(mediaServerItem);
+                if (localPort == 0) {
                     logger.warn("服务器端口资源不足");
                     try {
                         responseAck(request, Response.BUSY_HERE);
@@ -691,16 +704,11 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
                     }
                     return;
                 }
-                if (tcpActive != null) {
-                    sendRtpItem.setTcpActive(tcpActive);
-                }
-                sendRtpItem.setPlayType(InviteStreamType.PUSH);
                 // 写入redis， 超时时回复
                 sendRtpItem.setStatus(1);
-                sendRtpItem.setCallId(callIdHeader.getCallId());
-
                 sendRtpItem.setFromTag(request.getFromTag());
-                SIPResponse response = sendStreamAck(mediaServerItem, request, sendRtpItem, platform, evt);
+                sendRtpItem.setLocalIp(mediaServerItem.getSdpIp());
+                SIPResponse response = sendStreamAck(request, sendRtpItem, platform);
                 if (response != null) {
                     sendRtpItem.setToTag(response.getToTag());
                 }
@@ -708,210 +716,168 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
 
             } else {
                 // 不在线 拉起
-                notifyStreamOnline(evt, request, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
-                        mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+                notifyPushStreamOnline(sendRtpItem, mediaServerItem, platform, request);
             }
-
         } else {
+            SendRtpItem sendRtpItem = new SendRtpItem();
+            sendRtpItem.setRtcp(platform.isRtcp());
+            sendRtpItem.setTcp(mediaTransmissionTCP);
+            sendRtpItem.setTcpActive();
             // 其他平台内容
-            otherWvpPushStream(evt, request, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
-                    mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+            otherWvpPushStream(sendRtpItem, request, 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) {
-        if ("proxy".equals(gbStream.getStreamType())) {
-            // TODO 控制启用以使设备上线
-            logger.info("[ app={}, stream={} ]通道未推流，启用流后开始推流", gbStream.getApp(), gbStream.getStream());
-            // 监听流上线
-            HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed(gbStream.getApp(), gbStream.getStream(), true, "rtsp", mediaServerItem.getId());
-            zlmHttpHookSubscribe.addSubscribe(hookSubscribe, (mediaServerItemInUSe, hookParam) -> {
-                OnStreamChangedHookParam streamChangedHookParam = (OnStreamChangedHookParam)hookParam;
-                logger.info("[上级点播]拉流代理已经就绪， {}/{}", streamChangedHookParam.getApp(), streamChangedHookParam.getStream());
-                dynamicTask.stop(callIdHeader.getCallId());
-                pushProxyStream(evt, request, gbStream, platform, callIdHeader, mediaServerItem, port, tcpActive,
-                        mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
-            });
-            dynamicTask.startDelay(callIdHeader.getCallId(), () -> {
-                logger.info("[ app={}, stream={} ] 等待拉流代理流超时", gbStream.getApp(), gbStream.getStream());
-                zlmHttpHookSubscribe.removeSubscribe(hookSubscribe);
-            }, userSetting.getPlatformPlayTimeout());
-            boolean start = streamProxyService.start(gbStream.getApp(), gbStream.getStream());
-            if (!start) {
-                try {
-                    responseAck(request, Response.BUSY_HERE, "channel [" + gbStream.getGbId() + "] offline");
-                } catch (SipException | InvalidArgumentException | ParseException e) {
-                    logger.error("[命令发送失败] invite 通道未推流: {}", e.getMessage());
-                }
-                zlmHttpHookSubscribe.removeSubscribe(hookSubscribe);
-                dynamicTask.stop(callIdHeader.getCallId());
-            }
-        } else if ("push".equals(gbStream.getStreamType())) {
-            if (!platform.isStartOfflinePush()) {
-                // 平台设置中关闭了拉起离线的推流则直接回复
-                try {
-                    logger.info("[上级点播] 失败，推流设备未推流，channel: {}, 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;
+    private void notifyProxyStreamOnline(SendRtpItem sendRtpItem, MediaServerItem mediaServerItem, ParentPlatform platform, SIPRequest request) {
+        // TODO 控制启用以使设备上线
+        logger.info("[ app={}, stream={} ]通道未推流，启用流后开始推流", gbStream.getApp(), gbStream.getStream());
+        // 监听流上线
+        HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed(gbStream.getApp(), gbStream.getStream(), true, "rtsp", mediaServerItem.getId());
+        zlmHttpHookSubscribe.addSubscribe(hookSubscribe, (mediaServerItemInUSe, hookParam) -> {
+            OnStreamChangedHookParam streamChangedHookParam = (OnStreamChangedHookParam)hookParam;
+            logger.info("[上级点播]拉流代理已经就绪， {}/{}", streamChangedHookParam.getApp(), streamChangedHookParam.getStream());
+            dynamicTask.stop(callIdHeader.getCallId());
+            pushProxyStream(evt, request, gbStream, platform, callIdHeader, mediaServerItem, port, tcpActive,
+                    mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+        });
+        dynamicTask.startDelay(callIdHeader.getCallId(), () -> {
+            logger.info("[ app={}, stream={} ] 等待拉流代理流超时", gbStream.getApp(), gbStream.getStream());
+            zlmHttpHookSubscribe.removeSubscribe(hookSubscribe);
+        }, userSetting.getPlatformPlayTimeout());
+        boolean start = streamProxyService.start(gbStream.getApp(), gbStream.getStream());
+        if (!start) {
+            try {
+                responseAck(request, Response.BUSY_HERE, "channel [" + gbStream.getGbId() + "] offline");
+            } catch (SipException | InvalidArgumentException | ParseException e) {
+                logger.error("[命令发送失败] invite 通道未推流: {}", e.getMessage());
             }
-            // 发送redis消息以使设备上线
-            logger.info("[ app={}, stream={} ]通道未推流，发送redis信息控制设备开始推流", gbStream.getApp(), gbStream.getStream());
-
-            MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(1,
-                    gbStream.getApp(), gbStream.getStream(), gbStream.getGbId(), gbStream.getPlatformId(),
-                    platform.getName(), null, gbStream.getMediaServerId());
-            redisCatchStorage.sendStreamPushRequestedMsg(messageForPushChannel);
-            // 设置超时
-            dynamicTask.startDelay(callIdHeader.getCallId(), () -> {
-                logger.info("[ app={}, stream={} ] 等待设备开始推流超时", gbStream.getApp(), gbStream.getStream());
-                try {
-                    redisPushStreamResponseListener.removeEvent(gbStream.getApp(), gbStream.getStream());
-                    mediaListManager.removedChannelOnlineEventLister(gbStream.getApp(), gbStream.getStream());
-                    responseAck(request, Response.REQUEST_TIMEOUT); // 超时
-                } catch (SipException | InvalidArgumentException | ParseException e) {
-                    logger.error("未处理的异常 ", e);
-                }
-            }, userSetting.getPlatformPlayTimeout());
-            // 添加监听
-            int finalPort = port;
-            Boolean finalTcpActive = tcpActive;
+            zlmHttpHookSubscribe.removeSubscribe(hookSubscribe);
+            dynamicTask.stop(callIdHeader.getCallId());
+        }
+    }
 
-            // 添加在本机上线的通知
-            mediaListManager.addChannelOnlineEventLister(gbStream.getApp(), gbStream.getStream(), (app, stream, serverId) -> {
-                dynamicTask.stop(callIdHeader.getCallId());
+    /**
+     * 通知流上线
+     */
+    private void notifyPushStreamOnline(SendRtpItem sendRtpItem, MediaServerItem mediaServerItem, ParentPlatform platform, SIPRequest request) {
+        if (!platform.isStartOfflinePush()) {
+            // 平台设置中关闭了拉起离线的推流则直接回复
+            try {
+                logger.info("[上级点播] 失败，推流设备未推流，channel: {}, 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;
+        }
+        // 发送redis消息以使设备上线
+        logger.info("[ app={}, stream={} ]通道未推流，发送redis信息控制设备开始推流", gbStream.getApp(), gbStream.getStream());
+
+        MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(1,
+                gbStream.getApp(), gbStream.getStream(), gbStream.getGbId(), gbStream.getPlatformId(),
+                platform.getName(), userSetting.getServerId(), gbStream.getMediaServerId());
+        redisCatchStorage.sendStreamPushRequestedMsg(messageForPushChannel);
+        // 设置超时
+        dynamicTask.startDelay(callIdHeader.getCallId(), () -> {
+            logger.info("[ app={}, stream={} ] 等待设备开始推流超时", gbStream.getApp(), gbStream.getStream());
+            try {
                 redisPushStreamResponseListener.removeEvent(gbStream.getApp(), gbStream.getStream());
-                if (serverId.equals(userSetting.getServerId())) {
-                    SendRtpItem sendRtpItem = zlmServerFactory.createSendRtpItem(mediaServerItem, addressStr, finalPort, ssrc, requesterId,
-                            app, stream, channelId, mediaTransmissionTCP, platform.isRtcp());
-
-                    if (sendRtpItem == null) {
-                        logger.warn("上级点时创建sendRTPItem失败，可能是服务器端口资源不足");
-                        try {
-                            responseAck(request, Response.BUSY_HERE);
-                        } catch (SipException e) {
-                            logger.error("未处理的异常 ", e);
-                        } catch (InvalidArgumentException e) {
-                            logger.error("未处理的异常 ", e);
-                        } catch (ParseException e) {
-                            logger.error("未处理的异常 ", e);
-                        }
-                        return;
-                    }
-                    if (finalTcpActive != null) {
-                        sendRtpItem.setTcpActive(finalTcpActive);
+                mediaListManager.removedChannelOnlineEventLister(gbStream.getApp(), gbStream.getStream());
+                responseAck(request, Response.REQUEST_TIMEOUT); // 超时
+            } catch (SipException | InvalidArgumentException | ParseException e) {
+                logger.error("未处理的异常 ", e);
+            }
+        }, userSetting.getPlatformPlayTimeout());
+        // 写入redis待发流信息，供其他wvp读取并生成发流信息
+        SendRtpItem sendRtpItemTemp = new SendRtpItem();
+        sendRtpItemTemp.setIp(addressStr);
+        sendRtpItemTemp.setPort(port);
+        sendRtpItemTemp.setSsrc(ssrc);
+        sendRtpItemTemp.setPlatformId(requesterId);
+        sendRtpItemTemp.setPlatformName(platform.getName());
+        sendRtpItemTemp.setTcp(mediaTransmissionTCP);
+        sendRtpItemTemp.setRtcp(platform.isRtcp());
+        sendRtpItemTemp.setTcpActive(tcpActive);
+        sendRtpItemTemp.setPlayType(InviteStreamType.PUSH);
+        redisCatchStorage.addWaiteSendRtpItem(sendRtpItemTemp, userSetting.getPlatformPlayTimeout());
+        // 添加上线的通知
+        mediaListManager.addChannelOnlineEventLister(gbStream.getApp(), gbStream.getStream(), (sendRtpItemFromRedis) -> {
+            dynamicTask.stop(callIdHeader.getCallId());
+            redisPushStreamResponseListener.removeEvent(gbStream.getApp(), gbStream.getStream());
+            if (sendRtpItemFromRedis.getServerId().equals(userSetting.getServerId())) {
+
+                int localPort = sendRtpPortManager.getNextPort(mediaServerItem);
+                if (localPort == 0) {
+                    logger.warn("上级点时创建sendRTPItem失败，可能是服务器端口资源不足");
+                    try {
+                        responseAck(request, Response.BUSY_HERE);
+                    } catch (SipException e) {
+                        logger.error("未处理的异常 ", e);
+                    } catch (InvalidArgumentException e) {
+                        logger.error("未处理的异常 ", e);
+                    } catch (ParseException e) {
+                        logger.error("未处理的异常 ", e);
                     }
-                    sendRtpItem.setPlayType(InviteStreamType.PUSH);
-                    // 写入redis， 超时时回复
-                    sendRtpItem.setStatus(1);
-                    sendRtpItem.setCallId(callIdHeader.getCallId());
+                    return;
+                }
+                sendRtpItemTemp.setLocalPort(localPort);
+                sendRtpItemTemp.setLocalIp(ObjectUtils.isEmpty(platform.getSendStreamIp()): );
+                // 写入redis， 超时时回复
+                sendRtpItemTemp.setStatus(1);
+                sendRtpItemTemp.setCallId(callIdHeader.getCallId());
 
-                    sendRtpItem.setFromTag(request.getFromTag());
-                    SIPResponse response = sendStreamAck(mediaServerItem, request, sendRtpItem, platform, evt);
-                    if (response != null) {
-                        sendRtpItem.setToTag(response.getToTag());
-                    }
-                    redisCatchStorage.updateSendRTPSever(sendRtpItem);
-                } else {
-                    // 其他平台内容
-                    otherWvpPushStream(evt, request, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
-                            mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+                sendRtpItemTemp.setFromTag(request.getFromTag());
+                SIPResponse response = sendStreamAck(request, sendRtpItemTemp, platform);
+                if (response != null) {
+                    sendRtpItemTemp.setToTag(response.getToTag());
                 }
-            });
+                redisCatchStorage.updateSendRTPSever(sendRtpItemTemp);
+            } else {
+                // 其他平台内容
+                otherWvpPushStream(sendRtpItemFromRedis, request, platform);
+            }
+        });
 
-            // 添加回复的拒绝或者错误的通知
-            redisPushStreamResponseListener.addEvent(gbStream.getApp(), gbStream.getStream(), response -> {
-                if (response.getCode() != 0) {
-                    dynamicTask.stop(callIdHeader.getCallId());
-                    mediaListManager.removedChannelOnlineEventLister(gbStream.getApp(), gbStream.getStream());
-                    try {
-                        responseAck(request, Response.TEMPORARILY_UNAVAILABLE, response.getMsg());
-                    } catch (SipException | InvalidArgumentException | ParseException e) {
-                        logger.error("[命令发送失败] 国标级联 点播回复: {}", e.getMessage());
-                    }
+        // 添加回复的拒绝或者错误的通知
+        redisPushStreamResponseListener.addEvent(gbStream.getApp(), gbStream.getStream(), response -> {
+            if (response.getCode() != 0) {
+                dynamicTask.stop(callIdHeader.getCallId());
+                mediaListManager.removedChannelOnlineEventLister(gbStream.getApp(), gbStream.getStream());
+                try {
+                    responseAck(request, Response.TEMPORARILY_UNAVAILABLE, response.getMsg());
+                } catch (SipException | InvalidArgumentException | ParseException e) {
+                    logger.error("[命令发送失败] 国标级联 点播回复: {}", e.getMessage());
                 }
-            });
-        }
+            }
+        });
     }
 
+
+
     /**
      * 来自其他wvp的推流
      */
-    private void otherWvpPushStream(RequestEvent evt, SIPRequest request, GbStream gbStream, StreamPushItem streamPushItem, ParentPlatform platform,
-                                    CallIdHeader callIdHeader, MediaServerItem mediaServerItem,
-                                    int port, Boolean tcpActive, boolean mediaTransmissionTCP,
-                                    String channelId, String addressStr, String ssrc, String requesterId) {
+    private void otherWvpPushStream(SendRtpItem sendRtpItem, SIPRequest request, ParentPlatform platform) {
         logger.info("[级联点播]直播流来自其他平台，发送redis消息");
         // 发送redis消息
-        redisGbPlayMsgListener.sendMsg(streamPushItem.getServerId(), streamPushItem.getMediaServerId(),
-                streamPushItem.getApp(), streamPushItem.getStream(), addressStr, port, ssrc, requesterId,
-                channelId, mediaTransmissionTCP, platform.isRtcp(),platform.getName(), responseSendItemMsg -> {
-                    SendRtpItem sendRtpItem = responseSendItemMsg.getSendRtpItem();
-                    if (sendRtpItem == null || responseSendItemMsg.getMediaServerItem() == null) {
-                        logger.warn("服务器端口资源不足");
-                        try {
-                            responseAck(request, Response.BUSY_HERE);
-                        } catch (SipException e) {
-                            logger.error("未处理的异常 ", e);
-                        } catch (InvalidArgumentException e) {
-                            logger.error("未处理的异常 ", e);
-                        } catch (ParseException e) {
-                            logger.error("未处理的异常 ", e);
-                        }
-                        return;
-                    }
-                    // 收到sendItem
-                    if (tcpActive != null) {
-                        sendRtpItem.setTcpActive(tcpActive);
-                    }
-                    sendRtpItem.setPlayType(InviteStreamType.PUSH);
-                    // 写入redis， 超时时回复
-                    sendRtpItem.setStatus(1);
-                    sendRtpItem.setCallId(callIdHeader.getCallId());
-
-                    sendRtpItem.setFromTag(request.getFromTag());
-                    SIPResponse response = sendStreamAck(responseSendItemMsg.getMediaServerItem(), request, sendRtpItem, platform, evt);
-                    if (response != null) {
-                        sendRtpItem.setToTag(response.getToTag());
-                    }
-                    redisCatchStorage.updateSendRTPSever(sendRtpItem);
-                }, (wvpResult) -> {
-
-                    // 错误
-                    if (wvpResult.getCode() == RedisGbPlayMsgListener.ERROR_CODE_OFFLINE) {
-                        // 离线
-                        // 查询是否在本机上线了
-                        StreamPushItem currentStreamPushItem = streamPushService.getPush(streamPushItem.getApp(), streamPushItem.getStream());
-                        if (currentStreamPushItem.isPushIng()) {
-                            // 在线状态
-                            pushStream(evt, request, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
-                                    mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
-
-                        } else {
-                            // 不在线 拉起
-                            notifyStreamOnline(evt, request, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
-                                    mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
-                        }
-                    }
-                    try {
-                        responseAck(request, Response.BUSY_HERE);
-                    } catch (InvalidArgumentException | ParseException | SipException e) {
-                        logger.error("[命令发送失败] 国标级联 点播回复 BUSY_HERE: {}", e.getMessage());
-                    }
-                });
+        redisCatchStorage.sendStartSendRtp(sendRtpItem);
+        // 写入redis， 超时时回复
+        sendRtpItem.setStatus(1);
+        sendRtpItem.setCallId(request.getCallIdHeader().getCallId());
+        sendRtpItem.setFromTag(request.getFromTag());
+        SIPResponse response = sendStreamAck(request, sendRtpItem, platform);
+        if (response != null) {
+            sendRtpItem.setToTag(response.getToTag());
+        }
+        redisCatchStorage.updateSendRTPSever(sendRtpItem);
     }
 
-    public SIPResponse sendStreamAck(MediaServerItem mediaServerItem, SIPRequest request, SendRtpItem sendRtpItem, ParentPlatform platform, RequestEvent evt) {
+    public SIPResponse sendStreamAck(SIPRequest request, SendRtpItem sendRtpItem, ParentPlatform platform) {
 
-        String sdpIp = mediaServerItem.getSdpIp();
+        String sdpIp = sendRtpItem.getLocalIp();
         if (!ObjectUtils.isEmpty(platform.getSendStreamIp())) {
             sdpIp = platform.getSendStreamIp();
         }

src/main/java/com/genersoft/iot/vmp/media/zlm/dto/ChannelOnlineEvent.java

@@ -1,5 +1,7 @@
 package com.genersoft.iot.vmp.media.zlm.dto;
 
+import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
+
 import java.text.ParseException;
 
 /**
@@ -7,5 +9,5 @@ import java.text.ParseException;
  */
 public interface ChannelOnlineEvent {
 
-    void run(String app, String stream, String serverId) throws ParseException;
+    void run(SendRtpItem sendRtpItem) throws ParseException;
 }

src/main/java/com/genersoft/iot/vmp/service/bean/MessageForPushChannel.java

@@ -61,6 +61,7 @@ public class MessageForPushChannel {
         messageForPushChannel.setGbId(gbId);
         messageForPushChannel.setApp(app);
         messageForPushChannel.setStream(stream);
+        messageForPushChannel.setServerId(serverId);
         messageForPushChannel.setMediaServerId(mediaServerId);
         messageForPushChannel.setPlatFormId(platFormId);
         messageForPushChannel.setPlatFormName(platFormName);

src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java

@@ -1,474 +0,0 @@
-package com.genersoft.iot.vmp.service.redisMsg;
-
-import com.alibaba.fastjson2.JSON;
-import com.alibaba.fastjson2.JSONObject;
-import com.genersoft.iot.vmp.common.VideoManagerConstants;
-import com.genersoft.iot.vmp.conf.DynamicTask;
-import com.genersoft.iot.vmp.conf.UserSetting;
-import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
-import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
-import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
-import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
-import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
-import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
-import com.genersoft.iot.vmp.service.IMediaServerService;
-import com.genersoft.iot.vmp.service.bean.*;
-import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
-import com.genersoft.iot.vmp.utils.redis.RedisUtil;
-import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.data.redis.connection.Message;
-import org.springframework.data.redis.connection.MessageListener;
-import org.springframework.data.redis.core.RedisTemplate;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
-import org.springframework.stereotype.Component;
-
-import java.text.ParseException;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.UUID;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentLinkedQueue;
-
-
-/**
- * 监听下级发送推送信息，并发送国标推流消息上级
- * @author lin
- */
-@Component
-public class RedisGbPlayMsgListener implements MessageListener {
-
-    private final static Logger logger = LoggerFactory.getLogger(RedisGbPlayMsgListener.class);
-
-    public static final String WVP_PUSH_STREAM_KEY = "WVP_PUSH_STREAM";
-
-    /**
-     * 流媒体不存在的错误玛
-     */
-    public static final  int ERROR_CODE_MEDIA_SERVER_NOT_FOUND = -1;
-
-    /**
-     * 离线的错误玛
-     */
-    public static final  int ERROR_CODE_OFFLINE = -2;
-
-    /**
-     * 超时的错误玛
-     */
-    public static final  int ERROR_CODE_TIMEOUT = -3;
-
-    private Map<String, PlayMsgCallback> callbacks = new ConcurrentHashMap<>();
-    private Map<String, PlayMsgCallbackForStartSendRtpStream> callbacksForStartSendRtpStream = new ConcurrentHashMap<>();
-    private Map<String, PlayMsgErrorCallback> callbacksForError = new ConcurrentHashMap<>();
-
-    @Autowired
-    private UserSetting userSetting;
-
-
-    @Autowired
-    private RedisTemplate<Object, Object> redisTemplate;
-
-    @Autowired
-    private ZLMServerFactory zlmServerFactory;
-
-    @Autowired
-    private IMediaServerService mediaServerService;
-
-    @Autowired
-    private IRedisCatchStorage redisCatchStorage;
-
-
-    @Autowired
-    private DynamicTask dynamicTask;
-
-
-    @Autowired
-    private ZlmHttpHookSubscribe subscribe;
-
-    private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
-
-    @Qualifier("taskExecutor")
-    @Autowired
-    private ThreadPoolTaskExecutor taskExecutor;
-
-
-    public interface PlayMsgCallback{
-        void handler(ResponseSendItemMsg responseSendItemMsg) throws ParseException;
-    }
-
-    public interface PlayMsgCallbackForStartSendRtpStream{
-        void handler(JSONObject jsonObject);
-    }
-
-    public interface PlayMsgErrorCallback{
-        void handler(WVPResult wvpResult);
-    }
-
-    @Override
-    public void onMessage(Message message, byte[] bytes) {
-        boolean isEmpty = taskQueue.isEmpty();
-        taskQueue.offer(message);
-        if (isEmpty) {
-            taskExecutor.execute(() -> {
-                while (!taskQueue.isEmpty()) {
-                    Message msg = taskQueue.poll();
-                    try {
-                        WvpRedisMsg wvpRedisMsg = JSON.parseObject(msg.getBody(), WvpRedisMsg.class);
-                        logger.info("[收到REDIS通知] 消息： {}", JSON.toJSONString(wvpRedisMsg));
-                        if (!userSetting.getServerId().equals(wvpRedisMsg.getToId())) {
-                            continue;
-                        }
-                        if (WvpRedisMsg.isRequest(wvpRedisMsg)) {
-                            logger.info("[收到REDIS通知] 请求： {}", new String(msg.getBody()));
-
-                            switch (wvpRedisMsg.getCmd()){
-                                case WvpRedisMsgCmd.GET_SEND_ITEM:
-                                    RequestSendItemMsg content = JSON.parseObject(wvpRedisMsg.getContent(), RequestSendItemMsg.class);
-                                    requestSendItemMsgHand(content, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
-                                    break;
-                                case WvpRedisMsgCmd.REQUEST_PUSH_STREAM:
-                                    RequestPushStreamMsg param = JSON.to(RequestPushStreamMsg.class, wvpRedisMsg.getContent());
-                                    requestPushStreamMsgHand(param, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
-                                    break;
-                                case WvpRedisMsgCmd.REQUEST_STOP_PUSH_STREAM:
-                                    RequestStopPushStreamMsg streamMsg = JSON.to(RequestStopPushStreamMsg.class, wvpRedisMsg.getContent());
-                                    requestStopPushStreamMsgHand(streamMsg, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
-                                    break;
-                                default:
-                                    break;
-                            }
-
-                        }else {
-                            logger.info("[收到REDIS通知] 回复： {}", new String(msg.getBody()));
-                            switch (wvpRedisMsg.getCmd()){
-                                case WvpRedisMsgCmd.GET_SEND_ITEM:
-
-                                   WVPResult content  = JSON.to(WVPResult.class, wvpRedisMsg.getContent());
-
-                                    String key = wvpRedisMsg.getSerial();
-                                    switch (content.getCode()) {
-                                        case 0:
-                                           ResponseSendItemMsg responseSendItemMsg =JSON.to(ResponseSendItemMsg.class, content.getData());
-                                            PlayMsgCallback playMsgCallback = callbacks.get(key);
-                                            if (playMsgCallback != null) {
-                                                callbacksForError.remove(key);
-                                                try {
-                                                    playMsgCallback.handler(responseSendItemMsg);
-                                                } catch (ParseException e) {
-                                                    logger.error("[REDIS消息处理异常] ", e);
-                                                }
-                                            }
-                                            break;
-                                        case ERROR_CODE_MEDIA_SERVER_NOT_FOUND:
-                                        case ERROR_CODE_OFFLINE:
-                                        case ERROR_CODE_TIMEOUT:
-                                            PlayMsgErrorCallback errorCallback = callbacksForError.get(key);
-                                            if (errorCallback != null) {
-                                                callbacks.remove(key);
-                                                errorCallback.handler(content);
-                                            }
-                                            break;
-                                        default:
-                                            break;
-                                    }
-                                    break;
-                                case WvpRedisMsgCmd.REQUEST_PUSH_STREAM:
-                                    WVPResult wvpResult  = JSON.to(WVPResult.class, wvpRedisMsg.getContent());
-                                    String serial = wvpRedisMsg.getSerial();
-                                    switch (wvpResult.getCode()) {
-                                        case 0:
-                                            JSONObject jsonObject = (JSONObject)wvpResult.getData();
-                                            PlayMsgCallbackForStartSendRtpStream playMsgCallback = callbacksForStartSendRtpStream.get(serial);
-                                            if (playMsgCallback != null) {
-                                                callbacksForError.remove(serial);
-                                                playMsgCallback.handler(jsonObject);
-                                            }
-                                            break;
-                                        case ERROR_CODE_MEDIA_SERVER_NOT_FOUND:
-                                        case ERROR_CODE_OFFLINE:
-                                        case ERROR_CODE_TIMEOUT:
-                                            PlayMsgErrorCallback errorCallback = callbacksForError.get(serial);
-                                            if (errorCallback != null) {
-                                                callbacks.remove(serial);
-                                                errorCallback.handler(wvpResult);
-                                            }
-                                            break;
-                                        default:
-                                            break;
-                                    }
-                                    break;
-                                default:
-                                    break;
-                            }
-
-                        }
-                    }catch (Exception e) {
-                        logger.warn("[RedisGbPlayMsg] 发现未处理的异常, \r\n{}", JSON.toJSONString(message));
-                        logger.error("[RedisGbPlayMsg] 异常内容： ", e);
-                    }
-                }
-            });
-        }
-    }
-
-    /**
-     * 处理收到的请求推流的请求
-     */
-    private void requestPushStreamMsgHand(RequestPushStreamMsg requestPushStreamMsg, String fromId, String serial) {
-        MediaServerItem mediaInfo = mediaServerService.getOne(requestPushStreamMsg.getMediaServerId());
-        if (mediaInfo == null) {
-            // TODO 回复错误
-            return;
-        }
-        String is_Udp = requestPushStreamMsg.isTcp() ? "0" : "1";
-        Map<String, Object> param = new HashMap<>();
-        param.put("vhost","__defaultVhost__");
-        param.put("app",requestPushStreamMsg.getApp());
-        param.put("stream",requestPushStreamMsg.getStream());
-        param.put("ssrc", requestPushStreamMsg.getSsrc());
-        param.put("dst_url",requestPushStreamMsg.getIp());
-        param.put("dst_port", requestPushStreamMsg.getPort());
-        param.put("is_udp", is_Udp);
-        param.put("src_port", requestPushStreamMsg.getSrcPort());
-        param.put("pt", requestPushStreamMsg.getPt());
-        param.put("use_ps", requestPushStreamMsg.isPs() ? "1" : "0");
-        param.put("only_audio", requestPushStreamMsg.isOnlyAudio() ? "1" : "0");
-        JSONObject jsonObject = zlmServerFactory.startSendRtpStream(mediaInfo, param);
-        // 回复消息
-        responsePushStream(jsonObject, fromId, serial);
-    }
-
-    private void responsePushStream(JSONObject content, String toId, String serial) {
-
-        WVPResult<JSONObject> result = new WVPResult<>();
-        result.setCode(0);
-        result.setData(content);
-
-        WvpRedisMsg response = WvpRedisMsg.getResponseInstance(userSetting.getServerId(), toId,
-                WvpRedisMsgCmd.REQUEST_PUSH_STREAM, serial, JSON.toJSONString(result));
-        JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
-        redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
-    }
-
-    /**
-     * 处理收到的请求sendItem的请求
-     */
-    private void requestSendItemMsgHand(RequestSendItemMsg content, String toId, String serial) {
-        MediaServerItem mediaServerItem = mediaServerService.getOne(content.getMediaServerId());
-        if (mediaServerItem == null) {
-            logger.info("[回复推流信息] 流媒体{}不存在 ", content.getMediaServerId());
-
-            WVPResult<SendRtpItem> result = new WVPResult<>();
-            result.setCode(ERROR_CODE_MEDIA_SERVER_NOT_FOUND);
-            result.setMsg("流媒体不存在");
-
-            WvpRedisMsg response = WvpRedisMsg.getResponseInstance(userSetting.getServerId(), toId,
-                    WvpRedisMsgCmd.GET_SEND_ITEM, serial, JSON.toJSONString(result));
-
-            JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
-            redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
-            return;
-        }
-        // 确定流是否在线
-        Boolean streamReady = zlmServerFactory.isStreamReady(mediaServerItem, content.getApp(), content.getStream());
-        if (streamReady != null && streamReady) {
-            logger.info("[回复推流信息]  {}/{}", content.getApp(), content.getStream());
-            responseSendItem(mediaServerItem, content, toId, serial);
-        }else {
-            // 流已经离线
-            // 发送redis消息以使设备上线
-            logger.info("[ app={}, stream={} ]通道离线，发送redis信息控制设备开始推流",content.getApp(), content.getStream());
-
-            String taskKey = UUID.randomUUID().toString();
-            // 设置超时
-            dynamicTask.startDelay(taskKey, ()->{
-                logger.info("[ app={}, stream={} ] 等待设备开始推流超时", content.getApp(), content.getStream());
-                WVPResult<SendRtpItem> result = new WVPResult<>();
-                result.setCode(ERROR_CODE_TIMEOUT);
-                WvpRedisMsg response = WvpRedisMsg.getResponseInstance(
-                        userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, JSON.toJSONString(result)
-                );
-                JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
-                redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
-            }, userSetting.getPlatformPlayTimeout());
-
-            // 添加订阅
-            HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed(content.getApp(), content.getStream(), true, "rtsp", mediaServerItem.getId());
-
-            subscribe.addSubscribe(hookSubscribe, (mediaServerItemInUse, hookParam)->{
-                        dynamicTask.stop(taskKey);
-                        responseSendItem(mediaServerItem, content, toId, serial);
-                    });
-
-            MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(1, content.getApp(), content.getStream(),
-                    content.getChannelId(), content.getPlatformId(), content.getPlatformName(), content.getServerId(),
-                    content.getMediaServerId());
-
-            String key = VideoManagerConstants.VM_MSG_STREAM_PUSH_REQUESTED;
-            logger.info("[redis发送通知] 推流被请求 {}: {}/{}", key, messageForPushChannel.getApp(), messageForPushChannel.getStream());
-            redisTemplate.convertAndSend(key, JSON.toJSON(messageForPushChannel));
-        }
-    }
-
-    /**
-     * 将获取到的sendItem发送出去
-     */
-    private void responseSendItem(MediaServerItem mediaServerItem, RequestSendItemMsg content, String toId, String serial) {
-        SendRtpItem sendRtpItem = zlmServerFactory.createSendRtpItem(mediaServerItem, content.getIp(),
-                content.getPort(), content.getSsrc(), content.getPlatformId(),
-                content.getApp(), content.getStream(), content.getChannelId(),
-                content.getTcp(), content.getRtcp());
-
-        WVPResult<ResponseSendItemMsg> result = new WVPResult<>();
-        result.setCode(0);
-        ResponseSendItemMsg responseSendItemMsg = new ResponseSendItemMsg();
-        responseSendItemMsg.setSendRtpItem(sendRtpItem);
-        responseSendItemMsg.setMediaServerItem(mediaServerItem);
-        result.setData(responseSendItemMsg);
-        redisCatchStorage.updateSendRTPSever(sendRtpItem);
-
-        WvpRedisMsg response = WvpRedisMsg.getResponseInstance(
-                userSetting.getServerId(), toId, WvpRedisMsgCmd.GET_SEND_ITEM, serial, JSON.toJSONString(result)
-        );
-        JSONObject jsonObject = (JSONObject)JSON.toJSON(response);
-        redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
-    }
-
-    /**
-     * 发送消息要求下级生成推流信息
-     * @param serverId 下级服务ID
-     * @param app 应用名
-     * @param stream 流ID
-     * @param ip 目标IP
-     * @param port 目标端口
-     * @param ssrc  ssrc
-     * @param platformId 平台国标编号
-     * @param channelId 通道ID
-     * @param isTcp 是否使用TCP
-     * @param callback 得到信息的回调
-     */
-    public void sendMsg(String serverId, String mediaServerId, String app, String stream, String ip, int port, String ssrc,
-                        String platformId, String channelId, boolean isTcp, boolean rtcp, String platformName, PlayMsgCallback callback, PlayMsgErrorCallback errorCallback) {
-        RequestSendItemMsg requestSendItemMsg = RequestSendItemMsg.getInstance(
-                serverId, mediaServerId, app, stream, ip, port, ssrc, platformId, channelId, isTcp, rtcp, platformName);
-        requestSendItemMsg.setServerId(serverId);
-        String key = UUID.randomUUID().toString();
-        WvpRedisMsg redisMsg = WvpRedisMsg.getRequestInstance(userSetting.getServerId(), serverId, WvpRedisMsgCmd.GET_SEND_ITEM,
-                key, JSON.toJSONString(requestSendItemMsg));
-
-        JSONObject jsonObject = (JSONObject)JSON.toJSON(redisMsg);
-        logger.info("[请求推流SendItem] {}: {}", serverId, jsonObject);
-        callbacks.put(key, callback);
-        callbacksForError.put(key, errorCallback);
-        dynamicTask.startDelay(key, ()->{
-            callbacks.remove(key);
-            callbacksForError.remove(key);
-            WVPResult<Object> wvpResult = new WVPResult<>();
-            wvpResult.setCode(ERROR_CODE_TIMEOUT);
-            wvpResult.setMsg("timeout");
-            errorCallback.handler(wvpResult);
-        }, userSetting.getPlatformPlayTimeout());
-        redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
-    }
-
-    /**
-     * 发送请求推流的消息
-     * @param param 推流参数
-     * @param callback 回调
-     */
-    public void sendMsgForStartSendRtpStream(String serverId, RequestPushStreamMsg param, PlayMsgCallbackForStartSendRtpStream callback) {
-        String key = UUID.randomUUID().toString();
-        WvpRedisMsg redisMsg = WvpRedisMsg.getRequestInstance(userSetting.getServerId(), serverId,
-                WvpRedisMsgCmd.REQUEST_PUSH_STREAM, key, JSON.toJSONString(param));
-
-        JSONObject jsonObject = (JSONObject)JSON.toJSON(redisMsg);
-        logger.info("[REDIS 请求其他平台推流] {}: {}", serverId, jsonObject);
-        dynamicTask.startDelay(key, ()->{
-            callbacksForStartSendRtpStream.remove(key);
-            callbacksForError.remove(key);
-        }, userSetting.getPlatformPlayTimeout());
-        callbacksForStartSendRtpStream.put(key, callback);
-        callbacksForError.put(key, (wvpResult)->{
-            logger.info("[REDIS 请求其他平台推流] 失败: {}", wvpResult.getMsg());
-            callbacksForStartSendRtpStream.remove(key);
-            callbacksForError.remove(key);
-        });
-        redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
-    }
-
-    /**
-     * 发送请求推流的消息
-     */
-    public void sendMsgForStopSendRtpStream(String serverId, RequestStopPushStreamMsg streamMsg) {
-        String key = UUID.randomUUID().toString();
-        WvpRedisMsg redisMsg = WvpRedisMsg.getRequestInstance(userSetting.getServerId(), serverId,
-                WvpRedisMsgCmd.REQUEST_STOP_PUSH_STREAM, key, JSON.toJSONString(streamMsg));
-
-        JSONObject jsonObject = (JSONObject)JSON.toJSON(redisMsg);
-        logger.info("[REDIS 请求其他平台停止推流] {}: {}", serverId, jsonObject);
-        redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
-    }
-
-    private SendRtpItem querySendRTPServer(String platformGbId, String channelId, String streamId, String callId) {
-        if (platformGbId == null) {
-            platformGbId = "*";
-        }
-        if (channelId == null) {
-            channelId = "*";
-        }
-        if (streamId == null) {
-            streamId = "*";
-        }
-        if (callId == null) {
-            callId = "*";
-        }
-        String key = VideoManagerConstants.PLATFORM_SEND_RTP_INFO_PREFIX
-                + userSetting.getServerId() + "_*_"
-                + platformGbId + "_"
-                + channelId + "_"
-                + streamId + "_"
-                + callId;
-        List<Object> scan = RedisUtil.scan(redisTemplate, key);
-        if (scan.size() > 0) {
-            return (SendRtpItem)redisTemplate.opsForValue().get(scan.get(0));
-        }else {
-            return null;
-        }
-    }
-
-    /**
-     * 处理收到的请求推流的请求
-     */
-    private void requestStopPushStreamMsgHand(RequestStopPushStreamMsg streamMsg, String fromId, String serial) {
-        SendRtpItem sendRtpItem = streamMsg.getSendRtpItem();
-        if (sendRtpItem == null) {
-            logger.info("[REDIS 执行其他平台的请求停止推流] 失败： sendRtpItem为NULL");
-            return;
-        }
-        MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
-        if (mediaInfo == null) {
-            // TODO 回复错误
-            return;
-        }
-        Map<String, Object> param = new HashMap<>();
-        param.put("vhost","__defaultVhost__");
-        param.put("app",sendRtpItem.getApp());
-        param.put("stream",sendRtpItem.getStream());
-        param.put("ssrc", sendRtpItem.getSsrc());
-
-        if (zlmServerFactory.stopSendRtpStream(mediaInfo, param)) {
-            logger.info("[REDIS 执行其他平台的请求停止推流] 成功： {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream());
-            // 发送redis消息
-            MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,
-                    sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getChannelId(),
-                    sendRtpItem.getPlatformId(), streamMsg.getPlatformName(), userSetting.getServerId(), sendRtpItem.getMediaServerId());
-            messageForPushChannel.setPlatFormIndex(streamMsg.getPlatFormIndex());
-            redisCatchStorage.sendPlatformStopPlayMsg(messageForPushChannel);
-        }
-
-    }
-}

src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPlatformStartSendRtpListener.java

@@ -0,0 +1,81 @@
+package com.genersoft.iot.vmp.service.redisMsg;
+
+import com.alibaba.fastjson2.JSON;
+import com.genersoft.iot.vmp.conf.UserSetting;
+import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
+import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
+import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
+import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
+import com.genersoft.iot.vmp.media.zlm.dto.hook.HookParam;
+import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Component;
+import org.springframework.util.ObjectUtils;
+
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * 收到消息后开始给上级发流
+ * @author lin
+ */
+@Component
+public class RedisPlatformStartSendRtpListener implements MessageListener {
+
+    private final static Logger logger = LoggerFactory.getLogger(RedisPlatformStartSendRtpListener.class);
+
+    private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
+
+    @Autowired
+    private UserSetting userSetting;
+
+    @Autowired
+    private ZlmHttpHookSubscribe hookSubscribe;
+
+    @Qualifier("taskExecutor")
+    @Autowired
+    private ThreadPoolTaskExecutor taskExecutor;
+
+
+    @Override
+    public void onMessage(Message message, byte[] bytes) {
+        logger.info("[REDIS消息-收到上级等到设备推流的redis消息]： {}", new String(message.getBody()));
+        boolean isEmpty = taskQueue.isEmpty();
+        taskQueue.offer(message);
+        if (isEmpty) {
+            taskExecutor.execute(() -> {
+                while (!taskQueue.isEmpty()) {
+                    Message msg = taskQueue.poll();
+                    try {
+                        MessageForPushChannel messageForPushChannel = JSON.parseObject(new String(msg.getBody()), MessageForPushChannel.class);
+                        if (messageForPushChannel == null
+                                || ObjectUtils.isEmpty(messageForPushChannel.getApp())
+                                || ObjectUtils.isEmpty(messageForPushChannel.getStream())
+                        || userSetting.getServerId().equals(messageForPushChannel.getServerId())){
+                            continue;
+                        }
+
+                        // 监听流上线。 流上线直接发送sendRtpItem消息给实际的信令处理者
+                        HookSubscribeForStreamChange hook = HookSubscribeFactory.on_stream_changed(
+                                messageForPushChannel.getApp(), messageForPushChannel.getStream(), true, "rtsp",
+                                null);
+                        hookSubscribe.addSubscribe(hook, (MediaServerItem mediaServerItemInUse, HookParam hookParam) -> {
+                            // 读取redis中的上级点播信息，生成sendRtpItm发送出去
+
+                        });
+
+
+                    }catch (Exception e) {
+                        logger.warn("[REDIS消息-请求推流结果] 发现未处理的异常, \r\n{}", JSON.toJSONString(message));
+                        logger.error("[REDIS消息-请求推流结果] 异常内容： ", e);
+                    }
+                }
+            });
+        }
+    }
+}

src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPlatformWaitPushStreamOnlineListener.java

@@ -0,0 +1,81 @@
+package com.genersoft.iot.vmp.service.redisMsg;
+
+import com.alibaba.fastjson2.JSON;
+import com.genersoft.iot.vmp.conf.UserSetting;
+import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
+import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeFactory;
+import com.genersoft.iot.vmp.media.zlm.dto.HookSubscribeForStreamChange;
+import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
+import com.genersoft.iot.vmp.media.zlm.dto.hook.HookParam;
+import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Component;
+import org.springframework.util.ObjectUtils;
+
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * 上级等到设备推流的redis消息
+ * @author lin
+ */
+@Component
+public class RedisPlatformWaitPushStreamOnlineListener implements MessageListener {
+
+    private final static Logger logger = LoggerFactory.getLogger(RedisPlatformWaitPushStreamOnlineListener.class);
+
+    private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
+
+    @Autowired
+    private UserSetting userSetting;
+
+    @Autowired
+    private ZlmHttpHookSubscribe hookSubscribe;
+
+    @Qualifier("taskExecutor")
+    @Autowired
+    private ThreadPoolTaskExecutor taskExecutor;
+
+
+    @Override
+    public void onMessage(Message message, byte[] bytes) {
+        logger.info("[REDIS消息-收到上级等到设备推流的redis消息]： {}", new String(message.getBody()));
+        boolean isEmpty = taskQueue.isEmpty();
+        taskQueue.offer(message);
+        if (isEmpty) {
+            taskExecutor.execute(() -> {
+                while (!taskQueue.isEmpty()) {
+                    Message msg = taskQueue.poll();
+                    try {
+                        MessageForPushChannel messageForPushChannel = JSON.parseObject(new String(msg.getBody()), MessageForPushChannel.class);
+                        if (messageForPushChannel == null
+                                || ObjectUtils.isEmpty(messageForPushChannel.getApp())
+                                || ObjectUtils.isEmpty(messageForPushChannel.getStream())
+                        || userSetting.getServerId().equals(messageForPushChannel.getServerId())){
+                            continue;
+                        }
+
+                        // 监听流上线。 流上线直接发送sendRtpItem消息给实际的信令处理者
+                        HookSubscribeForStreamChange hook = HookSubscribeFactory.on_stream_changed(
+                                messageForPushChannel.getApp(), messageForPushChannel.getStream(), true, "rtsp",
+                                null);
+                        hookSubscribe.addSubscribe(hook, (MediaServerItem mediaServerItemInUse, HookParam hookParam) -> {
+                            // 读取redis中的上级点播信息，生成sendRtpItm发送出去
+
+                        });
+
+
+                    }catch (Exception e) {
+                        logger.warn("[REDIS消息-请求推流结果] 发现未处理的异常, \r\n{}", JSON.toJSONString(message));
+                        logger.error("[REDIS消息-请求推流结果] 异常内容： ", e);
+                    }
+                }
+            });
+        }
+    }
+}

src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java

@@ -217,4 +217,7 @@ public interface IRedisCatchStorage {
 
     void sendPushStreamClose(MessageForPushChannel messageForPushChannel);
 
+    void addWaiteSendRtpItem(SendRtpItem sendRtpItem, int platformPlayTimeout);
+
+    void sendStartSendRtp(SendRtpItem sendRtpItem);
 }

src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java

@@ -678,4 +678,16 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
         logger.info("[redis发送通知] 发送 停止向上级推流 {}: {}/{}->{}", key, msg.getApp(), msg.getStream(), msg.getPlatFormId());
         redisTemplate.convertAndSend(key, JSON.toJSON(msg));
     }
+
+    @Override
+    public void addWaiteSendRtpItem(SendRtpItem sendRtpItem, int platformPlayTimeout) {
+        String key = VideoManagerConstants.WAITE_SEND_PUSH_STREAM + sendRtpItem.getApp() + "_" + sendRtpItem.getStream();
+        redisTemplate.opsForValue().set(key, platformPlayTimeout);
+    }
+
+    @Override
+    public void sendStartSendRtp(SendRtpItem sendRtpItem) {
+        String key = VideoManagerConstants.START_SEND_PUSH_STREAM + sendRtpItem.getApp() + "_" + sendRtpItem.getStream();
+        redisTemplate.opsForValue().set(key, JSON.toJSONString(sendRtpItem));
+    }
 }