Shikong/wvp-gb28181-project
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/wvp-28181-2.0' into map

Browse Source 
# Conflicts:
#	src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/SubscribeListenerForPlatform.java
This commit is contained in:
648540858 2022-04-15 09:37:33 +08:00
commit 40a806329c
39 changed files with 1024 additions and 340 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
README.md
View File

@ -130,5 +130,14 @@ QQ私信一般不回, 精力有限.欢迎大家在群里讨论.觉得项目对
# 致谢
感谢作者[夏楚](https://github.com/xia-chu) 提供这么棒的开源流媒体服务框架
感谢作者[夏楚](https://github.com/xia-chu) 提供这么棒的开源流媒体服务框架,并在开发过程中给予支持与帮助。
感谢作者[dexter langhuihui](https://github.com/langhuihui) 开源这么好用的WEB播放器。
感谢作者[Kyle](https://gitee.com/kkkkk5G) 开源了好用的前端页面
感谢各位大佬的赞助以及对项目的指正与帮助。包括但不限于代码贡献、问题反馈、资金捐赠等各种方式的支持！以下排名不分先后：
[lawrencehj](https://github.com/lawrencehj) @陆丰-创奇科技 [swwhaha](https://github.com/swwheihei)
[hotcoffie](https://github.com/hotcoffie) [xiaomu](https://github.com/nikmu) [TristingChen](https://github.com/TristingChen)
[chenparty](https://github.com/chenparty) [Hotleave](https://github.com/hotleave) [ydwxb](https://github.com/ydwxb)
[ydpd](https://github.com/ydpd) [szy833](https://github.com/szy833) [ydwxb](https://github.com/ydwxb)
ps: 刚增加了这个名单，肯定遗漏了一些大佬，欢迎大佬联系我添加。

4
src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java
View File

@ -22,6 +22,9 @@ public class VideoManagerConstants {
public static final String DEVICE_PREFIX = "VMP_DEVICE_";
// 设备同步完成
public static final String DEVICE_SYNC_PREFIX = "VMP_DEVICE_SYNC_";
public static final String CACHEKEY_PREFIX = "VMP_CHANNEL_";
public static final String KEEPLIVEKEY_PREFIX = "VMP_KEEPALIVE_";
@ -69,6 +72,7 @@ public class VideoManagerConstants {
public static final String SYSTEM_INFO_NET_PREFIX = "VMP_SYSTEM_INFO_NET_";
//************************** redis 消息*********************************
// 流变化的通知

58
src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java
View File

@ -1,6 +1,9 @@
package com.genersoft.iot.vmp.conf;
import com.genersoft.iot.vmp.gb28181.task.ISubscribeTask;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd.CatalogResponseMessageHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@ -18,6 +21,8 @@ import java.util.concurrent.ScheduledFuture;
@Component
public class DynamicTask {
private Logger logger = LoggerFactory.getLogger(DynamicTask.class);
@Autowired
private ThreadPoolTaskScheduler threadPoolTaskScheduler;
@ -26,7 +31,12 @@ public class DynamicTask {
@Bean
public ThreadPoolTaskScheduler threadPoolTaskScheduler() {
return new ThreadPoolTaskScheduler();
ThreadPoolTaskScheduler schedulerPool = new ThreadPoolTaskScheduler();
schedulerPool.setPoolSize(300);
schedulerPool.setWaitForTasksToCompleteOnShutdown(true);
schedulerPool.setAwaitTerminationSeconds(10);
return schedulerPool;
}
/**
@ -37,11 +47,24 @@ public class DynamicTask {
* @return
*/
public void startCron(String key, Runnable task, int cycleForCatalog) {
stop(key);
ScheduledFuture future = futureMap.get(key);
if (future != null) {
if (future.isCancelled()) {
logger.info("任务【{}】已存在但是关闭状态！！！", key);
} else {
logger.info("任务【{}】已存在且已启动！！！", key);
return;
}
}
// scheduleWithFixedDelay 必须等待上一个任务结束才开始计时period， cycleForCatalog表示执行的间隔
ScheduledFuture future = threadPoolTaskScheduler.scheduleWithFixedDelay(task, cycleForCatalog * 1000L);
futureMap.put(key, future);
runnableMap.put(key, task);
future = threadPoolTaskScheduler.scheduleAtFixedRate(task, cycleForCatalog * 1000L);
if (future != null){
futureMap.put(key, future);
runnableMap.put(key, task);
logger.info("任务【{}】启动成功！！！", key);
}else {
logger.info("任务【{}】启动失败！！！", key);
}
}
/**
@ -53,13 +76,31 @@ public class DynamicTask {
*/
public void startDelay(String key, Runnable task, int delay) {
stop(key);
System.out.println("定时任务开始了");
Date starTime = new Date(System.currentTimeMillis() + delay);
ScheduledFuture future = futureMap.get(key);
if (future != null) {
if (future.isCancelled()) {
logger.info("任务【{}】已存在但是关闭状态！！！", key);
} else {
logger.info("任务【{}】已存在且已启动！！！", key);
return;
}
}
// scheduleWithFixedDelay 必须等待上一个任务结束才开始计时period， cycleForCatalog表示执行的间隔
ScheduledFuture future = threadPoolTaskScheduler.schedule(task, starTime);
futureMap.put(key, future);
future = threadPoolTaskScheduler.schedule(task, starTime);
if (future != null){
futureMap.put(key, future);
runnableMap.put(key, task);
logger.info("任务【{}】启动成功！！！", key);
}else {
logger.info("任务【{}】启动失败！！！", key);
}
}
public void stop(String key) {
System.out.println("定时任务结束了");
if (futureMap.get(key) != null && !futureMap.get(key).isCancelled()) {
futureMap.get(key).cancel(true);
Runnable runnable = runnableMap.get(key);
@ -78,4 +119,7 @@ public class DynamicTask {
return futureMap.keySet();
}
public Runnable get(String key) {
return runnableMap.get(key);
}
}

22
src/main/java/com/genersoft/iot/vmp/gb28181/bean/CatalogData.java
View File

@ -8,6 +8,12 @@ public class CatalogData {
private List<DeviceChannel> channelList;
private Date lastTime;
private Device device;
private String errorMsg;
public enum CatalogDataStatus{
ready, runIng, end
}
private CatalogDataStatus status;
public int getTotal() {
return total;
@ -40,4 +46,20 @@ public class CatalogData {
public void setDevice(Device device) {
this.device = device;
}
public String getErrorMsg() {
return errorMsg;
}
public void setErrorMsg(String errorMsg) {
this.errorMsg = errorMsg;
}
public CatalogDataStatus getStatus() {
return status;
}
public void setStatus(CatalogDataStatus status) {
this.status = status;
}
}

51
src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeHolder.java
View File

@ -1,5 +1,12 @@
package com.genersoft.iot.vmp.gb28181.bean;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.gb28181.task.impl.MobilePositionSubscribeHandlerTask;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
@ -9,12 +16,32 @@ import java.util.concurrent.ConcurrentHashMap;
@Component
public class SubscribeHolder {
@Autowired
private DynamicTask dynamicTask;
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Autowired
private ISIPCommanderForPlatform sipCommanderForPlatform;
@Autowired
private IVideoManagerStorage storager;
private final String taskOverduePrefix = "subscribe_overdue_";
private static ConcurrentHashMap<String, SubscribeInfo> catalogMap = new ConcurrentHashMap<>();
private static ConcurrentHashMap<String, SubscribeInfo> mobilePositionMap = new ConcurrentHashMap<>();
public void putCatalogSubscribe(String platformId, SubscribeInfo subscribeInfo) {
catalogMap.put(platformId, subscribeInfo);
// 添加订阅到期
String taskOverdueKey = taskOverduePrefix + "catalog_" + platformId;
dynamicTask.stop(taskOverdueKey);
// 添加任务处理订阅过期
dynamicTask.startDelay(taskOverdueKey, () -> removeCatalogSubscribe(subscribeInfo.getId()),
subscribeInfo.getExpires() * 1000);
}
public SubscribeInfo getCatalogSubscribe(String platformId) {
@ -23,10 +50,24 @@ public class SubscribeHolder {
public void removeCatalogSubscribe(String platformId) {
catalogMap.remove(platformId);
String taskOverdueKey = taskOverduePrefix + "catalog_" + platformId;
// 添加任务处理订阅过期
dynamicTask.stop(taskOverdueKey);
}
public void putMobilePositionSubscribe(String platformId, SubscribeInfo subscribeInfo) {
mobilePositionMap.put(platformId, subscribeInfo);
String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + "MobilePosition_" + platformId;
// 添加任务处理GPS定时推送
dynamicTask.startCron(key, new MobilePositionSubscribeHandlerTask(redisCatchStorage, sipCommanderForPlatform, storager, platformId, subscribeInfo.getSn(), key, this), subscribeInfo.getGpsInterval());
String taskOverdueKey = taskOverduePrefix + "MobilePosition_" + platformId;
dynamicTask.stop(taskOverdueKey);
// 添加任务处理订阅过期
dynamicTask.startDelay(taskOverdueKey, () -> {
System.out.println("订阅过期");
removeMobilePositionSubscribe(subscribeInfo.getId());
},
subscribeInfo.getExpires() * 1000);
}
public SubscribeInfo getMobilePositionSubscribe(String platformId) {
@ -35,6 +76,12 @@ public class SubscribeHolder {
public void removeMobilePositionSubscribe(String platformId) {
mobilePositionMap.remove(platformId);
String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + "MobilePosition_" + platformId;
// 结束任务处理GPS定时推送
dynamicTask.stop(key);
String taskOverdueKey = taskOverduePrefix + "MobilePosition_" + platformId;
// 添加任务处理订阅过期
dynamicTask.stop(taskOverdueKey);
}
public List<String> getAllCatalogSubscribePlatform() {
@ -48,7 +95,7 @@ public class SubscribeHolder {
}
public void removeAllSubscribe(String platformId) {
mobilePositionMap.remove(platformId);
catalogMap.remove(platformId);
removeMobilePositionSubscribe(platformId);
removeCatalogSubscribe(platformId);
}
}

24
src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeInfo.java
View File

@ -33,6 +33,14 @@ public class SubscribeInfo {
private ServerTransaction transaction;
private Dialog dialog;
/**
* 以下为可选字段
* @return
*/
private String sn;
private int gpsInterval;
public String getId() {
return id;
}
@ -88,4 +96,20 @@ public class SubscribeInfo {
public void setDialog(Dialog dialog) {
this.dialog = dialog;
}
public String getSn() {
return sn;
}
public void setSn(String sn) {
this.sn = sn;
}
public int getGpsInterval() {
return gpsInterval;
}
public void setGpsInterval(int gpsInterval) {
this.gpsInterval = gpsInterval;
}
}

34
src/main/java/com/genersoft/iot/vmp/gb28181/bean/SyncStatus.java Normal file
View File

@ -0,0 +1,34 @@
package com.genersoft.iot.vmp.gb28181.bean;
/**
* 摄像机同步状态
*/
public class SyncStatus {
private int total;
private int current;
private String errorMsg;
public int getTotal() {
return total;
}
public void setTotal(int total) {
this.total = total;
}
public int getCurrent() {
return current;
}
public void setCurrent(int current) {
this.current = current;
}
public String getErrorMsg() {
return errorMsg;
}
public void setErrorMsg(String errorMsg) {
this.errorMsg = errorMsg;
}
}

5
src/main/java/com/genersoft/iot/vmp/gb28181/event/online/OnlineEventListener.java
View File

@ -95,11 +95,12 @@ public class OnlineEventListener implements ApplicationListener<OnlineEvent> {
}
// 处理上线监听
storager.updateDevice(device);
List<DeviceChannel> deviceChannelList = storager.queryOnlineChannelsByDeviceId(device.getDeviceId());
eventPublisher.catalogEventPublish(null, deviceChannelList, CatalogEvent.ON);
// 上线添加订阅
if (device.getSubscribeCycleForCatalog() > 0) {
// 查询在线设备那些开启了订阅，为设备开启定时的目录订阅
deviceService.addCatalogSubscribe(device);
}
if (device.getSubscribeCycleForMobilePosition() > 0) {
deviceService.addMobilePositionSubscribe(device);
}
}

49
src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/SubscribeListenerForPlatform.java
View File

@ -1,49 +0,0 @@
package com.genersoft.iot.vmp.gb28181.event.subscribe;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.RedisKeyExpirationEventMessageListener;
import com.genersoft.iot.vmp.conf.UserSetting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.stereotype.Component;
/**
* 平台订阅到期事件
*/
@Component
public class SubscribeListenerForPlatform extends RedisKeyExpirationEventMessageListener {
private Logger logger = LoggerFactory.getLogger(SubscribeListenerForPlatform.class);
@Autowired
private UserSetting userSetting;
@Autowired
private DynamicTask dynamicTask;
public SubscribeListenerForPlatform(RedisMessageListenerContainer listenerContainer, UserSetting userSetting) {
super(listenerContainer, userSetting);
}
/**
* 监听失效的key
* @param message
* @param pattern
*/
@Override
public void onMessage(Message message, byte[] pattern) {
// 获取失效的key
String expiredKey = message.toString();
// 订阅到期
String PLATFORM_KEEPLIVEKEY_PREFIX = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetting.getServerId() + "_";
if (expiredKey.startsWith(PLATFORM_KEEPLIVEKEY_PREFIX)) {
// 取消定时任务
dynamicTask.stop(expiredKey);
}
}
}

2
src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/catalog/CatalogEventLister.java
View File

@ -61,8 +61,6 @@ public class CatalogEventLister implements ApplicationListener<CatalogEvent> {
if (event.getPlatformId() != null) {
parentPlatform = storager.queryParentPlatByServerGBId(event.getPlatformId());
if (parentPlatform != null && !parentPlatform.isStatus())return;
String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetting.getServerId() + "_Catalog_" + event.getPlatformId();
// subscribe = redisCatchStorage.getSubscribe(key);
subscribe = subscribeHolder.getCatalogSubscribe(event.getPlatformId());
if (subscribe == null) {

72
src/main/java/com/genersoft/iot/vmp/gb28181/session/CatalogDataCatch.java
View File

@ -3,6 +3,7 @@ package com.genersoft.iot.vmp.gb28181.session;
import com.genersoft.iot.vmp.gb28181.bean.CatalogData;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.SyncStatus;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
@ -25,6 +26,17 @@ public class CatalogDataCatch {
@Autowired
private IVideoManagerStorage storager;
public void addReady(String key) {
CatalogData catalogData = data.get(key);
if (catalogData == null || catalogData.getStatus().equals(CatalogData.CatalogDataStatus.end)) {
catalogData = new CatalogData();
catalogData.setChannelList(new ArrayList<>());
catalogData.setStatus(CatalogData.CatalogDataStatus.ready);
catalogData.setLastTime(new Date(System.currentTimeMillis()));
data.put(key, catalogData);
}
}
public void put(String key, int total, Device device, List<DeviceChannel> deviceChannelList) {
CatalogData catalogData = data.get(key);
if (catalogData == null) {
@ -32,10 +44,16 @@ public class CatalogDataCatch {
catalogData.setTotal(total);
catalogData.setDevice(device);
catalogData.setChannelList(new ArrayList<>());
catalogData.setStatus(CatalogData.CatalogDataStatus.runIng);
catalogData.setLastTime(new Date(System.currentTimeMillis()));
data.put(key, catalogData);
}else {
catalogData.setTotal(total);
catalogData.setDevice(device);
catalogData.setStatus(CatalogData.CatalogDataStatus.runIng);
catalogData.getChannelList().addAll(deviceChannelList);
catalogData.setLastTime(new Date(System.currentTimeMillis()));
}
catalogData.getChannelList().addAll(deviceChannelList);
catalogData.setLastTime(new Date(System.currentTimeMillis()));
}
public List<DeviceChannel> get(String key) {
@ -44,6 +62,22 @@ public class CatalogDataCatch {
return catalogData.getChannelList();
}
public int getTotal(String key) {
CatalogData catalogData = data.get(key);
if (catalogData == null) return 0;
return catalogData.getTotal();
}
public SyncStatus getSyncStatus(String key) {
CatalogData catalogData = data.get(key);
if (catalogData == null) return null;
SyncStatus syncStatus = new SyncStatus();
syncStatus.setCurrent(catalogData.getChannelList().size());
syncStatus.setTotal(catalogData.getTotal());
syncStatus.setErrorMsg(catalogData.getErrorMsg());
return syncStatus;
}
public void del(String key) {
data.remove(key);
}
@ -51,24 +85,32 @@ public class CatalogDataCatch {
@Scheduled(fixedRate = 5 * 1000) //每5秒执行一次, 发现数据5秒未更新则移除数据并认为数据接收超时
private void timerTask(){
Set<String> keys = data.keySet();
Calendar calendar = Calendar.getInstance();
calendar.setTime(new Date());
calendar.set(Calendar.SECOND, calendar.get(Calendar.SECOND) - 5);
Calendar calendarBefore5S = Calendar.getInstance();
calendarBefore5S.setTime(new Date());
calendarBefore5S.set(Calendar.SECOND, calendarBefore5S.get(Calendar.SECOND) - 5);
Calendar calendarBefore30S = Calendar.getInstance();
calendarBefore30S.setTime(new Date());
calendarBefore30S.set(Calendar.SECOND, calendarBefore30S.get(Calendar.SECOND) - 30);
for (String key : keys) {
CatalogData catalogData = data.get(key);
if (catalogData.getLastTime().before(calendar.getTime())) {
if (catalogData.getLastTime().before(calendarBefore5S.getTime())) { // 超过五秒收不到消息任务超时， 只更新这一部分数据
storager.resetChannels(catalogData.getDevice().getDeviceId(), catalogData.getChannelList());
RequestMessage msg = new RequestMessage();
msg.setKey(key);
WVPResult<Object> result = new WVPResult<>();
result.setCode(0);
result.setMsg("更新成功，共" + catalogData.getTotal() + "条，已更新" + catalogData.getChannelList().size() + "条");
result.setData(catalogData.getDevice());
msg.setData(result);
deferredResultHolder.invokeAllResult(msg);
String errorMsg = "更新成功，共" + catalogData.getTotal() + "条，已更新" + catalogData.getChannelList().size() + "条";
catalogData.setStatus(CatalogData.CatalogDataStatus.end);
catalogData.setErrorMsg(errorMsg);
}
if (catalogData.getLastTime().before(calendarBefore30S.getTime())) { // 超过三十秒，如果标记为end则删除
data.remove(key);
}
}
}
public void setChannelSyncEnd(String key, String errorMsg) {
CatalogData catalogData = data.get(key);
if (catalogData == null)return;
catalogData.setStatus(CatalogData.CatalogDataStatus.end);
catalogData.setErrorMsg(errorMsg);
}
}

4
src/main/java/com/genersoft/iot/vmp/gb28181/task/ISubscribeTask.java
View File

@ -1,5 +1,9 @@
package com.genersoft.iot.vmp.gb28181.task;
import javax.sip.DialogState;
public interface ISubscribeTask extends Runnable{
void stop();
DialogState getDialogState();
}

7
src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/CatalogSubscribeTask.java
View File

@ -5,6 +5,7 @@ import com.genersoft.iot.vmp.gb28181.task.ISubscribeTask;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
import javax.sip.Dialog;
import javax.sip.DialogState;
@ -72,4 +73,10 @@ public class CatalogSubscribeTask implements ISubscribeTask {
});
}
}
@Override
public DialogState getDialogState() {
if (dialog == null) return null;
return dialog.getState();
}
}

71
src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/MobilePositionSubscribeHandlerTask.java
View File

@ -1,16 +1,16 @@
package com.genersoft.iot.vmp.gb28181.task.impl;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.SubscribeHolder;
import com.genersoft.iot.vmp.gb28181.bean.SubscribeInfo;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.task.ISubscribeTask;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
import java.text.SimpleDateFormat;
import javax.sip.DialogState;
import java.util.List;
/**
@ -18,20 +18,21 @@ import java.util.List;
*/
public class MobilePositionSubscribeHandlerTask implements ISubscribeTask {
private Logger logger = LoggerFactory.getLogger(MobilePositionSubscribeHandlerTask.class);
private IRedisCatchStorage redisCatchStorage;
private IVideoManagerStorage storager;
private ISIPCommanderForPlatform sipCommanderForPlatform;
private SubscribeHolder subscribeHolder;
private String platformId;
private ParentPlatform platform;
private String sn;
private String key;
private final SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
public MobilePositionSubscribeHandlerTask(IRedisCatchStorage redisCatchStorage, ISIPCommanderForPlatform sipCommanderForPlatform, IVideoManagerStorage storager, String platformId, String sn, String key, SubscribeHolder subscribeInfo) {
System.out.println("MobilePositionSubscribeHandlerTask 初始化");
this.redisCatchStorage = redisCatchStorage;
this.storager = storager;
this.platformId = platformId;
this.platform = storager.queryParentPlatByServerGBId(platformId);
this.sn = sn;
this.key = key;
this.sipCommanderForPlatform = sipCommanderForPlatform;
@ -41,37 +42,45 @@ public class MobilePositionSubscribeHandlerTask implements ISubscribeTask {
@Override
public void run() {
SubscribeInfo subscribe = subscribeHolder.getMobilePositionSubscribe(platformId);
logger.info("执行MobilePositionSubscribeHandlerTask");
if (platform == null) return;
SubscribeInfo subscribe = subscribeHolder.getMobilePositionSubscribe(platform.getServerGBId());
if (subscribe != null) {
ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformId);
if (parentPlatform == null || parentPlatform.isStatus()) {
// TODO 暂时只处理视频流的回复,后续增加对国标设备的支持
List<GbStream> gbStreams = storager.queryGbStreamListInPlatform(platformId);
if (gbStreams.size() > 0) {
for (GbStream gbStream : gbStreams) {
String gbId = gbStream.getGbId();
GPSMsgInfo gpsMsgInfo = redisCatchStorage.getGpsMsgInfo(gbId);
if (gpsMsgInfo != null) {
// 发送GPS消息
sipCommanderForPlatform.sendNotifyMobilePosition(parentPlatform, gpsMsgInfo, subscribe);
}else {
// 没有在redis找到新的消息就使用数据库的消息
gpsMsgInfo = new GPSMsgInfo();
gpsMsgInfo.setId(gbId);
gpsMsgInfo.setLat(gbStream.getLongitude());
gpsMsgInfo.setLng(gbStream.getLongitude());
// 发送GPS消息
sipCommanderForPlatform.sendNotifyMobilePosition(parentPlatform, gpsMsgInfo, subscribe);
}
// if (!parentPlatform.isStatus()) {
// logger.info("发送订阅时发现平台已经离线：{}", platformId);
// return;
// }
// TODO 暂时只处理视频流的回复,后续增加对国标设备的支持
List<GbStream> gbStreams = storager.queryGbStreamListInPlatform(platform.getServerGBId());
if (gbStreams.size() == 0) {
logger.info("发送订阅时发现平台已经没有关联的直播流：{}", platform.getServerGBId());
return;
}
for (GbStream gbStream : gbStreams) {
String gbId = gbStream.getGbId();
GPSMsgInfo gpsMsgInfo = redisCatchStorage.getGpsMsgInfo(gbId);
if (gpsMsgInfo != null) { // 无最新位置不发送
logger.info("无最新位置不发送");
// 经纬度都为0不发送
if (gpsMsgInfo.getLng() == 0 && gpsMsgInfo.getLat() == 0) {
continue;
}
// 发送GPS消息
sipCommanderForPlatform.sendNotifyMobilePosition(platform, gpsMsgInfo, subscribe);
}
}
}
logger.info("结束执行MobilePositionSubscribeHandlerTask");
}
@Override
public void stop() {
}
@Override
public DialogState getDialogState() {
return null;
}
}

33
src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/MobilePositionSubscribeTask.java
View File

@ -6,10 +6,13 @@ import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import org.dom4j.Element;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
import javax.sip.Dialog;
import javax.sip.DialogState;
import javax.sip.ResponseEvent;
import java.util.Timer;
import java.util.TimerTask;
/**
* 移动位置订阅的定时更新
@ -20,6 +23,8 @@ public class MobilePositionSubscribeTask implements ISubscribeTask {
private ISIPCommander sipCommander;
private Dialog dialog;
private Timer timer ;
public MobilePositionSubscribeTask(Device device, ISIPCommander sipCommander) {
this.device = device;
this.sipCommander = sipCommander;
@ -27,10 +32,14 @@ public class MobilePositionSubscribeTask implements ISubscribeTask {
@Override
public void run() {
if (timer != null ) {
timer.cancel();
timer = null;
}
sipCommander.mobilePositionSubscribe(device, dialog, eventResult -> {
if (eventResult.dialog != null || eventResult.dialog.getState().equals(DialogState.CONFIRMED)) {
dialog = eventResult.dialog;
}
// if (eventResult.dialog != null || eventResult.dialog.getState().equals(DialogState.CONFIRMED)) {
// dialog = eventResult.dialog;
// }
ResponseEvent event = (ResponseEvent) eventResult.event;
if (event.getResponse().getRawContent() != null) {
// 成功
@ -43,6 +52,13 @@ public class MobilePositionSubscribeTask implements ISubscribeTask {
dialog = null;
// 失败
logger.warn("[移动位置订阅]失败，信令发送失败： {}-{} ", device.getDeviceId(), eventResult.msg);
timer = new Timer();
timer.schedule(new TimerTask() {
@Override
public void run() {
MobilePositionSubscribeTask.this.run();
}
}, 2000);
});
}
@ -56,8 +72,12 @@ public class MobilePositionSubscribeTask implements ISubscribeTask {
* COMPLETED-> Completed Dialog状态-已完成
* TERMINATED-> Terminated Dialog状态-终止
*/
logger.info("取消移动订阅时dialog状态为{}", dialog.getState());
if (timer != null ) {
timer.cancel();
timer = null;
}
if (dialog != null && dialog.getState().equals(DialogState.CONFIRMED)) {
logger.info("取消移动订阅时dialog状态为{}", dialog.getState());
device.setSubscribeCycleForMobilePosition(0);
sipCommander.mobilePositionSubscribe(device, dialog, eventResult -> {
ResponseEvent event = (ResponseEvent) eventResult.event;
@ -74,4 +94,9 @@ public class MobilePositionSubscribeTask implements ISubscribeTask {
});
}
}
@Override
public DialogState getDialogState() {
if (dialog == null) return null;
return dialog.getState();
}
}

1
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommanderForPlatform.java
View File

@ -46,6 +46,7 @@ public interface ISIPCommanderForPlatform {
* @return
*/
boolean catalogQuery(DeviceChannel channel, ParentPlatform parentPlatform, String sn, String fromTag, int size);
boolean catalogQuery(List<DeviceChannel> channels, ParentPlatform parentPlatform, String sn, String fromTag);
/**
* 向上级回复DeviceInfo查询信息

27
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
View File

@ -1566,17 +1566,28 @@ public class SIPCommander implements ISIPCommander {
cmdXml.append("<DeviceID>" + device.getDeviceId() + "</DeviceID>\r\n");
cmdXml.append("</Query>\r\n");
String tm = Long.toString(System.currentTimeMillis());
CallIdHeader callIdHeader = device.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
: udpSipProvider.getNewCallId();
Request request;
if (dialog != null) {
logger.info("发送目录订阅消息时 dialog的状态为： {}", dialog.getState());
request = dialog.createRequest(Request.SUBSCRIBE);
ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("Application", "MANSCDP+xml");
request.setContent(cmdXml.toString(), contentTypeHeader);
ExpiresHeader expireHeader = sipFactory.createHeaderFactory().createExpiresHeader(device.getSubscribeCycleForMobilePosition());
request.addHeader(expireHeader);
}else {
String tm = Long.toString(System.currentTimeMillis());
// 有效时间默认为60秒以上
Request request = headerProvider.createSubscribeRequest(device, cmdXml.toString(), "z9hG4bK-viaPos-" + tm,
"fromTagPos" + tm, null, device.getSubscribeCycleForCatalog(), "Catalog" ,
callIdHeader);
CallIdHeader callIdHeader = device.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
: udpSipProvider.getNewCallId();
// 有效时间默认为60秒以上
request = headerProvider.createSubscribeRequest(device, cmdXml.toString(), "z9hG4bK-viaPos-" + tm,
"fromTagPos" + tm, null, device.getSubscribeCycleForCatalog(), "Catalog" ,
callIdHeader);
}
transmitRequest(device, request, errorEvent, okEvent);
return true;
} catch ( NumberFormatException | ParseException | InvalidArgumentException | SipException e) {

116
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommanderFroPlatform.java
View File

@ -215,44 +215,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
return false;
}
try {
String characterSet = parentPlatform.getCharacterSet();
StringBuffer catalogXml = new StringBuffer(600);
catalogXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet +"\"?>\r\n");
catalogXml.append("<Response>\r\n");
catalogXml.append("<CmdType>Catalog</CmdType>\r\n");
catalogXml.append("<SN>" +sn + "</SN>\r\n");
catalogXml.append("<DeviceID>" + parentPlatform.getDeviceGBId() + "</DeviceID>\r\n");
catalogXml.append("<SumNum>" + size + "</SumNum>\r\n");
catalogXml.append("<DeviceList Num=\"1\">\r\n");
catalogXml.append("<Item>\r\n");
if (channel != null) {
catalogXml.append("<DeviceID>" + channel.getChannelId() + "</DeviceID>\r\n");
catalogXml.append("<Name>" + channel.getName() + "</Name>\r\n");
catalogXml.append("<Manufacturer>" + channel.getManufacture() + "</Manufacturer>\r\n");
catalogXml.append("<Model>" + channel.getModel() + "</Model>\r\n");
catalogXml.append("<Owner>" + channel.getOwner() + "</Owner>\r\n");
catalogXml.append("<CivilCode>" + channel.getCivilCode() + "</CivilCode>\r\n");
catalogXml.append("<Address>" + channel.getAddress() + "</Address>\r\n");
catalogXml.append("<Parental>" + channel.getParental() + "</Parental>\r\n");
if (channel.getParentId() != null) {
catalogXml.append("<ParentID>" + channel.getParentId() + "</ParentID>\r\n");
}
catalogXml.append("<Secrecy>" + channel.getSecrecy() + "</Secrecy>\r\n");
catalogXml.append("<RegisterWay>" + channel.getRegisterWay() + "</RegisterWay>\r\n");
catalogXml.append("<Status>" + (channel.getStatus() == 0?"OFF":"ON") + "</Status>\r\n");
catalogXml.append("<Longitude>" + channel.getLongitude() + "</Longitude>\r\n");
catalogXml.append("<Latitude>" + channel.getLatitude() + "</Latitude>\r\n");
catalogXml.append("<IPAddress>" + channel.getIpAddress() + "</IPAddress>\r\n");
catalogXml.append("<Port>" + channel.getPort() + "</Port>\r\n");
catalogXml.append("<Info>\r\n");
catalogXml.append("<PTZType>" + channel.getPTZType() + "</PTZType>\r\n");
catalogXml.append("</Info>\r\n");
}
catalogXml.append("</Item>\r\n");
catalogXml.append("</DeviceList>\r\n");
catalogXml.append("</Response>\r\n");
String catalogXml = getCatalogXml(channel, sn, parentPlatform, size);
// callid
CallIdHeader callIdHeader = parentPlatform.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
@ -268,6 +231,77 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
return true;
}
@Override
public boolean catalogQuery(List<DeviceChannel> channels, ParentPlatform parentPlatform, String sn, String fromTag) {
if ( parentPlatform ==null) {
return false;
}
sendCatalogResponse(channels, parentPlatform, sn, fromTag, 0);
return true;
}
private String getCatalogXml(DeviceChannel channel, String sn, ParentPlatform parentPlatform, int size) {
String characterSet = parentPlatform.getCharacterSet();
StringBuffer catalogXml = new StringBuffer(600);
catalogXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet +"\"?>\r\n");
catalogXml.append("<Response>\r\n");
catalogXml.append("<CmdType>Catalog</CmdType>\r\n");
catalogXml.append("<SN>" +sn + "</SN>\r\n");
catalogXml.append("<DeviceID>" + parentPlatform.getDeviceGBId() + "</DeviceID>\r\n");
catalogXml.append("<SumNum>" + size + "</SumNum>\r\n");
catalogXml.append("<DeviceList Num=\"1\">\r\n");
catalogXml.append("<Item>\r\n");
if (channel != null) {
catalogXml.append("<DeviceID>" + channel.getChannelId() + "</DeviceID>\r\n");
catalogXml.append("<Name>" + channel.getName() + "</Name>\r\n");
catalogXml.append("<Manufacturer>" + channel.getManufacture() + "</Manufacturer>\r\n");
catalogXml.append("<Model>" + channel.getModel() + "</Model>\r\n");
catalogXml.append("<Owner>" + channel.getOwner() + "</Owner>\r\n");
catalogXml.append("<CivilCode>" + channel.getCivilCode() + "</CivilCode>\r\n");
catalogXml.append("<Address>" + channel.getAddress() + "</Address>\r\n");
catalogXml.append("<Parental>" + channel.getParental() + "</Parental>\r\n");
if (channel.getParentId() != null) {
catalogXml.append("<ParentID>" + channel.getParentId() + "</ParentID>\r\n");
}
catalogXml.append("<Secrecy>" + channel.getSecrecy() + "</Secrecy>\r\n");
catalogXml.append("<RegisterWay>" + channel.getRegisterWay() + "</RegisterWay>\r\n");
catalogXml.append("<Status>" + (channel.getStatus() == 0?"OFF":"ON") + "</Status>\r\n");
catalogXml.append("<Longitude>" + channel.getLongitude() + "</Longitude>\r\n");
catalogXml.append("<Latitude>" + channel.getLatitude() + "</Latitude>\r\n");
catalogXml.append("<IPAddress>" + channel.getIpAddress() + "</IPAddress>\r\n");
catalogXml.append("<Port>" + channel.getPort() + "</Port>\r\n");
catalogXml.append("<Info>\r\n");
catalogXml.append("<PTZType>" + channel.getPTZType() + "</PTZType>\r\n");
catalogXml.append("</Info>\r\n");
}
catalogXml.append("</Item>\r\n");
catalogXml.append("</DeviceList>\r\n");
catalogXml.append("</Response>\r\n");
return catalogXml.toString();
}
private void sendCatalogResponse(List<DeviceChannel> channels, ParentPlatform parentPlatform, String sn, String fromTag, int index) {
if (index >= channels.size()) {
return;
}
try {
DeviceChannel deviceChannel = channels.get(index);
String catalogXml = getCatalogXml(deviceChannel, sn, parentPlatform, channels.size());
// callid
CallIdHeader callIdHeader = parentPlatform.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
: udpSipProvider.getNewCallId();
Request request = headerProviderPlarformProvider.createMessageRequest(parentPlatform, catalogXml, fromTag, callIdHeader);
transmitRequest(parentPlatform, request, null, eventResult -> {
int indexNext = index + 1;
sendCatalogResponse(channels, parentPlatform, sn, fromTag, indexNext);
});
} catch (SipException | ParseException | InvalidArgumentException e) {
e.printStackTrace();
}
}
/**
* 向上级回复DeviceInfo查询信息
* @param parentPlatform 平台信息
@ -351,7 +385,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
if (parentPlatform == null) {
return false;
}
logger.info("[发送 移动位置订阅] {}/{}->{},{}", parentPlatform.getServerGBId(), gpsMsgInfo.getId(), gpsMsgInfo.getLng(), gpsMsgInfo.getLat());
try {
String characterSet = parentPlatform.getCharacterSet();
StringBuffer deviceStatusXml = new StringBuffer(600);
@ -371,7 +405,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
CallIdHeader callIdHeader = parentPlatform.getTransport().equals("TCP") ? tcpSipProvider.getNewCallId()
: udpSipProvider.getNewCallId();
callIdHeader.setCallId(subscribeInfo.getCallId());
logger.info("[发送Notify-MobilePosition] {}/{}->{},{}", parentPlatform.getServerGBId(), gpsMsgInfo.getId(), gpsMsgInfo.getLng(), gpsMsgInfo.getLat());
sendNotify(parentPlatform, deviceStatusXml.toString(), subscribeInfo, eventResult -> {
logger.error("发送NOTIFY通知消息失败。错误：{} {}", eventResult.statusCode, eventResult.msg);
}, null);
@ -425,7 +459,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
// 设置编码， 防止中文乱码
messageFactory.setDefaultContentEncodingCharset(characterSet);
Dialog dialog = subscribeInfo.getDialog();
if (dialog == null) return;
if (dialog == null || !dialog.getState().equals(DialogState.CONFIRMED)) return;
SIPRequest notifyRequest = (SIPRequest)dialog.createRequest(Request.NOTIFY);
ContentTypeHeader contentTypeHeader = sipFactory.createHeaderFactory().createContentTypeHeader("Application", "MANSCDP+xml");
notifyRequest.setContent(catalogXmlContent, contentTypeHeader);

4
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/NotifyRequestProcessor.java
View File

@ -147,7 +147,7 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
} else {
mobilePosition.setAltitude(0.0);
}
logger.info("[收到Notify-MobilePosition]：{}/{}->{}.{}", mobilePosition.getDeviceId(), mobilePosition.getChannelId(),
logger.info("[收到 移动位置订阅]：{}/{}->{}.{}", mobilePosition.getDeviceId(), mobilePosition.getChannelId(),
mobilePosition.getLongitude(), mobilePosition.getLatitude());
mobilePosition.setReportSource("Mobile Position");
// 默认来源坐标系为WGS-84处理
@ -283,7 +283,7 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
Element eventElement = itemDevice.element("Event");
DeviceChannel channel = XmlUtil.channelContentHander(itemDevice);
channel.setDeviceId(device.getDeviceId());
logger.info("[收到Notify-Catalog]：{}/{}", device.getDeviceId(), channel.getChannelId());
logger.info("[收到 目录订阅]：{}/{}", device.getDeviceId(), channel.getChannelId());
switch (eventElement.getText().toUpperCase()) {
case CatalogEvent.ON: // 上线
logger.info("收到来自设备【{}】的通道【{}】上线通知", device.getDeviceId(), channel.getChannelId());

30
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/SubscribeRequestProcessor.java
View File

@ -137,6 +137,9 @@ public class SubscribeRequestProcessor extends SIPRequestProcessorParent impleme
String deviceID = XmlUtil.getText(rootElement, "DeviceID");
ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId);
SubscribeInfo subscribeInfo = new SubscribeInfo(evt, platformId);
if (platform == null) {
return;
}
if (evt.getServerTransaction() == null) {
ServerTransaction serverTransaction = platform.getTransport().equals("TCP") ? tcpSipProvider.getNewServerTransaction(evt.getRequest())
: udpSipProvider.getNewServerTransaction(evt.getRequest());
@ -146,8 +149,7 @@ public class SubscribeRequestProcessor extends SIPRequestProcessorParent impleme
subscribeInfo.setDialog(dialog);
}
String sn = XmlUtil.getText(rootElement, "SN");
String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetting.getServerId() + "_MobilePosition_" + platformId;
logger.info("[notify-MobilePosition]: {}", platformId);
logger.info("[回复 移动位置订阅]: {}", platformId);
StringBuilder resultXml = new StringBuilder(200);
resultXml.append("<?xml version=\"1.0\" ?>\r\n")
.append("<Response>\r\n")
@ -158,14 +160,25 @@ public class SubscribeRequestProcessor extends SIPRequestProcessorParent impleme
.append("</Response>\r\n");
if (subscribeInfo.getExpires() > 0) {
if (subscribeHolder.getMobilePositionSubscribe(platformId) != null) {
dynamicTask.stop(key);
}
String interval = XmlUtil.getText(rootElement, "Interval"); // GPS上报时间间隔
dynamicTask.startCron(key, new MobilePositionSubscribeHandlerTask(redisCatchStorage, sipCommanderForPlatform, storager, platformId, sn, key, subscribeHolder), Integer.parseInt(interval) -1 );
if (interval == null) {
subscribeInfo.setGpsInterval(5);
}else {
subscribeInfo.setGpsInterval(Integer.parseInt(interval));
}
subscribeInfo.setSn(sn);
subscribeHolder.putMobilePositionSubscribe(platformId, subscribeInfo);
// if (subscribeHolder.getMobilePositionSubscribe(platformId) == null ) {
// subscribeHolder.putMobilePositionSubscribe(platformId, subscribeInfo);
// }else {
// if (subscribeHolder.getMobilePositionSubscribe(platformId).getDialog() != null
// && subscribeHolder.getMobilePositionSubscribe(platformId).getDialog().getState() != null
// && !subscribeHolder.getMobilePositionSubscribe(platformId).getDialog().getState().equals(DialogState.CONFIRMED)) {
// subscribeHolder.putMobilePositionSubscribe(platformId, subscribeInfo);
// }
// }
}else if (subscribeInfo.getExpires() == 0) {
dynamicTask.stop(key);
subscribeHolder.removeMobilePositionSubscribe(platformId);
}
@ -199,8 +212,7 @@ public class SubscribeRequestProcessor extends SIPRequestProcessorParent impleme
subscribeInfo.setDialog(dialog);
}
String sn = XmlUtil.getText(rootElement, "SN");
String key = VideoManagerConstants.SIP_SUBSCRIBE_PREFIX + userSetting.getServerId() + "_Catalog_" + platformId;
logger.info("[notify-Catalog]: {}", platformId);
logger.info("[回复 目录订阅]: {}/{}", platformId, deviceID);
StringBuilder resultXml = new StringBuilder(200);
resultXml.append("<?xml version=\"1.0\" ?>\r\n")
.append("<Response>\r\n")

13
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/MessageRequestProcessor.java
View File

@ -12,6 +12,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorP
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import gov.nist.javax.sip.message.SIPRequest;
import org.dom4j.DocumentException;
import org.dom4j.Element;
import org.slf4j.Logger;
@ -23,6 +24,7 @@ import org.springframework.stereotype.Component;
import javax.sip.InvalidArgumentException;
import javax.sip.RequestEvent;
import javax.sip.SipException;
import javax.sip.address.SipURI;
import javax.sip.header.CSeqHeader;
import javax.sip.header.CallIdHeader;
import javax.sip.message.Response;
@ -81,6 +83,17 @@ public class MessageRequestProcessor extends SIPRequestProcessorParent implement
// 查询上级平台是否存在
ParentPlatform parentPlatform = storage.queryParentPlatByServerGBId(deviceId);
try {
if (device != null && parentPlatform != null) {
logger.warn("[重复]平台与设备编号重复：{}", deviceId);
SIPRequest request = (SIPRequest) evt.getRequest();
String hostAddress = request.getRemoteAddress().getHostAddress();
int remotePort = request.getRemotePort();
if (device.getHostAddress().equals(hostAddress + ":" + remotePort)) {
parentPlatform = null;
}else {
device = null;
}
}
if (device == null && parentPlatform == null) {
// 不存在则回复404
responseAck(evt, Response.NOT_FOUND, "device "+ deviceId +" not found");

13
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/control/cmd/DeviceControlQueryMessageHandler.java
View File

@ -23,6 +23,7 @@ import javax.sip.*;
import javax.sip.address.SipURI;
import javax.sip.header.HeaderAddress;
import javax.sip.header.ToHeader;
import javax.sip.message.Response;
import java.text.ParseException;
import java.util.Iterator;
@ -103,6 +104,18 @@ public class DeviceControlQueryMessageHandler extends SIPRequestProcessorParent
if (!StringUtils.isEmpty(getText(rootElement,"PTZCmd")) && !parentPlatform.getServerGBId().equals(targetGBId)) {
String cmdString = getText(rootElement,"PTZCmd");
Device deviceForPlatform = storager.queryVideoDeviceByPlatformIdAndChannelId(parentPlatform.getServerGBId(), channelId);
if (deviceForPlatform == null) {
try {
responseAck(evt, Response.NOT_FOUND);
return;
} catch (SipException e) {
e.printStackTrace();
} catch (InvalidArgumentException e) {
e.printStackTrace();
} catch (ParseException e) {
e.printStackTrace();
}
}
cmder.fronEndCmd(deviceForPlatform, channelId, cmdString, eventResult -> {
// 失败的回复
try {

27
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/CatalogNotifyMessageHandler.java
View File

@ -18,6 +18,7 @@ import javax.sip.SipException;
import javax.sip.header.FromHeader;
import javax.sip.message.Response;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.List;
@Component
@ -58,7 +59,8 @@ public class CatalogNotifyMessageHandler extends SIPRequestProcessorParent imple
List<DeviceChannelInPlatform> deviceChannels = storage.queryChannelListInParentPlatform(parentPlatform.getServerGBId());
// 查询关联的直播通道
List<GbStream> gbStreams = storage.queryGbStreamListInPlatform(parentPlatform.getServerGBId());
int size = deviceChannels.size() + gbStreams.size();
List<DeviceChannel> allChannels = new ArrayList<>();
// 回复目录信息
List<PlatformCatalog> catalogs = storage.queryCatalogInPlatform(parentPlatform.getServerGBId());
if (catalogs.size() > 0) {
@ -81,9 +83,7 @@ public class CatalogNotifyMessageHandler extends SIPRequestProcessorParent imple
deviceChannel.setModel("live");
deviceChannel.setOwner("wvp-pro");
deviceChannel.setSecrecy("0");
cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size);
// 防止发送过快
Thread.sleep(100);
allChannels.add(deviceChannel);
}
}
// 回复级联的通道
@ -96,9 +96,7 @@ public class CatalogNotifyMessageHandler extends SIPRequestProcessorParent imple
deviceChannel.setParental(0);
deviceChannel.setParentId(channel.getCatalogId());
deviceChannel.setCivilCode(parentPlatform.getDeviceGBId().substring(0, 6));
cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size);
// 防止发送过快
Thread.sleep(100);
allChannels.add(deviceChannel);
}
}
// 回复直播的通道
@ -114,7 +112,8 @@ public class CatalogNotifyMessageHandler extends SIPRequestProcessorParent imple
deviceChannel.setLatitude(gbStream.getLatitude());
deviceChannel.setDeviceId(parentPlatform.getDeviceGBId());
deviceChannel.setManufacture("wvp-pro");
deviceChannel.setStatus(gbStream.isStatus()?1:0);
// deviceChannel.setStatus(gbStream.isStatus()?1:0);
deviceChannel.setStatus(1);
deviceChannel.setParentId(gbStream.getCatalogId());
deviceChannel.setRegisterWay(1);
deviceChannel.setCivilCode(parentPlatform.getDeviceGBId().substring(0,6));
@ -122,16 +121,16 @@ public class CatalogNotifyMessageHandler extends SIPRequestProcessorParent imple
deviceChannel.setOwner("wvp-pro");
deviceChannel.setParental(0);
deviceChannel.setSecrecy("0");
cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size);
// 防止发送过快
Thread.sleep(100);
allChannels.add(deviceChannel);
}
}
if (size == 0) {
if (allChannels.size() > 0) {
cmderFroPlatform.catalogQuery(allChannels, parentPlatform, sn, fromHeader.getTag());
}else {
// 回复无通道
cmderFroPlatform.catalogQuery(null, parentPlatform, sn, fromHeader.getTag(), size);
cmderFroPlatform.catalogQuery(null, parentPlatform, sn, fromHeader.getTag(), 0);
}
} catch (SipException | InvalidArgumentException | ParseException | InterruptedException e) {
} catch (SipException | InvalidArgumentException | ParseException e) {
e.printStackTrace();
}

37
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/query/cmd/CatalogQueryMessageHandler.java
View File

@ -22,6 +22,7 @@ import javax.sip.SipException;
import javax.sip.header.FromHeader;
import javax.sip.message.Response;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.List;
@Component
@ -45,6 +46,9 @@ public class CatalogQueryMessageHandler extends SIPRequestProcessorParent implem
@Autowired
private EventPublisher publisher;
@Autowired
private IVideoManagerStorage storage;
@Override
public void afterPropertiesSet() throws Exception {
queryMessageHandler.addHandler(cmdType, this);
@ -71,10 +75,11 @@ public class CatalogQueryMessageHandler extends SIPRequestProcessorParent implem
List<GbStream> gbStreams = storager.queryGbStreamListInPlatform(parentPlatform.getServerGBId());
// 回复目录信息
List<PlatformCatalog> catalogs = storager.queryCatalogInPlatform(parentPlatform.getServerGBId());
int size = catalogs.size() + deviceChannelInPlatforms.size() + gbStreams.size();
List<DeviceChannel> allChannels = new ArrayList<>();
if (catalogs.size() > 0) {
for (PlatformCatalog catalog : catalogs) {
if (catalog.getParentId().equals(parentPlatform.getServerGBId())) {
if (catalog.getParentId().equals(catalog.getPlatformId())) {
catalog.setParentId(parentPlatform.getDeviceGBId());
}
DeviceChannel deviceChannel = new DeviceChannel();
@ -92,9 +97,7 @@ public class CatalogQueryMessageHandler extends SIPRequestProcessorParent implem
deviceChannel.setModel("live");
deviceChannel.setOwner("wvp-pro");
deviceChannel.setSecrecy("0");
cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size);
// 防止发送过快
Thread.sleep(100);
allChannels.add(deviceChannel);
}
}
// 回复级联的通道
@ -103,20 +106,18 @@ public class CatalogQueryMessageHandler extends SIPRequestProcessorParent implem
if (channel.getCatalogId().equals(parentPlatform.getServerGBId())) {
channel.setCatalogId(parentPlatform.getDeviceGBId());
}
DeviceChannel deviceChannel = storager.queryChannel(channel.getDeviceId(), channel.getChannelId());
DeviceChannel deviceChannel = storage.queryChannel(channel.getDeviceId(), channel.getChannelId());
deviceChannel.setParental(0);
deviceChannel.setParentId(channel.getCatalogId());
deviceChannel.setCivilCode(parentPlatform.getDeviceGBId().substring(0, 6));
cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size);
// 防止发送过快
Thread.sleep(100);
allChannels.add(deviceChannel);
}
}
// 回复直播的通道
if (gbStreams.size() > 0) {
for (GbStream gbStream : gbStreams) {
if (gbStream.getCatalogId().equals(parentPlatform.getServerGBId())) {
gbStream.setCatalogId(parentPlatform.getDeviceGBId());
gbStream.setCatalogId(null);
}
DeviceChannel deviceChannel = new DeviceChannel();
deviceChannel.setChannelId(gbStream.getGbId());
@ -125,7 +126,8 @@ public class CatalogQueryMessageHandler extends SIPRequestProcessorParent implem
deviceChannel.setLatitude(gbStream.getLatitude());
deviceChannel.setDeviceId(parentPlatform.getDeviceGBId());
deviceChannel.setManufacture("wvp-pro");
deviceChannel.setStatus(gbStream.isStatus()?1:0);
// deviceChannel.setStatus(gbStream.isStatus()?1:0);
deviceChannel.setStatus(1);
deviceChannel.setParentId(gbStream.getCatalogId());
deviceChannel.setRegisterWay(1);
deviceChannel.setCivilCode(parentPlatform.getDeviceGBId().substring(0,6));
@ -133,15 +135,14 @@ public class CatalogQueryMessageHandler extends SIPRequestProcessorParent implem
deviceChannel.setOwner("wvp-pro");
deviceChannel.setParental(0);
deviceChannel.setSecrecy("0");
cmderFroPlatform.catalogQuery(deviceChannel, parentPlatform, sn, fromHeader.getTag(), size);
// 防止发送过快
Thread.sleep(100);
allChannels.add(deviceChannel);
}
}
if (size == 0) {
if (allChannels.size() > 0) {
cmderFroPlatform.catalogQuery(allChannels, parentPlatform, sn, fromHeader.getTag());
}else {
// 回复无通道
cmderFroPlatform.catalogQuery(null, parentPlatform, sn, fromHeader.getTag(), size);
cmderFroPlatform.catalogQuery(null, parentPlatform, sn, fromHeader.getTag(), 0);
}
} catch (SipException e) {
e.printStackTrace();
@ -149,8 +150,6 @@ public class CatalogQueryMessageHandler extends SIPRequestProcessorParent implem
e.printStackTrace();
} catch (ParseException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}

33
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/CatalogResponseMessageHandler.java
View File

@ -116,16 +116,15 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp
continue;
}
//by brewswang
if (NumericUtil.isDouble(XmlUtil.getText(itemDevice, "Longitude"))) {//如果包含位置信息，就更新一下位置
processNotifyMobilePosition(evt, itemDevice);
}
// if (NumericUtil.isDouble(XmlUtil.getText(itemDevice, "Longitude"))) {//如果包含位置信息，就更新一下位置
// processNotifyMobilePosition(evt, itemDevice);
// }
DeviceChannel deviceChannel = XmlUtil.channelContentHander(itemDevice);
deviceChannel.setDeviceId(device.getDeviceId());
logger.debug("收到来自设备【{}】的通道: {}【{}】", device.getDeviceId(), deviceChannel.getName(), deviceChannel.getChannelId());
channelList.add(deviceChannel);
}
logger.info("收到来自设备【{}】的通道: {}个，{}/{}", device.getDeviceId(), channelList.size(), catalogDataCatch.get(key) == null ? 0 :catalogDataCatch.get(key).size(), sumNum);
catalogDataCatch.put(key, sumNum, device, channelList);
if (catalogDataCatch.get(key).size() == sumNum) {
// 数据已经完整接收
@ -147,9 +146,6 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp
}
// 回复200 OK
responseAck(evt, Response.OK);
if (offLineDetector.isOnline(device.getDeviceId())) {
publisher.onlineEventPublish(device, VideoManagerConstants.EVENT_ONLINE_MESSAGE);
}
}
} catch (DocumentException e) {
e.printStackTrace();
@ -231,4 +227,23 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp
e.printStackTrace();
}
}
public SyncStatus getChannelSyncProgress(String deviceId) {
String key = DeferredResultHolder.CALLBACK_CMD_CATALOG + deviceId;
if (catalogDataCatch.get(key) == null) {
return null;
}else {
return catalogDataCatch.getSyncStatus(key);
}
}
public void setChannelSyncReady(String deviceId) {
String key = DeferredResultHolder.CALLBACK_CMD_CATALOG + deviceId;
catalogDataCatch.addReady(key);
}
public void setChannelSyncEnd(String deviceId, String errorMsg) {
String key = DeferredResultHolder.CALLBACK_CMD_CATALOG + deviceId;
catalogDataCatch.setChannelSyncEnd(key, errorMsg);
}
}

4
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
View File

@ -396,7 +396,7 @@ public class ZLMHttpHookListener {
}
}
if (gbStreams.size() > 0) {
eventPublisher.catalogEventPublishForStream(null, gbStreams, CatalogEvent.ON);
// eventPublisher.catalogEventPublishForStream(null, gbStreams, CatalogEvent.ON);
}
}else {
@ -408,7 +408,7 @@ public class ZLMHttpHookListener {
}
GbStream gbStream = storager.getGbStream(app, streamId);
if (gbStream != null) {
eventPublisher.catalogEventPublishForStream(null, gbStream, CatalogEvent.OFF);
// eventPublisher.catalogEventPublishForStream(null, gbStream, CatalogEvent.OFF);
}
zlmMediaListManager.removeMedia(app, streamId);
}

1
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaListManager.java
View File

@ -204,6 +204,7 @@ public class ZLMMediaListManager {
if (streamProxyItem == null) {
result = storager.removeMedia(app, streamId);
}else {
// TODO 暂不设置为离线
result =storager.mediaOutline(app, streamId);
}
return result;

21
src/main/java/com/genersoft/iot/vmp/service/IDeviceService.java
View File

@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.service;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.SyncStatus;
/**
* 设备相关业务处理
@ -34,4 +35,24 @@ public interface IDeviceService {
* @return
*/
boolean removeMobilePositionSubscribe(Device device);
/**
* 移除移动位置订阅
* @param deviceId 设备ID
* @return
*/
SyncStatus getChannelSyncStatus(String deviceId);
/**
* 设置通道同步状态
* @param deviceId 设备ID
*/
void setChannelSyncReady(String deviceId);
/**
* 设置同步结束
* @param deviceId 设备ID
* @param errorMsg 错误信息
*/
void setChannelSyncEnd(String deviceId, String errorMsg);
}

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

@ -3,14 +3,19 @@ package com.genersoft.iot.vmp.service.impl;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd.CatalogResponseMessageHandler;
import com.genersoft.iot.vmp.service.IDeviceService;
import com.genersoft.iot.vmp.gb28181.task.impl.CatalogSubscribeTask;
import com.genersoft.iot.vmp.gb28181.task.impl.MobilePositionSubscribeTask;
import com.genersoft.iot.vmp.gb28181.bean.SyncStatus;
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.stereotype.Service;
import javax.sip.DialogState;
/**
* 设备业务（目录订阅）
*/
@ -25,24 +30,28 @@ public class DeviceServiceImpl implements IDeviceService {
@Autowired
private ISIPCommander sipCommander;
@Autowired
private CatalogResponseMessageHandler catalogResponseMessageHandler;
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Override
public boolean addCatalogSubscribe(Device device) {
if (device == null || device.getSubscribeCycleForCatalog() < 0) {
return false;
}
if (dynamicTask.contains(device.getDeviceId() + "catalog")) {
// 存在则停止现有的，开启新的
dynamicTask.stop(device.getDeviceId() + "catalog");
CatalogSubscribeTask task = (CatalogSubscribeTask)dynamicTask.get(device.getDeviceId() + "catalog");
if (task != null && task.getDialogState() != null && task.getDialogState().equals(DialogState.CONFIRMED)) { // 已存在不需要再次添加
return true;
}
logger.info("[添加目录订阅] 设备{}", device.getDeviceId());
// 添加目录订阅
CatalogSubscribeTask catalogSubscribeTask = new CatalogSubscribeTask(device, sipCommander);
catalogSubscribeTask.run();
// 提前开始刷新订阅
int subscribeCycleForCatalog = device.getSubscribeCycleForCatalog();
int subscribeCycleForCatalog = Math.max(device.getSubscribeCycleForCatalog(),30);
// 设置最小值为30
subscribeCycleForCatalog = Math.max(subscribeCycleForCatalog, 30);
dynamicTask.startCron(device.getDeviceId() + "catalog", catalogSubscribeTask, subscribeCycleForCatalog);
dynamicTask.startCron(device.getDeviceId() + "catalog", catalogSubscribeTask, subscribeCycleForCatalog -1);
return true;
}
@ -61,18 +70,16 @@ public class DeviceServiceImpl implements IDeviceService {
if (device == null || device.getSubscribeCycleForMobilePosition() < 0) {
return false;
}
if (dynamicTask.contains(device.getDeviceId() + "mobile_position")) {
// 存在则停止现有的，开启新的
dynamicTask.stop(device.getDeviceId() + "mobile_position");
}
logger.info("[添加移动位置订阅] 设备{}", device.getDeviceId());
MobilePositionSubscribeTask task = (MobilePositionSubscribeTask)dynamicTask.get(device.getDeviceId() + "mobile_position");
if (task != null && task.getDialogState() != null && task.getDialogState().equals(DialogState.CONFIRMED)) { // 已存在不需要再次添加
return true;
}
// 添加目录订阅
MobilePositionSubscribeTask mobilePositionSubscribeTask = new MobilePositionSubscribeTask(device, sipCommander);
mobilePositionSubscribeTask.run();
// 提前开始刷新订阅
int subscribeCycleForCatalog = device.getSubscribeCycleForCatalog();
// 设置最小值为30
subscribeCycleForCatalog = Math.max(subscribeCycleForCatalog, 30);
int subscribeCycleForCatalog = Math.max(device.getSubscribeCycleForMobilePosition(),30);
dynamicTask.startCron(device.getDeviceId() + "mobile_position" , mobilePositionSubscribeTask, subscribeCycleForCatalog -1 );
return true;
}
@ -86,4 +93,19 @@ public class DeviceServiceImpl implements IDeviceService {
dynamicTask.stop(device.getDeviceId() + "mobile_position");
return true;
}
@Override
public SyncStatus getChannelSyncStatus(String deviceId) {
return catalogResponseMessageHandler.getChannelSyncProgress(deviceId);
}
@Override
public void setChannelSyncReady(String deviceId) {
catalogResponseMessageHandler.setChannelSyncReady(deviceId);
}
@Override
public void setChannelSyncEnd(String deviceId, String errorMsg) {
catalogResponseMessageHandler.setChannelSyncEnd(deviceId, errorMsg);
}
}

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

@ -100,7 +100,8 @@ public class GbStreamServiceImpl implements IGbStreamService {
deviceChannel.setLatitude(gbStream.getLatitude());
deviceChannel.setDeviceId(deviceGBId);
deviceChannel.setManufacture("wvp-pro");
deviceChannel.setStatus(gbStream.isStatus()?1:0);
// deviceChannel.setStatus(gbStream.isStatus()?1:0);
deviceChannel.setStatus(1);
deviceChannel.setParentId(catalogId ==null?gbStream.getCatalogId():catalogId);
deviceChannel.setRegisterWay(1);
deviceChannel.setCivilCode(deviceGBId.substring(0, 6));

1
src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java
View File

@ -216,4 +216,5 @@ public interface IRedisCatchStorage {
void sendMobilePositionMsg(JSONObject jsonObject);
void sendStreamPushRequestedMsg(MessageForPushChannel messageForPushChannel);
}

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

@ -638,4 +638,5 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
logger.info("[redis 推流被请求通知] {}: {}-{}", key, msg.getApp(), msg.getStream());
redis.convertAndSend(key, (JSONObject)JSON.toJSON(msg));
}
}

2
src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStorageImpl.java
View File

@ -445,8 +445,6 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage {
device.setOnline(1);
logger.info("更新设备在线: " + deviceId);
redisCatchStorage.updateDevice(device);
List<DeviceChannel> deviceChannelList = deviceChannelMapper.queryOnlineChannelsByDeviceId(deviceId);
eventPublisher.catalogEventPublish(null, deviceChannelList, CatalogEvent.ON);
return deviceMapper.update(device) > 0;
}

126
src/main/java/com/genersoft/iot/vmp/utils/Coordtransform.java Normal file
View File

@ -0,0 +1,126 @@
package com.genersoft.iot.vmp.utils;
/**
* 坐标转换
* 一个提供了百度坐标（BD09）、国测局坐标（火星坐标，GCJ02）、和WGS84坐标系之间的转换的工具类
* 参考https://github.com/wandergis/coordtransform 写的Java版本
* @author Xinconan
* @date 2016-03-18
* @url https://github.com/xinconan/coordtransform
*/
public class Coordtransform {
private static double x_PI = 3.14159265358979324 * 3000.0 / 180.0;
private static double PI = 3.1415926535897932384626;
private static double a = 6378245.0;
private static double ee = 0.00669342162296594323;
/**
* 百度坐标系 (BD-09) 与 火星坐标系 (GCJ-02)的转换
* 即 百度 转 谷歌、高德
* @param bd_lon
* @param bd_lat
* @return Double[lon,lat]
*/
public static Double[] BD09ToGCJ02(Double bd_lon,Double bd_lat){
double x = bd_lon - 0.0065;
double y = bd_lat - 0.006;
double z = Math.sqrt(x * x + y * y) - 0.00002 * Math.sin(y * x_PI);
double theta = Math.atan2(y, x) - 0.000003 * Math.cos(x * x_PI);
Double[] arr = new Double[2];
arr[0] = z * Math.cos(theta);
arr[1] = z * Math.sin(theta);
return arr;
}
/**
* 火星坐标系 (GCJ-02) 与百度坐标系 (BD-09) 的转换
* 即谷歌、高德 转 百度
* @param gcj_lon
* @param gcj_lat
* @return Double[lon,lat]
*/
public static Double[] GCJ02ToBD09(Double gcj_lon,Double gcj_lat){
double z = Math.sqrt(gcj_lon * gcj_lon + gcj_lat * gcj_lat) + 0.00002 * Math.sin(gcj_lat * x_PI);
double theta = Math.atan2(gcj_lat, gcj_lon) + 0.000003 * Math.cos(gcj_lon * x_PI);
Double[] arr = new Double[2];
arr[0] = z * Math.cos(theta) + 0.0065;
arr[1] = z * Math.sin(theta) + 0.006;
return arr;
}
/**
* WGS84转GCJ02
* @param wgs_lon
* @param wgs_lat
* @return Double[lon,lat]
*/
public static Double[] WGS84ToGCJ02(Double wgs_lon,Double wgs_lat){
if(outOfChina(wgs_lon, wgs_lat)){
return new Double[]{wgs_lon,wgs_lat};
}
double dlat = transformlat(wgs_lon - 105.0, wgs_lat - 35.0);
double dlng = transformlng(wgs_lon - 105.0, wgs_lat - 35.0);
double radlat = wgs_lat / 180.0 * PI;
double magic = Math.sin(radlat);
magic = 1 - ee * magic * magic;
double sqrtmagic = Math.sqrt(magic);
dlat = (dlat * 180.0) / ((a * (1 - ee)) / (magic * sqrtmagic) * PI);
dlng = (dlng * 180.0) / (a / sqrtmagic * Math.cos(radlat) * PI);
Double[] arr = new Double[2];
arr[0] = wgs_lon + dlng;
arr[1] = wgs_lat + dlat;
return arr;
}
/**
* GCJ02转WGS84
* @param gcj_lon
* @param gcj_lat
* @return Double[lon,lat]
*/
public static Double[] GCJ02ToWGS84(Double gcj_lon,Double gcj_lat){
if(outOfChina(gcj_lon, gcj_lat)){
return new Double[]{gcj_lon,gcj_lat};
}
double dlat = transformlat(gcj_lon - 105.0, gcj_lat - 35.0);
double dlng = transformlng(gcj_lon - 105.0, gcj_lat - 35.0);
double radlat = gcj_lat / 180.0 * PI;
double magic = Math.sin(radlat);
magic = 1 - ee * magic * magic;
double sqrtmagic = Math.sqrt(magic);
dlat = (dlat * 180.0) / ((a * (1 - ee)) / (magic * sqrtmagic) * PI);
dlng = (dlng * 180.0) / (a / sqrtmagic * Math.cos(radlat) * PI);
double mglat = gcj_lat + dlat;
double mglng = gcj_lon + dlng;
return new Double[]{gcj_lon * 2 - mglng, gcj_lat * 2 - mglat};
}
private static Double transformlat(double lng, double lat) {
double ret = -100.0 + 2.0 * lng + 3.0 * lat + 0.2 * lat * lat + 0.1 * lng * lat + 0.2 * Math.sqrt(Math.abs(lng));
ret += (20.0 * Math.sin(6.0 * lng * PI) + 20.0 * Math.sin(2.0 * lng * PI)) * 2.0 / 3.0;
ret += (20.0 * Math.sin(lat * PI) + 40.0 * Math.sin(lat / 3.0 * PI)) * 2.0 / 3.0;
ret += (160.0 * Math.sin(lat / 12.0 * PI) + 320 * Math.sin(lat * PI / 30.0)) * 2.0 / 3.0;
return ret;
}
private static Double transformlng(double lng,double lat) {
double ret = 300.0 + lng + 2.0 * lat + 0.1 * lng * lng + 0.1 * lng * lat + 0.1 * Math.sqrt(Math.abs(lng));
ret += (20.0 * Math.sin(6.0 * lng * PI) + 20.0 * Math.sin(2.0 * lng * PI)) * 2.0 / 3.0;
ret += (20.0 * Math.sin(lng * PI) + 40.0 * Math.sin(lng / 3.0 * PI)) * 2.0 / 3.0;
ret += (150.0 * Math.sin(lng / 12.0 * PI) + 300.0 * Math.sin(lng / 30.0 * PI)) * 2.0 / 3.0;
return ret;
}
/**
* outOfChina
* @描述: 判断是否在国内，不在国内则不做偏移
* @param lng
* @param lat
* @return {boolean}
*/
private static boolean outOfChina(Double lng,Double lat) {
return (lng < 72.004 || lng > 137.8347) || ((lat < 0.8293 || lat > 55.8271) || false);
};
}

91
src/main/java/com/genersoft/iot/vmp/utils/GpsUtil.java
View File

@ -17,48 +17,57 @@ public class GpsUtil {
public static BaiduPoint Wgs84ToBd09(String xx, String yy) {
try {
Socket s = new Socket("api.map.baidu.com", 80);
BufferedReader br = new BufferedReader(new InputStreamReader(s.getInputStream(), "UTF-8"));
OutputStream out = s.getOutputStream();
StringBuffer sb = new StringBuffer("GET /ag/coord/convert?from=0&to=4");
sb.append("&x=" + xx + "&y=" + yy);
sb.append("&callback=BMap.Convertor.cbk_3976 HTTP/1.1\r\n");
sb.append("User-Agent: Java/1.6.0_20\r\n");
sb.append("Host: api.map.baidu.com:80\r\n");
sb.append("Accept: text/html, image/gif, image/jpeg, *; q=.2, */*; q=.2\r\n");
sb.append("Connection: Close\r\n");
sb.append("\r\n");
out.write(sb.toString().getBytes());
String json = "";
String tmp = "";
while ((tmp = br.readLine()) != null) {
// logger.info(tmp);
json += tmp;
}
// try {
// Socket s = new Socket("api.map.baidu.com", 80);
// BufferedReader br = new BufferedReader(new InputStreamReader(s.getInputStream(), "UTF-8"));
// OutputStream out = s.getOutputStream();
// StringBuffer sb = new StringBuffer("GET /ag/coord/convert?from=0&to=4");
// sb.append("&x=" + xx + "&y=" + yy);
// sb.append("&callback=BMap.Convertor.cbk_3976 HTTP/1.1\r\n");
// sb.append("User-Agent: Java/1.6.0_20\r\n");
// sb.append("Host: api.map.baidu.com:80\r\n");
// sb.append("Accept: text/html, image/gif, image/jpeg, *; q=.2, */*; q=.2\r\n");
// sb.append("Connection: Close\r\n");
// sb.append("\r\n");
// out.write(sb.toString().getBytes());
// String json = "";
// String tmp = "";
// while ((tmp = br.readLine()) != null) {
// // logger.info(tmp);
// json += tmp;
// }
//
// s.close();
// int start = json.indexOf("cbk_3976");
// int end = json.lastIndexOf("}");
// if (start != -1 && end != -1 && json.contains("\"x\":\"")) {
// json = json.substring(start, end);
// String[] point = json.split(",");
// String x = point[1].split(":")[1].replace("\"", "");
// String y = point[2].split(":")[1].replace("\"", "");
// BaiduPoint bdPoint= new BaiduPoint();
// bdPoint.setBdLng(new String(decode(x)));
// bdPoint.setBdLat(new String(decode(y)));
// return bdPoint;
// //return (new String(decode(x)) + "," + new String(decode(y)));
// } else {
// logger.info("gps坐标无效！！");
// }
// out.close();
// br.close();
// } catch (Exception e) {
// e.printStackTrace();
// }
s.close();
int start = json.indexOf("cbk_3976");
int end = json.lastIndexOf("}");
if (start != -1 && end != -1 && json.contains("\"x\":\"")) {
json = json.substring(start, end);
String[] point = json.split(",");
String x = point[1].split(":")[1].replace("\"", "");
String y = point[2].split(":")[1].replace("\"", "");
BaiduPoint bdPoint= new BaiduPoint();
bdPoint.setBdLng(new String(decode(x)));
bdPoint.setBdLat(new String(decode(y)));
return bdPoint;
//return (new String(decode(x)) + "," + new String(decode(y)));
} else {
logger.info("gps坐标无效！！");
}
out.close();
br.close();
} catch (Exception e) {
e.printStackTrace();
}
return null;
double lng = Double.parseDouble(xx);
double lat = Double.parseDouble(yy);
Double[] gcj02 = Coordtransform.WGS84ToGCJ02(lng, lat);
Double[] doubles = Coordtransform.GCJ02ToBD09(gcj02[0], gcj02[1]);
BaiduPoint bdPoint= new BaiduPoint();
bdPoint.setBdLng(doubles[0] + "");
bdPoint.setBdLat(doubles[1] + "");
return bdPoint;
}
/**

110
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/device/DeviceQuery.java
View File

@ -4,7 +4,13 @@ import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.SubscribeHolder;
import com.genersoft.iot.vmp.gb28181.bean.SyncStatus;
import com.genersoft.iot.vmp.gb28181.event.DeviceOffLineDetector;
import com.genersoft.iot.vmp.gb28181.task.ISubscribeTask;
import com.genersoft.iot.vmp.gb28181.task.impl.CatalogSubscribeTask;
import com.genersoft.iot.vmp.gb28181.task.impl.MobilePositionSubscribeHandlerTask;
import com.genersoft.iot.vmp.gb28181.task.impl.MobilePositionSubscribeTask;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
@ -18,6 +24,7 @@ import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation;
import org.kxml2.wap.wv.WV;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@ -27,9 +34,8 @@ import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.context.request.async.DeferredResult;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import javax.sip.DialogState;
import java.util.*;
@Api(tags = "国标设备查询", value = "国标设备查询")
@SuppressWarnings("rawtypes")
@ -61,6 +67,9 @@ public class DeviceQuery {
@Autowired
private DynamicTask dynamicTask;
@Autowired
private SubscribeHolder subscribeHolder;
/**
* 使用ID查询国标设备
* @param deviceId 国标ID
@ -149,48 +158,30 @@ public class DeviceQuery {
@ApiImplicitParam(name="deviceId", value = "设备id", required = true, dataTypeClass = String.class),
})
@PostMapping("/devices/{deviceId}/sync")
public DeferredResult<ResponseEntity<Device>> devicesSync(@PathVariable String deviceId){
public WVPResult<SyncStatus> devicesSync(@PathVariable String deviceId){
if (logger.isDebugEnabled()) {
logger.debug("设备通道信息同步API调用，deviceId：" + deviceId);
}
Device device = storager.queryVideoDevice(deviceId);
String key = DeferredResultHolder.CALLBACK_CMD_CATALOG + deviceId;
String uuid = UUID.randomUUID().toString();
// 默认超时时间为30分钟
DeferredResult<ResponseEntity<Device>> result = new DeferredResult<ResponseEntity<Device>>(30*60*1000L);
result.onTimeout(()->{
logger.warn("设备[{}]通道信息同步超时", deviceId);
// 释放rtpserver
RequestMessage msg = new RequestMessage();
msg.setKey(key);
msg.setId(uuid);
WVPResult<Object> wvpResult = new WVPResult<>();
wvpResult.setCode(-1);
wvpResult.setData(device);
wvpResult.setMsg("更新超时");
msg.setData(wvpResult);
resultHolder.invokeAllResult(msg);
});
// 等待其他相同请求返回时一起返回
if (resultHolder.exist(key, null)) {
return result;
SyncStatus syncStatus = deviceService.getChannelSyncStatus(deviceId);
// 已存在则返回进度
if (syncStatus != null && syncStatus.getErrorMsg() == null) {
WVPResult<SyncStatus> wvpResult = new WVPResult<>();
wvpResult.setCode(0);
wvpResult.setData(syncStatus);
return wvpResult;
}
cmder.catalogQuery(device, event -> {
RequestMessage msg = new RequestMessage();
msg.setKey(key);
msg.setId(uuid);
WVPResult<Object> wvpResult = new WVPResult<>();
wvpResult.setCode(-1);
wvpResult.setData(device);
wvpResult.setMsg(String.format("同步通道失败，错误码： %s, %s", event.statusCode, event.msg));
msg.setData(wvpResult);
resultHolder.invokeAllResult(msg);
SyncStatus syncStatusReady = new SyncStatus();
deviceService.setChannelSyncReady(deviceId);
cmder.catalogQuery(device, event -> {
String errorMsg = String.format("同步通道失败，错误码： %s, %s", event.statusCode, event.msg);
deviceService.setChannelSyncEnd(deviceId, errorMsg);
});
resultHolder.put(key, uuid, result);
return result;
WVPResult<SyncStatus> wvpResult = new WVPResult<>();
wvpResult.setCode(0);
wvpResult.setMsg("开始同步");
return wvpResult;
}
/**
@ -467,4 +458,47 @@ public class DeviceQuery {
public WVPResult<List<DeviceChannelTree>> tree(@PathVariable String deviceId) {
return WVPResult.Data(storager.tree(deviceId));
}
@GetMapping("/{deviceId}/sync_status")
@ApiOperation(value = "获取通道同步进度", notes = "获取通道同步进度")
public WVPResult<SyncStatus> getSyncStatus(@PathVariable String deviceId) {
SyncStatus channelSyncStatus = deviceService.getChannelSyncStatus(deviceId);
WVPResult<SyncStatus> wvpResult = new WVPResult<>();
if (channelSyncStatus == null) {
wvpResult.setCode(-1);
wvpResult.setMsg("同步尚未开始");
}else {
wvpResult.setCode(0);
wvpResult.setData(channelSyncStatus);
if (channelSyncStatus.getErrorMsg() != null) {
wvpResult.setMsg(channelSyncStatus.getErrorMsg());
}
}
return wvpResult;
}
@GetMapping("/{deviceId}/subscribe_info")
@ApiOperation(value = "获取设备的订阅状态", notes = "获取设备的订阅状态")
public WVPResult<Map<String, String>> getSubscribeInfo(@PathVariable String deviceId) {
Set<String> allKeys = dynamicTask.getAllKeys();
Map<String, String> dialogStateMap = new HashMap<>();
for (String key : allKeys) {
if (key.startsWith(deviceId)) {
ISubscribeTask subscribeTask = (ISubscribeTask)dynamicTask.get(key);
DialogState dialogState = subscribeTask.getDialogState();
if (dialogState == null) {
continue;
}
if (subscribeTask instanceof CatalogSubscribeTask) {
dialogStateMap.put("catalog", dialogState.toString());
}else if (subscribeTask instanceof MobilePositionSubscribeTask) {
dialogStateMap.put("mobilePosition", dialogState.toString());
}
}
}
WVPResult<Map<String, String>> wvpResult = new WVPResult<>();
wvpResult.setCode(0);
wvpResult.setData(dialogStateMap);
return wvpResult;
}
}

42
src/main/java/com/genersoft/iot/vmp/vmanager/server/ServerController.java
View File

@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.VManageBootstrap;
import com.genersoft.iot.vmp.common.VersionPo;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.VersionInfo;
@ -27,6 +28,7 @@ import javax.sip.ObjectInUseException;
import javax.sip.SipProvider;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
@SuppressWarnings("rawtypes")
@Api(tags = "服务控制")
@ -42,13 +44,16 @@ public class ServerController {
private IMediaServerService mediaServerService;
@Autowired
VersionInfo versionInfo;
private VersionInfo versionInfo;
@Autowired
SipConfig sipConfig;
private SipConfig sipConfig;
@Autowired
UserSetting userSetting;
private UserSetting userSetting;
@Autowired
private DynamicTask dynamicTask;
@Value("${server.port}")
private int serverPort;
@ -248,4 +253,35 @@ public class ServerController {
result.setData(jsonObject);
return result;
}
// @ApiOperation("当前进行中的动态任务")
// @GetMapping(value = "/dynamicTask")
// @ResponseBody
// public WVPResult<JSONObject> getDynamicTask(){
// WVPResult<JSONObject> result = new WVPResult<>();
// result.setCode(0);
// result.setMsg("success");
//
// JSONObject jsonObject = new JSONObject();
//
// Set<String> allKeys = dynamicTask.getAllKeys();
// jsonObject.put("server.port", serverPort);
// if (StringUtils.isEmpty(type)) {
// jsonObject.put("sip", JSON.toJSON(sipConfig));
// jsonObject.put("base", JSON.toJSON(userSetting));
// }else {
// switch (type){
// case "sip":
// jsonObject.put("sip", sipConfig);
// break;
// case "base":
// jsonObject.put("base", userSetting);
// break;
// default:
// break;
// }
// }
// result.setData(jsonObject);
// return result;
// }
}

62
web_src/src/components/DeviceList.vue
View File

@ -57,7 +57,7 @@
<el-table-column label="操作" width="450" align="center" fixed="right">
<template slot-scope="scope">
<el-button size="mini" :loading="scope.row.loading" v-if="scope.row.online!=0" icon="el-icon-refresh" @click="refDevice(scope.row)">刷新</el-button>
<el-button size="mini" v-if="scope.row.online!=0" icon="el-icon-refresh" @click="refDevice(scope.row)" @mouseover="getTooltipContent(scope.row.deviceId)">刷新</el-button>
<el-button-group>
<el-button size="mini" icon="el-icon-video-camera-solid" v-bind:disabled="scope.row.online==0" type="primary" @click="showChannelList(scope.row)">通道</el-button>
<el-button size="mini" icon="el-icon-location" v-bind:disabled="scope.row.online==0" type="primary" @click="showDevicePosition(scope.row)">定位</el-button>
@ -78,6 +78,7 @@
:total="total">
</el-pagination>
<deviceEdit ref="deviceEdit" ></deviceEdit>
<syncChannelProgress ref="syncChannelProgress" ></syncChannelProgress>
</el-main>
</el-container>
</div>
@ -86,11 +87,13 @@
<script>
import uiHeader from './UiHeader.vue'
import deviceEdit from './dialog/deviceEdit.vue'
import syncChannelProgress from './dialog/SyncChannelProgress.vue'
export default {
name: 'app',
components: {
uiHeader,
deviceEdit
deviceEdit,
syncChannelProgress,
},
data() {
return {
@ -104,7 +107,7 @@
currentPage:1,
count:15,
total:0,
getDeviceListLoading: false
getDeviceListLoading: false,
};
},
computed: {
@ -117,8 +120,6 @@
});
this.currentDeviceChannelsLenth = channels.length;
}
console.log("数据：" + JSON.stringify(channels));
return channels;
}
},
@ -153,13 +154,11 @@
count: that.count
}
}).then(function (res) {
console.log(res);
console.log(res.data.list);
that.total = res.data.total;
that.deviceList = res.data.list;
that.getDeviceListLoading = false;
}).catch(function (error) {
console.log(error);
console.error(error);
that.getDeviceListLoading = false;
});
@ -182,7 +181,7 @@
}).then((res)=>{
this.getDeviceList();
}).catch((error) =>{
console.log(error);
console.error(error);
});
}).catch(() => {
@ -191,11 +190,9 @@
},
showChannelList: function(row) {
console.log(JSON.stringify(row))
this.$router.push(`/channelList/${row.deviceId}/0/15/1`);
},
showDevicePosition: function(row) {
console.log(JSON.stringify(row))
this.$router.push(`/devicePosition/${row.deviceId}/0/15/1`);
},
@ -203,12 +200,11 @@
//刷新设备信息
refDevice: function(itemData) {
console.log("刷新对应设备:" + itemData.deviceId);
var that = this;
that.$set(itemData,"loading", true);
let that = this;
this.$axios({
method: 'post',
url: '/api/device/query/devices/' + itemData.deviceId + '/sync'
}).then(function(res) {
}).then((res) => {
console.log("刷新设备结果："+JSON.stringify(res));
if (res.data.code !==0) {
that.$message({
@ -217,24 +213,44 @@
type: 'error'
});
}else{
that.$message({
showClose: true,
message: res.data.msg,
type: 'success'
});
// that.$message({
// showClose: true,
// message: res.data.msg,
// type: 'success'
// });
this.$refs.syncChannelProgress.openDialog(itemData.deviceId)
}
that.initData()
that.$set(itemData,"loading", true);
}).catch(function(e) {
}).catch((e) => {
console.error(e)
that.$message({
showClose: true,
message: e,
type: 'error'
});
that.$set(itemData,"loading", true);
});
},
getTooltipContent: async function (deviceId){
let result = "";
await this.$axios({
method: 'get',
async: false,
url:`/api/device/query/${deviceId}/sync_status/`,
}).then((res) => {
if (res.data.code == 0) {
if (res.data.data.errorMsg !== null) {
result = res.data.data.errorMsg
} else if (res.data.msg !== null) {
result = res.data.msg
} else {
result = `同步中...[${res.data.data.current}/${res.data.data.total}]`;
}
}
})
return result;
},
//通知设备上传媒体流
sendDevicePush: function(itemData) {
// let deviceId = this.currentDevice.deviceId;
@ -251,7 +267,6 @@
// });
},
transportChange: function (row) {
console.log(row);
console.log(`修改传输方式为 ${row.streamMode}：${row.deviceId} `);
let that = this;
this.$axios({
@ -263,7 +278,6 @@
});
},
edit: function (row) {
console.log(row);
this.$refs.deviceEdit.openDialog(row, ()=>{
this.$refs.deviceEdit.close();
this.$message({

102
web_src/src/components/dialog/SyncChannelProgress.vue Normal file
View File

@ -0,0 +1,102 @@
<template>
<div id="SyncChannelProgress" v-loading="isLoging">
<el-dialog
width="240px"
top="13%"
:append-to-body="true"
:close-on-click-modal="false"
:visible.sync="showDialog"
:destroy-on-close="true"
:show-close="true"
@close="close()"
style="text-align: center">
<el-progress type="circle" :percentage="percentage" :status="syncStatus"></el-progress>
<div style="text-align: center">
{{msg}}
</div>
</el-dialog>
</div>
</template>
<script>
export default {
name: "SyncChannelProgress",
computed: {},
props: ['platformId'],
created() {},
data() {
return {
syncStatus: null,
percentage: 0,
total: 0,
current: 0,
showDialog: false,
isLoging: false,
syncFlag: false,
deviceId: null,
timmer: null,
msg: "正在同步",
};
},
methods: {
openDialog: function (deviceId) {
console.log("deviceId: " + deviceId)
this.deviceId = deviceId;
this.showDialog = true;
this.msg = "";
this.percentage= 0;
this.total= 0;
this.current= 0;
this.syncFlag= false;
this.syncStatus = null;
this.getProgress()
},
getProgress(){
this.$axios({
method: 'get',
url:`/api/device/query/${this.deviceId}/sync_status/`,
}).then((res) => {
if (res.data.code == 0) {
if (!this.syncFlag) {
this.syncFlag = true;
}
if (res.data.data == null) {
this.syncStatus = "success"
this.percentage = 100;
this.msg = '同步成功';
}else if (res.data.data.total == 0){
this.msg = `等待同步中`;
this.timmer = setTimeout(this.getProgress, 300)
}else if (res.data.data.errorMsg !== null ){
this.msg = res.data.data.errorMsg;
this.syncStatus = "exception"
}else {
this.total = res.data.data.total;
this.current = res.data.data.current;
this.percentage = Math.floor(Number(res.data.data.current)/Number(res.data.data.total)* 10000)/100;
this.msg = `同步中...[${res.data.data.current}/${res.data.data.total}]`;
this.timmer = setTimeout(this.getProgress, 300)
}
}else {
if (this.syncFlag) {
this.syncStatus = "success"
this.percentage = 100;
this.msg = '同步成功';
}else {
this.syncStatus = "error"
this.msg = res.data.msg;
}
}
}).catch((error) =>{
console.log(error);
this.syncStatus = "error"
this.msg = error.response.data.msg;
});
},
close: function (){
window.clearTimeout(this.timmer)
}
},
};
</script>