Thirdparty
/
wvp-GB28181-pro
mirror of https://github.com/648540858/wvp-GB28181-pro
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

优化消息处理中存在可能异常的处理流程

2.6.5
648540858 2022-10-17 12:39:58 +08:00
parent c4fc4abf7c
commit 0ff4ed217d
12 changed files with 414 additions and 306 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -132,7 +132,11 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
if (requesterId == null || channelId == null) { if (requesterId == null || channelId == null) {
logger.info("无法从FromHeader的Address中获取到平台id，返回400"); logger.info("无法从FromHeader的Address中获取到平台id，返回400");
// 参数不全， 发400，请求错误 // 参数不全， 发400，请求错误
try {
responseAck(serverTransaction, Response.BAD_REQUEST); responseAck(serverTransaction, Response.BAD_REQUEST);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] invite BAD_REQUEST: {}", e.getMessage());
}
return; return;
} }
@ -141,6 +145,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
ParentPlatform platform = storager.queryParentPlatByServerGBId(requesterId); ParentPlatform platform = storager.queryParentPlatByServerGBId(requesterId);
if (platform == null) { if (platform == null) {
inviteFromDeviceHandle(serverTransaction, requesterId); inviteFromDeviceHandle(serverTransaction, requesterId);
} else { } else {
// 查询平台下是否有该通道 // 查询平台下是否有该通道
DeviceChannel channel = storager.queryChannelInParentPlatform(requesterId, channelId); DeviceChannel channel = storager.queryChannelInParentPlatform(requesterId, channelId);
@ -158,7 +163,11 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
// return; // return;
// } // }
// 通道存在，发100，TRYING // 通道存在，发100，TRYING
try {
responseAck(serverTransaction, Response.TRYING); responseAck(serverTransaction, Response.TRYING);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] invite TRYING: {}", e.getMessage());
}
} else if (channel == null && gbStream != null) { } else if (channel == null && gbStream != null) {
String mediaServerId = gbStream.getMediaServerId(); String mediaServerId = gbStream.getMediaServerId();
@ -166,13 +175,21 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
if (mediaServerItem == null) { if (mediaServerItem == null) {
if ("proxy".equals(gbStream.getStreamType())) { if ("proxy".equals(gbStream.getStreamType())) {
logger.info("[ app={}, stream={} ]找不到zlm {}，返回410", gbStream.getApp(), gbStream.getStream(), mediaServerId); logger.info("[ app={}, stream={} ]找不到zlm {}，返回410", gbStream.getApp(), gbStream.getStream(), mediaServerId);
try {
responseAck(serverTransaction, Response.GONE); responseAck(serverTransaction, Response.GONE);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] invite GONE: {}", e.getMessage());
}
return; return;
} else { } else {
streamPushItem = streamPushService.getPush(gbStream.getApp(), gbStream.getStream()); streamPushItem = streamPushService.getPush(gbStream.getApp(), gbStream.getStream());
if (streamPushItem == null || streamPushItem.getServerId().equals(userSetting.getServerId())) { if (streamPushItem == null || streamPushItem.getServerId().equals(userSetting.getServerId())) {
logger.info("[ app={}, stream={} ]找不到zlm {}，返回410", gbStream.getApp(), gbStream.getStream(), mediaServerId); logger.info("[ app={}, stream={} ]找不到zlm {}，返回410", gbStream.getApp(), gbStream.getStream(), mediaServerId);
try {
responseAck(serverTransaction, Response.GONE); responseAck(serverTransaction, Response.GONE);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] invite GONE: {}", e.getMessage());
}
return; return;
} }
} }
@ -181,25 +198,47 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
streamPushItem = streamPushService.getPush(gbStream.getApp(), gbStream.getStream()); streamPushItem = streamPushService.getPush(gbStream.getApp(), gbStream.getStream());
if (streamPushItem == null) { if (streamPushItem == null) {
logger.info("[ app={}, stream={} ]找不到zlm {}，返回410", gbStream.getApp(), gbStream.getStream(), mediaServerId); logger.info("[ app={}, stream={} ]找不到zlm {}，返回410", gbStream.getApp(), gbStream.getStream(), mediaServerId);
try {
responseAck(serverTransaction, Response.GONE); responseAck(serverTransaction, Response.GONE);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] invite GONE: {}", e.getMessage());
}
return; return;
} }
}else if("proxy".equals(gbStream.getStreamType())){ }else if("proxy".equals(gbStream.getStreamType())){
proxyByAppAndStream = streamProxyService.getStreamProxyByAppAndStream(gbStream.getApp(), gbStream.getStream()); proxyByAppAndStream = streamProxyService.getStreamProxyByAppAndStream(gbStream.getApp(), gbStream.getStream());
if (proxyByAppAndStream == null) { if (proxyByAppAndStream == null) {
logger.info("[ app={}, stream={} ]找不到zlm {}，返回410", gbStream.getApp(), gbStream.getStream(), mediaServerId); logger.info("[ app={}, stream={} ]找不到zlm {}，返回410", gbStream.getApp(), gbStream.getStream(), mediaServerId);
try {
responseAck(serverTransaction, Response.GONE); responseAck(serverTransaction, Response.GONE);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] invite GONE: {}", e.getMessage());
}
return; return;
} }
} }
} }
responseAck(serverTransaction, Response.CALL_IS_BEING_FORWARDED); // 通道存在，发181，呼叫转接中 try {
responseAck(serverTransaction, Response.CALL_IS_BEING_FORWARDED);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] invite CALL_IS_BEING_FORWARDED: {}", e.getMessage());
}
} else if (catalog != null) { } else if (catalog != null) {
responseAck(serverTransaction, Response.BAD_REQUEST, "catalog channel can not play"); // 目录不支持点播 try {
// 目录不支持点播
responseAck(serverTransaction, Response.BAD_REQUEST, "catalog channel can not play");
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] invite 目录不支持点播: {}", e.getMessage());
}
return; return;
} else { } else {
logger.info("通道不存在，返回404"); logger.info("通道不存在，返回404");
responseAck(serverTransaction, Response.NOT_FOUND); // 通道不存在，发404，资源不存在 try {
// 通道不存在，发404，资源不存在
responseAck(serverTransaction, Response.NOT_FOUND);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] invite 通道不存在: {}", e.getMessage());
}
return; return;
} }
// 解析sdp消息, 使用jainsip 自带的sdp解析方式 // 解析sdp消息, 使用jainsip 自带的sdp解析方式
@ -270,7 +309,12 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
if (port == -1) { if (port == -1) {
logger.info("不支持的媒体格式，返回415"); logger.info("不支持的媒体格式，返回415");
// 回复不支持的格式 // 回复不支持的格式
responseAck(serverTransaction, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415 try {
// 不支持的格式，发415
responseAck(serverTransaction, Response.UNSUPPORTED_MEDIA_TYPE);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] invite 不支持的格式: {}", e.getMessage());
}
return; return;
} }
String username = sdp.getOrigin().getUsername(); String username = sdp.getOrigin().getUsername();
@ -283,13 +327,21 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
device = storager.queryVideoDeviceByPlatformIdAndChannelId(requesterId, channelId); device = storager.queryVideoDeviceByPlatformIdAndChannelId(requesterId, channelId);
if (device == null) { if (device == null) {
logger.warn("点播平台{}的通道{}时未找到设备信息", requesterId, channel); logger.warn("点播平台{}的通道{}时未找到设备信息", requesterId, channel);
try {
responseAck(serverTransaction, Response.SERVER_INTERNAL_ERROR); responseAck(serverTransaction, Response.SERVER_INTERNAL_ERROR);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] invite 未找到设备信息: {}", e.getMessage());
}
return; return;
} }
mediaServerItem = playService.getNewMediaServerItem(device); mediaServerItem = playService.getNewMediaServerItem(device);
if (mediaServerItem == null) { if (mediaServerItem == null) {
logger.warn("未找到可用的zlm"); logger.warn("未找到可用的zlm");
try {
responseAck(serverTransaction, Response.BUSY_HERE); responseAck(serverTransaction, Response.BUSY_HERE);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] invite BUSY_HERE: {}", e.getMessage());
}
return; return;
} }
SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId, SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
@ -301,7 +353,11 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
} }
if (sendRtpItem == null) { if (sendRtpItem == null) {
logger.warn("服务器端口资源不足"); logger.warn("服务器端口资源不足");
try {
responseAck(serverTransaction, Response.BUSY_HERE); responseAck(serverTransaction, Response.BUSY_HERE);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] invite 服务器端口资源不足: {}", e.getMessage());
}
return; return;
} }
sendRtpItem.setCallId(callIdHeader.getCallId()); sendRtpItem.setCallId(callIdHeader.getCallId());
@ -474,13 +530,8 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
} }
} }
} }
} catch (SipException | InvalidArgumentException | ParseException e) {
e.printStackTrace();
logger.warn("sdp解析错误");
e.printStackTrace();
} catch (SdpParseException e) { } catch (SdpParseException e) {
e.printStackTrace(); logger.error("sdp解析错误", e);
} catch (SdpException e) { } catch (SdpException e) {
e.printStackTrace(); e.printStackTrace();
} }
@ -492,7 +543,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
private void pushProxyStream(RequestEvent evt, ServerTransaction serverTransaction, GbStream gbStream, ParentPlatform platform, private void pushProxyStream(RequestEvent evt, ServerTransaction serverTransaction, GbStream gbStream, ParentPlatform platform,
CallIdHeader callIdHeader, MediaServerItem mediaServerItem, CallIdHeader callIdHeader, MediaServerItem mediaServerItem,
int port, Boolean tcpActive, boolean mediaTransmissionTCP, int port, Boolean tcpActive, boolean mediaTransmissionTCP,
String channelId, String addressStr, String ssrc, String requesterId) throws InvalidArgumentException, ParseException, SipException { String channelId, String addressStr, String ssrc, String requesterId) {
Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream()); Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream());
if (streamReady) { if (streamReady) {
// 自平台内容 // 自平台内容
@ -502,7 +553,11 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
if (sendRtpItem == null) { if (sendRtpItem == null) {
logger.warn("服务器端口资源不足"); logger.warn("服务器端口资源不足");
try {
responseAck(serverTransaction, Response.BUSY_HERE); responseAck(serverTransaction, Response.BUSY_HERE);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] invite 服务器端口资源不足: {}", e.getMessage());
}
return; return;
} }
if (tcpActive != null) { if (tcpActive != null) {
@ -527,7 +582,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
private void pushStream(RequestEvent evt, ServerTransaction serverTransaction, GbStream gbStream, StreamPushItem streamPushItem, ParentPlatform platform, private void pushStream(RequestEvent evt, ServerTransaction serverTransaction, GbStream gbStream, StreamPushItem streamPushItem, ParentPlatform platform,
CallIdHeader callIdHeader, MediaServerItem mediaServerItem, CallIdHeader callIdHeader, MediaServerItem mediaServerItem,
int port, Boolean tcpActive, boolean mediaTransmissionTCP, int port, Boolean tcpActive, boolean mediaTransmissionTCP,
String channelId, String addressStr, String ssrc, String requesterId) throws InvalidArgumentException, ParseException, SipException { String channelId, String addressStr, String ssrc, String requesterId) {
// 推流 // 推流
if (streamPushItem.isSelf()) { if (streamPushItem.isSelf()) {
Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream()); Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream());
@ -539,7 +594,11 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
if (sendRtpItem == null) { if (sendRtpItem == null) {
logger.warn("服务器端口资源不足"); logger.warn("服务器端口资源不足");
try {
responseAck(serverTransaction, Response.BUSY_HERE); responseAck(serverTransaction, Response.BUSY_HERE);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] invite 服务器端口资源不足: {}", e.getMessage());
}
return; return;
} }
if (tcpActive != null) { if (tcpActive != null) {
@ -577,15 +636,23 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
private void notifyStreamOnline(RequestEvent evt, ServerTransaction serverTransaction, GbStream gbStream, StreamPushItem streamPushItem, ParentPlatform platform, private void notifyStreamOnline(RequestEvent evt, ServerTransaction serverTransaction, GbStream gbStream, StreamPushItem streamPushItem, ParentPlatform platform,
CallIdHeader callIdHeader, MediaServerItem mediaServerItem, CallIdHeader callIdHeader, MediaServerItem mediaServerItem,
int port, Boolean tcpActive, boolean mediaTransmissionTCP, int port, Boolean tcpActive, boolean mediaTransmissionTCP,
String channelId, String addressStr, String ssrc, String requesterId) throws InvalidArgumentException, ParseException, SipException { String channelId, String addressStr, String ssrc, String requesterId) {
if ("proxy".equals(gbStream.getStreamType())) { if ("proxy".equals(gbStream.getStreamType())) {
// TODO 控制启用以使设备上线 // TODO 控制启用以使设备上线
logger.info("[ app={}, stream={} ]通道未推流，启用流后开始推流", gbStream.getApp(), gbStream.getStream()); logger.info("[ app={}, stream={} ]通道未推流，启用流后开始推流", gbStream.getApp(), gbStream.getStream());
try {
responseAck(serverTransaction, Response.BAD_REQUEST, "channel [" + gbStream.getGbId() + "] offline"); responseAck(serverTransaction, Response.BAD_REQUEST, "channel [" + gbStream.getGbId() + "] offline");
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] invite 通道未推流: {}", e.getMessage());
}
} else if ("push".equals(gbStream.getStreamType())) { } else if ("push".equals(gbStream.getStreamType())) {
if (!platform.isStartOfflinePush()) { if (!platform.isStartOfflinePush()) {
// 平台设置中关闭了拉起离线的推流则直接回复 // 平台设置中关闭了拉起离线的推流则直接回复
try {
responseAck(serverTransaction, Response.TEMPORARILY_UNAVAILABLE, "channel stream not pushing"); responseAck(serverTransaction, Response.TEMPORARILY_UNAVAILABLE, "channel stream not pushing");
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] invite 通道未推流: {}", e.getMessage());
}
return; return;
} }
// 发送redis消息以使设备上线 // 发送redis消息以使设备上线
@ -713,7 +780,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
} }
redisCatchStorage.updateSendRTPSever(sendRtpItem); redisCatchStorage.updateSendRTPSever(sendRtpItem);
}, (wvpResult) -> { }, (wvpResult) -> {
try {
// 错误 // 错误
if (wvpResult.getCode() == RedisGbPlayMsgListener.ERROR_CODE_OFFLINE) { if (wvpResult.getCode() == RedisGbPlayMsgListener.ERROR_CODE_OFFLINE) {
// 离线 // 离线
@ -730,21 +797,11 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId); mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
} }
} }
} catch (InvalidArgumentException | ParseException | SipException e) {
logger.error("[命令发送失败] 国标级联 点播回复: {}", e.getMessage());
}
try { try {
responseAck(serverTransaction, Response.BUSY_HERE); responseAck(serverTransaction, Response.BUSY_HERE);
} catch (SipException e) { } catch (InvalidArgumentException | ParseException | SipException e) {
e.printStackTrace(); logger.error("[命令发送失败] 国标级联 点播回复 BUSY_HERE: {}", e.getMessage());
} catch (InvalidArgumentException e) {
e.printStackTrace();
} catch (ParseException e) {
e.printStackTrace();
} }
return;
}); });
} }
@ -782,14 +839,17 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
return null; return null;
} }
public void inviteFromDeviceHandle(ServerTransaction serverTransaction, String requesterId) throws InvalidArgumentException, ParseException, SipException, SdpException { public void inviteFromDeviceHandle(ServerTransaction serverTransaction, String requesterId) {
// 非上级平台请求，查询是否设备请求（通常为接收语音广播的设备） // 非上级平台请求，查询是否设备请求（通常为接收语音广播的设备）
Device device = redisCatchStorage.getDevice(requesterId); Device device = redisCatchStorage.getDevice(requesterId);
if (device != null) { if (device != null) {
logger.info("收到设备" + requesterId + "的语音广播Invite请求"); logger.info("收到设备" + requesterId + "的语音广播Invite请求");
try {
responseAck(serverTransaction, Response.TRYING); responseAck(serverTransaction, Response.TRYING);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] invite BAD_REQUEST: {}", e.getMessage());
}
String contentString = new String(serverTransaction.getRequest().getRawContent()); String contentString = new String(serverTransaction.getRequest().getRawContent());
// jainSip不支持y=字段， 移除移除以解析。 // jainSip不支持y=字段， 移除移除以解析。
String substring = contentString; String substring = contentString;
@ -803,8 +863,9 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
if (ssrcIndex > 0) { if (ssrcIndex > 0) {
substring = contentString.substring(0, ssrcIndex); substring = contentString.substring(0, ssrcIndex);
} }
SessionDescription sdp = SdpFactory.getInstance().createSessionDescription(substring); SessionDescription sdp = null;
try {
sdp = SdpFactory.getInstance().createSessionDescription(substring);
// 获取支持的格式 // 获取支持的格式
Vector mediaDescriptions = sdp.getMediaDescriptions(true); Vector mediaDescriptions = sdp.getMediaDescriptions(true);
// 查看是否支持PS 负载96 // 查看是否支持PS 负载96
@ -838,16 +899,29 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
if (port == -1) { if (port == -1) {
logger.info("不支持的媒体格式，返回415"); logger.info("不支持的媒体格式，返回415");
// 回复不支持的格式 // 回复不支持的格式
try {
responseAck(serverTransaction, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415 responseAck(serverTransaction, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] invite 不支持的媒体格式，返回415， {}", e.getMessage());
}
return; return;
} }
String username = sdp.getOrigin().getUsername(); String username = sdp.getOrigin().getUsername();
String addressStr = sdp.getOrigin().getAddress(); String addressStr = sdp.getOrigin().getAddress();
logger.info("设备{}请求语音流，地址：{}:{}，ssrc：{}", username, addressStr, port, ssrc); logger.info("设备{}请求语音流，地址：{}:{}，ssrc：{}", username, addressStr, port, ssrc);
} catch (SdpException e) {
logger.error("[SDP解析异常]", e);
}
} else { } else {
logger.warn("来自无效设备/平台的请求"); logger.warn("来自无效设备/平台的请求");
responseAck(serverTransaction, Response.BAD_REQUEST); try {
responseAck(serverTransaction, Response.BAD_REQUEST);; // 不支持的格式，发415
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] invite 来自无效设备/平台的请求， {}", e.getMessage());
}
} }
} }
} }

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

