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

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

Browse Source
This commit is contained in:
648540858
2023-02-01 09:47:58 +08:00
committed by GitHub
parent 141b228267 44b250eed2
commit bdf799fb64
26 changed files with 1016 additions and 862 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -109,12 +109,18 @@ public class CatalogDataCatch {
for (String deviceId : keys) {
CatalogData catalogData = data.get(deviceId);
if ( catalogData.getLastTime().isBefore(instantBefore5S)) { // 超过五秒收不到消息任务超时， 只更新这一部分数据
if ( catalogData.getLastTime().isBefore(instantBefore5S)) {
// 超过五秒收不到消息任务超时， 只更新这一部分数据, 收到数据与声明的总数一致，则重置通道数据，数据不全则只对收到的数据做更新操作
if (catalogData.getStatus().equals(CatalogData.CatalogDataStatus.runIng)) {
storager.resetChannels(catalogData.getDevice().getDeviceId(), catalogData.getChannelList());
if (catalogData.getTotal() == catalogData.getChannelList().size()) {
storager.resetChannels(catalogData.getDevice().getDeviceId(), catalogData.getChannelList());
}else {
storager.updateChannels(catalogData.getDevice().getDeviceId(), catalogData.getChannelList());
}
String errorMsg = "更新成功，共" + catalogData.getTotal() + "条，已更新" + catalogData.getChannelList().size() + "条";
catalogData.setErrorMsg(errorMsg);
if (catalogData.getTotal() != catalogData.getChannelList().size()) {
String errorMsg = "更新成功，共" + catalogData.getTotal() + "条，已更新" + catalogData.getChannelList().size() + "条";
catalogData.setErrorMsg(errorMsg);
}
}else if (catalogData.getStatus().equals(CatalogData.CatalogDataStatus.ready)) {
String errorMsg = "同步失败，等待回复超时";

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

@@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.gb28181.transmit.cmd.impl;
import com.alibaba.fastjson2.JSON;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.gb28181.SipLayer;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
@@ -61,6 +62,9 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
@Autowired
private SIPSender sipSender;
@Autowired
private DynamicTask dynamicTask;
@Override
public void register(ParentPlatform parentPlatform, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent) throws InvalidArgumentException, ParseException, SipException {
register(parentPlatform, null, null, errorEvent, okEvent, false, true);
@@ -109,13 +113,14 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
public String keepalive(ParentPlatform parentPlatform,SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent) throws SipException, InvalidArgumentException, ParseException {
String characterSet = parentPlatform.getCharacterSet();
StringBuffer keepaliveXml = new StringBuffer(200);
keepaliveXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n");
keepaliveXml.append("<Notify>\r\n");
keepaliveXml.append("<CmdType>Keepalive</CmdType>\r\n");
keepaliveXml.append("<SN>" + (int)((Math.random()*9+1)*100000) + "</SN>\r\n");
keepaliveXml.append("<DeviceID>" + parentPlatform.getDeviceGBId() + "</DeviceID>\r\n");
keepaliveXml.append("<Status>OK</Status>\r\n");
keepaliveXml.append("</Notify>\r\n");
keepaliveXml.append("<?xml version=\"1.0\" encoding=\"")
.append(characterSet).append("\"?>\r\n")
.append("<Notify>\r\n")
.append("<CmdType>Keepalive</CmdType>\r\n")
.append("<SN>" + (int)((Math.random()*9+1)*100000) + "</SN>\r\n")
.append("<DeviceID>" + parentPlatform.getDeviceGBId() + "</DeviceID>\r\n")
.append("<Status>OK</Status>\r\n")
.append("</Notify>\r\n");
CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(parentPlatform.getDeviceIp(),parentPlatform.getTransport());
@@ -133,7 +138,6 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
* 向上级回复通道信息
* @param channel 通道信息
* @param parentPlatform 平台信息
* @return
*/
@Override
public void catalogQuery(DeviceChannel channel, ParentPlatform parentPlatform, String sn, String fromTag, int size) throws SipException, InvalidArgumentException, ParseException {
@@ -160,18 +164,18 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
if ( parentPlatform ==null) {
return ;
}
sendCatalogResponse(channels, parentPlatform, sn, fromTag, 0);
sendCatalogResponse(channels, parentPlatform, sn, fromTag, 0, true);
}
private String getCatalogXml(List<DeviceChannel> channels, 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=\"" + channels.size() +"\">\r\n");
catalogXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet +"\"?>\r\n")
.append("<Response>\r\n")
.append("<CmdType>Catalog</CmdType>\r\n")
.append("<SN>" +sn + "</SN>\r\n")
.append("<DeviceID>" + parentPlatform.getDeviceGBId() + "</DeviceID>\r\n")
.append("<SumNum>" + size + "</SumNum>\r\n")
.append("<DeviceList Num=\"" + channels.size() +"\">\r\n");
if (channels.size() > 0) {
for (DeviceChannel channel : channels) {
if (parentPlatform.getServerGBId().equals(channel.getParentId())) {
@@ -222,7 +226,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
return catalogXml.toString();
}
private void sendCatalogResponse(List<DeviceChannel> channels, ParentPlatform parentPlatform, String sn, String fromTag, int index) throws SipException, InvalidArgumentException, ParseException {
private void sendCatalogResponse(List<DeviceChannel> channels, ParentPlatform parentPlatform, String sn, String fromTag, int index, boolean sendAfterResponse) throws SipException, InvalidArgumentException, ParseException {
if (index >= channels.size()) {
return;
}
@@ -236,15 +240,49 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
// callid
CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(parentPlatform.getDeviceIp(),parentPlatform.getTransport());
Request request = headerProviderPlatformProvider.createMessageRequest(parentPlatform, catalogXml, fromTag, SipUtils.getNewViaTag(), callIdHeader);
sipSender.transmitRequest(parentPlatform.getDeviceIp(), request, null, eventResult -> {
int indexNext = index + parentPlatform.getCatalogGroup();
try {
sendCatalogResponse(channels, parentPlatform, sn, fromTag, indexNext);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 目录查询回复: {}", e.getMessage());
}
});
SIPRequest request = (SIPRequest)headerProviderPlatformProvider.createMessageRequest(parentPlatform, catalogXml, fromTag, SipUtils.getNewViaTag(), callIdHeader);
String timeoutTaskKey = "catalog_task_" + parentPlatform.getServerGBId() + sn;
String callId = request.getCallIdHeader().getCallId();
if (sendAfterResponse) {
// 默认按照收到200回复后发送下一条， 如果超时收不到回复，就以30毫秒的间隔直接发送。
dynamicTask.startDelay(timeoutTaskKey, ()->{
sipSubscribe.removeOkSubscribe(callId);
int indexNext = index + parentPlatform.getCatalogGroup();
try {
sendCatalogResponse(channels, parentPlatform, sn, fromTag, indexNext, false);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 目录查询回复: {}", e.getMessage());
}
}, 3000);
sipSender.transmitRequest(parentPlatform.getDeviceIp(), request, eventResult -> {
logger.error("[目录推送失败] 国标级联 platform : {}, code: {}, msg: {}, 停止发送", parentPlatform.getServerGBId(), eventResult.statusCode, eventResult.msg);
dynamicTask.stop(timeoutTaskKey);
}, eventResult -> {
dynamicTask.stop(timeoutTaskKey);
int indexNext = index + parentPlatform.getCatalogGroup();
try {
sendCatalogResponse(channels, parentPlatform, sn, fromTag, indexNext, true);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 目录查询回复: {}", e.getMessage());
}
});
}else {
sipSender.transmitRequest(parentPlatform.getDeviceIp(), request, eventResult -> {
logger.error("[目录推送失败] 国标级联 platform : {}, code: {}, msg: {}, 停止发送", parentPlatform.getServerGBId(), eventResult.statusCode, eventResult.msg);
dynamicTask.stop(timeoutTaskKey);
}, null);
dynamicTask.startDelay(timeoutTaskKey, ()->{
int indexNext = index + parentPlatform.getCatalogGroup();
try {
sendCatalogResponse(channels, parentPlatform, sn, fromTag, indexNext, false);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 目录查询回复: {}", e.getMessage());
}
}, 30);
}
}
/**
@@ -294,15 +332,15 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
String statusStr = (status==1)?"ONLINE":"OFFLINE";
String characterSet = parentPlatform.getCharacterSet();
StringBuffer deviceStatusXml = new StringBuffer(600);
deviceStatusXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n");
deviceStatusXml.append("<Response>\r\n");
deviceStatusXml.append("<CmdType>DeviceStatus</CmdType>\r\n");
deviceStatusXml.append("<SN>" +sn + "</SN>\r\n");
deviceStatusXml.append("<DeviceID>" + channelId + "</DeviceID>\r\n");
deviceStatusXml.append("<Result>OK</Result>\r\n");
deviceStatusXml.append("<Online>"+statusStr+"</Online>\r\n");
deviceStatusXml.append("<Status>OK</Status>\r\n");
deviceStatusXml.append("</Response>\r\n");
deviceStatusXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n")
.append("<Response>\r\n")
.append("<CmdType>DeviceStatus</CmdType>\r\n")
.append("<SN>" +sn + "</SN>\r\n")
.append("<DeviceID>" + channelId + "</DeviceID>\r\n")
.append("<Result>OK</Result>\r\n")
.append("<Online>"+statusStr+"</Online>\r\n")
.append("<Status>OK</Status>\r\n")
.append("</Response>\r\n");
CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(parentPlatform.getDeviceIp(),parentPlatform.getTransport());
@@ -321,18 +359,18 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
String characterSet = parentPlatform.getCharacterSet();
StringBuffer deviceStatusXml = new StringBuffer(600);
deviceStatusXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n");
deviceStatusXml.append("<Notify>\r\n");
deviceStatusXml.append("<CmdType>MobilePosition</CmdType>\r\n");
deviceStatusXml.append("<SN>" + (int)((Math.random()*9+1)*100000) + "</SN>\r\n");
deviceStatusXml.append("<DeviceID>" + gpsMsgInfo.getId() + "</DeviceID>\r\n");
deviceStatusXml.append("<Time>" + gpsMsgInfo.getTime() + "</Time>\r\n");
deviceStatusXml.append("<Longitude>" + gpsMsgInfo.getLng() + "</Longitude>\r\n");
deviceStatusXml.append("<Latitude>" + gpsMsgInfo.getLat() + "</Latitude>\r\n");
deviceStatusXml.append("<Speed>" + gpsMsgInfo.getSpeed() + "</Speed>\r\n");
deviceStatusXml.append("<Direction>" + gpsMsgInfo.getDirection() + "</Direction>\r\n");
deviceStatusXml.append("<Altitude>" + gpsMsgInfo.getAltitude() + "</Altitude>\r\n");
deviceStatusXml.append("</Notify>\r\n");
deviceStatusXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n")
.append("<Notify>\r\n")
.append("<CmdType>MobilePosition</CmdType>\r\n")
.append("<SN>" + (int)((Math.random()*9+1)*100000) + "</SN>\r\n")
.append("<DeviceID>" + gpsMsgInfo.getId() + "</DeviceID>\r\n")
.append("<Time>" + gpsMsgInfo.getTime() + "</Time>\r\n")
.append("<Longitude>" + gpsMsgInfo.getLng() + "</Longitude>\r\n")
.append("<Latitude>" + gpsMsgInfo.getLat() + "</Latitude>\r\n")
.append("<Speed>" + gpsMsgInfo.getSpeed() + "</Speed>\r\n")
.append("<Direction>" + gpsMsgInfo.getDirection() + "</Direction>\r\n")
.append("<Altitude>" + gpsMsgInfo.getAltitude() + "</Altitude>\r\n")
.append("</Notify>\r\n");
sendNotify(parentPlatform, deviceStatusXml.toString(), subscribeInfo, eventResult -> {
logger.error("发送NOTIFY通知消息失败。错误：{} {}", eventResult.statusCode, eventResult.msg);
@@ -349,21 +387,21 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
deviceAlarm.getLongitude(), deviceAlarm.getLatitude(), JSON.toJSONString(deviceAlarm));
String characterSet = parentPlatform.getCharacterSet();
StringBuffer deviceStatusXml = new StringBuffer(600);
deviceStatusXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n");
deviceStatusXml.append("<Notify>\r\n");
deviceStatusXml.append("<CmdType>Alarm</CmdType>\r\n");
deviceStatusXml.append("<SN>" + (int)((Math.random()*9+1)*100000) + "</SN>\r\n");
deviceStatusXml.append("<DeviceID>" + deviceAlarm.getChannelId() + "</DeviceID>\r\n");
deviceStatusXml.append("<AlarmPriority>" + deviceAlarm.getAlarmPriority() + "</AlarmPriority>\r\n");
deviceStatusXml.append("<AlarmMethod>" + deviceAlarm.getAlarmMethod() + "</AlarmMethod>\r\n");
deviceStatusXml.append("<AlarmTime>" + deviceAlarm.getAlarmTime() + "</AlarmTime>\r\n");
deviceStatusXml.append("<AlarmDescription>" + deviceAlarm.getAlarmDescription() + "</AlarmDescription>\r\n");
deviceStatusXml.append("<Longitude>" + deviceAlarm.getLongitude() + "</Longitude>\r\n");
deviceStatusXml.append("<Latitude>" + deviceAlarm.getLatitude() + "</Latitude>\r\n");
deviceStatusXml.append("<info>\r\n");
deviceStatusXml.append("<AlarmType>" + deviceAlarm.getAlarmType() + "</AlarmType>\r\n");
deviceStatusXml.append("</info>\r\n");
deviceStatusXml.append("</Notify>\r\n");
deviceStatusXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n")
.append("<Notify>\r\n")
.append("<CmdType>Alarm</CmdType>\r\n")
.append("<SN>" + (int)((Math.random()*9+1)*100000) + "</SN>\r\n")
.append("<DeviceID>" + deviceAlarm.getChannelId() + "</DeviceID>\r\n")
.append("<AlarmPriority>" + deviceAlarm.getAlarmPriority() + "</AlarmPriority>\r\n")
.append("<AlarmMethod>" + deviceAlarm.getAlarmMethod() + "</AlarmMethod>\r\n")
.append("<AlarmTime>" + deviceAlarm.getAlarmTime() + "</AlarmTime>\r\n")
.append("<AlarmDescription>" + deviceAlarm.getAlarmDescription() + "</AlarmDescription>\r\n")
.append("<Longitude>" + deviceAlarm.getLongitude() + "</Longitude>\r\n")
.append("<Latitude>" + deviceAlarm.getLatitude() + "</Latitude>\r\n")
.append("<info>\r\n")
.append("<AlarmType>" + deviceAlarm.getAlarmType() + "</AlarmType>\r\n")
.append("</info>\r\n")
.append("</Notify>\r\n");
CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(parentPlatform.getDeviceIp(),parentPlatform.getTransport());
@@ -422,13 +460,13 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
StringBuffer catalogXml = new StringBuffer(600);
String characterSet = parentPlatform.getCharacterSet();
catalogXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n");
catalogXml.append("<Notify>\r\n");
catalogXml.append("<CmdType>Catalog</CmdType>\r\n");
catalogXml.append("<SN>" + (int) ((Math.random() * 9 + 1) * 100000) + "</SN>\r\n");
catalogXml.append("<DeviceID>" + parentPlatform.getDeviceGBId() + "</DeviceID>\r\n");
catalogXml.append("<SumNum>1</SumNum>\r\n");
catalogXml.append("<DeviceList Num=\"" + channels.size() + "\">\r\n");
catalogXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n")
.append("<Notify>\r\n")
.append("<CmdType>Catalog</CmdType>\r\n")
.append("<SN>" + (int) ((Math.random() * 9 + 1) * 100000) + "</SN>\r\n")
.append("<DeviceID>" + parentPlatform.getDeviceGBId() + "</DeviceID>\r\n")
.append("<SumNum>1</SumNum>\r\n")
.append("<DeviceList Num=\"" + channels.size() + "\">\r\n");
if (channels.size() > 0) {
for (DeviceChannel channel : channels) {
if (parentPlatform.getServerGBId().equals(channel.getParentId())) {
@@ -449,16 +487,16 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
catalogXml.append("<Parental>" + channel.getParental() + "</Parental>\r\n");
if (channel.getParental() == 0) {
// 通道项
catalogXml.append("<Manufacturer>" + channel.getManufacture() + "</Manufacturer>\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("<Manufacturer>" + channel.getManufacture() + "</Manufacturer>\r\n")
.append("<Secrecy>" + channel.getSecrecy() + "</Secrecy>\r\n")
.append("<RegisterWay>" + channel.getRegisterWay() + "</RegisterWay>\r\n")
.append("<Status>" + (channel.getStatus() == 0 ? "OFF" : "ON") + "</Status>\r\n");
if (channel.getChannelType() != 2) { // 业务分组/虚拟组织/行政区划 不设置以下属性
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("<Model>" + channel.getModel() + "</Model>\r\n")
.append("<Owner> " + channel.getOwner()+ "</Owner>\r\n")
.append("<CivilCode>" + channel.getCivilCode() + "</CivilCode>\r\n")
.append("<Address>" + channel.getAddress() + "</Address>\r\n");
}
if (!"presence".equals(subscribeInfo.getEventType())) {
catalogXml.append("<Event>" + type + "</Event>\r\n");
@@ -468,8 +506,8 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
catalogXml.append("</Item>\r\n");
}
}
catalogXml.append("</DeviceList>\r\n");
catalogXml.append("</Notify>\r\n");
catalogXml.append("</DeviceList>\r\n")
.append("</Notify>\r\n");
return catalogXml.toString();
}
@@ -515,26 +553,26 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
String characterSet = parentPlatform.getCharacterSet();
StringBuffer catalogXml = new StringBuffer(600);
catalogXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n");
catalogXml.append("<Notify>\r\n");
catalogXml.append("<CmdType>Catalog</CmdType>\r\n");
catalogXml.append("<SN>" + (int) ((Math.random() * 9 + 1) * 100000) + "</SN>\r\n");
catalogXml.append("<DeviceID>" + parentPlatform.getDeviceGBId() + "</DeviceID>\r\n");
catalogXml.append("<SumNum>1</SumNum>\r\n");
catalogXml.append("<DeviceList Num=\" " + channels.size() + " \">\r\n");
catalogXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n")
.append("<Notify>\r\n")
.append("<CmdType>Catalog</CmdType>\r\n")
.append("<SN>" + (int) ((Math.random() * 9 + 1) * 100000) + "</SN>\r\n")
.append("<DeviceID>" + parentPlatform.getDeviceGBId() + "</DeviceID>\r\n")
.append("<SumNum>1</SumNum>\r\n")
.append("<DeviceList Num=\" " + channels.size() + " \">\r\n");
if (channels.size() > 0) {
for (DeviceChannel channel : channels) {
if (parentPlatform.getServerGBId().equals(channel.getParentId())) {
channel.setParentId(parentPlatform.getDeviceGBId());
}
catalogXml.append("<Item>\r\n");
catalogXml.append("<DeviceID>" + channel.getChannelId() + "</DeviceID>\r\n");
catalogXml.append("<Event>" + type + "</Event>\r\n");
catalogXml.append("</Item>\r\n");
catalogXml.append("<Item>\r\n")
.append("<DeviceID>" + channel.getChannelId() + "</DeviceID>\r\n")
.append("<Event>" + type + "</Event>\r\n")
.append("</Item>\r\n");
}
}
catalogXml.append("</DeviceList>\r\n");
catalogXml.append("</Notify>\r\n");
catalogXml.append("</DeviceList>\r\n")
.append("</Notify>\r\n");
return catalogXml.toString();
}
@Override
@@ -544,12 +582,12 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
}
String characterSet = parentPlatform.getCharacterSet();
StringBuffer recordXml = new StringBuffer(600);
recordXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n");
recordXml.append("<Response>\r\n");
recordXml.append("<CmdType>RecordInfo</CmdType>\r\n");
recordXml.append("<SN>" +recordInfo.getSn() + "</SN>\r\n");
recordXml.append("<DeviceID>" + recordInfo.getDeviceId() + "</DeviceID>\r\n");
recordXml.append("<SumNum>" + recordInfo.getSumNum() + "</SumNum>\r\n");
recordXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n")
.append("<Response>\r\n")
.append("<CmdType>RecordInfo</CmdType>\r\n")
.append("<SN>" +recordInfo.getSn() + "</SN>\r\n")
.append("<DeviceID>" + recordInfo.getDeviceId() + "</DeviceID>\r\n")
.append("<SumNum>" + recordInfo.getSumNum() + "</SumNum>\r\n");
if (recordInfo.getRecordList() == null ) {
recordXml.append("<RecordList Num=\"0\">\r\n");
}else {
@@ -558,12 +596,12 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
for (RecordItem recordItem : recordInfo.getRecordList()) {
recordXml.append("<Item>\r\n");
if (deviceChannel != null) {
recordXml.append("<DeviceID>" + recordItem.getDeviceId() + "</DeviceID>\r\n");
recordXml.append("<Name>" + recordItem.getName() + "</Name>\r\n");
recordXml.append("<StartTime>" + DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(recordItem.getStartTime()) + "</StartTime>\r\n");
recordXml.append("<EndTime>" + DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(recordItem.getEndTime()) + "</EndTime>\r\n");
recordXml.append("<Secrecy>" + recordItem.getSecrecy() + "</Secrecy>\r\n");
recordXml.append("<Type>" + recordItem.getType() + "</Type>\r\n");
recordXml.append("<DeviceID>" + recordItem.getDeviceId() + "</DeviceID>\r\n")
.append("<Name>" + recordItem.getName() + "</Name>\r\n")
.append("<StartTime>" + DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(recordItem.getStartTime()) + "</StartTime>\r\n")
.append("<EndTime>" + DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(recordItem.getEndTime()) + "</EndTime>\r\n")
.append("<Secrecy>" + recordItem.getSecrecy() + "</Secrecy>\r\n")
.append("<Type>" + recordItem.getType() + "</Type>\r\n");
if (!ObjectUtils.isEmpty(recordItem.getFileSize())) {
recordXml.append("<FileSize>" + recordItem.getFileSize() + "</FileSize>\r\n");
}
@@ -576,8 +614,8 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
}
}
recordXml.append("</RecordList>\r\n");
recordXml.append("</Response>\r\n");
recordXml.append("</RecordList>\r\n")
.append("</Response>\r\n");
// callid
CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(parentPlatform.getDeviceIp(),parentPlatform.getTransport());
@@ -596,13 +634,13 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
String characterSet = parentPlatform.getCharacterSet();
StringBuffer mediaStatusXml = new StringBuffer(200);
mediaStatusXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n");
mediaStatusXml.append("<Notify>\r\n");
mediaStatusXml.append("<CmdType>MediaStatus</CmdType>\r\n");
mediaStatusXml.append("<SN>" + (int)((Math.random()*9+1)*100000) + "</SN>\r\n");
mediaStatusXml.append("<DeviceID>" + sendRtpItem.getChannelId() + "</DeviceID>\r\n");
mediaStatusXml.append("<NotifyType>121</NotifyType>\r\n");
mediaStatusXml.append("</Notify>\r\n");
mediaStatusXml.append("<?xml version=\"1.0\" encoding=\"" + characterSet + "\"?>\r\n")
.append("<Notify>\r\n")
.append("<CmdType>MediaStatus</CmdType>\r\n")
.append("<SN>" + (int)((Math.random()*9+1)*100000) + "</SN>\r\n")
.append("<DeviceID>" + sendRtpItem.getChannelId() + "</DeviceID>\r\n")
.append("<NotifyType>121</NotifyType>\r\n")
.append("</Notify>\r\n");
SIPRequest messageRequest = (SIPRequest)headerProviderPlatformProvider.createMessageRequest(parentPlatform, mediaStatusXml.toString(),
sendRtpItem);

65
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/SIPRequestProcessorParent.java
View File

@@ -3,7 +3,6 @@ package com.genersoft.iot.vmp.gb28181.transmit.event.request;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.SIPSender;
import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import gov.nist.javax.sip.SipProviderImpl;
import gov.nist.javax.sip.message.SIPRequest;
import gov.nist.javax.sip.message.SIPResponse;
import org.apache.commons.lang3.ArrayUtils;
@@ -14,19 +13,17 @@ import org.dom4j.io.SAXReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import javax.sip.*;
import javax.sip.address.Address;
import javax.sip.address.AddressFactory;
import javax.sip.address.SipURI;
import javax.sip.header.*;
import javax.sip.header.ContentTypeHeader;
import javax.sip.header.ExpiresHeader;
import javax.sip.header.HeaderFactory;
import javax.sip.message.MessageFactory;
import javax.sip.message.Request;
import javax.sip.message.Response;
import java.io.ByteArrayInputStream;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Arrays;
@@ -44,15 +41,6 @@ public abstract class SIPRequestProcessorParent {
@Autowired
private SIPSender sipSender;
public AddressFactory getAddressFactory() {
try {
return SipFactory.getInstance().createAddressFactory();
} catch (PeerUnavailableException e) {
e.printStackTrace();
}
return null;
}
public HeaderFactory getHeaderFactory() {
try {
return SipFactory.getInstance().createHeaderFactory();
@@ -93,53 +81,6 @@ public abstract class SIPRequestProcessorParent {
return responseAck(sipRequest, statusCode, msg, null);
}
// public SIPResponse responseAck(ServerTransaction serverTransaction, int statusCode, String msg, ResponseAckExtraParam responseAckExtraParam) throws SipException, InvalidArgumentException, ParseException {
// if (serverTransaction == null) {
// logger.warn("[回复消息] ServerTransaction 为null");
// return null;
// }
// ToHeader toHeader = (ToHeader) serverTransaction.getRequest().getHeader(ToHeader.NAME);
// if (toHeader.getTag() == null) {
// toHeader.setTag(SipUtils.getNewTag());
// }
// SIPResponse response = (SIPResponse)getMessageFactory().createResponse(statusCode, serverTransaction.getRequest());
// if (msg != null) {
// response.setReasonPhrase(msg);
// }
// if (responseAckExtraParam != null) {
// if (responseAckExtraParam.sipURI != null && serverTransaction.getRequest().getMethod().equals(Request.INVITE)) {
// logger.debug("responseSdpAck SipURI: {}:{}", responseAckExtraParam.sipURI.getHost(), responseAckExtraParam.sipURI.getPort());
// Address concatAddress = SipFactory.getInstance().createAddressFactory().createAddress(
// SipFactory.getInstance().createAddressFactory().createSipURI(responseAckExtraParam.sipURI.getUser(), responseAckExtraParam.sipURI.getHost()+":"+responseAckExtraParam.sipURI.getPort()
// ));
// response.addHeader(SipFactory.getInstance().createHeaderFactory().createContactHeader(concatAddress));
// }
// if (responseAckExtraParam.contentTypeHeader != null) {
// response.setContent(responseAckExtraParam.content, responseAckExtraParam.contentTypeHeader);
// }
//
// if (serverTransaction.getRequest().getMethod().equals(Request.SUBSCRIBE)) {
// if (responseAckExtraParam.expires == -1) {
// logger.error("[参数不全] 2xx的SUBSCRIBE回复，必须设置Expires header");
// }else {
// ExpiresHeader expiresHeader = SipFactory.getInstance().createHeaderFactory().createExpiresHeader(responseAckExtraParam.expires);
// response.addHeader(expiresHeader);
// }
// }
// }else {
// if (serverTransaction.getRequest().getMethod().equals(Request.SUBSCRIBE)) {
// logger.error("[参数不全] 2xx的SUBSCRIBE回复，必须设置Expires header");
// }
// }
// serverTransaction.sendResponse(response);
// if (statusCode >= 200 && !"NOTIFY".equalsIgnoreCase(serverTransaction.getRequest().getMethod())) {
// if (serverTransaction.getDialog() != null) {
// serverTransaction.getDialog().delete();
// }
// }
// return response;
// }
public SIPResponse responseAck(SIPRequest sipRequest, int statusCode, String msg, ResponseAckExtraParam responseAckExtraParam) throws SipException, InvalidArgumentException, ParseException {
if (sipRequest.getToHeader().getTag() == null) {
sipRequest.getToHeader().setTag(SipUtils.getNewTag());

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

@@ -228,7 +228,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
}
return;
} else {
logger.info("通道不存在，返回404");
logger.info("通道不存在，返回404: {}", channelId);
try {
// 通道不存在，发404，资源不存在
responseAck(request, Response.NOT_FOUND);

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

@@ -1,9 +1,10 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.query.cmd;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.IMessageHandler;
@@ -63,7 +64,6 @@ public class CatalogQueryMessageHandler extends SIPRequestProcessorParent implem
@Override
public void handForPlatform(RequestEvent evt, ParentPlatform parentPlatform, Element rootElement) {
String key = DeferredResultHolder.CALLBACK_CMD_CATALOG + parentPlatform.getServerGBId();
FromHeader fromHeader = (FromHeader) evt.getRequest().getHeader(FromHeader.NAME);
try {
// 回复200 OK

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

File diff suppressed because it is too large Load Diff

36
src/main/java/com/genersoft/iot/vmp/media/zlm/dto/hook/HookResult.java Normal file
View File

@@ -0,0 +1,36 @@
package com.genersoft.iot.vmp.media.zlm.dto.hook;
public class HookResult {
private int code;
private String msg;
public HookResult() {
}
public HookResult(int code, String msg) {
this.code = code;
this.msg = msg;
}
public static HookResult SUCCESS(){
return new HookResult(0, "success");
}
public int getCode() {
return code;
}
public void setCode(int code) {
this.code = code;
}
public String getMsg() {
return msg;
}
public void setMsg(String msg) {
this.msg = msg;
}
}

44
src/main/java/com/genersoft/iot/vmp/media/zlm/dto/hook/HookResultForOnPublish.java Normal file
View File

@@ -0,0 +1,44 @@
package com.genersoft.iot.vmp.media.zlm.dto.hook;
public class HookResultForOnPublish extends HookResult{
private boolean enable_audio;
private boolean enable_mp4;
private int mp4_max_second;
public HookResultForOnPublish() {
}
public static HookResultForOnPublish SUCCESS(){
return new HookResultForOnPublish(0, "success");
}
public HookResultForOnPublish(int code, String msg) {
setCode(code);
setMsg(msg);
}
public boolean isEnable_audio() {
return enable_audio;
}
public void setEnable_audio(boolean enable_audio) {
this.enable_audio = enable_audio;
}
public boolean isEnable_mp4() {
return enable_mp4;
}
public void setEnable_mp4(boolean enable_mp4) {
this.enable_mp4 = enable_mp4;
}
public int getMp4_max_second() {
return mp4_max_second;
}
public void setMp4_max_second(int mp4_max_second) {
this.mp4_max_second = mp4_max_second;
}
}

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

@@ -152,6 +152,10 @@ public class GbStreamServiceImpl implements IGbStreamService {
@Override
public void sendCatalogMsg(GbStream gbStream, String type) {
if (gbStream == null || type == null) {
logger.warn("[发送目录订阅]类型：流信息或类型为NULL");
return;
}
List<GbStream> gbStreams = new ArrayList<>();
if (gbStream.getGbId() != null) {
gbStreams.add(gbStream);

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

@@ -184,7 +184,9 @@ public class StreamPushServiceImpl implements IStreamPushService {
@Override
public boolean stop(String app, String streamId) {
StreamPushItem streamPushItem = streamPushMapper.selectOne(app, streamId);
gbStreamService.sendCatalogMsg(streamPushItem, CatalogEvent.DEL);
if (streamPushItem != null) {
gbStreamService.sendCatalogMsg(streamPushItem, CatalogEvent.DEL);
}
platformGbStreamMapper.delByAppAndStream(app, streamId);
gbStreamMapper.del(app, streamId);

3
src/main/java/com/genersoft/iot/vmp/storager/IVideoManagerStorage.java
View File

@@ -2,7 +2,6 @@ package com.genersoft.iot.vmp.storager;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.storager.dao.dto.ChannelSourceInfo;
import com.genersoft.iot.vmp.vmanager.gb28181.platform.bean.ChannelReduce;
@@ -330,6 +329,8 @@ public interface IVideoManagerStorage {
*/
boolean resetChannels(String deviceId, List<DeviceChannel> deviceChannelList);
boolean updateChannels(String deviceId, List<DeviceChannel> deviceChannelList);
/**
* 获取目录信息
* @param platformId

2
src/main/java/com/genersoft/iot/vmp/storager/dao/UserMapper.java
View File

@@ -50,7 +50,7 @@ public interface UserMapper {
@ResultMap(value="roleMap")
List<User> selectAll();
@Select("select * from (select user.*, concat(#{callId}_', pushKey) as str1 from user) as u where md5(u.str1) = #{sign}")
@Select("select * from (select user.*, concat(concat(#{callId}, '_'), pushKey) as str1 from user) as u where md5(u.str1) = #{sign}")
List<User> checkPushAuthorityByCallIdAndSign(String callId, String sign);
@Select("select * from user where md5(pushKey) = #{sign}")

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

@@ -194,6 +194,119 @@ public class VideoManagerStorageImpl implements IVideoManagerStorage {
}
@Override
public boolean updateChannels(String deviceId, List<DeviceChannel> deviceChannelList) {
if (CollectionUtils.isEmpty(deviceChannelList)) {
return false;
}
List<DeviceChannel> allChannels = deviceChannelMapper.queryAllChannels(deviceId);
Map<String,DeviceChannel> allChannelMap = new ConcurrentHashMap<>();
if (allChannels.size() > 0) {
for (DeviceChannel deviceChannel : allChannels) {
allChannelMap.put(deviceChannel.getChannelId(), deviceChannel);
}
}
List<DeviceChannel> addChannels = new ArrayList<>();
List<DeviceChannel> updateChannels = new ArrayList<>();
TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
// 数据去重
StringBuilder stringBuilder = new StringBuilder();
Map<String, Integer> subContMap = new HashMap<>();
if (deviceChannelList.size() > 0) {
// 数据去重
Set<String> gbIdSet = new HashSet<>();
for (DeviceChannel deviceChannel : deviceChannelList) {
if (!gbIdSet.contains(deviceChannel.getChannelId())) {
gbIdSet.add(deviceChannel.getChannelId());
if (allChannelMap.containsKey(deviceChannel.getChannelId())) {
deviceChannel.setStreamId(allChannelMap.get(deviceChannel.getChannelId()).getStreamId());
deviceChannel.setHasAudio(allChannelMap.get(deviceChannel.getChannelId()).isHasAudio());
updateChannels.add(deviceChannel);
}else {
addChannels.add(deviceChannel);
}
if (!ObjectUtils.isEmpty(deviceChannel.getParentId())) {
if (subContMap.get(deviceChannel.getParentId()) == null) {
subContMap.put(deviceChannel.getParentId(), 1);
}else {
Integer count = subContMap.get(deviceChannel.getParentId());
subContMap.put(deviceChannel.getParentId(), count++);
}
}
}else {
stringBuilder.append(deviceChannel.getChannelId()).append(",");
}
}
if (addChannels.size() > 0) {
for (DeviceChannel channel : addChannels) {
if (subContMap.get(channel.getChannelId()) != null){
channel.setSubCount(subContMap.get(channel.getChannelId()));
}
}
}
if (updateChannels.size() > 0) {
for (DeviceChannel channel : updateChannels) {
if (subContMap.get(channel.getChannelId()) != null){
channel.setSubCount(subContMap.get(channel.getChannelId()));
}
}
}
}
if (stringBuilder.length() > 0) {
logger.info("[目录查询]收到的数据存在重复： {}" , stringBuilder);
}
if(CollectionUtils.isEmpty(updateChannels) && CollectionUtils.isEmpty(addChannels) ){
logger.info("通道更新，数据为空={}" , deviceChannelList);
return false;
}
try {
int limitCount = 300;
boolean result = false;
if (addChannels.size() > 0) {
if (addChannels.size() > limitCount) {
for (int i = 0; i < addChannels.size(); i += limitCount) {
int toIndex = i + limitCount;
if (i + limitCount > addChannels.size()) {
toIndex = addChannels.size();
}
result = result || deviceChannelMapper.batchAdd(addChannels.subList(i, toIndex)) < 0;
}
}else {
result = result || deviceChannelMapper.batchAdd(addChannels) < 0;
}
}
if (updateChannels.size() > 0) {
if (updateChannels.size() > limitCount) {
for (int i = 0; i < updateChannels.size(); i += limitCount) {
int toIndex = i + limitCount;
if (i + limitCount > updateChannels.size()) {
toIndex = updateChannels.size();
}
result = result || deviceChannelMapper.batchUpdate(updateChannels.subList(i, toIndex)) < 0;
}
}else {
result = result || deviceChannelMapper.batchUpdate(updateChannels) < 0;
}
}
if (result) {
//事务回滚
dataSourceTransactionManager.rollback(transactionStatus);
}else {
//手动提交
dataSourceTransactionManager.commit(transactionStatus);
}
return true;
}catch (Exception e) {
e.printStackTrace();
dataSourceTransactionManager.rollback(transactionStatus);
return false;
}
}
@Override
public void deviceChannelOnline(String deviceId, String channelId) {
deviceChannelMapper.online(deviceId, channelId);

7
src/main/java/com/genersoft/iot/vmp/utils/redis/RedisUtil.java
View File

@@ -1,14 +1,13 @@
package com.genersoft.iot.vmp.utils.redis;
import java.util.*;
import java.util.concurrent.TimeUnit;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.utils.SpringBeanFactory;
import gov.nist.javax.sip.stack.UDPMessageChannel;
import org.springframework.data.redis.core.*;
import org.springframework.util.CollectionUtils;
import java.util.*;
import java.util.concurrent.TimeUnit;
/**
* Redis工具类
* @author swwheihei

6
src/main/java/com/genersoft/iot/vmp/vmanager/user/UserController.java
View File

@@ -11,17 +11,13 @@ import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
import com.github.pagehelper.PageInfo;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.tags.Tag;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.security.authentication.AuthenticationManager;
import org.springframework.util.DigestUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.*;
import javax.security.sasl.AuthenticationException;
@@ -90,7 +86,7 @@ public class UserController {
@PostMapping("/add")
@Operation(summary = "停止视频回放")
@Operation(summary = "添加用户")
@Parameter(name = "username", description = "用户名", required = true)
@Parameter(name = "password", description = "密码（未md5加密的密码）", required = true)
@Parameter(name = "roleId", description = "角色ID", required = true)