更新级联查看直播视频及代理拉流视频bug

gushouzheng
1 parent d09d8f11
Showing 3 changed files with 83 additions and 10 deletions
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java
@@ -17,9 +17,11 @@ import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
 import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
 import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
 import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
+import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
 import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
 import com.genersoft.iot.vmp.service.IMediaServerService;
 import com.genersoft.iot.vmp.service.IPlayService;
+import com.genersoft.iot.vmp.service.IStreamProxyService;
 import com.genersoft.iot.vmp.service.IStreamPushService;
 import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
 import com.genersoft.iot.vmp.service.bean.SSRCInfo;
@@ -65,6 +67,8 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
  
     @Autowired
     private IStreamPushService streamPushService;
+    @Autowired
+    private IStreamProxyService streamProxyService;
  
     @Autowired
     private IRedisCatchStorage redisCatchStorage;
@@ -145,6 +149,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
  
                 MediaServerItem mediaServerItem = null;
                 StreamPushItem streamPushItem = null;
+                StreamProxyItem proxyByAppAndStream =null;
                 // 不是通道可能是直播流
                 if (channel != null && gbStream == null) {
                     if (channel.getStatus() == 0) {
@@ -178,6 +183,13 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
                                 responseAck(evt, Response.GONE);
                                 return;
                             }
+                        }else if("proxy".equals(gbStream.getStreamType())){
+                            proxyByAppAndStream = streamProxyService.getStreamProxyByAppAndStream(gbStream.getApp(), gbStream.getStream());
+                            if (proxyByAppAndStream == null) {
+                                logger.info("[ app={}, stream={} ]找不到zlm {}，返回410", gbStream.getApp(), gbStream.getStream(), mediaServerId);
+                                responseAck(evt, Response.GONE);
+                                return;
+                            }
                         }
                     }
                     responseAck(evt, Response.CALL_IS_BEING_FORWARDED); // 通道存在，发181，呼叫转接中
@@ -419,14 +431,33 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
                         }
                     }
                 } else if (gbStream != null) {
-                    if (streamPushItem != null && streamPushItem.isPushIng()) {
-                        // 推流状态
-                        pushStream(evt, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
-                                mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
-                    } else {
-                        // 未推流 拉起
-                        notifyStreamOnline(evt, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
-                                mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+                    if("push".equals(gbStream.getStreamType())) {
+                        if (streamPushItem != null && streamPushItem.isPushIng()) {
+                            // 推流状态
+                            pushStream(evt, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
+                                    mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+                        } else {
+                            // 未推流 拉起
+                            notifyStreamOnline(evt, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
+                                    mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+                        }
+                    }else if ("proxy".equals(gbStream.getStreamType())){
+                        if(null != proxyByAppAndStream &&proxyByAppAndStream.isStatus()){
+                            pushProxyStream(evt, gbStream,  platform, callIdHeader, mediaServerItem, port, tcpActive,
+                                    mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+                        }else{
+                            //开启代理拉流
+                            boolean start1 = streamProxyService.start(gbStream.getApp(), gbStream.getStream());
+                            if(start1) {
+                                pushProxyStream(evt, gbStream,  platform, callIdHeader, mediaServerItem, port, tcpActive,
+                                        mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+                            }else{
+                                //失败后通知
+                                notifyStreamOnline(evt, gbStream, null, platform, callIdHeader, mediaServerItem, port, tcpActive,
+                                        mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
+                            }
+                        }
+
                     }
                 }
             }
@@ -445,7 +476,39 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
     /**
      * 安排推流
      */
+    private void pushProxyStream(RequestEvent evt, GbStream gbStream, ParentPlatform platform,
+                            CallIdHeader callIdHeader, MediaServerItem mediaServerItem,
+                            int port, Boolean tcpActive, boolean mediaTransmissionTCP,
+                            String channelId, String addressStr, String ssrc, String requesterId) throws InvalidArgumentException, ParseException, SipException {
+            Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream());
+            if (streamReady) {
+                // 自平台内容
+                SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
+                        gbStream.getApp(), gbStream.getStream(), channelId,
+                        mediaTransmissionTCP);
  
+                if (sendRtpItem == null) {
+                    logger.warn("服务器端口资源不足");
+                    responseAck(evt, Response.BUSY_HERE);
+                    return;
+                }
+                if (tcpActive != null) {
+                    sendRtpItem.setTcpActive(tcpActive);
+                }
+                sendRtpItem.setPlayType(InviteStreamType.PUSH);
+                // 写入redis， 超时时回复
+                sendRtpItem.setStatus(1);
+                sendRtpItem.setCallId(callIdHeader.getCallId());
+                byte[] dialogByteArray = SerializeUtils.serialize(evt.getDialog());
+                sendRtpItem.setDialog(dialogByteArray);
+                byte[] transactionByteArray = SerializeUtils.serialize(evt.getServerTransaction());
+                sendRtpItem.setTransaction(transactionByteArray);
+                redisCatchStorage.updateSendRTPSever(sendRtpItem);
+                sendStreamAck(mediaServerItem, sendRtpItem, platform, evt);
+
+        }
+
+    }
     private void pushStream(RequestEvent evt, GbStream gbStream, StreamPushItem streamPushItem, ParentPlatform platform,
                             CallIdHeader callIdHeader, MediaServerItem mediaServerItem,
                             int port, Boolean tcpActive, boolean mediaTransmissionTCP,
@@ -490,7 +553,6 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
         }
  
     }
-
     /**
      * 通知流上线
      */
@@ -277,7 +277,11 @@ public class MediaServerServiceImpl implements IMediaServerService {
             return null;
         }
         String key = VideoManagerConstants.MEDIA_SERVER_PREFIX + userSetting.getServerId() + "_" + mediaServerId;
-        return (MediaServerItem)redisUtil.get(key);
+        MediaServerItem serverItem=(MediaServerItem)redisUtil.get(key);
+        if(null==serverItem){
+            serverItem=mediaServerMapper.queryOne(mediaServerId);
+        }
+        return serverItem;
     }
  
     @Override
@@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import com.alibaba.fastjson.TypeReference;
+import com.genersoft.iot.vmp.conf.MediaConfig;
 import com.genersoft.iot.vmp.conf.UserSetting;
 import com.genersoft.iot.vmp.gb28181.bean.*;
 import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
@@ -78,6 +79,10 @@ public class StreamPushServiceImpl implements IStreamPushService {
     @Autowired
     TransactionDefinition transactionDefinition;
  
+    @Autowired
+    private MediaConfig mediaConfig;
+
+
     @Override
     public List<StreamPushItem> handleJSON(String jsonData, MediaServerItem mediaServerItem) {
         if (jsonData == null) {
@@ -142,6 +147,8 @@ public class StreamPushServiceImpl implements IStreamPushService {
         stream.setStreamType("push");
         stream.setStatus(true);
         stream.setCreateTime(DateUtil.getNow());
+        stream.setStreamType("push");
+        stream.setMediaServerId(mediaConfig.getId());
         int add = gbStreamMapper.add(stream);
         return add > 0;
     }