guzijian / wvp-GB28181-pro · Commits

Commit ec0ec5eb54723cc9aeabb4f313da2a101ab98bd2

Authored by 648540858
1 parent e7bdcc1f

修复并发点播时可能出现的rtpServer开启但是还未收到流的情况，编码类型136，137,138默认开启音频通道

Inline Side-by-side
Showing 11 changed files with 87 additions and 35 deletions
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
  View file @ec0ec5e
... ... @@ -356,7 +356,7 @@ public class SIPCommander implements ISIPCommander {
356 356 // String streamMode = device.getStreamMode().toUpperCase();
357 357  
358 358 logger.info("{} 分配的ZLM为: {} [{}:{}]", stream, mediaServerItem.getId(), mediaServerItem.getIp(), ssrcInfo.getPort());
359   - HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", stream, true, "rtmp", mediaServerItem.getId());
  359 + HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", stream, true, "rtsp", mediaServerItem.getId());
360 360 subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json)->{
361 361 if (event != null) {
362 362 event.response(mediaServerItemInUse, json);
... ... @@ -524,7 +524,7 @@ public class SIPCommander implements ISIPCommander {
524 524  
525 525 CallIdHeader callIdHeader = device.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
526 526 : udpSipProvider.getNewCallId();
527   - HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtmp", mediaServerItem.getId());
  527 + HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", ssrcInfo.getStream(), true, "rtsp", mediaServerItem.getId());
528 528 // 添加订阅
529 529 subscribe.addSubscribe(hookSubscribe, (MediaServerItem mediaServerItemInUse, JSONObject json)->{
530 530 if (hookEvent != null) {
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
  View file @ec0ec5e
... ... @@ -16,9 +16,7 @@ import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
16 16 import com.genersoft.iot.vmp.media.zlm.ZlmHttpHookSubscribe;
17 17 import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
18 18 import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
19   -import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
20   -import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
21   -import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
  19 +import com.genersoft.iot.vmp.media.zlm.dto.*;
22 20 import com.genersoft.iot.vmp.service.IMediaServerService;
23 21 import com.genersoft.iot.vmp.service.IPlayService;
24 22 import com.genersoft.iot.vmp.service.IStreamProxyService;
... ... @@ -91,6 +89,9 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
91 89 private IMediaServerService mediaServerService;
92 90  
93 91 @Autowired
  92 + private ZlmHttpHookSubscribe zlmHttpHookSubscribe;
  93 +
  94 + @Autowired
94 95 private SIPProcessorObserver sipProcessorObserver;
95 96  
96 97 @Autowired
... ... @@ -400,7 +401,14 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
400 401 if (playTransaction != null) {
401 402 Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, "rtp", playTransaction.getStream());
402 403 if (!streamReady) {
403   - playTransaction = null;
  404 + boolean hasRtpServer = mediaServerService.checkRtpServer(mediaServerItem, "rtp", playTransaction.getStream());
  405 + if (hasRtpServer) {
  406 + logger.info("[上级点播]已经开启rtpServer但是尚未收到流，开启监听流的到来");
  407 + HookSubscribeForStreamChange hookSubscribe = HookSubscribeFactory.on_stream_changed("rtp", playTransaction.getStream(), true, "rtsp", mediaServerItem.getId());
  408 + zlmHttpHookSubscribe.addSubscribe(hookSubscribe, hookEvent);
  409 + }else {
  410 + playTransaction = null;
  411 + }
404 412 }
405 413 }
406 414 if (playTransaction == null) {
... ... @@ -564,7 +572,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
564 572 } else if ("push".equals(gbStream.getStreamType())) {
565 573 if (!platform.isStartOfflinePush()) {
566 574 // 平台设置中关闭了拉起离线的推流则直接回复
567   - responseAck(evt, Response.TEMPORARILY_UNAVAILABLE, "channel unavailable");
  575 + responseAck(evt, Response.TEMPORARILY_UNAVAILABLE, "channel stream not pushing");
568 576 return;
569 577 }
570 578 // 发送redis消息以使设备上线
... ...
src/main/java/com/genersoft/iot/vmp/gb28181/utils/XmlUtil.java
  View file @ec0ec5e
... ... @@ -203,6 +203,12 @@ public class XmlUtil {
203 203 return null;
204 204 }
205 205 deviceChannel.setChannelId(channelId);
  206 + int channelTypeCode = Integer.parseInt(channelId.substring(10, 13));
  207 + if (channelTypeCode == 136 || channelTypeCode == 137 || channelTypeCode == 138) {
  208 + deviceChannel.setHasAudio(true);
  209 + }else {
  210 + deviceChannel.setHasAudio(false);
  211 + }
206 212 if (event != null && !event.equals(CatalogEvent.ADD) && !event.equals(CatalogEvent.UPDATE)) {
207 213 // 除了ADD和update情况下需要识别全部内容，
208 214 return deviceChannel;
... ... @@ -396,7 +402,6 @@ public class XmlUtil {
396 402 } else {
397 403 deviceChannel.setPTZType(Integer.parseInt(XmlUtil.getText(itemDevice, "PTZType")));
398 404 }
399   - deviceChannel.setHasAudio(true); // 默认含有音频，播放时再检查是否有音频及是否AAC
400 405 return deviceChannel;
401 406 }
402 407 }
403 408 \ No newline at end of file
... ...
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java
  View file @ec0ec5e
... ... @@ -96,6 +96,10 @@ public class ZLMRTPServerFactory {
96 96 if(rtpInfo.getInteger("code") == 0){
97 97 if (rtpInfo.getBoolean("exist")) {
98 98 result = rtpInfo.getInteger("local_port");
  99 + if (result == 0) {
  100 + // 此时说明rtpServer已经创建但是流还没有推上来
  101 +
  102 + }
99 103 return result;
100 104 }
101 105 }else if(rtpInfo.getInteger("code") == -2){
... ...
src/main/java/com/genersoft/iot/vmp/service/IMediaServerService.java
  View file @ec0ec5e
... ... @@ -83,4 +83,6 @@ public interface IMediaServerService {
83 83 MediaServerItem getDefaultMediaServer();
84 84  
85 85 void updateMediaServerKeepalive(String mediaServerId, JSONObject data);
  86 +
  87 + boolean checkRtpServer(MediaServerItem mediaServerItem, String rtp, String stream);
86 88 }
... ...
src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
  View file @ec0ec5e
... ... @@ -147,9 +147,11 @@ public class MediaServerServiceImpl implements IMediaServerService {
147 147 if (streamId == null) {
148 148 streamId = String.format("%08x", Integer.parseInt(ssrc)).toUpperCase();
149 149 }
150   - int rtpServerPort = mediaServerItem.getRtpProxyPort();
  150 + int rtpServerPort;
151 151 if (mediaServerItem.isRtpEnable()) {
152 152 rtpServerPort = zlmrtpServerFactory.createRTPServer(mediaServerItem, streamId, ssrcCheck?Integer.parseInt(ssrc):0, port);
  153 + } else {
  154 + rtpServerPort = mediaServerItem.getRtpProxyPort();
153 155 }
154 156 RedisUtil.set(key, mediaServerItem);
155 157 return new SSRCInfo(rtpServerPort, ssrc, streamId);
... ... @@ -681,4 +683,13 @@ public class MediaServerServiceImpl implements IMediaServerService {
681 683 }
682 684 }
683 685 }
  686 +
  687 + @Override
  688 + public boolean checkRtpServer(MediaServerItem mediaServerItem, String app, String stream) {
  689 + JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaServerItem, stream);
  690 + if(rtpInfo.getInteger("code") == 0){
  691 + return rtpInfo.getBoolean("exist");
  692 + }
  693 + return false;
  694 + }
684 695 }
... ...
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
  View file @ec0ec5e
... ... @@ -164,17 +164,30 @@ public class PlayServiceImpl implements IPlayService {
164 164 JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(mediaInfo, streamId);
165 165 if(rtpInfo.getInteger("code") == 0){
166 166 if (rtpInfo.getBoolean("exist")) {
167   -
168   - WVPResult wvpResult = new WVPResult();
169   - wvpResult.setCode(ErrorCode.SUCCESS.getCode());
170   - wvpResult.setMsg(ErrorCode.SUCCESS.getMsg());
171   - wvpResult.setData(streamInfo);
172   - msg.setData(wvpResult);
173   -
174   - resultHolder.invokeAllResult(msg);
175   - if (hookEvent != null) {
176   - hookEvent.response(mediaServerItem, JSONObject.parseObject(JSON.toJSONString(streamInfo)));
  167 + int localPort = rtpInfo.getInteger("local_port");
  168 + if (localPort == 0) {
  169 + logger.warn("[点播]，点播时发现rtpServerC存在，但是尚未开始推流");
  170 + // 此时说明rtpServer已经创建但是流还没有推上来
  171 + WVPResult wvpResult = new WVPResult();
  172 + wvpResult.setCode(ErrorCode.ERROR100.getCode());
  173 + wvpResult.setMsg("点播已经在进行中，请稍候重试");
  174 + msg.setData(wvpResult);
  175 +
  176 + resultHolder.invokeAllResult(msg);
  177 + return playResult;
  178 + }else {
  179 + WVPResult wvpResult = new WVPResult();
  180 + wvpResult.setCode(ErrorCode.SUCCESS.getCode());
  181 + wvpResult.setMsg(ErrorCode.SUCCESS.getMsg());
  182 + wvpResult.setData(streamInfo);
  183 + msg.setData(wvpResult);
  184 +
  185 + resultHolder.invokeAllResult(msg);
  186 + if (hookEvent != null) {
  187 + hookEvent.response(mediaServerItem, JSONObject.parseObject(JSON.toJSONString(streamInfo)));
  188 + }
177 189 }
  190 +
178 191 }else {
179 192 redisCatchStorage.stopPlay(streamInfo);
180 193 storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
... ... @@ -187,7 +200,6 @@ public class PlayServiceImpl implements IPlayService {
187 200 streamInfo = null;
188 201  
189 202 }
190   -
191 203 }
192 204 if (streamInfo == null) {
193 205 String streamId = null;
... ...
src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java
  View file @ec0ec5e
... ... @@ -143,15 +143,12 @@ public interface DeviceChannelMapper {
143 143 @Update(value = {"UPDATE device_channel SET status=0 WHERE deviceId=#{deviceId}"})
144 144 void offlineByDeviceId(String deviceId);
145 145  
146   - @Update(value = {"UPDATE device_channel SET status=1 WHERE deviceId=#{deviceId} AND channelId=#{channelId}"})
147   - void online(String deviceId, String channelId);
148   -
149 146 @Insert("<script> " +
150 147 "insert into device_channel " +
151 148 "(channelId, deviceId, name, manufacture, model, owner, civilCode, block, subCount, " +
152 149 " address, parental, parentId, safetyWay, registerWay, certNum, certifiable, errCode, secrecy, " +
153 150 " ipAddress, port, password, PTZType, status, streamId, longitude, latitude, longitudeGcj02, latitudeGcj02, " +
154   - " longitudeWgs84, latitudeWgs84, createTime, updateTime, businessGroupId, gpsTime) " +
  151 + " longitudeWgs84, latitudeWgs84, hasAudio, createTime, updateTime, businessGroupId, gpsTime) " +
155 152 "values " +
156 153 "<foreach collection='addChannels' index='index' item='item' separator=','> " +
157 154 "('${item.channelId}', '${item.deviceId}', '${item.name}', '${item.manufacture}', '${item.model}', " +
... ... @@ -160,7 +157,7 @@ public interface DeviceChannelMapper {
160 157 "'${item.certNum}', ${item.certifiable}, ${item.errCode}, '${item.secrecy}', " +
161 158 "'${item.ipAddress}', ${item.port}, '${item.password}', ${item.PTZType}, ${item.status}, " +
162 159 "'${item.streamId}', ${item.longitude}, ${item.latitude},${item.longitudeGcj02}, " +
163   - "${item.latitudeGcj02},${item.longitudeWgs84}, ${item.latitudeWgs84},'${item.createTime}', '${item.updateTime}', " +
  160 + "${item.latitudeGcj02},${item.longitudeWgs84}, ${item.latitudeWgs84}, ${item.hasAudio},'${item.createTime}', '${item.updateTime}', " +
164 161 "'${item.businessGroupId}', '${item.gpsTime}') " +
165 162 "</foreach> " +
166 163 "ON DUPLICATE KEY UPDATE " +
... ... @@ -193,11 +190,15 @@ public interface DeviceChannelMapper {
193 190 "latitudeGcj02=VALUES(latitudeGcj02), " +
194 191 "longitudeWgs84=VALUES(longitudeWgs84), " +
195 192 "latitudeWgs84=VALUES(latitudeWgs84), " +
  193 + "hasAudio=VALUES(hasAudio), " +
196 194 "businessGroupId=VALUES(businessGroupId), " +
197 195 "gpsTime=VALUES(gpsTime)" +
198 196 "</script>")
199 197 int batchAdd(List<DeviceChannel> addChannels);
200 198  
  199 + @Update(value = {"UPDATE device_channel SET status=1 WHERE deviceId=#{deviceId} AND channelId=#{channelId}"})
  200 + void online(String deviceId, String channelId);
  201 +
201 202 @Update({"<script>" +
202 203 "<foreach collection='updateChannels' item='item' separator=';'>" +
203 204 " UPDATE" +
... ... @@ -341,4 +342,7 @@ public interface DeviceChannelMapper {
341 342 " left join platform_catalog pc on pgc.catalogId = pc.id and pgc.platformId = pc.platformId" +
342 343 " where pgc.platformId=#{serverGBId}")
343 344 List<DeviceChannel> queryChannelWithCatalog(String serverGBId);
  345 +
  346 + @Select("select * from device_channel where deviceId = #{deviceId}")
  347 + List<DeviceChannel> queryAllChannels(String deviceId);
344 348 }
... ...
src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java
  View file @ec0ec5e
... ... @@ -111,11 +111,11 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage {
111 111 if (CollectionUtils.isEmpty(deviceChannelList)) {
112 112 return false;
113 113 }
114   - List<DeviceChannel> allChannelInPlay = deviceChannelMapper.getAllChannelInPlay();
115   - Map<String,DeviceChannel> allChannelMapInPlay = new ConcurrentHashMap<>();
116   - if (allChannelInPlay.size() > 0) {
117   - for (DeviceChannel deviceChannel : allChannelInPlay) {
118   - allChannelMapInPlay.put(deviceChannel.getChannelId(), deviceChannel);
  114 + List<DeviceChannel> allChannels = deviceChannelMapper.queryAllChannels(deviceId);
  115 + Map<String,DeviceChannel> allChannelMap = new ConcurrentHashMap<>();
  116 + if (allChannels.size() > 0) {
  117 + for (DeviceChannel deviceChannel : allChannels) {
  118 + allChannelMap.put(deviceChannel.getChannelId(), deviceChannel);
119 119 }
120 120 }
121 121 TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
... ... @@ -123,15 +123,17 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage {
123 123 List<DeviceChannel> channels = new ArrayList<>();
124 124 StringBuilder stringBuilder = new StringBuilder();
125 125 Map<String, Integer> subContMap = new HashMap<>();
126   - if (deviceChannelList.size() > 1) {
  126 + if (deviceChannelList.size() > 0) {
127 127 // 数据去重
128 128 Set<String> gbIdSet = new HashSet<>();
129 129 for (DeviceChannel deviceChannel : deviceChannelList) {
130 130 if (!gbIdSet.contains(deviceChannel.getChannelId())) {
131 131 gbIdSet.add(deviceChannel.getChannelId());
132   - if (allChannelMapInPlay.containsKey(deviceChannel.getChannelId())) {
133   - deviceChannel.setStreamId(allChannelMapInPlay.get(deviceChannel.getChannelId()).getStreamId());
  132 + if (allChannelMap.containsKey(deviceChannel.getChannelId())) {
  133 + deviceChannel.setStreamId(allChannelMap.get(deviceChannel.getChannelId()).getStreamId());
  134 + deviceChannel.setHasAudio(allChannelMap.get(deviceChannel.getChannelId()).isHasAudio());
134 135 }
  136 +
135 137 channels.add(deviceChannel);
136 138 if (!ObjectUtils.isEmpty(deviceChannel.getParentId())) {
137 139 if (subContMap.get(deviceChannel.getParentId()) == null) {
... ... @@ -153,8 +155,6 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage {
153 155 }
154 156 }
155 157  
156   - }else {
157   - channels = deviceChannelList;
158 158 }
159 159 if (stringBuilder.length() > 0) {
160 160 logger.info("[目录查询]收到的数据存在重复： {}" , stringBuilder);
... ...
src/main/java/com/genersoft/iot/vmp/vmanager/user/UserController.java
  View file @ec0ec5e
... ... @@ -43,6 +43,7 @@ public class UserController {
43 43 private IRoleService roleService;
44 44  
45 45 @GetMapping("/login")
  46 + @PostMapping("/login")
46 47 @Operation(summary = "登录")
47 48 @Parameter(name = "username", description = "用户名", required = true)
48 49 @Parameter(name = "password", description = "密码（32位md5加密）", required = true)
... ...
src/main/resources/logback-spring-local.xml
  View file @ec0ec5e
... ... @@ -98,6 +98,11 @@
98 98 <appender-ref ref="STDOUT" />
99 99 </root>
100 100  
  101 + <logger name="wvp" level="debug" additivity="true">
  102 + <appender-ref ref="RollingFileError"/>
  103 + <appender-ref ref="RollingFile"/>
  104 + </logger>
  105 +
101 106 <logger name="GB28181_SIP" level="debug" additivity="true">
102 107 <appender-ref ref="RollingFileError"/>
103 108 <appender-ref ref="sipRollingFile"/>
... ...