@ -93,10 +93,13 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
@Override @Override
public void process(RequestEvent evt) { public void process(RequestEvent evt) {
try {
taskQueue.offer(new HandlerCatchData(evt, null, null));
ServerTransaction serverTransaction = getServerTransaction(evt); ServerTransaction serverTransaction = getServerTransaction(evt);
try {
responseAck(serverTransaction, Response.OK); responseAck(serverTransaction, Response.OK);
}catch (SipException | InvalidArgumentException | ParseException e) {
e.printStackTrace();
}
taskQueue.offer(new HandlerCatchData(evt, null, null));
if (!taskQueueHandlerRun) { if (!taskQueueHandlerRun) {
taskQueueHandlerRun = true; taskQueueHandlerRun = true;
taskExecutor.execute(()-> { taskExecutor.execute(()-> {
@ -129,11 +132,6 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
taskQueueHandlerRun = false; taskQueueHandlerRun = false;
}); });
} }
} catch (SipException | InvalidArgumentException | ParseException e) {
e.printStackTrace();
} finally {
taskQueueHandlerRun = false;
}
} }
/** /**

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

@ -112,10 +112,10 @@ public class DeviceControlQueryMessageHandler extends SIPRequestProcessorParent
if (deviceForPlatform == null) { if (deviceForPlatform == null) {
try { try {
responseAck(serverTransaction, Response.NOT_FOUND); responseAck(serverTransaction, Response.NOT_FOUND);
return;
} catch (SipException | InvalidArgumentException | ParseException e) { } catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 错误信息: {}", e.getMessage()); logger.error("[命令发送失败] 错误信息: {}", e.getMessage());
} }
return;
} }
try { try {
cmder.fronEndCmd(deviceForPlatform, channelId, cmdString, eventResult -> { cmder.fronEndCmd(deviceForPlatform, channelId, cmdString, eventResult -> {

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

@ -52,7 +52,12 @@ public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent imp
// 未注册的设备不做处理 // 未注册的设备不做处理
return; return;
} }
// 回复200 OK
try { try {
responseAck(getServerTransaction(evt), Response.OK);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 心跳回复: {}", e.getMessage());
}
// 判断RPort是否改变，改变则说明路由nat信息变化，修改设备信息 // 判断RPort是否改变，改变则说明路由nat信息变化，修改设备信息
// 获取到通信地址等信息 // 获取到通信地址等信息
ViaHeader viaHeader = (ViaHeader) evt.getRequest().getHeader(ViaHeader.NAME); ViaHeader viaHeader = (ViaHeader) evt.getRequest().getHeader(ViaHeader.NAME);
@ -68,8 +73,7 @@ public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent imp
device.setHostAddress(received.concat(":").concat(String.valueOf(rPort))); device.setHostAddress(received.concat(":").concat(String.valueOf(rPort)));
} }
device.setKeepaliveTime(DateUtil.getNow()); device.setKeepaliveTime(DateUtil.getNow());
// 回复200 OK
responseAck(getServerTransaction(evt), Response.OK);
if (device.getOnline() == 1) { if (device.getOnline() == 1) {
deviceService.updateDevice(device); deviceService.updateDevice(device);
}else { }else {
@ -78,9 +82,6 @@ public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent imp
deviceService.online(device); deviceService.online(device);
} }
} }
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 心跳回复: {}", e.getMessage());
}
} }
@Override @Override

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

@ -81,8 +81,12 @@ public class MobilePositionNotifyMessageHandler extends SIPRequestProcessorParen
try { try {
Element rootElementAfterCharset = getRootElement(sipMsgInfo.getEvt(), sipMsgInfo.getDevice().getCharset()); Element rootElementAfterCharset = getRootElement(sipMsgInfo.getEvt(), sipMsgInfo.getDevice().getCharset());
if (rootElementAfterCharset == null) { if (rootElementAfterCharset == null) {
try {
logger.warn("[ 移动设备位置数据通知 ] content cannot be null, {}", sipMsgInfo.getEvt().getRequest()); logger.warn("[ 移动设备位置数据通知 ] content cannot be null, {}", sipMsgInfo.getEvt().getRequest());
responseAck(getServerTransaction(sipMsgInfo.getEvt()), Response.BAD_REQUEST); responseAck(getServerTransaction(sipMsgInfo.getEvt()), Response.BAD_REQUEST);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 移动设备位置数据通知 内容为空: {}", e.getMessage());
}
continue; continue;
} }
MobilePosition mobilePosition = new MobilePosition(); MobilePosition mobilePosition = new MobilePosition();
@ -133,7 +137,11 @@ public class MobilePositionNotifyMessageHandler extends SIPRequestProcessorParen
} }
storager.updateChannelPosition(deviceChannel); storager.updateChannelPosition(deviceChannel);
//回复 200 OK //回复 200 OK
try {
responseAck(getServerTransaction(sipMsgInfo.getEvt()), Response.OK); responseAck(getServerTransaction(sipMsgInfo.getEvt()), Response.OK);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 移动设备位置数据回复200: {}", e.getMessage());
}
// 发送redis消息。 通知位置信息的变化 // 发送redis消息。 通知位置信息的变化
JSONObject jsonObject = new JSONObject(); JSONObject jsonObject = new JSONObject();
@ -147,7 +155,7 @@ public class MobilePositionNotifyMessageHandler extends SIPRequestProcessorParen
jsonObject.put("speed", mobilePosition.getSpeed()); jsonObject.put("speed", mobilePosition.getSpeed());
redisCatchStorage.sendMobilePositionMsg(jsonObject); redisCatchStorage.sendMobilePositionMsg(jsonObject);
} catch (DocumentException | SipException | InvalidArgumentException | ParseException e) { } catch (DocumentException e) {
e.printStackTrace(); e.printStackTrace();
} }

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

@ -67,6 +67,9 @@ public class CatalogQueryMessageHandler extends SIPRequestProcessorParent implem
try { try {
// 回复200 OK // 回复200 OK
responseAck(getServerTransaction(evt), Response.OK); responseAck(getServerTransaction(evt), Response.OK);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 目录查询回复200OK: {}", e.getMessage());
}
Element snElement = rootElement.element("SN"); Element snElement = rootElement.element("SN");
String sn = snElement.getText(); String sn = snElement.getText();
// 准备回复通道信息 // 准备回复通道信息
@ -94,6 +97,7 @@ public class CatalogQueryMessageHandler extends SIPRequestProcessorParent implem
if (gbStreams.size() > 0) { if (gbStreams.size() > 0) {
allChannels.addAll(gbStreams); allChannels.addAll(gbStreams);
} }
try {
if (allChannels.size() > 0) { if (allChannels.size() > 0) {
cmderFroPlatform.catalogQuery(allChannels, parentPlatform, sn, fromHeader.getTag()); cmderFroPlatform.catalogQuery(allChannels, parentPlatform, sn, fromHeader.getTag());
}else { }else {
@ -101,9 +105,11 @@ public class CatalogQueryMessageHandler extends SIPRequestProcessorParent implem
cmderFroPlatform.catalogQuery(null, parentPlatform, sn, fromHeader.getTag(), 0); cmderFroPlatform.catalogQuery(null, parentPlatform, sn, fromHeader.getTag(), 0);
} }
} catch (SipException | InvalidArgumentException | ParseException e) { } catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 目录查询: {}", e.getMessage()); logger.error("[命令发送失败] 国标级联 目录查询回复: {}", e.getMessage());
} }
} }
private DeviceChannel getChannelForPlatform(ParentPlatform platform) { private DeviceChannel getChannelForPlatform(ParentPlatform platform) {

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

@ -53,6 +53,9 @@ public class ConfigDownloadResponseMessageHandler extends SIPRequestProcessorPar
try { try {
// 回复200 OK // 回复200 OK
responseAck(getServerTransaction(evt), Response.OK); responseAck(getServerTransaction(evt), Response.OK);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 设备配置查询: {}", e.getMessage());
}
// 此处是对本平台发出DeviceControl指令的应答 // 此处是对本平台发出DeviceControl指令的应答
JSONObject json = new JSONObject(); JSONObject json = new JSONObject();
XmlUtil.node2Json(element, json); XmlUtil.node2Json(element, json);
@ -63,9 +66,7 @@ public class ConfigDownloadResponseMessageHandler extends SIPRequestProcessorPar
msg.setKey(key); msg.setKey(key);
msg.setData(json); msg.setData(json);
deferredResultHolder.invokeAllResult(msg); deferredResultHolder.invokeAllResult(msg);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 设备配置查询: {}", e.getMessage());
}
} }

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

@ -47,6 +47,9 @@ public class DeviceControlResponseMessageHandler extends SIPRequestProcessorPare
// 此处是对本平台发出DeviceControl指令的应答 // 此处是对本平台发出DeviceControl指令的应答
try { try {
responseAck(getServerTransaction(evt), Response.OK); responseAck(getServerTransaction(evt), Response.OK);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 设备控制: {}", e.getMessage());
}
JSONObject json = new JSONObject(); JSONObject json = new JSONObject();
String channelId = getText(element, "DeviceID"); String channelId = getText(element, "DeviceID");
XmlUtil.node2Json(element, json); XmlUtil.node2Json(element, json);
@ -58,9 +61,7 @@ public class DeviceControlResponseMessageHandler extends SIPRequestProcessorPare
msg.setKey(key); msg.setKey(key);
msg.setData(json); msg.setData(json);
deferredResultHolder.invokeAllResult(msg); deferredResultHolder.invokeAllResult(msg);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 设备控制: {}", e.getMessage());
}
} }
@Override @Override

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

@ -78,9 +78,14 @@ public class DeviceInfoResponseMessageHandler extends SIPRequestProcessorParent
ServerTransaction serverTransaction = getServerTransaction(evt); ServerTransaction serverTransaction = getServerTransaction(evt);
try { try {
rootElement = getRootElement(evt, device.getCharset()); rootElement = getRootElement(evt, device.getCharset());
if (rootElement == null) { if (rootElement == null) {
logger.warn("[ 接收到DeviceInfo应答消息 ] content cannot be null, {}", evt.getRequest()); logger.warn("[ 接收到DeviceInfo应答消息 ] content cannot be null, {}", evt.getRequest());
try {
responseAck(serverTransaction, Response.BAD_REQUEST); responseAck(serverTransaction, Response.BAD_REQUEST);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] DeviceInfo应答消息 BAD_REQUEST: {}", e.getMessage());
}
return; return;
} }
Element deviceIdElement = rootElement.element("DeviceID"); Element deviceIdElement = rootElement.element("DeviceID");
@ -100,17 +105,16 @@ public class DeviceInfoResponseMessageHandler extends SIPRequestProcessorParent
msg.setKey(key); msg.setKey(key);
msg.setData(device); msg.setData(device);
deferredResultHolder.invokeAllResult(msg); deferredResultHolder.invokeAllResult(msg);
} catch (DocumentException e) {
throw new RuntimeException(e);
}
try {
// 回复200 OK // 回复200 OK
responseAck(serverTransaction, Response.OK); responseAck(serverTransaction, Response.OK);
} catch (DocumentException e) { } catch (SipException | InvalidArgumentException | ParseException e) {
e.printStackTrace(); logger.error("[命令发送失败] DeviceInfo应答消息 200: {}", e.getMessage());
} catch (InvalidArgumentException e) {
e.printStackTrace();
} catch (ParseException e) {
e.printStackTrace();
} catch (SipException e) {
e.printStackTrace();
} }
} }
@Override @Override

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

@ -71,7 +71,11 @@ public class MobilePositionResponseMessageHandler extends SIPRequestProcessorPar
rootElement = getRootElement(evt, device.getCharset()); rootElement = getRootElement(evt, device.getCharset());
if (rootElement == null) { if (rootElement == null) {
logger.warn("[ 移动设备位置数据查询回复 ] content cannot be null, {}", evt.getRequest()); logger.warn("[ 移动设备位置数据查询回复 ] content cannot be null, {}", evt.getRequest());
try {
responseAck(serverTransaction, Response.BAD_REQUEST); responseAck(serverTransaction, Response.BAD_REQUEST);
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 移动设备位置数据查询 BAD_REQUEST: {}", e.getMessage());
}
return; return;
} }
MobilePosition mobilePosition = new MobilePosition(); MobilePosition mobilePosition = new MobilePosition();
@ -133,8 +137,13 @@ public class MobilePositionResponseMessageHandler extends SIPRequestProcessorPar
jsonObject.put("speed", mobilePosition.getSpeed()); jsonObject.put("speed", mobilePosition.getSpeed());
redisCatchStorage.sendMobilePositionMsg(jsonObject); redisCatchStorage.sendMobilePositionMsg(jsonObject);
//回复 200 OK //回复 200 OK
try {
responseAck(serverTransaction, Response.OK); responseAck(serverTransaction, Response.OK);
} catch (DocumentException | SipException | InvalidArgumentException | ParseException e) { } catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 移动设备位置数据查询 200: {}", e.getMessage());
}
} catch (DocumentException e) {
e.printStackTrace(); e.printStackTrace();
} }
} }

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

@ -58,7 +58,11 @@ public class PresetQueryResponseMessageHandler extends SIPRequestProcessorParent
if (rootElement == null) { if (rootElement == null) {
logger.warn("[ 设备预置位查询应答 ] content cannot be null, {}", evt.getRequest()); logger.warn("[ 设备预置位查询应答 ] content cannot be null, {}", evt.getRequest());
try {
responseAck(serverTransaction, Response.BAD_REQUEST); responseAck(serverTransaction, Response.BAD_REQUEST);
} catch (InvalidArgumentException | ParseException | SipException e) {
logger.error("[命令发送失败] 设备预置位查询应答处理: {}", e.getMessage());
}
return; return;
} }
Element presetListNumElement = rootElement.element("PresetList"); Element presetListNumElement = rootElement.element("PresetList");
@ -67,7 +71,11 @@ public class PresetQueryResponseMessageHandler extends SIPRequestProcessorParent
String deviceId = getText(rootElement, "DeviceID"); String deviceId = getText(rootElement, "DeviceID");
String key = DeferredResultHolder.CALLBACK_CMD_PRESETQUERY + deviceId; String key = DeferredResultHolder.CALLBACK_CMD_PRESETQUERY + deviceId;
if (snElement == null || presetListNumElement == null) { if (snElement == null || presetListNumElement == null) {
try {
responseAck(serverTransaction, Response.BAD_REQUEST, "xml error"); responseAck(serverTransaction, Response.BAD_REQUEST, "xml error");
} catch (InvalidArgumentException | ParseException | SipException e) {
logger.error("[命令发送失败] 设备预置位查询应答处理: {}", e.getMessage());
}
return; return;
} }
int sumNum = Integer.parseInt(presetListNumElement.attributeValue("Num")); int sumNum = Integer.parseInt(presetListNumElement.attributeValue("Num"));
@ -94,12 +102,14 @@ public class PresetQueryResponseMessageHandler extends SIPRequestProcessorParent
requestMessage.setKey(key); requestMessage.setKey(key);
requestMessage.setData(presetQuerySipReqList); requestMessage.setData(presetQuerySipReqList);
deferredResultHolder.invokeAllResult(requestMessage); deferredResultHolder.invokeAllResult(requestMessage);
try {
responseAck(serverTransaction, Response.OK); responseAck(serverTransaction, Response.OK);
} catch (DocumentException e) {
logger.error("[解析xml]失败: ", e);
} catch (InvalidArgumentException | ParseException | SipException e) { } catch (InvalidArgumentException | ParseException | SipException e) {
logger.error("[命令发送失败] 设备预置位查询应答处理: {}", e.getMessage()); logger.error("[命令发送失败] 设备预置位查询应答处理: {}", e.getMessage());
} }
} catch (DocumentException e) {
logger.error("[解析xml]失败: ", e);
}
} }
@Override @Override

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

@ -69,10 +69,12 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent
@Override @Override
public void handForDevice(RequestEvent evt, Device device, Element rootElement) { public void handForDevice(RequestEvent evt, Device device, Element rootElement) {
// 回复200 OK
try { try {
// 回复200 OK
responseAck(getServerTransaction(evt), Response.OK); responseAck(getServerTransaction(evt), Response.OK);
}catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 国标录像: {}", e.getMessage());
}
taskQueue.offer(new HandlerCatchData(evt, device, rootElement)); taskQueue.offer(new HandlerCatchData(evt, device, rootElement));
if (!taskQueueHandlerRun) { if (!taskQueueHandlerRun) {
taskQueueHandlerRun = true; taskQueueHandlerRun = true;
@ -153,12 +155,6 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent
taskQueueHandlerRun = false; taskQueueHandlerRun = false;
}); });
} }
} catch (SipException | InvalidArgumentException | ParseException e) {
logger.error("[命令发送失败] 国标级联 国标录像: {}", e.getMessage());
} finally {
taskQueueHandlerRun = false;
}
} }
@Override @Override