XW-AIOT/wvp-platform
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch '648540858:wvp-28181-2.0' into wvp-28181-2.0

Browse Source
This commit is contained in:
hotcoffie
2022-05-06 10:59:33 +08:00
committed by GitHub
parent c592740efa c286ecb455
commit bfaa519e49
61 changed files with 596 additions and 393 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java
View File

@@ -12,6 +12,8 @@ import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
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.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import javax.sip.DialogState;
@@ -43,7 +45,7 @@ public class DeviceServiceImpl implements IDeviceService {
}
logger.info("[添加目录订阅] 设备{}", device.getDeviceId());
// 添加目录订阅
CatalogSubscribeTask catalogSubscribeTask = new CatalogSubscribeTask(device, sipCommander);
CatalogSubscribeTask catalogSubscribeTask = new CatalogSubscribeTask(device, sipCommander, dynamicTask);
// 提前开始刷新订阅
int subscribeCycleForCatalog = Math.max(device.getSubscribeCycleForCatalog(),30);
// 设置最小值为30
@@ -68,10 +70,10 @@ public class DeviceServiceImpl implements IDeviceService {
}
logger.info("[添加移动位置订阅] 设备{}", device.getDeviceId());
// 添加目录订阅
MobilePositionSubscribeTask mobilePositionSubscribeTask = new MobilePositionSubscribeTask(device, sipCommander);
// 提前开始刷新订阅
MobilePositionSubscribeTask mobilePositionSubscribeTask = new MobilePositionSubscribeTask(device, sipCommander, dynamicTask);
// 设置最小值为30
int subscribeCycleForCatalog = Math.max(device.getSubscribeCycleForMobilePosition(),30);
// 提前开始刷新订阅
dynamicTask.startCron(device.getDeviceId() + "mobile_position" , mobilePositionSubscribeTask, subscribeCycleForCatalog -1 );
return true;
}

4
src/main/java/com/genersoft/iot/vmp/service/impl/GbStreamServiceImpl.java
View File

