From ec0ec5eb54723cc9aeabb4f313da2a101ab98bd2 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期三, 07 九月 2022 16:18:35 +0800
Subject: [PATCH] 修复并发点播时可能出现的rtpServer开启但是还未收到流的情况,编码类型136,137,138默认开启音频通道

---
 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java |   18 ++++++--
 src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java                      |   20 +++++-----
 src/main/java/com/genersoft/iot/vmp/gb28181/utils/XmlUtil.java                                      |    7 +++
 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java                     |    4 +-
 src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java                              |    4 ++
 src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java                               |   30 ++++++++++----
 src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java                        |   13 ++++++
 src/main/java/com/genersoft/iot/vmp/vmanager/user/UserController.java                               |    1 
 src/main/java/com/genersoft/iot/vmp/service/IMediaServerService.java                                |    2 +
 src/main/resources/logback-spring-local.xml                                                         |    5 ++
 src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java                           |   14 ++++--
 11 files changed, 85 insertions(+), 33 deletions(-)

diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
index 4d1e568..4c40f54 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
@@ -356,7 +356,7 @@
 //			String streamMode = device.getStreamMode().toUpperCase();
 
 			logger.info("{} 鍒嗛厤鐨刏LM涓�: {} [{}:{}]", stream, mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort());
-			HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", stream, true, "rtmp", mediaServerItem.getId());
+			HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", stream, true, "rtsp", mediaServerItem.getId());
 			subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json)->{
 				if (event != null) {
 					event.response(mediaServerItemInUse, json);
@@ -524,7 +524,7 @@
 
 			CallIdHeader callIdHeader = device.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
 					: udpSipProvider.getNewCallId();
-			HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtmp", mediaServerItem.getId());
+			HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());
 			// 娣诲姞璁㈤槄
 			subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json)->{
 						if (hookEvent != null) {
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
index a6956da..82b3ba4 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
@@ -16,9 +16,7 @@
 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.media.zlm.dto.*;
 import com.genersoft.iot.vmp.service.IMediaServerService;
 import com.genersoft.iot.vmp.service.IPlayService;
 import com.genersoft.iot.vmp.service.IStreamProxyService;
@@ -89,6 +87,9 @@
 
     @Autowired
     private IMediaServerService mediaServerService;
+
+    @Autowired
+    private ZlmHttpHookSubscribe zlmHttpHookSubscribe;
 
     @Autowired
     private SIPProcessorObserver sipProcessorObserver;
@@ -400,7 +401,14 @@
                         if (playTransaction != null) {
                             Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, "rtp", playTransaction.getStream());
                             if (!streamReady) {
-                                playTransaction = null;
+                                boolean hasRtpServer = mediaServerService.checkRtpServer(mediaServerItem, "rtp", playTransaction.getStream());
+                                if (hasRtpServer) {
+                                    logger.info("[涓婄骇鐐规挱]宸茬粡寮�鍚痳tpServer浣嗘槸灏氭湭鏀跺埌娴侊紝寮�鍚洃鍚祦鐨勫埌鏉�");
+                                    HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", playTransaction.getStream(), true, "rtsp", mediaServerItem.getId());
+                                    zlmHttpHookSubscribe.addSubscribe(hookSubscribe, hookEvent);
+                                }else {
+                                    playTransaction = null;
+                                }
                             }
                         }
                         if (playTransaction == null) {
@@ -564,7 +572,7 @@
         } else if ("push".equals(gbStream.getStreamType())) {
             if (!platform.isStartOfflinePush()) {
                 // 骞冲彴璁剧疆涓叧闂簡鎷夎捣绂荤嚎鐨勬帹娴佸垯鐩存帴鍥炲
-                responseAck(evt, Response.TEMPORARILY_UNAVAILABLE, "channel unavailable");
+                responseAck(evt, Response.TEMPORARILY_UNAVAILABLE, "channel stream not pushing");
                 return;
             }
             // 鍙戦�乺edis娑堟伅浠ヤ娇璁惧涓婄嚎
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/utils/XmlUtil.java b/src/main/java/com/genersoft/iot/vmp/gb28181/utils/XmlUtil.java
index 733f78a..2d568a1 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/utils/XmlUtil.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/utils/XmlUtil.java
@@ -203,6 +203,12 @@
             return null;
         }
         deviceChannel.setChannelId(channelId);
+        int channelTypeCode = Integer.parseInt(channelId.substring(10, 13));
+        if (channelTypeCode == 136 || channelTypeCode == 137 || channelTypeCode == 138) {
+            deviceChannel.setHasAudio(true);
+        }else {
+            deviceChannel.setHasAudio(false);
+        }
         if (event != null && !event.equals(CatalogEvent.ADD) && !event.equals(CatalogEvent.UPDATE)) {
             // 闄や簡ADD鍜寀pdate鎯呭喌涓嬮渶瑕佽瘑鍒叏閮ㄥ唴瀹癸紝
             return deviceChannel;
@@ -396,7 +402,6 @@
         } else {
             deviceChannel.setPTZType(Integer.parseInt(XmlUtil.getText(itemDevice, "PTZType")));
         }
-        deviceChannel.setHasAudio(true); // 榛樿鍚湁闊抽锛屾挱鏀炬椂鍐嶆鏌ユ槸鍚︽湁闊抽鍙婃槸鍚AC
         return deviceChannel;
     }
 }
