roouzenn/wvp
Template
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source 
# Conflicts:
#	pom.xml
#	src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderProvider.java
#	src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java
#	src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/ByeRequestProcessor.java
#	src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java
#	src/main/java/com/genersoft/iot/vmp/media/zlm/AssistRESTfulUtils.java
#	src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
#	src/main/java/com/genersoft/iot/vmp/media/zlm/dto/HookSubscribeFactory.java
#	src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java
#	src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java
#	src/main/java/com/genersoft/iot/vmp/service/impl/PlatformServiceImpl.java
#	src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
#	src/main/java/com/genersoft/iot/vmp/storager/dao/PlatformChannelMapper.java
#	src/main/resources/all-application.yml
#	src/main/resources/application-dev.yml
This commit is contained in:
648540858
2024-02-07 20:30:37 +08:00
parent fc8bb378de b079039f81
commit 2e7c9a7341
144 changed files with 6070 additions and 28330 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPSender.java
View File

@@ -66,17 +66,17 @@ public class SIPSender {
// 添加错误订阅
if (errorEvent != null) {
sipSubscribe.addErrorSubscribe(callIdHeader.getCallId(), (eventResult -> {
errorEvent.response(eventResult);
sipSubscribe.removeErrorSubscribe(eventResult.callId);
sipSubscribe.removeOkSubscribe(eventResult.callId);
errorEvent.response(eventResult);
}));
}
// 添加订阅
if (okEvent != null) {
sipSubscribe.addOkSubscribe(callIdHeader.getCallId(), eventResult -> {
okEvent.response(eventResult);
sipSubscribe.removeOkSubscribe(eventResult.callId);
sipSubscribe.removeErrorSubscribe(eventResult.callId);
okEvent.response(eventResult);
});
}
if ("TCP".equals(transport)) {

2
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/SIPRequestHeaderProvider.java
View File

@@ -164,6 +164,7 @@ public class SIPRequestHeaderProvider {
Request request = null;
//请求行
SipURI requestLine = SipFactory.getInstance().createAddressFactory().createSipURI(channelId, device.getHostAddress());
// SipURI requestLine = SipFactory.getInstance().createAddressFactory().createSipURI(device.getDeviceId(), device.getHostAddress());
// via
ArrayList<ViaHeader> viaHeaders = new ArrayList<ViaHeader>();
ViaHeader viaHeader = SipFactory.getInstance().createHeaderFactory().createViaHeader(sipLayer.getLocalIp(device.getLocalIp()), sipConfig.getPort(), device.getTransport(), SipUtils.getNewViaTag());
@@ -174,6 +175,7 @@ public class SIPRequestHeaderProvider {
FromHeader fromHeader = SipFactory.getInstance().createHeaderFactory().createFromHeader(fromAddress, transactionInfo.getFromTag());
//to
SipURI toSipURI = SipFactory.getInstance().createAddressFactory().createSipURI(channelId,device.getHostAddress());
// SipURI toSipURI = SipFactory.getInstance().createAddressFactory().createSipURI(device.getDeviceId(),device.getHostAddress());
Address toAddress = SipFactory.getInstance().createAddressFactory().createAddress(toSipURI);
ToHeader toHeader = SipFactory.getInstance().createHeaderFactory().createToHeader(toAddress, transactionInfo.getToTag());

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

@@ -40,6 +40,7 @@ import javax.sip.SipFactory;
import javax.sip.header.CallIdHeader;
import javax.sip.message.Request;
import java.text.ParseException;
import java.util.List;
/**
* @description:设备能力接口，用于定义设备的控制、查询能力
@@ -374,7 +375,8 @@ public class SIPCommander implements ISIPCommander {
}), e -> {
ResponseEvent responseEvent = (ResponseEvent) e.event;
SIPResponse response = (SIPResponse) responseEvent.getResponse();
streamSession.put(device.getDeviceId(), channelId, "play", stream, ssrcInfo.getSsrc(), mediaServerItem.getId(), response,
String callId = response.getCallIdHeader().getCallId();
streamSession.put(device.getDeviceId(), channelId, callId, stream, ssrcInfo.getSsrc(), mediaServerItem.getId(), response,
InviteSessionType.PLAY);
okEvent.response(e);
});
@@ -676,22 +678,26 @@ public class SIPCommander implements ISIPCommander {
*/
@Override
public void streamByeCmd(Device device, String channelId, String stream, String callId, SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException, SsrcTransactionNotFoundException {
SsrcTransaction ssrcTransaction;
if (callId != null) {
ssrcTransaction = streamSession.getSsrcTransaction(null, null, callId, null);
}else {
ssrcTransaction = streamSession.getSsrcTransaction(device.getDeviceId(), channelId, null, stream);
if (device == null) {
logger.warn("[发送BYE] device为null");
return;
}
if (ssrcTransaction == null) {
List<SsrcTransaction> ssrcTransactionList = streamSession.getSsrcTransactionForAll(device.getDeviceId(), channelId, callId, stream);
if (ssrcTransactionList == null || ssrcTransactionList.isEmpty()) {
logger.info("[发送BYE] 未找到事务信息,设备： device: {}, channel: {}", device.getDeviceId(), channelId);
throw new SsrcTransactionNotFoundException(device.getDeviceId(), channelId, callId, stream);
}
mediaServerService.releaseSsrc(ssrcTransaction.getMediaServerId(), ssrcTransaction.getSsrc());
mediaServerService.closeRTPServer(ssrcTransaction.getMediaServerId(), ssrcTransaction.getStream());
streamSession.remove(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId(), ssrcTransaction.getStream());
for (SsrcTransaction ssrcTransaction : ssrcTransactionList) {
logger.info("[发送BYE] 设备： device: {}, channel: {}, callId: {}", device.getDeviceId(), channelId, ssrcTransaction.getCallId());
mediaServerService.releaseSsrc(ssrcTransaction.getMediaServerId(), ssrcTransaction.getSsrc());
Request byteRequest = headerProvider.createByteRequest(device, channelId, ssrcTransaction.getSipTransactionInfo());
sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), byteRequest, null, okEvent);
mediaServerService.closeRTPServer(ssrcTransaction.getMediaServerId(), ssrcTransaction.getStream());
streamSession.removeByCallId(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId(), ssrcTransaction.getCallId());
Request byteRequest = headerProvider.createByteRequest(device, channelId, ssrcTransaction.getSipTransactionInfo());
sipSender.transmitRequest(sipLayer.getLocalIp(device.getLocalIp()), byteRequest, null, okEvent);
}
}
@Override
@@ -1000,7 +1006,7 @@ public class SIPCommander implements ISIPCommander {
catalogXml.append("<DeviceID>" + device.getDeviceId() + "</DeviceID>\r\n");
catalogXml.append("</Query>\r\n");
Request request = headerProvider.createMessageRequest(device, catalogXml.toString(), null, SipUtils.getNewFromTag(), null,sipSender.getNewCallIdHeader(sipLayer.getLocalIp(device.getLocalIp()),device.getTransport()));

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

@@ -579,7 +579,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
@Override
public void sendNotifyForCatalogAddOrUpdate(String type, ParentPlatform parentPlatform, List<DeviceChannel> deviceChannels, SubscribeInfo subscribeInfo, Integer index) throws InvalidArgumentException, ParseException, NoSuchFieldException, SipException, IllegalAccessException {
if (parentPlatform == null || deviceChannels == null || deviceChannels.size() == 0 || subscribeInfo == null) {
if (parentPlatform == null || deviceChannels == null || deviceChannels.isEmpty() || subscribeInfo == null) {
return;
}
if (index == null) {
@@ -597,6 +597,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
Integer finalIndex = index;
String catalogXmlContent = getCatalogXmlContentForCatalogAddOrUpdate(parentPlatform, channels,
deviceChannels.size(), type, subscribeInfo);
logger.info("[发送NOTIFY通知]类型： {}，发送数量： {}", type, channels.size());
sendNotify(parentPlatform, catalogXmlContent, subscribeInfo, eventResult -> {
logger.error("发送NOTIFY通知消息失败。错误：{} {}", eventResult.statusCode, eventResult.msg);
}, (eventResult -> {
@@ -620,7 +621,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
SIPRequest notifyRequest = headerProviderPlatformProvider.createNotifyRequest(parentPlatform, catalogXmlContent, subscribeInfo);
sipSender.transmitRequest(parentPlatform.getDeviceIp(), notifyRequest);
sipSender.transmitRequest(parentPlatform.getDeviceIp(), notifyRequest, errorEvent, okEvent);
}
private String getCatalogXmlContentForCatalogAddOrUpdate(ParentPlatform parentPlatform, List<DeviceChannel> channels, int sumNum, String type, SubscribeInfo subscribeInfo) {
@@ -632,9 +633,9 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
.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("<SumNum>"+ sumNum +"</SumNum>\r\n")
.append("<DeviceList Num=\"" + channels.size() + "\">\r\n");
if (channels.size() > 0) {
if (!channels.isEmpty()) {
for (DeviceChannel channel : channels) {
if (parentPlatform.getServerGBId().equals(channel.getParentId())) {
channel.setParentId(parentPlatform.getDeviceGBId());
@@ -701,6 +702,7 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
}else {
channels = deviceChannels.subList(index, deviceChannels.size());
}
logger.info("[发送NOTIFY通知]类型： {}，发送数量： {}", type, channels.size());
Integer finalIndex = index;
String catalogXmlContent = getCatalogXmlContentForCatalogOther(parentPlatform, channels, type);
sendNotify(parentPlatform, catalogXmlContent, subscribeInfo, eventResult -> {
@@ -747,13 +749,14 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
if ( parentPlatform ==null) {
return ;
}
logger.info("[国标级联] 发送录像数据通道： {}", recordInfo.getChannelId());
String characterSet = parentPlatform.getCharacterSet();
StringBuffer recordXml = new StringBuffer(600);
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.getChannelId() + "</DeviceID>\r\n")
.append("<DeviceID>" + deviceChannel.getChannelId() + "</DeviceID>\r\n")
.append("<SumNum>" + recordInfo.getSumNum() + "</SumNum>\r\n");
if (recordInfo.getRecordList() == null ) {
recordXml.append("<RecordList Num=\"0\">\r\n");
@@ -763,7 +766,7 @@ 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("<DeviceID>" + deviceChannel.getChannelId() + "</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")
@@ -783,12 +786,14 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
recordXml.append("</RecordList>\r\n")
.append("</Response>\r\n");
logger.info("[国标级联] 发送录像数据通道：{}, 内容： {}", recordInfo.getChannelId(), recordXml);
// callid
CallIdHeader callIdHeader = sipSender.getNewCallIdHeader(parentPlatform.getDeviceIp(),parentPlatform.getTransport());
Request request = headerProviderPlatformProvider.createMessageRequest(parentPlatform, recordXml.toString(), fromTag, SipUtils.getNewViaTag(), callIdHeader);
sipSender.transmitRequest(parentPlatform.getDeviceIp(), request);
sipSender.transmitRequest(parentPlatform.getDeviceIp(), request, null, eventResult -> {
logger.info("[国标级联] 发送录像数据通道：{}, 发送成功", recordInfo.getChannelId());
});
}

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

@@ -128,6 +128,9 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
redisCatchStorage.deleteSendRTPServer(sendRtpItem.getPlatformId(), sendRtpItem.getChannelId(),
callIdHeader.getCallId(), null);
zlmServerFactory.stopSendRtpStream(mediaInfo, param);
if (userSetting.getUseCustomSsrcForParentInvite()) {
mediaServerService.releaseSsrc(mediaInfo.getId(), sendRtpItem.getSsrc());
}
if (sendRtpItem.getPlayType().equals(InviteStreamType.PUSH)) {
ParentPlatform platform = platformService.queryPlatformByServerGBId(sendRtpItem.getPlatformId());
if (platform != null) {
@@ -167,14 +170,12 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
}
}
// 发流端发送的停止
SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(null, null, callIdHeader.getCallId(), null);
if (ssrcTransaction == null ) {
logger.info("[收到bye] 但是无法获取推流信息和发流信息，忽略此请求");
logger.info(request.toString());
return;
}
// 可能是设备发送的停止
SsrcTransaction ssrcTransaction = streamSession.getSsrcTransactionByCallId(callIdHeader.getCallId());
if (ssrcTransaction == null) {
return;
}
logger.info("[收到bye] 来自设备：{}, 通道已停止推流: {}", ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId());
ParentPlatform platform = platformService.queryPlatformByServerGBId(ssrcTransaction.getDeviceId());
if (platform != null ) {
@@ -216,7 +217,7 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
if (mediaServerItem != null) {
mediaServerService.releaseSsrc(mediaServerItem.getId(), ssrcTransaction.getSsrc());
}
streamSession.remove(device.getDeviceId(), channel.getChannelId(), ssrcTransaction.getStream());
streamSession.removeByCallId(device.getDeviceId(), channel.getChannelId(), ssrcTransaction.getCallId());
if (ssrcTransaction.getType() == InviteSessionType.BROADCAST) {
// 查找来源的对讲设备，发送停止
Device sourceDevice = storager.queryVideoDeviceByPlatformIdAndChannelId(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId());

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

@@ -152,7 +152,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
String requesterId = SipUtils.getUserIdFromFromHeader(request);
CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME);
if (requesterId == null || channelId == null) {
logger.info("无法从FromHeader的Address中获取到平台id，返回400");
logger.info("无法从请求中获取到平台id，返回400");
// 参数不全， 发400，请求错误
try {
responseAck(request, Response.BAD_REQUEST);
@@ -162,6 +162,8 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
return;
}
logger.info("[INVITE] requesterId: {}, callId: {}, 来自：{}：{}",
requesterId, callIdHeader.getCallId(), request.getRemoteAddress(), request.getRemotePort());
// 查询请求是否来自上级平台\设备
ParentPlatform platform = storager.queryParentPlatByServerGBId(requesterId);
@@ -409,7 +411,16 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
// 非严格模式端口不统一, 增加兼容性，修改为一个不为0的端口
localPort = new Random().nextInt(65535) + 1;
}
content.append("m=video " + localPort + " RTP/AVP 96\r\n");
if (sendRtpItem.isTcp()) {
content.append("m=video " + localPort + " TCP/RTP/AVP 96\r\n");
if (!sendRtpItem.isTcpActive()) {
content.append("a=setup:active\r\n");
} else {
content.append("a=setup:passive\r\n");
}
}else {
content.append("m=video " + localPort + " RTP/AVP 96\r\n");
}
content.append("a=sendonly\r\n");
content.append("a=rtpmap:96 PS/90000\r\n");
content.append("y=" + sendRtpItem.getSsrc() + "\r\n");
@@ -524,7 +535,10 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
}
});
} else {
sendRtpItem.setPlayType(InviteStreamType.PLAY);
String streamId = String.format("%s_%s", device.getDeviceId(), channelId);
sendRtpItem.setStreamId(streamId);
redisCatchStorage.updateSendRTPSever(sendRtpItem);
SSRCInfo ssrcInfo = playService.play(mediaServerItem, device.getDeviceId(), channelId, ssrc, ((code, msg, data) -> {
if (code == InviteErrorCode.SUCCESS.getCode()) {
hookEvent.run(code, msg, data);
@@ -536,9 +550,6 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
errorEvent.run(code, msg, data);
}
}));
sendRtpItem.setPlayType(InviteStreamType.PLAY);
String streamId = String.format("%s_%s", device.getDeviceId(), channelId);
sendRtpItem.setStream(streamId);
sendRtpItem.setSsrc(ssrcInfo.getSsrc());
redisCatchStorage.updateSendRTPSever(sendRtpItem);
@@ -721,8 +732,6 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
zlmHttpHookSubscribe.removeSubscribe(hookSubscribe);
dynamicTask.stop(callIdHeader.getCallId());
}
} else if ("push".equals(gbStream.getStreamType())) {
if (!platform.isStartOfflinePush()) {
// 平台设置中关闭了拉起离线的推流则直接回复
@@ -745,13 +754,10 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
dynamicTask.startDelay(callIdHeader.getCallId(), () -> {
logger.info("[ app={}, stream={} ] 等待设备开始推流超时", gbStream.getApp(), gbStream.getStream());
try {
redisPushStreamResponseListener.removeEvent(gbStream.getApp(), gbStream.getStream());
mediaListManager.removedChannelOnlineEventLister(gbStream.getApp(), gbStream.getStream());
responseAck(request, Response.REQUEST_TIMEOUT); // 超时
} catch (SipException e) {
logger.error("未处理的异常 ", e);
} catch (InvalidArgumentException e) {
logger.error("未处理的异常 ", e);
} catch (ParseException e) {
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("未处理的异常 ", e);
}
}, userSetting.getPlatformPlayTimeout());
@@ -762,6 +768,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
// 添加在本机上线的通知
mediaListManager.addChannelOnlineEventLister(gbStream.getApp(), gbStream.getStream(), (app, stream, serverId) -> {
dynamicTask.stop(callIdHeader.getCallId());
redisPushStreamResponseListener.removeEvent(gbStream.getApp(), gbStream.getStream());
if (serverId.equals(userSetting.getServerId())) {
SendRtpItem sendRtpItem = zlmServerFactory.createSendRtpItem(mediaServerItem, addressStr, finalPort, ssrc, requesterId,
app, stream, channelId, mediaTransmissionTCP, platform.isRtcp());
@@ -826,7 +833,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
// 发送redis消息
redisGbPlayMsgListener.sendMsg(streamPushItem.getServerId(), streamPushItem.getMediaServerId(),
streamPushItem.getApp(), streamPushItem.getStream(), addressStr, port, ssrc, requesterId,
channelId, mediaTransmissionTCP, platform.isRtcp(), null, responseSendItemMsg -> {
channelId, mediaTransmissionTCP, platform.isRtcp(),platform.getName(), responseSendItemMsg -> {
SendRtpItem sendRtpItem = responseSendItemMsg.getSendRtpItem();
if (sendRtpItem == null || responseSendItemMsg.getMediaServerItem() == null) {
logger.warn("服务器端口资源不足");

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

@@ -13,6 +13,7 @@ import com.genersoft.iot.vmp.gb28181.utils.SipUtils;
import com.genersoft.iot.vmp.gb28181.utils.XmlUtil;
import com.genersoft.iot.vmp.service.IDeviceChannelService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.utils.DateUtil;
import org.dom4j.DocumentException;
import org.dom4j.Element;
import org.slf4j.Logger;
@@ -185,6 +186,7 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
// 判断此通道是否存在
DeviceChannel deviceChannel = deviceChannelService.getOne(deviceId, channel.getChannelId());
if (deviceChannel != null) {
logger.info("[增加通道] 已存在，不发送通知只更新，设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
channel.setId(deviceChannel.getId());
updateChannelMap.put(channel.getChannelId(), channel);
if (updateChannelMap.keySet().size() > 300) {
@@ -222,6 +224,7 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
DeviceChannel deviceChannelForUpdate = deviceChannelService.getOne(deviceId, channel.getChannelId());
if (deviceChannelForUpdate != null) {
channel.setId(deviceChannelForUpdate.getId());
channel.setUpdateTime(DateUtil.getNow());
updateChannelMap.put(channel.getChannelId(), channel);
if (updateChannelMap.keySet().size() > 300) {
executeSaveForUpdate();
@@ -244,11 +247,11 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
// 转发变化信息
eventPublisher.catalogEventPublish(null, channel, event);
if (updateChannelMap.keySet().size() > 0
|| addChannelMap.keySet().size() > 0
|| updateChannelOnlineList.size() > 0
|| updateChannelOfflineList.size() > 0
|| deleteChannelList.size() > 0) {
if (!updateChannelMap.keySet().isEmpty()
|| !addChannelMap.keySet().isEmpty()
|| !updateChannelOnlineList.isEmpty()
|| !updateChannelOfflineList.isEmpty()
|| !deleteChannelList.isEmpty()) {
if (!dynamicTask.contains(talkKey)) {
dynamicTask.startDelay(talkKey, this::executeSave, 1000);
@@ -262,16 +265,36 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
}
private void executeSave(){
executeSaveForAdd();
executeSaveForUpdate();
executeSaveForDelete();
executeSaveForOnline();
executeSaveForOffline();
try {
executeSaveForAdd();
} catch (Exception e) {
logger.error("[存储收到的增加通道] 异常： ", e );
}
try {
executeSaveForUpdate();
} catch (Exception e) {
logger.error("[存储收到的更新通道] 异常： ", e );
}
try {
executeSaveForDelete();
} catch (Exception e) {
logger.error("[存储收到的删除通道] 异常： ", e );
}
try {
executeSaveForOnline();
} catch (Exception e) {
logger.error("[存储收到的通道上线] 异常： ", e );
}
try {
executeSaveForOffline();
} catch (Exception e) {
logger.error("[存储收到的通道离线] 异常： ", e );
}
dynamicTask.stop(talkKey);
}
private void executeSaveForUpdate(){
if (updateChannelMap.values().size() > 0) {
if (!updateChannelMap.values().isEmpty()) {
ArrayList<DeviceChannel> deviceChannels = new ArrayList<>(updateChannelMap.values());
updateChannelMap.clear();
deviceChannelService.batchUpdateChannel(deviceChannels);
@@ -280,7 +303,7 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
}
private void executeSaveForAdd(){
if (addChannelMap.values().size() > 0) {
if (!addChannelMap.values().isEmpty()) {
ArrayList<DeviceChannel> deviceChannels = new ArrayList<>(addChannelMap.values());
addChannelMap.clear();
deviceChannelService.batchAddChannel(deviceChannels);
@@ -288,21 +311,21 @@ public class NotifyRequestForCatalogProcessor extends SIPRequestProcessorParent
}
private void executeSaveForDelete(){
if (deleteChannelList.size() > 0) {
if (!deleteChannelList.isEmpty()) {
deviceChannelService.deleteChannels(deleteChannelList);
deleteChannelList.clear();
}
}
private void executeSaveForOnline(){
if (updateChannelOnlineList.size() > 0) {
if (!updateChannelOnlineList.isEmpty()) {
deviceChannelService.channelsOnline(updateChannelOnlineList);
updateChannelOnlineList.clear();
}
}
private void executeSaveForOffline(){
if (updateChannelOfflineList.size() > 0) {
if (!updateChannelOfflineList.isEmpty()) {
deviceChannelService.channelsOffline(updateChannelOfflineList);
updateChannelOfflineList.clear();
}

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

@@ -106,23 +106,27 @@ public class RegisterRequestProcessor extends SIPRequestProcessorParent implemen
String title = registerFlag ? "[注册请求]": "[注销请求]";
logger.info(title + "设备：{}, 开始处理: {}", deviceId, requestAddress);
if (device != null &&
device.getSipTransactionInfo() != null &&
request.getCallIdHeader().getCallId().equals(device.getSipTransactionInfo().getCallId())) {
device.getSipTransactionInfo() != null &&
request.getCallIdHeader().getCallId().equals(device.getSipTransactionInfo().getCallId())) {
logger.info(title + "设备：{}, 注册续订: {}",device.getDeviceId(), device.getDeviceId());
device.setExpires(request.getExpires().getExpires());
device.setIp(remoteAddressInfo.getIp());
device.setPort(remoteAddressInfo.getPort());
device.setHostAddress(remoteAddressInfo.getIp().concat(":").concat(String.valueOf(remoteAddressInfo.getPort())));
device.setLocalIp(request.getLocalAddress().getHostAddress());
Response registerOkResponse = getRegisterOkResponse(request);
// 判断TCP还是UDP
ViaHeader reqViaHeader = (ViaHeader) request.getHeader(ViaHeader.NAME);
String transport = reqViaHeader.getTransport();
device.setTransport("TCP".equalsIgnoreCase(transport) ? "TCP" : "UDP");
sipSender.transmitRequest(request.getLocalAddress().getHostAddress(), registerOkResponse);
device.setRegisterTime(DateUtil.getNow());
SipTransactionInfo sipTransactionInfo = new SipTransactionInfo((SIPResponse)registerOkResponse);
deviceService.online(device, sipTransactionInfo);
if (registerFlag) {
device.setExpires(request.getExpires().getExpires());
device.setIp(remoteAddressInfo.getIp());
device.setPort(remoteAddressInfo.getPort());
device.setHostAddress(remoteAddressInfo.getIp().concat(":").concat(String.valueOf(remoteAddressInfo.getPort())));
device.setLocalIp(request.getLocalAddress().getHostAddress());
Response registerOkResponse = getRegisterOkResponse(request);
// 判断TCP还是UDP
ViaHeader reqViaHeader = (ViaHeader) request.getHeader(ViaHeader.NAME);
String transport = reqViaHeader.getTransport();
device.setTransport("TCP".equalsIgnoreCase(transport) ? "TCP" : "UDP");
sipSender.transmitRequest(request.getLocalAddress().getHostAddress(), registerOkResponse);
device.setRegisterTime(DateUtil.getNow());
SipTransactionInfo sipTransactionInfo = new SipTransactionInfo((SIPResponse)registerOkResponse);
deviceService.online(device, sipTransactionInfo);
}else {
deviceService.offline(deviceId, "主动注销");
}
return;
}
String password = (device != null && !ObjectUtils.isEmpty(device.getPassword()))? device.getPassword() : sipConfig.getPassword();

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

@@ -61,7 +61,7 @@ public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent imp
return;
}
SIPRequest request = (SIPRequest) evt.getRequest();
logger.info("[收到心跳]， device: {}, callId: {}", device.getDeviceId(), request.getCallIdHeader().getCallId());
logger.info("[收到心跳] device: {}, callId: {}", device.getDeviceId(), request.getCallIdHeader().getCallId());
// 回复200 OK
try {
@@ -76,10 +76,15 @@ public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent imp
RemoteAddressInfo remoteAddressInfo = SipUtils.getRemoteAddressFromRequest(request, userSetting.getSipUseSourceIpAsRemoteAddress());
if (!device.getIp().equalsIgnoreCase(remoteAddressInfo.getIp()) || device.getPort() != remoteAddressInfo.getPort()) {
logger.info("[心跳] 设备{}地址变化, 远程地址为: {}:{}", device.getDeviceId(), remoteAddressInfo.getIp(), remoteAddressInfo.getPort());
logger.info("[收到心跳] 设备{}地址变化, 远程地址为: {}:{}", device.getDeviceId(), remoteAddressInfo.getIp(), remoteAddressInfo.getPort());
device.setPort(remoteAddressInfo.getPort());
device.setHostAddress(remoteAddressInfo.getIp().concat(":").concat(String.valueOf(remoteAddressInfo.getPort())));
device.setIp(remoteAddressInfo.getIp());
// 设备地址变化会引起目录订阅任务失效，需要重新添加
if (device.getSubscribeCycleForCatalog() > 0) {
deviceService.removeCatalogSubscribe(device);
deviceService.addCatalogSubscribe(device);
}
}
if (device.getKeepaliveTime() == null) {
device.setKeepaliveIntervalTime(60);

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

@@ -102,6 +102,7 @@ public class RecordInfoQueryMessageHandler extends SIPRequestProcessorParent imp
// 接收录像数据
recordEndEventListener.addEndEventHandler(deviceChannel.getDeviceId(), channelId, (recordInfo)->{
try {
logger.info("[国标级联] 录像查询收到数据， 通道： {}，准备转发===", channelId);
cmderFroPlatform.recordInfo(deviceChannel, parentPlatform, request.getFromTag(), recordInfo);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 回复录像数据: {}", e.getMessage());