@@ -70,7 +70,9 @@ public class GbStreamServiceImpl implements IGbStreamService {
boolean result = false;
TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
ParentPlatform parentPlatform = platformMapper.getParentPlatByServerGBId(platformId);
if (catalogId == null) catalogId = parentPlatform.getCatalogId();
if (catalogId == null) {
catalogId = parentPlatform.getCatalogId();
}
try {
List<DeviceChannel> deviceChannelList = new ArrayList<>();
for (GbStream gbStream : gbStreams) {

4
src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java
View File

@@ -54,7 +54,9 @@ public class MediaServiceImpl implements IMediaService {
if (mediaList != null) {
if (mediaList.getInteger("code") == 0) {
JSONArray data = mediaList.getJSONArray("data");
if (data == null) return null;
if (data == null) {
return null;
}
JSONObject mediaJSON = JSON.parseObject(JSON.toJSONString(data.get(0)), JSONObject.class);
JSONArray tracks = mediaJSON.getJSONArray("tracks");
streamInfo = getStreamInfoByAppAndStream(mediaInfo, app, stream, tracks);

147
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
View File

@@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
@@ -83,6 +84,9 @@ public class PlayServiceImpl implements IPlayService {
@Autowired
private UserSetting userSetting;
@Autowired
private DynamicTask dynamicTask;
@@ -97,7 +101,7 @@ public class PlayServiceImpl implements IPlayService {
String uuid = UUID.randomUUID().toString();
msg.setId(uuid);
playResult.setUuid(uuid);
DeferredResult<ResponseEntity<String>> result = new DeferredResult<>(userSetting.getPlayTimeout());
DeferredResult<ResponseEntity<String>> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue());
playResult.setResult(result);
// 录像查询以channelId作为deviceId查询
resultHolder.put(key, uuid, result);
@@ -234,30 +238,27 @@ public class PlayServiceImpl implements IPlayService {
}
// 超时处理
Timer timer = new Timer();
String timeOutTaskKey = UUID.randomUUID().toString();
SSRCInfo finalSsrcInfo = ssrcInfo;
timer.schedule(new TimerTask() {
@Override
public void run() {
logger.warn(String.format("设备点播超时，deviceId：%s ，channelId：%s", device.getDeviceId(), channelId));
dynamicTask.startDelay( timeOutTaskKey,()->{
logger.warn(String.format("设备点播超时，deviceId：%s ，channelId：%s", device.getDeviceId(), channelId));
SIPDialog dialog = streamSession.getDialogByStream(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
if (dialog != null) {
timeoutCallback.run(1, "收流超时");
// 点播超时回复BYE 同时释放ssrc以及此次点播的资源
cmder.streamByeCmd(device.getDeviceId(), channelId, finalSsrcInfo.getStream(), null);
}else {
timeoutCallback.run(0, "点播超时");
mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc());
mediaServerService.closeRTPServer(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
streamSession.remove(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
}
SIPDialog dialog = streamSession.getDialogByStream(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
if (dialog != null) {
timeoutCallback.run(1, "收流超时");
// 点播超时回复BYE 同时释放ssrc以及此次点播的资源
cmder.streamByeCmd(device.getDeviceId(), channelId, finalSsrcInfo.getStream(), null);
}else {
timeoutCallback.run(0, "点播超时");
mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc());
mediaServerService.closeRTPServer(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
streamSession.remove(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
}
}, userSetting.getPlayTimeout());
}, userSetting.getPlayTimeout()*1000);
final String ssrc = ssrcInfo.getSsrc();
cmder.playStreamCmd(mediaServerItem, ssrcInfo, device, channelId, (MediaServerItem mediaServerItemInuse, JSONObject response) -> {
logger.info("收到订阅消息： " + response.toJSONString());
timer.cancel();
dynamicTask.stop(timeOutTaskKey);
// hook响应
onPublishHandlerForPlay(mediaServerItemInuse, response, device.getDeviceId(), channelId, uuid);
hookEvent.response(mediaServerItemInuse, response);
@@ -289,7 +290,7 @@ public class PlayServiceImpl implements IPlayService {
}
}
}, (event) -> {
timer.cancel();
dynamicTask.stop(timeOutTaskKey);
mediaServerService.closeRTPServer(device.getDeviceId(), channelId, finalSsrcInfo.getStream());
// 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), finalSsrcInfo.getSsrc());
@@ -331,7 +332,9 @@ public class PlayServiceImpl implements IPlayService {
@Override
public MediaServerItem getNewMediaServerItem(Device device) {
if (device == null) return null;
if (device == null) {
return null;
}
String mediaServerId = device.getMediaServerId();
MediaServerItem mediaServerItem;
if (mediaServerId == null) {
@@ -350,7 +353,9 @@ public class PlayServiceImpl implements IPlayService {
String endTime,InviteStreamCallback inviteStreamCallback,
PlayBackCallback callback) {
Device device = storager.queryVideoDevice(deviceId);
if (device == null) return null;
if (device == null) {
return null;
}
MediaServerItem newMediaServerItem = getNewMediaServerItem(device);
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, true);
@@ -362,7 +367,9 @@ public class PlayServiceImpl implements IPlayService {
String deviceId, String channelId, String startTime,
String endTime, InviteStreamCallback infoCallBack,
PlayBackCallback playBackCallback) {
if (mediaServerItem == null || ssrcInfo == null) return null;
if (mediaServerItem == null || ssrcInfo == null) {
return null;
}
String uuid = UUID.randomUUID().toString();
String key = DeferredResultHolder.CALLBACK_CMD_PLAYBACK + deviceId + channelId;
DeferredResult<ResponseEntity<String>> result = new DeferredResult<>(30000L);
@@ -377,34 +384,31 @@ public class PlayServiceImpl implements IPlayService {
msg.setId(uuid);
msg.setKey(key);
PlayBackResult<RequestMessage> playBackResult = new PlayBackResult<>();
Timer timer = new Timer();
timer.schedule(new TimerTask() {
@Override
public void run() {
logger.warn(String.format("设备回放超时，deviceId：%s ，channelId：%s", deviceId, channelId));
playBackResult.setCode(-1);
playBackResult.setData(msg);
playBackCallback.call(playBackResult);
SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, ssrcInfo.getStream());
String playBackTimeOutTaskKey = UUID.randomUUID().toString();
dynamicTask.startDelay(playBackTimeOutTaskKey, ()->{
logger.warn(String.format("设备回放超时，deviceId：%s ，channelId：%s", deviceId, channelId));
playBackResult.setCode(-1);
playBackResult.setData(msg);
playBackCallback.call(playBackResult);
SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, ssrcInfo.getStream());
// 点播超时回复BYE 同时释放ssrc以及此次点播的资源
if (dialog != null) {
// 点播超时回复BYE 同时释放ssrc以及此次点播的资源
if (dialog != null) {
// 点播超时回复BYE 同时释放ssrc以及此次点播的资源
cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream(), null);
}else {
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
mediaServerService.closeRTPServer(deviceId, channelId, ssrcInfo.getStream());
streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
}
cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream(), null);
// 回复之前所有的点播请求
playBackCallback.call(playBackResult);
}else {
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
mediaServerService.closeRTPServer(deviceId, channelId, ssrcInfo.getStream());
streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
}
}, userSetting.getPlayTimeout());
cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream(), null);
// 回复之前所有的点播请求
playBackCallback.call(playBackResult);
}, userSetting.getPlayTimeout()*1000);
cmder.playbackStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, infoCallBack,
(InviteStreamInfo inviteStreamInfo) -> {
logger.info("收到订阅消息： " + inviteStreamInfo.getResponse().toJSONString());
timer.cancel();
dynamicTask.stop(playBackTimeOutTaskKey);
StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId);
if (streamInfo == null) {
logger.warn("设备回放API调用失败！");
@@ -422,7 +426,7 @@ public class PlayServiceImpl implements IPlayService {
playBackResult.setResponse(inviteStreamInfo.getResponse());
playBackCallback.call(playBackResult);
}, event -> {
timer.cancel();
dynamicTask.stop(playBackTimeOutTaskKey);
msg.setData(String.format("回放失败， 错误码： %s, %s", event.statusCode, event.msg));
playBackResult.setCode(-1);
playBackResult.setData(msg);
@@ -436,7 +440,9 @@ public class PlayServiceImpl implements IPlayService {
@Override
public DeferredResult<ResponseEntity<String>> download(String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, InviteStreamCallback infoCallBack, PlayBackCallback hookCallBack) {
Device device = storager.queryVideoDevice(deviceId);
if (device == null) return null;
if (device == null) {
return null;
}
MediaServerItem newMediaServerItem = getNewMediaServerItem(device);
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, true);
@@ -445,7 +451,9 @@ public class PlayServiceImpl implements IPlayService {
@Override
public DeferredResult<ResponseEntity<String>> download(MediaServerItem mediaServerItem, SSRCInfo ssrcInfo, String deviceId, String channelId, String startTime, String endTime, int downloadSpeed, InviteStreamCallback infoCallBack, PlayBackCallback hookCallBack) {
if (mediaServerItem == null || ssrcInfo == null) return null;
if (mediaServerItem == null || ssrcInfo == null) {
return null;
}
String uuid = UUID.randomUUID().toString();
String key = DeferredResultHolder.CALLBACK_CMD_DOWNLOAD + deviceId + channelId;
DeferredResult<ResponseEntity<String>> result = new DeferredResult<>(30000L);
@@ -464,34 +472,31 @@ public class PlayServiceImpl implements IPlayService {
PlayBackResult<RequestMessage> downloadResult = new PlayBackResult<>();
downloadResult.setData(msg);
Timer timer = new Timer();
timer.schedule(new TimerTask() {
@Override
public void run() {
logger.warn(String.format("录像下载请求超时，deviceId：%s ，channelId：%s", deviceId, channelId));
wvpResult.setCode(-1);
wvpResult.setMsg("录像下载请求超时");
downloadResult.setCode(-1);
hookCallBack.call(downloadResult);
SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, ssrcInfo.getStream());
String downLoadTimeOutTaskKey = UUID.randomUUID().toString();
dynamicTask.startDelay(downLoadTimeOutTaskKey, ()->{
logger.warn(String.format("录像下载请求超时，deviceId：%s ，channelId：%s", deviceId, channelId));
wvpResult.setCode(-1);
wvpResult.setMsg("录像下载请求超时");
downloadResult.setCode(-1);
hookCallBack.call(downloadResult);
SIPDialog dialog = streamSession.getDialogByStream(deviceId, channelId, ssrcInfo.getStream());
// 点播超时回复BYE 同时释放ssrc以及此次点播的资源
if (dialog != null) {
// 点播超时回复BYE 同时释放ssrc以及此次点播的资源
if (dialog != null) {
// 点播超时回复BYE 同时释放ssrc以及此次点播的资源
cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream(), null);
}else {
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
mediaServerService.closeRTPServer(deviceId, channelId, ssrcInfo.getStream());
streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
}
cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream(), null);
// 回复之前所有的点播请求
hookCallBack.call(downloadResult);
}else {
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcInfo.getSsrc());
mediaServerService.closeRTPServer(deviceId, channelId, ssrcInfo.getStream());
streamSession.remove(deviceId, channelId, ssrcInfo.getStream());
}
}, userSetting.getPlayTimeout());
cmder.streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream(), null);
// 回复之前所有的点播请求
hookCallBack.call(downloadResult);
}, userSetting.getPlayTimeout()*1000);
cmder.downloadStreamCmd(mediaServerItem, ssrcInfo, device, channelId, startTime, endTime, downloadSpeed, infoCallBack,
inviteStreamInfo -> {
logger.info("收到订阅消息： " + inviteStreamInfo.getResponse().toJSONString());
timer.cancel();
dynamicTask.stop(downLoadTimeOutTaskKey);
StreamInfo streamInfo = onPublishHandler(inviteStreamInfo.getMediaServerItem(), inviteStreamInfo.getResponse(), deviceId, channelId);
streamInfo.setStartTime(startTime);
streamInfo.setEndTime(endTime);
@@ -512,7 +517,7 @@ public class PlayServiceImpl implements IPlayService {
downloadResult.setResponse(inviteStreamInfo.getResponse());
hookCallBack.call(downloadResult);
}, event -> {
timer.cancel();
dynamicTask.stop(downLoadTimeOutTaskKey);
downloadResult.setCode(-1);
wvpResult.setCode(-1);
wvpResult.setMsg(String.format("录像下载失败， 错误码： %s, %s", event.statusCode, event.msg));

8
src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
View File

@@ -194,7 +194,9 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
@Override
public JSONObject removeStreamProxyFromZlm(StreamProxyItem param) {
if (param ==null) return null;
if (param ==null) {
return null;
}
MediaServerItem mediaServerItem = mediaServerService.getOne(param.getMediaServerId());
JSONObject result = zlmresTfulUtils.closeStreams(mediaServerItem, param.getApp(), param.getStream());
return result;
@@ -230,7 +232,9 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
StreamProxyItem streamProxy = videoManagerStorager.queryStreamProxy(app, stream);
if (!streamProxy.isEnable() && streamProxy != null) {
JSONObject jsonObject = addStreamProxyToZlm(streamProxy);
if (jsonObject == null) return false;
if (jsonObject == null) {
return false;
}
if (jsonObject.getInteger("code") == 0) {
result = true;
streamProxy.setEnable(true);

8
src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java
View File

@@ -66,7 +66,9 @@ public class StreamPushServiceImpl implements IStreamPushService {
@Override
public List<StreamPushItem> handleJSON(String jsonData, MediaServerItem mediaServerItem) {
if (jsonData == null) return null;
if (jsonData == null) {
return null;
}
Map<String, StreamPushItem> result = new HashMap<>();
@@ -219,7 +221,9 @@ public class StreamPushServiceImpl implements IStreamPushService {
}
}
zlmresTfulUtils.getMediaList(mediaServerItem, (mediaList ->{
if (mediaList == null) return;
if (mediaList == null) {
return;
}
String dataStr = mediaList.getString("data");
Integer code = mediaList.getInteger("code");

4
src/main/java/com/genersoft/iot/vmp/service/impl/UserServiceImpl.java
View File

@@ -34,7 +34,9 @@ public class UserServiceImpl implements IUserService {
@Override
public int addUser(User user) {
User userByUsername = userMapper.getUserByUsername(user.getUsername());
if (userByUsername != null) return 0;
if (userByUsername != null) {
return 0;
}
return userMapper.add(user);
}
@Override