\ No newline at end of file
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 6c70096..6caff71 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
@@ -96,6 +96,10 @@
         if(rtpInfo.getInteger("code") == 0){
             if (rtpInfo.getBoolean("exist")) {
                 result = rtpInfo.getInteger("local_port");
+                if (result == 0) {
+                    // 姝ゆ椂璇存槑rtpServer宸茬粡鍒涘缓浣嗘槸娴佽繕娌℃湁鎺ㄤ笂鏉�
+
+                }
                 return result;
             }
         }else if(rtpInfo.getInteger("code") == -2){
diff --git a/src/main/java/com/genersoft/iot/vmp/service/IMediaServerService.java b/src/main/java/com/genersoft/iot/vmp/service/IMediaServerService.java
index f1163ca..55a4005 100644
--- a/src/main/java/com/genersoft/iot/vmp/service/IMediaServerService.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/IMediaServerService.java
@@ -83,4 +83,6 @@
     MediaServerItem getDefaultMediaServer();
 
     void updateMediaServerKeepalive(String mediaServerId, JSONObject data);
+
+    boolean checkRtpServer(MediaServerItem mediaServerItem, String rtp, String stream);
 }
diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
index 64a411a..702967d 100644
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
@@ -147,9 +147,11 @@
             if (streamId == null) {
                 streamId = String.format("%08x", Integer.parseInt(ssrc)).toUpperCase();
             }
-            int rtpServerPort = mediaServerItem.getRtpProxyPort();
+            int rtpServerPort;
             if (mediaServerItem.isRtpEnable()) {
                 rtpServerPort = zlmrtpServerFactory.createRTPServer(mediaServerItem, streamId, ssrcCheck?Integer.parseInt(ssrc):0, port);
+            } else {
+                rtpServerPort = mediaServerItem.getRtpProxyPort();
             }
             RedisUtil.set(key, mediaServerItem);
             return new SSRCInfo(rtpServerPort, ssrc, streamId);
@@ -681,4 +683,13 @@
             }
         }
     }
+
+    @Override
+    public boolean checkRtpServer(MediaServerItem mediaServerItem, String app, String stream) {
+        JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaServerItem, stream);
+        if(rtpInfo.getInteger("code") == 0){
+            return rtpInfo.getBoolean("exist");
+        }
+        return false;
+    }
 }
diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
index aa01922..5abb342 100644
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
@@ -164,17 +164,30 @@
             JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaInfo, streamId);
             if(rtpInfo.getInteger("code") == 0){
                 if (rtpInfo.getBoolean("exist")) {
+                    int localPort = rtpInfo.getInteger("local_port");
+                    if (localPort == 0) {
+                        logger.warn("[鐐规挱]锛岀偣鎾椂鍙戠幇rtpServerC瀛樺湪锛屼絾鏄皻鏈紑濮嬫帹娴�");
+                        // 姝ゆ椂璇存槑rtpServer宸茬粡鍒涘缓浣嗘槸娴佽繕娌℃湁鎺ㄤ笂鏉�
+                        WVPResult wvpResult = new WVPResult();
+                        wvpResult.setCode(ErrorCode.ERROR100.getCode());
+                        wvpResult.setMsg("鐐规挱宸茬粡鍦ㄨ繘琛屼腑锛岃绋嶅�欓噸璇�");
+                        msg.setData(wvpResult);
 
-                    WVPResult wvpResult = new WVPResult();
-                    wvpResult.setCode(ErrorCode.SUCCESS.getCode());
-                    wvpResult.setMsg(ErrorCode.SUCCESS.getMsg());
-                    wvpResult.setData(streamInfo);
-                    msg.setData(wvpResult);
+                        resultHolder.invokeAllResult(msg);
+                        return playResult;
+                    }else {
+                        WVPResult wvpResult = new WVPResult();
+                        wvpResult.setCode(ErrorCode.SUCCESS.getCode());
+                        wvpResult.setMsg(ErrorCode.SUCCESS.getMsg());
+                        wvpResult.setData(streamInfo);
+                        msg.setData(wvpResult);
 
-                    resultHolder.invokeAllResult(msg);
-                    if (hookEvent != null) {
-                        hookEvent.response(mediaServerItem, JSONObject.parseObject(JSON.toJSONString(streamInfo)));
+                        resultHolder.invokeAllResult(msg);
+                        if (hookEvent != null) {
+                            hookEvent.response(mediaServerItem, JSONObject.parseObject(JSON.toJSONString(streamInfo)));
+                        }
                     }
+
                 }else {
                     redisCatchStorage.stopPlay(streamInfo);
                     storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
@@ -187,7 +200,6 @@
                 streamInfo = null;
 
             }
-
         }
         if (streamInfo == null) {
             String streamId = null;
diff --git a/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java b/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java
index bcebb94..25745c4 100644
--- a/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java
+++ b/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java
@@ -143,15 +143,12 @@
     @Update(value = {"UPDATE device_channel SET status=0 WHERE deviceId=#{deviceId}"})
     void offlineByDeviceId(String deviceId);
 
-    @Update(value = {"UPDATE device_channel SET status=1 WHERE deviceId=#{deviceId} AND channelId=#{channelId}"})
-    void online(String deviceId,  String channelId);
-
     @Insert("<script> " +
             "insert into device_channel " +
             "(channelId, deviceId, name, manufacture, model, owner, civilCode, block, subCount, " +
             "  address, parental, parentId, safetyWay, registerWay, certNum, certifiable, errCode, secrecy, " +
             "  ipAddress, port, password, PTZType, status, streamId, longitude, latitude, longitudeGcj02, latitudeGcj02, " +
-            "  longitudeWgs84, latitudeWgs84, createTime, updateTime, businessGroupId, gpsTime) " +
+            "  longitudeWgs84, latitudeWgs84, hasAudio, createTime, updateTime, businessGroupId, gpsTime) " +
             "values " +
             "<foreach collection='addChannels' index='index' item='item' separator=','> " +
             "('${item.channelId}', '${item.deviceId}', '${item.name}', '${item.manufacture}', '${item.model}', " +
@@ -160,7 +157,7 @@
             "'${item.certNum}', ${item.certifiable}, ${item.errCode}, '${item.secrecy}', " +
             "'${item.ipAddress}', ${item.port}, '${item.password}', ${item.PTZType}, ${item.status}, " +
             "'${item.streamId}', ${item.longitude}, ${item.latitude},${item.longitudeGcj02}, " +
-            "${item.latitudeGcj02},${item.longitudeWgs84}, ${item.latitudeWgs84},'${item.createTime}', '${item.updateTime}', " +
+            "${item.latitudeGcj02},${item.longitudeWgs84}, ${item.latitudeWgs84}, ${item.hasAudio},'${item.createTime}', '${item.updateTime}', " +
             "'${item.businessGroupId}', '${item.gpsTime}') " +
             "</foreach> " +
             "ON DUPLICATE KEY UPDATE " +
@@ -193,10 +190,14 @@
             "latitudeGcj02=VALUES(latitudeGcj02), " +
             "longitudeWgs84=VALUES(longitudeWgs84), " +
             "latitudeWgs84=VALUES(latitudeWgs84), " +
+            "hasAudio=VALUES(hasAudio), " +
             "businessGroupId=VALUES(businessGroupId), " +
             "gpsTime=VALUES(gpsTime)" +
             "</script>")
     int batchAdd(List<DeviceChannel> addChannels);
+
+    @Update(value = {"UPDATE device_channel SET status=1 WHERE deviceId=#{deviceId} AND channelId=#{channelId}"})
+    void online(String deviceId,  String channelId);
 
     @Update({"<script>" +
             "<foreach collection='updateChannels' item='item' separator=';'>" +
@@ -341,4 +342,7 @@
             " left join platform_catalog pc on pgc.catalogId = pc.id and pgc.platformId = pc.platformId" +
             " where pgc.platformId=#{serverGBId}")
     List<DeviceChannel> queryChannelWithCatalog(String serverGBId);
+
+    @Select("select * from device_channel where deviceId = #{deviceId}")
+    List<DeviceChannel> queryAllChannels(String deviceId);
 }
diff --git a/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java b/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java
index e8e0e02..88cb8c0 100644
--- a/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java
@@ -111,11 +111,11 @@
 		if (CollectionUtils.isEmpty(deviceChannelList)) {
 			return false;
 		}
-		List<DeviceChannel> allChannelInPlay = deviceChannelMapper.getAllChannelInPlay();
-		Map<String,DeviceChannel> allChannelMapInPlay = new ConcurrentHashMap<>();
-		if (allChannelInPlay.size() > 0) {
-			for (DeviceChannel deviceChannel : allChannelInPlay) {
-				allChannelMapInPlay.put(deviceChannel.getChannelId(), deviceChannel);
+		List<DeviceChannel> allChannels = deviceChannelMapper.queryAllChannels(deviceId);
+		Map<String,DeviceChannel> allChannelMap = new ConcurrentHashMap<>();
+		if (allChannels.size() > 0) {
+			for (DeviceChannel deviceChannel : allChannels) {
+				allChannelMap.put(deviceChannel.getChannelId(), deviceChannel);
 			}
 		}
 		TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
@@ -123,15 +123,17 @@
 		List<DeviceChannel> channels = new ArrayList<>();
 		StringBuilder stringBuilder = new StringBuilder();
 		Map<String, Integer> subContMap = new HashMap<>();
-		if (deviceChannelList.size() > 1) {
+		if (deviceChannelList.size() > 0) {
 			// 鏁版嵁鍘婚噸
 			Set<String> gbIdSet = new HashSet<>();
 			for (DeviceChannel deviceChannel : deviceChannelList) {
 				if (!gbIdSet.contains(deviceChannel.getChannelId())) {
 					gbIdSet.add(deviceChannel.getChannelId());
-					if (allChannelMapInPlay.containsKey(deviceChannel.getChannelId())) {
-						deviceChannel.setStreamId(allChannelMapInPlay.get(deviceChannel.getChannelId()).getStreamId());
+					if (allChannelMap.containsKey(deviceChannel.getChannelId())) {
+						deviceChannel.setStreamId(allChannelMap.get(deviceChannel.getChannelId()).getStreamId());
+						deviceChannel.setHasAudio(allChannelMap.get(deviceChannel.getChannelId()).isHasAudio());
 					}
+
 					channels.add(deviceChannel);
 					if (!ObjectUtils.isEmpty(deviceChannel.getParentId())) {
 						if (subContMap.get(deviceChannel.getParentId()) == null) {
@@ -153,8 +155,6 @@
 				}
 			}
 
-		}else {
-			channels = deviceChannelList;
 		}
 		if (stringBuilder.length() > 0) {
 			logger.info("[鐩綍鏌ヨ]鏀跺埌鐨勬暟鎹瓨鍦ㄩ噸澶嶏細 {}" , stringBuilder);
diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/user/UserController.java b/src/main/java/com/genersoft/iot/vmp/vmanager/user/UserController.java
index 2b44dd7..0052fd3 100644
--- a/src/main/java/com/genersoft/iot/vmp/vmanager/user/UserController.java
+++ b/src/main/java/com/genersoft/iot/vmp/vmanager/user/UserController.java
@@ -43,6 +43,7 @@
     private IRoleService roleService;
 
     @GetMapping("/login")
+    @PostMapping("/login")
     @Operation(summary = "鐧诲綍")
     @Parameter(name = "username", description = "鐢ㄦ埛鍚�", required = true)
     @Parameter(name = "password", description = "瀵嗙爜锛�32浣峬d5鍔犲瘑锛�", required = true)
diff --git a/src/main/resources/logback-spring-local.xml b/src/main/resources/logback-spring-local.xml
index b9d3b39..724d05e 100644
--- a/src/main/resources/logback-spring-local.xml
+++ b/src/main/resources/logback-spring-local.xml
@@ -98,6 +98,11 @@
 		<appender-ref ref="STDOUT" />
 	</root>
 
+	<logger name="wvp" level="debug" additivity="true">
+		<appender-ref ref="RollingFileError"/>
+		<appender-ref ref="RollingFile"/>
+	</logger>
+
 	<logger name="GB28181_SIP" level="debug" additivity="true">
 		<appender-ref ref="RollingFileError"/>
 		<appender-ref ref="sipRollingFile"/>

--
Gitblit v1.8.0