From 37c61c4fb0ecc45f34dd04f7701690c716115bf0 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Fri, 13 May 2022 15:32:30 +0800 Subject: [PATCH 1/6] =?UTF-8?q?=E6=94=AF=E6=8C=81=E6=B5=B7=E5=BA=B7?= =?UTF-8?q?=E6=91=84=E5=83=8F=E5=A4=B4=E5=9B=BD=E6=A0=87=E5=BD=95=E5=83=8F?= =?UTF-8?q?=E6=9F=A5=E7=9C=8B=E6=92=AD=E6=94=BE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../genersoft/iot/vmp/gb28181/SipLayer.java | 8 +- .../iot/vmp/gb28181/bean/RecordInfo.java | 14 ++- .../vmp/gb28181/session/RecordDataCatch.java | 94 ++++++++++++++++++ .../callback/CheckForAllRecordsThread.java | 76 -------------- .../transmit/cmd/impl/SIPCommander.java | 37 +++---- .../request/impl/InviteRequestProcessor.java | 4 +- .../impl/RegisterRequestProcessor.java | 12 +-- .../cmd/RecordInfoResponseMessageHandler.java | 99 +++++++++---------- .../iot/vmp/media/zlm/ZLMRunner.java | 2 +- .../zlm/event/ZLMStatusEventListener.java | 5 +- .../iot/vmp/service/IMediaServerService.java | 2 +- .../vmp/service/impl/DeviceServiceImpl.java | 2 +- .../service/impl/MediaServerServiceImpl.java | 18 ++-- .../iot/vmp/service/impl/PlayServiceImpl.java | 8 +- .../com/genersoft/iot/vmp/utils/DateUtil.java | 57 +++++------ .../gb28181/record/GBRecordController.java | 33 ++++++- web_src/src/components/channelList.vue | 2 +- .../src/components/dialog/devicePlayer.vue | 16 ++- 18 files changed, 275 insertions(+), 214 deletions(-) create mode 100644 src/main/java/com/genersoft/iot/vmp/gb28181/session/RecordDataCatch.java delete mode 100644 src/main/java/com/genersoft/iot/vmp/gb28181/transmit/callback/CheckForAllRecordsThread.java diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java b/src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java index 2b93d702..a3428b10 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java @@ -81,11 +81,11 @@ public class SipLayer{ tcpSipProvider.setDialogErrorsAutomaticallyHandled(); tcpSipProvider.addSipListener(sipProcessorObserver); // tcpSipProvider.setAutomaticDialogSupportEnabled(false); - logger.info("Sip Server TCP 启动成功 port {" + sipConfig.getMonitorIp() + ":" + sipConfig.getPort() + "}"); + logger.info("[Sip Server] TCP 启动成功 {}:{}", sipConfig.getMonitorIp(), sipConfig.getPort()); } catch (TransportNotSupportedException e) { e.printStackTrace(); } catch (InvalidArgumentException e) { - logger.error("无法使用 [ {}:{} ]作为SIP[ TCP ]服务，可排查: 1. sip.monitor-ip 是否为本机网卡IP; 2. sip.port 是否已被占用" + logger.error("[Sip Server] 无法使用 [ {}:{} ]作为SIP[ TCP ]服务，可排查: 1. sip.monitor-ip 是否为本机网卡IP; 2. sip.port 是否已被占用" , sipConfig.getMonitorIp(), sipConfig.getPort()); } catch (TooManyListenersException e) { e.printStackTrace(); @@ -108,14 +108,14 @@ public class SipLayer{ } catch (TransportNotSupportedException e) { e.printStackTrace(); } catch (InvalidArgumentException e) { - logger.error("无法使用 [ {}:{} ]作为SIP[ UDP ]服务，可排查: 1. sip.monitor-ip 是否为本机网卡IP; 2. sip.port 是否已被占用" + logger.error("[Sip Server] 无法使用 [ {}:{} ]作为SIP[ UDP ]服务，可排查: 1. sip.monitor-ip 是否为本机网卡IP; 2. sip.port 是否已被占用" , sipConfig.getMonitorIp(), sipConfig.getPort()); } catch (TooManyListenersException e) { e.printStackTrace(); } catch (ObjectInUseException e) { e.printStackTrace(); } - logger.info("Sip Server UDP 启动成功 port [" + sipConfig.getMonitorIp() + ":" + sipConfig.getPort() + "]"); + logger.info("[Sip Server] UDP 启动成功 {}:{}", sipConfig.getMonitorIp(), sipConfig.getPort()); return udpSipProvider; } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/RecordInfo.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/RecordInfo.java index 24fc2212..55bf71f2 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/RecordInfo.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/RecordInfo.java @@ -1,8 +1,6 @@ package com.genersoft.iot.vmp.gb28181.bean; - -//import gov.nist.javax.sip.header.SIPDate; - +import java.util.Date; import java.util.List; /** @@ -21,6 +19,8 @@ public class RecordInfo { private String name; private int sumNum; + + private Date lastTime; private List recordList; @@ -71,4 +71,12 @@ public class RecordInfo { public void setSn(String sn) { this.sn = sn; } + + public Date getLastTime() { + return lastTime; + } + + public void setLastTime(Date lastTime) { + this.lastTime = lastTime; + } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/session/RecordDataCatch.java b/src/main/java/com/genersoft/iot/vmp/gb28181/session/RecordDataCatch.java new file mode 100644 index 00000000..0d166c12 --- /dev/null +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/session/RecordDataCatch.java @@ -0,0 +1,94 @@ +package com.genersoft.iot.vmp.gb28181.session; + +import com.genersoft.iot.vmp.gb28181.bean.*; +import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder; +import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage; +import com.genersoft.iot.vmp.storager.IVideoManagerStorage; +import com.genersoft.iot.vmp.utils.DateUtil; +import com.genersoft.iot.vmp.vmanager.bean.WVPResult; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +import java.text.SimpleDateFormat; +import java.util.*; +import java.util.concurrent.ConcurrentHashMap; + +/** + * @author lin + */ +@Component +public class RecordDataCatch { + + public static Map data = new ConcurrentHashMap<>(); + + @Autowired + private DeferredResultHolder deferredResultHolder; + + + public int put(String deviceId, String sn, int sumNum, List recordItems) { + String key = deviceId + sn; + RecordInfo recordInfo = data.get(key); + if (recordInfo == null) { + recordInfo = new RecordInfo(); + recordInfo.setDeviceId(deviceId); + recordInfo.setSn(sn.trim()); + recordInfo.setSumNum(sumNum); + recordInfo.setRecordList(Collections.synchronizedList(new ArrayList<>())); + recordInfo.setLastTime(new Date(System.currentTimeMillis())); + recordInfo.getRecordList().addAll(recordItems); + data.put(key, recordInfo); + }else { + // 同一个设备的通道同步请求只考虑一个，其他的直接忽略 + if (!Objects.equals(sn.trim(), recordInfo.getSn())) { + return 0; + } + recordInfo.getRecordList().addAll(recordItems); + recordInfo.setLastTime(new Date(System.currentTimeMillis())); + } + return recordInfo.getRecordList().size(); + } + + @Scheduled(fixedRate = 5 * 1000) //每5秒执行一次, 发现数据5秒未更新则移除数据并认为数据接收超时 + private void timerTask(){ + Set keys = data.keySet(); + Calendar calendarBefore5S = Calendar.getInstance(); + calendarBefore5S.setTime(new Date()); + calendarBefore5S.set(Calendar.SECOND, calendarBefore5S.get(Calendar.SECOND) - 5); + + for (String key : keys) { + RecordInfo recordInfo = data.get(key); + // 超过五秒收不到消息任务超时， 只更新这一部分数据 + if ( recordInfo.getLastTime().before(calendarBefore5S.getTime())) { + // 处理录像数据， 返回给前端 + String msgKey = DeferredResultHolder.CALLBACK_CMD_RECORDINFO + recordInfo.getDeviceId() + recordInfo.getSn(); + + WVPResult wvpResult = new WVPResult<>(); + wvpResult.setCode(0); + wvpResult.setMsg("success"); + // 对数据进行排序 + Collections.sort(recordInfo.getRecordList()); + wvpResult.setData(recordInfo); + + RequestMessage msg = new RequestMessage(); + msg.setKey(msgKey); + msg.setData(wvpResult); + deferredResultHolder.invokeAllResult(msg); + data.remove(key); + } + } + } + + public boolean isComplete(String deviceId, String sn) { + RecordInfo recordInfo = data.get(deviceId + sn); + return recordInfo != null && recordInfo.getRecordList().size() == recordInfo.getSumNum(); + } + + public RecordInfo getRecordInfo(String deviceId, String sn) { + return data.get(deviceId + sn); + } + + public void remove(String deviceId, String sn) { + data.remove(deviceId + sn); + } +} diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/callback/CheckForAllRecordsThread.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/callback/CheckForAllRecordsThread.java deleted file mode 100644 index 8a2e9009..00000000 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/callback/CheckForAllRecordsThread.java +++ /dev/null @@ -1,76 +0,0 @@ -package com.genersoft.iot.vmp.gb28181.transmit.callback; - -import com.genersoft.iot.vmp.gb28181.bean.RecordInfo; -import com.genersoft.iot.vmp.gb28181.bean.RecordItem; -import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd.RecordInfoResponseMessageHandler; -import com.genersoft.iot.vmp.utils.redis.RedisUtil; -import org.slf4j.Logger; - -import java.util.ArrayList; -import java.util.Comparator; -import java.util.List; -import java.util.concurrent.TimeUnit; - -@SuppressWarnings("unchecked") -public class CheckForAllRecordsThread extends Thread { - - private String key; - - private RecordInfo recordInfo; - - private RedisUtil redis; - - private Logger logger; - - private DeferredResultHolder deferredResultHolder; - - public CheckForAllRecordsThread(String key, RecordInfo recordInfo) { - this.key = key; - this.recordInfo = recordInfo; - } - - @Override - public void run() { - - String cacheKey = this.key; - - for (long stop = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); stop > System.nanoTime();) { - List cacheKeys = redis.scan(cacheKey + "_*"); - List totalRecordList = new ArrayList(); - for (int i = 0; i < cacheKeys.size(); i++) { - totalRecordList.addAll((List) redis.get(cacheKeys.get(i).toString())); - } - if (totalRecordList.size() < this.recordInfo.getSumNum()) { - logger.info("已获取" + totalRecordList.size() + "项录像数据，共" + this.recordInfo.getSumNum() + "项"); - } else { - logger.info("录像数据已全部获取，共 {} 项", this.recordInfo.getSumNum()); - this.recordInfo.setRecordList(totalRecordList); - for (int i = 0; i < cacheKeys.size(); i++) { - redis.del(cacheKeys.get(i).toString()); - } - break; - } - } - // 自然顺序排序, 元素进行升序排列 - this.recordInfo.getRecordList().sort(Comparator.naturalOrder()); - RequestMessage msg = new RequestMessage(); - msg.setKey(DeferredResultHolder.CALLBACK_CMD_RECORDINFO + recordInfo.getDeviceId() + recordInfo.getSn()); - msg.setData(recordInfo); - deferredResultHolder.invokeAllResult(msg); - logger.info("处理完成，返回结果"); - RecordInfoResponseMessageHandler.threadNameList.remove(cacheKey); - } - - public void setRedis(RedisUtil redis) { - this.redis = redis; - } - - public void setDeferredResultHolder(DeferredResultHolder deferredResultHolder) { - this.deferredResultHolder = deferredResultHolder; - } - - public void setLogger(Logger logger) { - this.logger = logger; - } - -} diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java index 567bcac7..8a6e9314 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java @@ -370,7 +370,7 @@ public class SIPCommander implements ISIPCommander { // StringBuffer content = new StringBuffer(200); content.append("v=0\r\n"); - content.append("o="+ sipConfig.getId()+" 0 0 IN IP4 "+ mediaServerItem.getSdpIp() +"\r\n"); + content.append("o="+ channelId+" 0 0 IN IP4 "+ mediaServerItem.getSdpIp() +"\r\n"); content.append("s=Play\r\n"); content.append("c=IN IP4 "+ mediaServerItem.getSdpIp() +"\r\n"); content.append("t=0 0\r\n"); @@ -389,8 +389,7 @@ public class SIPCommander implements ISIPCommander { content.append("a=rtpmap:126 H264/90000\r\n"); content.append("a=rtpmap:125 H264S/90000\r\n"); content.append("a=fmtp:125 profile-level-id=42e01e\r\n"); - content.append("a=rtpmap:99 MP4V-ES/90000\r\n"); - content.append("a=fmtp:99 profile-level-id=3\r\n"); + content.append("a=rtpmap:99 H265/90000\r\n"); content.append("a=rtpmap:98 H264/90000\r\n"); content.append("a=rtpmap:97 MPEG4/90000\r\n"); if("TCP-PASSIVE".equals(streamMode)){ // tcp被动模式 @@ -402,16 +401,17 @@ public class SIPCommander implements ISIPCommander { } }else { if("TCP-PASSIVE".equals(streamMode)) { - content.append("m=video "+ ssrcInfo.getPort() +" TCP/RTP/AVP 96 98 97\r\n"); + content.append("m=video "+ ssrcInfo.getPort() +" TCP/RTP/AVP 96 97 98 99\r\n"); }else if ("TCP-ACTIVE".equals(streamMode)) { - content.append("m=video "+ ssrcInfo.getPort() +" TCP/RTP/AVP 96 98 97\r\n"); + content.append("m=video "+ ssrcInfo.getPort() +" TCP/RTP/AVP 96 97 98 99\r\n"); }else if("UDP".equals(streamMode)) { - content.append("m=video "+ ssrcInfo.getPort() +" RTP/AVP 96 98 97\r\n"); + content.append("m=video "+ ssrcInfo.getPort() +" RTP/AVP 96 97 98 99\r\n"); } content.append("a=recvonly\r\n"); content.append("a=rtpmap:96 PS/90000\r\n"); content.append("a=rtpmap:98 H264/90000\r\n"); content.append("a=rtpmap:97 MPEG4/90000\r\n"); + content.append("a=rtpmap:99 H265/90000\r\n"); if ("TCP-PASSIVE".equals(streamMode)) { // tcp被动模式 content.append("a=setup:passive\r\n"); content.append("a=connection:new\r\n"); @@ -467,7 +467,7 @@ public class SIPCommander implements ISIPCommander { StringBuffer content = new StringBuffer(200); content.append("v=0\r\n"); - content.append("o="+sipConfig.getId()+" 0 0 IN IP4 " + mediaServerItem.getSdpIp() + "\r\n"); + content.append("o="+channelId+" 0 0 IN IP4 " + mediaServerItem.getSdpIp() + "\r\n"); content.append("s=Playback\r\n"); content.append("u="+channelId+":0\r\n"); content.append("c=IN IP4 "+mediaServerItem.getSdpIp()+"\r\n"); @@ -490,8 +490,7 @@ public class SIPCommander implements ISIPCommander { content.append("a=rtpmap:126 H264/90000\r\n"); content.append("a=rtpmap:125 H264S/90000\r\n"); content.append("a=fmtp:125 profile-level-id=42e01e\r\n"); - content.append("a=rtpmap:99 MP4V-ES/90000\r\n"); - content.append("a=fmtp:99 profile-level-id=3\r\n"); + content.append("a=rtpmap:99 H265/90000\r\n"); content.append("a=rtpmap:98 H264/90000\r\n"); content.append("a=rtpmap:97 MPEG4/90000\r\n"); if("TCP-PASSIVE".equals(streamMode)){ // tcp被动模式 @@ -503,16 +502,17 @@ public class SIPCommander implements ISIPCommander { } }else { if("TCP-PASSIVE".equals(streamMode)) { - content.append("m=video "+ ssrcInfo.getPort() +" TCP/RTP/AVP 96 98 97\r\n"); + content.append("m=video "+ ssrcInfo.getPort() +" TCP/RTP/AVP 96 97 98 99\r\n"); }else if ("TCP-ACTIVE".equals(streamMode)) { - content.append("m=video "+ ssrcInfo.getPort() +" TCP/RTP/AVP 96 98 97\r\n"); + content.append("m=video "+ ssrcInfo.getPort() +" TCP/RTP/AVP 96 97 98 99\r\n"); }else if("UDP".equals(streamMode)) { - content.append("m=video "+ ssrcInfo.getPort() +" RTP/AVP 96 98 97\r\n"); + content.append("m=video "+ ssrcInfo.getPort() +" RTP/AVP 96 97 98 99\r\n"); } content.append("a=recvonly\r\n"); content.append("a=rtpmap:96 PS/90000\r\n"); - content.append("a=rtpmap:98 H264/90000\r\n"); content.append("a=rtpmap:97 MPEG4/90000\r\n"); + content.append("a=rtpmap:98 H264/90000\r\n"); + content.append("a=rtpmap:99 H265/90000\r\n"); if("TCP-PASSIVE".equals(streamMode)){ // tcp被动模式 content.append("a=setup:passive\r\n"); content.append("a=connection:new\r\n"); @@ -577,7 +577,7 @@ public class SIPCommander implements ISIPCommander { StringBuffer content = new StringBuffer(200); content.append("v=0\r\n"); - content.append("o="+sipConfig.getId()+" 0 0 IN IP4 " + mediaServerItem.getSdpIp() + "\r\n"); + content.append("o="+channelId+" 0 0 IN IP4 " + mediaServerItem.getSdpIp() + "\r\n"); content.append("s=Download\r\n"); content.append("u="+channelId+":0\r\n"); content.append("c=IN IP4 "+mediaServerItem.getSdpIp()+"\r\n"); @@ -613,16 +613,17 @@ public class SIPCommander implements ISIPCommander { } }else { if("TCP-PASSIVE".equals(streamMode)) { - content.append("m=video "+ ssrcInfo.getPort() +" TCP/RTP/AVP 96 98 97\r\n"); + content.append("m=video "+ ssrcInfo.getPort() +" TCP/RTP/AVP 96 97 98 99\r\n"); }else if ("TCP-ACTIVE".equals(streamMode)) { - content.append("m=video "+ ssrcInfo.getPort() +" TCP/RTP/AVP 96 98 97\r\n"); + content.append("m=video "+ ssrcInfo.getPort() +" TCP/RTP/AVP 96 97 98 99\r\n"); }else if("UDP".equals(streamMode)) { - content.append("m=video "+ ssrcInfo.getPort() +" RTP/AVP 96 98 97\r\n"); + content.append("m=video "+ ssrcInfo.getPort() +" RTP/AVP 96 97 98 99\r\n"); } content.append("a=recvonly\r\n"); content.append("a=rtpmap:96 PS/90000\r\n"); - content.append("a=rtpmap:98 H264/90000\r\n"); content.append("a=rtpmap:97 MPEG4/90000\r\n"); + content.append("a=rtpmap:98 H264/90000\r\n"); + content.append("a=rtpmap:99 H265/90000\r\n"); if("TCP-PASSIVE".equals(streamMode)){ // tcp被动模式 content.append("a=setup:passive\r\n"); content.append("a=connection:new\r\n"); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java index 1ba1b3e7..7c63279d 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java @@ -331,7 +331,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements sendRtpItem.setApp("rtp"); if ("Playback".equals(sessionName)) { sendRtpItem.setPlayType(InviteStreamType.PLAYBACK); - SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, true); + SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, null, true, true); sendRtpItem.setStreamId(ssrcInfo.getStream()); // 写入redis， 超时时回复 redisCatchStorage.updateSendRTPSever(sendRtpItem); @@ -372,7 +372,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements if (mediaServerItem.isRtpEnable()) { streamId = String.format("%s_%s", device.getDeviceId(), channelId); } - SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, true); + SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, true, false); sendRtpItem.setStreamId(ssrcInfo.getStream()); // 写入redis， 超时时回复 redisCatchStorage.updateSendRTPSever(sendRtpItem); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/RegisterRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/RegisterRequestProcessor.java index 594c41a1..1c054401 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/RegisterRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/RegisterRequestProcessor.java @@ -81,7 +81,7 @@ public class RegisterRequestProcessor extends SIPRequestProcessorParent implemen try { RequestEventExt evtExt = (RequestEventExt) evt; String requestAddress = evtExt.getRemoteIpAddress() + ":" + evtExt.getRemotePort(); - logger.info("[{}] 收到注册请求，开始处理", requestAddress); + logger.info("[注册请求] 开始处理: {}", requestAddress); Request request = evt.getRequest(); ExpiresHeader expiresHeader = (ExpiresHeader) request.getHeader(Expires.NAME); Response response = null; @@ -95,7 +95,7 @@ public class RegisterRequestProcessor extends SIPRequestProcessorParent implemen AuthorizationHeader authHead = (AuthorizationHeader) request.getHeader(AuthorizationHeader.NAME); if (authHead == null) { - logger.info("[{}] 未携带授权头 回复401", requestAddress); + logger.info("[注册请求] 未携带授权头 回复401: {}", requestAddress); response = getMessageFactory().createResponse(Response.UNAUTHORIZED, request); new DigestServerAuthenticationHelper().generateChallenge(getHeaderFactory(), response, sipConfig.getDomain()); sendResponse(evt, response); @@ -111,7 +111,7 @@ public class RegisterRequestProcessor extends SIPRequestProcessorParent implemen // 注册失败 response = getMessageFactory().createResponse(Response.FORBIDDEN, request); response.setReasonPhrase("wrong password"); - logger.info("[{}] 密码/SIP服务器ID错误, 回复403", requestAddress); + logger.info("[注册请求] 密码/SIP服务器ID错误, 回复403: {}", requestAddress); sendResponse(evt, response); return; } @@ -176,11 +176,11 @@ public class RegisterRequestProcessor extends SIPRequestProcessorParent implemen // 注册成功 // 保存到redis if (registerFlag) { - logger.info("[{}] 注册成功! deviceId:" + deviceId, requestAddress); + logger.info("[注册成功] deviceId: {}->{}", deviceId, requestAddress); device.setRegisterTime(DateUtil.getNow()); deviceService.online(device); } else { - logger.info("[{}] 注销成功! deviceId:" + deviceId, requestAddress); + logger.info("[注销成功] deviceId: {}->{}" ,deviceId, requestAddress); deviceService.offline(deviceId); } } catch (SipException | InvalidArgumentException | NoSuchAlgorithmException | ParseException e) { @@ -192,7 +192,7 @@ public class RegisterRequestProcessor extends SIPRequestProcessorParent implemen private void sendResponse(RequestEvent evt, Response response) throws InvalidArgumentException, SipException { ServerTransaction serverTransaction = getServerTransaction(evt); if (serverTransaction == null) { - logger.warn("回复失败：{}", response); + logger.warn("[回复失败]：{}", response); return; } serverTransaction.sendResponse(response); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java index 2de8ef16..4509e424 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/response/cmd/RecordInfoResponseMessageHandler.java @@ -5,14 +5,14 @@ import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform; import com.genersoft.iot.vmp.gb28181.bean.RecordInfo; import com.genersoft.iot.vmp.gb28181.bean.RecordItem; import com.genersoft.iot.vmp.gb28181.event.EventPublisher; -import com.genersoft.iot.vmp.gb28181.transmit.callback.CheckForAllRecordsThread; +import com.genersoft.iot.vmp.gb28181.session.RecordDataCatch; import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder; import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage; import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent; import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.IMessageHandler; import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.ResponseMessageHandler; import com.genersoft.iot.vmp.utils.DateUtil; -import com.genersoft.iot.vmp.utils.redis.RedisUtil; +import com.genersoft.iot.vmp.vmanager.bean.WVPResult; import org.dom4j.DocumentException; import org.dom4j.Element; import org.slf4j.Logger; @@ -20,19 +20,20 @@ import org.slf4j.LoggerFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; +import org.springframework.util.StringUtils; import javax.sip.InvalidArgumentException; import javax.sip.RequestEvent; import javax.sip.SipException; import javax.sip.message.Response; import java.text.ParseException; -import java.util.ArrayList; -import java.util.Iterator; -import java.util.List; -import java.util.UUID; +import java.util.*; import static com.genersoft.iot.vmp.gb28181.utils.XmlUtil.getText; +/** + * @author lin + */ @Component public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent implements InitializingBean, IMessageHandler { @@ -45,11 +46,13 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent private ResponseMessageHandler responseMessageHandler; @Autowired - private RedisUtil redis; + private RecordDataCatch recordDataCatch; @Autowired private DeferredResultHolder deferredResultHolder; + + @Autowired private EventPublisher eventPublisher; @@ -66,32 +69,22 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent responseAck(evt, Response.OK); rootElement = getRootElement(evt, device.getCharset()); - String uuid = UUID.randomUUID().toString().replace("-", ""); - RecordInfo recordInfo = new RecordInfo(); String sn = getText(rootElement, "SN"); - String key = DeferredResultHolder.CALLBACK_CMD_RECORDINFO + device.getDeviceId() + sn; - recordInfo.setDeviceId(device.getDeviceId()); - recordInfo.setSn(sn); - recordInfo.setName(getText(rootElement, "Name")); - if (getText(rootElement, "SumNum") == null || getText(rootElement, "SumNum") == "") { - recordInfo.setSumNum(0); - } else { - recordInfo.setSumNum(Integer.parseInt(getText(rootElement, "SumNum"))); + + String sumNumStr = getText(rootElement, "SumNum"); + int sumNum = 0; + if (!StringUtils.isEmpty(sumNumStr)) { + sumNum = Integer.parseInt(sumNumStr); } Element recordListElement = rootElement.element("RecordList"); - if (recordListElement == null || recordInfo.getSumNum() == 0) { + if (recordListElement == null || sumNum == 0) { logger.info("无录像数据"); - eventPublisher.recordEndEventPush(recordInfo); - RequestMessage msg = new RequestMessage(); - msg.setKey(key); - msg.setData(recordInfo); - deferredResultHolder.invokeAllResult(msg); + recordDataCatch.put(device.getDeviceId(), sn, sumNum, new ArrayList<>()); + releaseRequest(device.getDeviceId(), sn); } else { Iterator recordListIterator = recordListElement.elementIterator(); - List recordList = new ArrayList(); if (recordListIterator != null) { - RecordItem record = new RecordItem(); - logger.info("处理录像列表数据..."); + List recordList = new ArrayList<>(); // 遍历DeviceList while (recordListIterator.hasNext()) { Element itemRecord = recordListIterator.next(); @@ -100,43 +93,31 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent logger.info("记录为空，下一个..."); continue; } - record = new RecordItem(); + RecordItem record = new RecordItem(); record.setDeviceId(getText(itemRecord, "DeviceID")); record.setName(getText(itemRecord, "Name")); record.setFilePath(getText(itemRecord, "FilePath")); record.setFileSize(getText(itemRecord, "FileSize")); record.setAddress(getText(itemRecord, "Address")); - record.setStartTime( - DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(getText(itemRecord, "StartTime"))); - record.setEndTime( - DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(getText(itemRecord, "EndTime"))); + + String startTimeStr = getText(itemRecord, "StartTime"); + record.setStartTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(startTimeStr)); + + String endTimeStr = getText(itemRecord, "EndTime"); + record.setEndTime(DateUtil.ISO8601Toyyyy_MM_dd_HH_mm_ss(endTimeStr)); + record.setSecrecy(itemRecord.element("Secrecy") == null ? 0 : Integer.parseInt(getText(itemRecord, "Secrecy"))); record.setType(getText(itemRecord, "Type")); record.setRecorderId(getText(itemRecord, "RecorderID")); recordList.add(record); } - recordInfo.setRecordList(recordList); + int count = recordDataCatch.put(device.getDeviceId(), sn, sumNum, recordList); + logger.info("[国标录像]， {}->{}: {}/{}", device.getDeviceId(), sn, count, sumNum); } - eventPublisher.recordEndEventPush(recordInfo); - // 改用单独线程统计已获取录像文件数量，避免多包并行分别统计不完整的问题 - String cacheKey = CACHE_RECORDINFO_KEY + device.getDeviceId() + sn; - redis.set(cacheKey + "_" + uuid, recordList, 90); - if (!threadNameList.contains(cacheKey)) { - threadNameList.add(cacheKey); - CheckForAllRecordsThread chk = new CheckForAllRecordsThread(cacheKey, recordInfo); - chk.setName(cacheKey); - chk.setDeferredResultHolder(deferredResultHolder); - chk.setRedis(redis); - chk.setLogger(logger); - chk.start(); - if (logger.isDebugEnabled()) { - logger.debug("Start Thread " + cacheKey + "."); - } - } else { - if (logger.isDebugEnabled()) { - logger.debug("Thread " + cacheKey + " already started."); - } + + if (recordDataCatch.isComplete(device.getDeviceId(), sn)){ + releaseRequest(device.getDeviceId(), sn); } } } catch (SipException e) { @@ -154,4 +135,20 @@ public class RecordInfoResponseMessageHandler extends SIPRequestProcessorParent public void handForPlatform(RequestEvent evt, ParentPlatform parentPlatform, Element element) { } + + public void releaseRequest(String deviceId, String sn){ + String key = DeferredResultHolder.CALLBACK_CMD_RECORDINFO + deviceId + sn; + WVPResult wvpResult = new WVPResult<>(); + wvpResult.setCode(0); + wvpResult.setMsg("success"); + // 对数据进行排序 + Collections.sort(recordDataCatch.getRecordInfo(deviceId, sn).getRecordList()); + wvpResult.setData(recordDataCatch.getRecordInfo(deviceId, sn)); + + RequestMessage msg = new RequestMessage(); + msg.setKey(key); + msg.setData(wvpResult); + deferredResultHolder.invokeAllResult(msg); + recordDataCatch.remove(deviceId, sn); + } } diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRunner.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRunner.java index 02612093..138af7ae 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRunner.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRunner.java @@ -89,7 +89,7 @@ public class ZLMRunner implements CommandLineRunner { }); // 获取zlm信息 - logger.info("[zlm接入]等待默认zlm中..."); + logger.info("[zlm] 等待默认zlm中..."); // 获取所有的zlm， 并开启主动连接 List all = mediaServerService.getAllFromDatabase(); diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/event/ZLMStatusEventListener.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/event/ZLMStatusEventListener.java index b193add9..f4a9febb 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/event/ZLMStatusEventListener.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/event/ZLMStatusEventListener.java @@ -39,8 +39,7 @@ public class ZLMStatusEventListener { @Async @EventListener public void onApplicationEvent(ZLMOnlineEvent event) { - - logger.info("【ZLM上线】ID：" + event.getMediaServerId()); + logger.info("[ZLM] 上线 ID：" + event.getMediaServerId()); streamPushService.zlmServerOnline(event.getMediaServerId()); streamProxyService.zlmServerOnline(event.getMediaServerId()); @@ -50,7 +49,7 @@ public class ZLMStatusEventListener { @EventListener public void onApplicationEvent(ZLMOfflineEvent event) { - logger.info("ZLM离线事件触发，ID：" + event.getMediaServerId()); + logger.info("[ZLM] 离线，ID：" + event.getMediaServerId()); // 处理ZLM离线 mediaServerService.zlmServerOffline(event.getMediaServerId()); streamProxyService.zlmServerOffline(event.getMediaServerId()); diff --git a/src/main/java/com/genersoft/iot/vmp/service/IMediaServerService.java b/src/main/java/com/genersoft/iot/vmp/service/IMediaServerService.java index 4614ee7d..e216c654 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/IMediaServerService.java +++ b/src/main/java/com/genersoft/iot/vmp/service/IMediaServerService.java @@ -44,7 +44,7 @@ public interface IMediaServerService { void updateVmServer(List mediaServerItemList); - SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, boolean ssrcCheck); + SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, boolean ssrcCheck, boolean isPlayback); SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, String ssrc, boolean ssrcCheck, boolean isPlayback); diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java index 9ac4d316..5cfa8525 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java @@ -66,7 +66,7 @@ public class DeviceServiceImpl implements IDeviceService { @Override public void online(Device device) { - logger.info("[设备上线]，deviceId：" + device.getDeviceId()); + logger.info("[设备上线] deviceId：{}->{}:{}", device.getDeviceId(), device.getIp(), device.getPort()); Device deviceInRedis = redisCatchStorage.getDevice(device.getDeviceId()); Device deviceInDb = deviceMapper.getDeviceByDeviceId(device.getDeviceId()); diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java index a690c1bd..5468faef 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServerServiceImpl.java @@ -95,7 +95,7 @@ public class MediaServerServiceImpl implements IMediaServerService { */ @Override public void updateVmServer(List mediaServerItemList) { - logger.info("[缓存初始化] Media Server "); + logger.info("[zlm] 缓存初始化 "); for (MediaServerItem mediaServerItem : mediaServerItemList) { if (StringUtils.isEmpty(mediaServerItem.getId())) { continue; @@ -116,8 +116,8 @@ public class MediaServerServiceImpl implements IMediaServerService { } @Override - public SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, boolean ssrcCheck) { - return openRTPServer(mediaServerItem, streamId, null, ssrcCheck,false); + public SSRCInfo openRTPServer(MediaServerItem mediaServerItem, String streamId, boolean ssrcCheck, boolean isPlayback) { + return openRTPServer(mediaServerItem, streamId, null, ssrcCheck,isPlayback); } @Override @@ -352,7 +352,7 @@ public class MediaServerServiceImpl implements IMediaServerService { */ @Override public void zlmServerOnline(ZLMServerConfig zlmServerConfig) { - logger.info("[ ZLM：{} ]-[ {}:{} ]正在连接", + logger.info("[ZLM] 正在连接 : {} -> {}:{}", zlmServerConfig.getGeneralMediaServerId(), zlmServerConfig.getIp(), zlmServerConfig.getHttpPort()); MediaServerItem serverItem = mediaServerMapper.queryOne(zlmServerConfig.getGeneralMediaServerId()); @@ -405,7 +405,7 @@ public class MediaServerServiceImpl implements IMediaServerService { setZLMConfig(serverItem, "0".equals(zlmServerConfig.getHookEnable())); } publisher.zlmOnlineEventPublish(serverItem.getId()); - logger.info("[ ZLM：{} ]-[ {}:{} ]连接成功", + logger.info("[ZLM] 连接成功 {} - {}:{} ", zlmServerConfig.getGeneralMediaServerId(), zlmServerConfig.getIp(), zlmServerConfig.getHttpPort()); } @@ -483,7 +483,7 @@ public class MediaServerServiceImpl implements IMediaServerService { */ @Override public void setZLMConfig(MediaServerItem mediaServerItem, boolean restart) { - logger.info("[ ZLM：{} ]-[ {}:{} ]正在设置zlm", + logger.info("[ZLM] 正在设置 ：{} -> {}:{}", mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort()); String protocol = sslEnabled ? "https" : "http"; String hookPrex = String.format("%s://%s:%s/index/hook", protocol, mediaServerItem.getHookIp(), serverPort); @@ -527,17 +527,17 @@ public class MediaServerServiceImpl implements IMediaServerService { if (responseJSON != null && responseJSON.getInteger("code") == 0) { if (restart) { - logger.info("[ ZLM：{} ]-[ {}:{} ]设置zlm成功, 开始重启以保证配置生效", + logger.info("[ZLM] 设置成功,开始重启以保证配置生效 {} -> {}:{}", mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort()); zlmresTfulUtils.restartServer(mediaServerItem); }else { - logger.info("[ ZLM：{} ]-[ {}:{} ]设置zlm成功", + logger.info("[ZLM] 设置成功 {} -> {}:{}", mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort()); } }else { - logger.info("[ ZLM：{} ]-[ {}:{} ]设置zlm失败", + logger.info("[ZLM] 设置zlm失败 {} -> {}:{}", mediaServerItem.getId(), mediaServerItem.getIp(), mediaServerItem.getHttpPort()); } diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java index 4372f8eb..eb3831a9 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java @@ -193,7 +193,7 @@ public class PlayServiceImpl implements IPlayService { if (mediaServerItem.isRtpEnable()) { streamId = String.format("%s_%s", device.getDeviceId(), channelId); } - SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, device.isSsrcCheck()); + SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, device.isSsrcCheck(), false); play(mediaServerItem, ssrcInfo, device, channelId, (mediaServerItemInUse, response)->{ if (hookEvent != null) { hookEvent.response(mediaServerItem, response); @@ -237,7 +237,7 @@ public class PlayServiceImpl implements IPlayService { streamId = String.format("%s_%s", device.getDeviceId(), channelId); } if (ssrcInfo == null) { - ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, device.isSsrcCheck()); + ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, streamId, device.isSsrcCheck(), false); } // 超时处理 @@ -360,7 +360,7 @@ public class PlayServiceImpl implements IPlayService { return null; } MediaServerItem newMediaServerItem = getNewMediaServerItem(device); - SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, true); + SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, true, true); return playBack(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, inviteStreamCallback, callback); } @@ -447,7 +447,7 @@ public class PlayServiceImpl implements IPlayService { return null; } MediaServerItem newMediaServerItem = getNewMediaServerItem(device); - SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, true); + SSRCInfo ssrcInfo = mediaServerService.openRTPServer(newMediaServerItem, null, true, true); return download(newMediaServerItem, ssrcInfo, deviceId, channelId, startTime, endTime, downloadSpeed,infoCallBack, hookCallBack); } diff --git a/src/main/java/com/genersoft/iot/vmp/utils/DateUtil.java b/src/main/java/com/genersoft/iot/vmp/utils/DateUtil.java index 1be0f407..15313cdc 100644 --- a/src/main/java/com/genersoft/iot/vmp/utils/DateUtil.java +++ b/src/main/java/com/genersoft/iot/vmp/utils/DateUtil.java @@ -1,8 +1,15 @@ package com.genersoft.iot.vmp.utils; -import java.text.ParseException; + import java.text.SimpleDateFormat; -import java.util.Date; +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeParseException; +import java.time.temporal.TemporalAccessor; + import java.util.Locale; /** @@ -12,46 +19,40 @@ import java.util.Locale; public class DateUtil { private static final String yyyy_MM_dd_T_HH_mm_ss_SSSXXX = "yyyy-MM-dd'T'HH:mm:ss"; - private static final String yyyy_MM_dd_HH_mm_ss = "yyyy-MM-dd HH:mm:ss"; + public static final String yyyy_MM_dd_HH_mm_ss = "yyyy-MM-dd HH:mm:ss"; public static final SimpleDateFormat formatISO8601 = new SimpleDateFormat(yyyy_MM_dd_T_HH_mm_ss_SSSXXX, Locale.getDefault()); public static final SimpleDateFormat format = new SimpleDateFormat(yyyy_MM_dd_HH_mm_ss, Locale.getDefault()); - public static String yyyy_MM_dd_HH_mm_ssToISO8601(String formatTime) { + public static final DateTimeFormatter formatterISO8601 = DateTimeFormatter.ofPattern(yyyy_MM_dd_T_HH_mm_ss_SSSXXX, Locale.getDefault()).withZone(ZoneId.systemDefault()); + public static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern(yyyy_MM_dd_HH_mm_ss, Locale.getDefault()).withZone(ZoneId.systemDefault()); - try { - return formatISO8601.format(format.parse(formatTime)); - } catch (ParseException e) { - e.printStackTrace(); - } - return ""; + public static String yyyy_MM_dd_HH_mm_ssToISO8601(String formatTime) { + return formatterISO8601.format(formatter.parse(formatTime)); } public static String ISO8601Toyyyy_MM_dd_HH_mm_ss(String formatTime) { + return formatter.format(formatterISO8601.parse(formatTime)); - try { - return format.format(formatISO8601.parse(formatTime)); - } catch (ParseException e) { - e.printStackTrace(); - } - return ""; } public static long yyyy_MM_dd_HH_mm_ssToTimestamp(String formatTime) { - //设置要读取的时间字符串格式 - Date date; - try { - date = format.parse(formatTime); - Long timestamp=date.getTime()/1000; - //转换为Date类 - return timestamp; - } catch (ParseException e) { - e.printStackTrace(); - } - return 0; + TemporalAccessor temporalAccessor = formatter.parse(formatTime); + Instant instant = Instant.from(temporalAccessor); + return instant.getEpochSecond(); } public static String getNow() { - return format.format(System.currentTimeMillis()); + LocalDateTime nowDateTime = LocalDateTime.now(); + return formatter.format(nowDateTime); + } + + public static boolean verification(String timeStr, DateTimeFormatter dateTimeFormatter) { + try { + LocalDate.parse(timeStr, dateTimeFormatter); + return true; + }catch (DateTimeParseException exception) { + return false; + } } } diff --git a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/record/GBRecordController.java b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/record/GBRecordController.java index 6fdadf26..8cb923a7 100644 --- a/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/record/GBRecordController.java +++ b/src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/record/GBRecordController.java @@ -5,6 +5,8 @@ import com.genersoft.iot.vmp.common.StreamInfo; import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage; import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.IPlayService; +import com.genersoft.iot.vmp.utils.DateUtil; +import com.genersoft.iot.vmp.vmanager.bean.WVPResult; import io.swagger.annotations.Api; import io.swagger.annotations.ApiImplicitParam; import io.swagger.annotations.ApiImplicitParams; @@ -27,6 +29,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder; import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; +import java.time.LocalDate; import java.util.UUID; @Api(tags = "国标录像") @@ -60,15 +63,32 @@ public class GBRecordController { @ApiImplicitParam(name = "endTime", value = "结束时间", dataTypeClass = String.class), }) @GetMapping("/query/{deviceId}/{channelId}") - public DeferredResult> recordinfo(@PathVariable String deviceId,@PathVariable String channelId, String startTime, String endTime){ + public DeferredResult>> recordinfo(@PathVariable String deviceId, @PathVariable String channelId, String startTime, String endTime){ if (logger.isDebugEnabled()) { logger.debug(String.format("录像信息查询 API调用，deviceId：%s ，startTime：%s， endTime：%s",deviceId, startTime, endTime)); } + DeferredResult>> result = new DeferredResult<>(); + if (!DateUtil.verification(startTime, DateUtil.formatter)){ + WVPResult wvpResult = new WVPResult<>(); + wvpResult.setCode(-1); + wvpResult.setMsg("startTime error, format is " + DateUtil.yyyy_MM_dd_HH_mm_ss); + + ResponseEntity> resultResponseEntity = new ResponseEntity<>(wvpResult, HttpStatus.OK); + result.setResult(resultResponseEntity); + return result; + } + if (!DateUtil.verification(endTime, DateUtil.formatter)){ + WVPResult wvpResult = new WVPResult<>(); + wvpResult.setCode(-1); + wvpResult.setMsg("endTime error, format is " + DateUtil.yyyy_MM_dd_HH_mm_ss); + ResponseEntity> resultResponseEntity = new ResponseEntity<>(wvpResult, HttpStatus.OK); + result.setResult(resultResponseEntity); + return result; + } Device device = storager.queryVideoDevice(deviceId); // 指定超时时间 1分钟30秒 - DeferredResult> result = new DeferredResult<>(90*1000L); String uuid = UUID.randomUUID().toString(); int sn = (int)((Math.random()*9+1)*100000); String key = DeferredResultHolder.CALLBACK_CMD_RECORDINFO + deviceId + sn; @@ -76,7 +96,10 @@ public class GBRecordController { msg.setId(uuid); msg.setKey(key); cmder.recordInfoQuery(device, channelId, startTime, endTime, sn, null, null, null, (eventResult -> { - msg.setData("查询录像失败, status: " + eventResult.statusCode + ", message: " + eventResult.msg ); + WVPResult wvpResult = new WVPResult<>(); + wvpResult.setCode(-1); + wvpResult.setMsg("查询录像失败, status: " + eventResult.statusCode + ", message: " + eventResult.msg); + msg.setData(wvpResult); resultHolder.invokeResult(msg); })); @@ -84,6 +107,10 @@ public class GBRecordController { resultHolder.put(key, uuid, result); result.onTimeout(()->{ msg.setData("timeout"); + WVPResult wvpResult = new WVPResult<>(); + wvpResult.setCode(-1); + wvpResult.setMsg("timeout"); + msg.setData(wvpResult); resultHolder.invokeResult(msg); }); return result; diff --git a/web_src/src/components/channelList.vue b/web_src/src/components/channelList.vue index f7e4e406..9c7ab8ef 100644 --- a/web_src/src/components/channelList.vue +++ b/web_src/src/components/channelList.vue @@ -244,7 +244,7 @@ export default { }); }, queryRecords: function (itemData) { - var format = moment().format("YYYY-M-D"); + var format = moment().format("yyyy-MM-DD"); let deviceId = this.deviceId; let channelId = itemData.channelId; this.$refs.devicePlayer.openDialog("record", deviceId, channelId, {date: format}) diff --git a/web_src/src/components/dialog/devicePlayer.vue b/web_src/src/components/dialog/devicePlayer.vue index f0b45125..135bd85b 100644 --- a/web_src/src/components/dialog/devicePlayer.vue +++ b/web_src/src/components/dialog/devicePlayer.vue @@ -453,9 +453,19 @@ export default { method: 'get', url: '/api/gb_record/query/' + this.deviceId + '/' + this.channelId + '?startTime=' + startTime + '&endTime=' + endTime }).then(function (res) { - // 处理时间信息 - that.videoHistory.searchHistoryResult = res.data.recordList; - that.recordsLoading = false; + console.log(res) + if(res.data.code === 0) { + // 处理时间信息 + that.videoHistory.searchHistoryResult = res.data.data.recordList; + that.recordsLoading = false; + }else { + this.$message({ + showClose: true, + message: res.data.msg, + type: "error", + }); + } + }).catch(function (e) { console.log(e.message); // that.videoHistory.searchHistoryResult = falsificationData.recordData; From 91170a292e888cdcdcf34d00a4fabae6b512116e Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Fri, 13 May 2022 15:41:50 +0800 Subject: [PATCH 2/6] =?UTF-8?q?=E4=BC=98=E5=8C=96=E6=B5=B7=E5=BA=B7?= =?UTF-8?q?=E5=9B=BD=E6=A0=87=E5=BD=95=E5=83=8F=E6=92=AD=E6=94=BE=E7=9A=84?= =?UTF-8?q?=E6=8E=A7=E5=88=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/main/java/com/genersoft/iot/vmp/utils/DateUtil.java | 2 +- web_src/src/components/dialog/devicePlayer.vue | 6 +++++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/utils/DateUtil.java b/src/main/java/com/genersoft/iot/vmp/utils/DateUtil.java index 15313cdc..9d37dcd3 100644 --- a/src/main/java/com/genersoft/iot/vmp/utils/DateUtil.java +++ b/src/main/java/com/genersoft/iot/vmp/utils/DateUtil.java @@ -14,7 +14,7 @@ import java.util.Locale; /** * 全局时间工具类 - * @author swwheihei + * @author lin */ public class DateUtil { diff --git a/web_src/src/components/dialog/devicePlayer.vue b/web_src/src/components/dialog/devicePlayer.vue index 135bd85b..921e4136 100644 --- a/web_src/src/components/dialog/devicePlayer.vue +++ b/web_src/src/components/dialog/devicePlayer.vue @@ -681,7 +681,11 @@ export default { this.$axios({ method: 'get', url: `/api/playback/seek/${this.streamId }/` + Math.floor(this.seekTime * val / 100000) - }).then(function (res) {}); + }).then( (res)=> { + setTimeout(()=>{ + this.$refs.videoPlayer.play(this.videoUrl) + }, 600) + }); } } From 84a4bc354fa467711512ff61686c578b576dfe98 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Fri, 13 May 2022 15:45:02 +0800 Subject: [PATCH 3/6] =?UTF-8?q?=E5=8E=BB=E9=99=A4=E8=BF=87=E6=97=B6?= =?UTF-8?q?=E6=B3=A8=E9=87=8A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 1 - pom.xml | 7 ------- src/main/resources/all-application.yml | 2 +- src/main/resources/application-dev.yml | 1 - src/main/resources/application-docker.yml | 2 +- 5 files changed, 2 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index f9ef44a5..7ba443fe 100644 --- a/README.md +++ b/README.md @@ -106,7 +106,6 @@ https://gitee.com/pan648540858/wvp-GB28181-pro.git - [X] 添加RTMP视频 - [X] 云端录像（需要部署单独服务配合使用） - [X] 多流媒体节点，自动选择负载最低的节点使用。 -- [X] 支持使用mysql作为数据库，默认sqlite3,开箱即用。 - [X] WEB端支持播放H264与H265，音频支持G.711A/G.711U/AAC,覆盖国标常用编码格式。 [//]: # (# docker快速体验) diff --git a/pom.xml b/pom.xml index eade1f08..1546e39c 100644 --- a/pom.xml +++ b/pom.xml @@ -101,13 +101,6 @@ 8.0.22 - - - org.xerial - sqlite-jdbc - 3.32.3.2 - - com.github.pagehelper diff --git a/src/main/resources/all-application.yml b/src/main/resources/all-application.yml index 3f148d6f..1233a890 100644 --- a/src/main/resources/all-application.yml +++ b/src/main/resources/all-application.yml @@ -28,7 +28,7 @@ spring: poolMaxIdle: 500 # [可选] 最大的等待时间(秒) poolMaxWait: 5 - # [可选] jdbc数据库配置, 项目使用sqlite作为数据库，一般不需要配置 + # [必选] jdbc数据库配置 datasource: type: com.alibaba.druid.pool.DruidDataSource driver-class-name: com.mysql.cj.jdbc.Driver diff --git a/src/main/resources/application-dev.yml b/src/main/resources/application-dev.yml index b1950c5d..35314314 100644 --- a/src/main/resources/application-dev.yml +++ b/src/main/resources/application-dev.yml @@ -16,7 +16,6 @@ spring: password: face2020 # [可选] 超时时间 timeout: 10000 - # [可选] jdbc数据库配置, 项目使用sqlite作为数据库，一般不需要配置 # mysql数据源 datasource: type: com.alibaba.druid.pool.DruidDataSource diff --git a/src/main/resources/application-docker.yml b/src/main/resources/application-docker.yml index 6487f9a9..1653a586 100644 --- a/src/main/resources/application-docker.yml +++ b/src/main/resources/application-docker.yml @@ -16,7 +16,7 @@ spring: password: ${REDIS_PWD:root} # [可选] 超时时间 timeout: 10000 - # [可选] jdbc数据库配置, 项目使用sqlite作为数据库，一般不需要配置 + # [必选] jdbc数据库配置 datasource: # 使用mysql 打开23-28行注释， 删除29-36行 name: wvp From fe029b566d67c3a9edd82f6ac2d4bcb845acc69b Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Fri, 13 May 2022 21:05:33 +0800 Subject: [PATCH 4/6] =?UTF-8?q?=E4=BC=98=E5=8C=96=E6=B5=B7=E5=BA=B7?= =?UTF-8?q?=E5=BD=95=E5=83=8F=E4=B8=8B=E8=BD=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../iot/vmp/gb28181/bean/SubscribeHolder.java | 1 - .../session/VideoStreamSessionManager.java | 4 ++-- .../gb28181/transmit/cmd/impl/SIPCommander.java | 16 +++++++++++++--- .../cmd/MediaStatusNotifyMessageHandler.java | 5 ++--- web_src/src/components/dialog/recordDownload.vue | 1 + 5 files changed, 18 insertions(+), 9 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeHolder.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeHolder.java index 2736be2d..f191c005 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeHolder.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeHolder.java @@ -66,7 +66,6 @@ public class SubscribeHolder { dynamicTask.stop(taskOverdueKey); // 添加任务处理订阅过期 dynamicTask.startDelay(taskOverdueKey, () -> { - System.out.println("订阅过期"); removeMobilePositionSubscribe(subscribeInfo.getId()); }, subscribeInfo.getExpires() * 1000); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/session/VideoStreamSessionManager.java b/src/main/java/com/genersoft/iot/vmp/gb28181/session/VideoStreamSessionManager.java index 8d72a282..85bc39dc 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/session/VideoStreamSessionManager.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/session/VideoStreamSessionManager.java @@ -76,8 +76,8 @@ public class VideoStreamSessionManager { } - public ClientTransaction getTransactionByStream(String deviceId, String channelId, String stream){ - SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId, null, stream); + public ClientTransaction getTransaction(String deviceId, String channelId, String stream, String callId){ + SsrcTransaction ssrcTransaction = getSsrcTransaction(deviceId, channelId, callId, stream); if (ssrcTransaction == null) { return null; } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java index 8a6e9314..3bab0ebd 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java @@ -652,6 +652,14 @@ public class SIPCommander implements ISIPCommander { (MediaServerItem mediaServerItemInUse, JSONObject json)->{ hookEvent.call(new InviteStreamInfo(mediaServerItem, json, callIdHeader.getCallId(), "rtp", ssrcInfo.getStream())); subscribe.removeSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey); + subscribeKey.put("regist", false); + subscribeKey.put("schema", "rtmp"); + // 添加流注销的订阅，注销了后向设备发送bye + subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey, + (MediaServerItem mediaServerItemForEnd, JSONObject jsonForEnd)->{ + logger.info("[录像]下载结束， 发送BYE"); + streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream(), callIdHeader.getCallId()); + }); }); Request request = headerProvider.createPlaybackInviteRequest(device, channelId, content.toString(), null, "fromplybck" + tm, null, callIdHeader, ssrcInfo.getSsrc()); @@ -684,10 +692,10 @@ public class SIPCommander implements ISIPCommander { @Override public void streamByeCmd(String deviceId, String channelId, String stream, String callId, SipSubscribe.Event okEvent) { try { - SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(deviceId, channelId, null, stream); - ClientTransaction transaction = streamSession.getTransactionByStream(deviceId, channelId, stream); + SsrcTransaction ssrcTransaction = streamSession.getSsrcTransaction(deviceId, channelId, callId, stream); + ClientTransaction transaction = streamSession.getTransaction(deviceId, channelId, stream, callId); - if (transaction == null) { + if (transaction == null ) { logger.warn("[ {} -> {}]停止视频流的时候发现事务已丢失", deviceId, channelId); SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult<>(); if (okEvent != null) { @@ -1664,6 +1672,7 @@ public class SIPCommander implements ISIPCommander { sipSubscribe.addErrorSubscribe(callIdHeader.getCallId(), (eventResult -> { errorEvent.response(eventResult); sipSubscribe.removeErrorSubscribe(eventResult.callId); + sipSubscribe.removeOkSubscribe(eventResult.callId); })); } // 添加订阅 @@ -1671,6 +1680,7 @@ public class SIPCommander implements ISIPCommander { sipSubscribe.addOkSubscribe(callIdHeader.getCallId(), eventResult ->{ okEvent.response(eventResult); sipSubscribe.removeOkSubscribe(eventResult.callId); + sipSubscribe.removeErrorSubscribe(eventResult.callId); }); } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/MediaStatusNotifyMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/MediaStatusNotifyMessageHandler.java index e36a705a..4cf97685 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/MediaStatusNotifyMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/MediaStatusNotifyMessageHandler.java @@ -60,10 +60,9 @@ public class MediaStatusNotifyMessageHandler extends SIPRequestProcessorParent i CallIdHeader callIdHeader = (CallIdHeader)evt.getRequest().getHeader(CallIdHeader.NAME); String NotifyType =getText(rootElement, "NotifyType"); if (NotifyType.equals("121")){ - logger.info("媒体播放完毕，通知关流"); + logger.info("[录像流]推送完毕，收到关流通知"); String channelId =getText(rootElement, "DeviceID"); -// redisCatchStorage.stopPlayback(device.getDeviceId(), channelId, null, callIdHeader.getCallId()); -// redisCatchStorage.stopDownload(device.getDeviceId(), channelId, null, callIdHeader.getCallId()); + // 查询是设备 StreamInfo streamInfo = redisCatchStorage.queryDownload(device.getDeviceId(), channelId, null, callIdHeader.getCallId()); // 设置进度100% streamInfo.setProgress(1); diff --git a/web_src/src/components/dialog/recordDownload.vue b/web_src/src/components/dialog/recordDownload.vue index 6b7ca1f6..c50e4ce1 100644 --- a/web_src/src/components/dialog/recordDownload.vue +++ b/web_src/src/components/dialog/recordDownload.vue @@ -172,6 +172,7 @@ export default { isEnd: true, } }).then((res) => { + console.log(res) if (res.data.code == 0) { this.percentage = parseFloat(res.data.data.percentage)*100 if (res.data.data[0].percentage === '1') { From a14ce866324af892897dba9a4a5311b02df4c533 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Fri, 13 May 2022 21:10:02 +0800 Subject: [PATCH 5/6] =?UTF-8?q?=E4=BC=98=E5=8C=96=E5=BD=95=E5=83=8F?= =?UTF-8?q?=E7=BB=93=E6=9D=9F=E5=8F=91=E9=80=81BYE=E5=88=A4=E6=96=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java index 3bab0ebd..f2e524e4 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/impl/SIPCommander.java @@ -657,8 +657,11 @@ public class SIPCommander implements ISIPCommander { // 添加流注销的订阅，注销了后向设备发送bye subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey, (MediaServerItem mediaServerItemForEnd, JSONObject jsonForEnd)->{ - logger.info("[录像]下载结束， 发送BYE"); - streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream(), callIdHeader.getCallId()); + ClientTransaction transaction = streamSession.getTransaction(device.getDeviceId(), channelId, ssrcInfo.getStream(), callIdHeader.getCallId()); + if (transaction != null) { + logger.info("[录像]下载结束， 发送BYE"); + streamByeCmd(device.getDeviceId(), channelId, ssrcInfo.getStream(), callIdHeader.getCallId()); + } }); }); From 7e9e27f562597f0d0026653ec6bac03cd2f4353e Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: Fri, 13 May 2022 22:15:31 +0800 Subject: [PATCH 6/6] =?UTF-8?q?=E4=BD=BF=E7=94=A8=20java.time.Instant?= =?UTF-8?q?=E4=BB=A3=E6=9B=BF=20java.util.Date?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../genersoft/iot/vmp/conf/DynamicTask.java | 19 +++++++----- .../DigestServerAuthenticationHelper.java | 10 ++----- .../iot/vmp/gb28181/bean/CatalogData.java | 8 ++--- .../iot/vmp/gb28181/bean/RecordInfo.java | 8 ++--- .../iot/vmp/gb28181/bean/RecordItem.java | 25 ++++++++-------- .../iot/vmp/gb28181/event/SipSubscribe.java | 28 +++++++++--------- .../vmp/gb28181/session/CatalogDataCatch.java | 27 +++++++---------- .../vmp/gb28181/session/RecordDataCatch.java | 17 +++++------ .../request/impl/InviteRequestProcessor.java | 14 ++++----- .../cmd/KeepaliveNotifyMessageHandler.java | 2 -- .../vmp/service/impl/DeviceServiceImpl.java | 29 ++++--------------- 11 files changed, 78 insertions(+), 109 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java b/src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java index 2812f92c..3b021de4 100644 --- a/src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java +++ b/src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java @@ -1,7 +1,6 @@ package com.genersoft.iot.vmp.conf; import com.genersoft.iot.vmp.gb28181.task.ISubscribeTask; -import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd.CatalogResponseMessageHandler; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -9,25 +8,27 @@ import org.springframework.context.annotation.Bean; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.stereotype.Component; -import java.util.Date; +import java.time.Instant; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; /** * 动态定时任务 + * @author lin */ @Component public class DynamicTask { - private Logger logger = LoggerFactory.getLogger(DynamicTask.class); + private final Logger logger = LoggerFactory.getLogger(DynamicTask.class); @Autowired private ThreadPoolTaskScheduler threadPoolTaskScheduler; - private Map> futureMap = new ConcurrentHashMap<>(); - private Map runnableMap = new ConcurrentHashMap<>(); + private final Map> futureMap = new ConcurrentHashMap<>(); + private final Map runnableMap = new ConcurrentHashMap<>(); @Bean public ThreadPoolTaskScheduler threadPoolTaskScheduler() { @@ -47,7 +48,7 @@ public class DynamicTask { * @return */ public void startCron(String key, Runnable task, int cycleForCatalog) { - ScheduledFuture future = futureMap.get(key); + ScheduledFuture future = futureMap.get(key); if (future != null) { if (future.isCancelled()) { logger.debug("任务【{}】已存在但是关闭状态！！！", key); @@ -76,7 +77,9 @@ public class DynamicTask { */ public void startDelay(String key, Runnable task, int delay) { stop(key); - Date starTime = new Date(System.currentTimeMillis() + delay); + + // 获取执行的时刻 + Instant startInstant = Instant.now().plusMillis(TimeUnit.MILLISECONDS.toMillis(delay)); ScheduledFuture future = futureMap.get(key); if (future != null) { @@ -88,7 +91,7 @@ public class DynamicTask { } } // scheduleWithFixedDelay 必须等待上一个任务结束才开始计时period， cycleForCatalog表示执行的间隔 - future = threadPoolTaskScheduler.schedule(task, starTime); + future = threadPoolTaskScheduler.schedule(task, startInstant); if (future != null){ futureMap.put(key, future); runnableMap.put(key, task); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/auth/DigestServerAuthenticationHelper.java b/src/main/java/com/genersoft/iot/vmp/gb28181/auth/DigestServerAuthenticationHelper.java index f6284f5a..a0e16bff 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/auth/DigestServerAuthenticationHelper.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/auth/DigestServerAuthenticationHelper.java @@ -27,8 +27,7 @@ package com.genersoft.iot.vmp.gb28181.auth; import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; -import java.text.DecimalFormat; -import java.util.Date; +import java.time.Instant; import java.util.Random; import javax.sip.address.URI; @@ -90,17 +89,12 @@ public class DigestServerAuthenticationHelper { * @return a generated nonce. */ private String generateNonce() { - // Get the time of day and run MD5 over it. - Date date = new Date(); - long time = date.getTime(); + long time = Instant.now().toEpochMilli(); Random rand = new Random(); long pad = rand.nextLong(); - // String nonceString = (new Long(time)).toString() - // + (new Long(pad)).toString(); String nonceString = Long.valueOf(time).toString() + Long.valueOf(pad).toString(); byte mdbytes[] = messageDigest.digest(nonceString.getBytes()); - // Convert the mdbytes array into a hex string. return toHexString(mdbytes); } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/CatalogData.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/CatalogData.java index 338f8ad5..8a96d356 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/CatalogData.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/CatalogData.java @@ -1,13 +1,13 @@ package com.genersoft.iot.vmp.gb28181.bean; -import java.util.Date; +import java.time.Instant; import java.util.List; public class CatalogData { private int sn; // 命令序列号 private int total; private List channelList; - private Date lastTime; + private Instant lastTime; private Device device; private String errorMsg; @@ -41,11 +41,11 @@ public class CatalogData { this.channelList = channelList; } - public Date getLastTime() { + public Instant getLastTime() { return lastTime; } - public void setLastTime(Date lastTime) { + public void setLastTime(Instant lastTime) { this.lastTime = lastTime; } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/RecordInfo.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/RecordInfo.java index 55bf71f2..2121db7a 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/RecordInfo.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/RecordInfo.java @@ -1,6 +1,6 @@ package com.genersoft.iot.vmp.gb28181.bean; -import java.util.Date; +import java.time.Instant; import java.util.List; /** @@ -20,7 +20,7 @@ public class RecordInfo { private int sumNum; - private Date lastTime; + private Instant lastTime; private List recordList; @@ -72,11 +72,11 @@ public class RecordInfo { this.sn = sn; } - public Date getLastTime() { + public Instant getLastTime() { return lastTime; } - public void setLastTime(Date lastTime) { + public void setLastTime(Instant lastTime) { this.lastTime = lastTime; } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/RecordItem.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/RecordItem.java index 3349cdc5..a47147a1 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/RecordItem.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/RecordItem.java @@ -5,7 +5,8 @@ import com.genersoft.iot.vmp.utils.DateUtil; import org.jetbrains.annotations.NotNull; import java.text.ParseException; -import java.util.Date; +import java.time.Instant; +import java.time.temporal.TemporalAccessor; /** * @description:设备录像bean @@ -116,17 +117,17 @@ public class RecordItem implements Comparable{ @Override public int compareTo(@NotNull RecordItem recordItem) { - try { - Date startTime_now = DateUtil.format.parse(startTime); - Date startTime_param = DateUtil.format.parse(recordItem.getStartTime()); - if (startTime_param.compareTo(startTime_now) > 0) { - return -1; - }else { - return 1; - } - } catch (ParseException e) { - e.printStackTrace(); + TemporalAccessor startTimeNow = DateUtil.formatter.parse(startTime); + TemporalAccessor startTimeParam = DateUtil.formatter.parse(recordItem.getStartTime()); + Instant startTimeParamInstant = Instant.from(startTimeParam); + Instant startTimeNowInstant = Instant.from(startTimeNow); + if (startTimeNowInstant.equals(startTimeParamInstant)) { + return 0; + }else if (Instant.from(startTimeParam).isAfter(Instant.from(startTimeNow)) ) { + return -1; + }else { + return 1; } - return 0; + } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/event/SipSubscribe.java b/src/main/java/com/genersoft/iot/vmp/gb28181/event/SipSubscribe.java index bc775e46..3d817c34 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/event/SipSubscribe.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/event/SipSubscribe.java @@ -9,11 +9,14 @@ import org.springframework.stereotype.Component; import javax.sip.*; import javax.sip.header.CallIdHeader; import javax.sip.message.Response; -import java.util.Calendar; -import java.util.Date; +import java.time.Instant; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +/** + * @author lin + */ @Component public class SipSubscribe { @@ -23,28 +26,25 @@ public class SipSubscribe { private Map okSubscribes = new ConcurrentHashMap<>(); - private Map okTimeSubscribes = new ConcurrentHashMap<>(); - private Map errorTimeSubscribes = new ConcurrentHashMap<>(); + private Map okTimeSubscribes = new ConcurrentHashMap<>(); + private Map errorTimeSubscribes = new ConcurrentHashMap<>(); // @Scheduled(cron="*/5 * * * * ?") //每五秒执行一次 -// @Scheduled(fixedRate= 100 * 60 * 60 ) + // @Scheduled(fixedRate= 100 * 60 * 60 ) @Scheduled(cron="0 0/5 * * * ?") //每5分钟执行一次 public void execute(){ logger.info("[定时任务] 清理过期的SIP订阅信息"); - Calendar calendar = Calendar.getInstance(); - calendar.setTime(new Date()); - calendar.set(Calendar.MINUTE, calendar.get(Calendar.MINUTE) - 5); + + Instant instant = Instant.now().minusMillis(TimeUnit.MINUTES.toMillis(5)); for (String key : okTimeSubscribes.keySet()) { - if (okTimeSubscribes.get(key).before(calendar.getTime())){ -// logger.info("[定时任务] 清理过期的订阅信息： {}", key); + if (okTimeSubscribes.get(key).isBefore(instant)){ okSubscribes.remove(key); okTimeSubscribes.remove(key); } } for (String key : errorTimeSubscribes.keySet()) { - if (errorTimeSubscribes.get(key).before(calendar.getTime())){ -// logger.info("[定时任务] 清理过期的订阅信息： {}", key); + if (errorTimeSubscribes.get(key).isBefore(instant)){ errorSubscribes.remove(key); errorTimeSubscribes.remove(key); } @@ -117,12 +117,12 @@ public class SipSubscribe { public void addErrorSubscribe(String key, SipSubscribe.Event event) { errorSubscribes.put(key, event); - errorTimeSubscribes.put(key, new Date()); + errorTimeSubscribes.put(key, Instant.now()); } public void addOkSubscribe(String key, SipSubscribe.Event event) { okSubscribes.put(key, event); - okTimeSubscribes.put(key, new Date()); + okTimeSubscribes.put(key, Instant.now()); } public SipSubscribe.Event getErrorSubscribe(String key) { diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/session/CatalogDataCatch.java b/src/main/java/com/genersoft/iot/vmp/gb28181/session/CatalogDataCatch.java index 75b9f59a..7ed3c116 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/session/CatalogDataCatch.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/session/CatalogDataCatch.java @@ -4,25 +4,21 @@ import com.genersoft.iot.vmp.gb28181.bean.CatalogData; import com.genersoft.iot.vmp.gb28181.bean.Device; import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel; import com.genersoft.iot.vmp.gb28181.bean.SyncStatus; -import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder; -import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; -import com.genersoft.iot.vmp.vmanager.bean.WVPResult; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; +import java.time.Instant; import java.util.*; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; @Component public class CatalogDataCatch { public static Map data = new ConcurrentHashMap<>(); - @Autowired - private DeferredResultHolder deferredResultHolder; - @Autowired private IVideoManagerStorage storager; @@ -34,7 +30,7 @@ public class CatalogDataCatch { catalogData.setDevice(device); catalogData.setSn(sn); catalogData.setStatus(CatalogData.CatalogDataStatus.ready); - catalogData.setLastTime(new Date(System.currentTimeMillis())); + catalogData.setLastTime(Instant.now()); data.put(device.getDeviceId(), catalogData); } } @@ -48,7 +44,7 @@ public class CatalogDataCatch { catalogData.setDevice(device); catalogData.setChannelList(Collections.synchronizedList(new ArrayList<>())); catalogData.setStatus(CatalogData.CatalogDataStatus.runIng); - catalogData.setLastTime(new Date(System.currentTimeMillis())); + catalogData.setLastTime(Instant.now()); data.put(deviceId, catalogData); }else { // 同一个设备的通道同步请求只考虑一个，其他的直接忽略 @@ -59,7 +55,7 @@ public class CatalogDataCatch { catalogData.setDevice(device); catalogData.setStatus(CatalogData.CatalogDataStatus.runIng); catalogData.getChannelList().addAll(deviceChannelList); - catalogData.setLastTime(new Date(System.currentTimeMillis())); + catalogData.setLastTime(Instant.now()); } } @@ -102,16 +98,13 @@ public class CatalogDataCatch { @Scheduled(fixedRate = 5 * 1000) //每5秒执行一次, 发现数据5秒未更新则移除数据并认为数据接收超时 private void timerTask(){ Set keys = data.keySet(); - Calendar calendarBefore5S = Calendar.getInstance(); - calendarBefore5S.setTime(new Date()); - calendarBefore5S.set(Calendar.SECOND, calendarBefore5S.get(Calendar.SECOND) - 5); - Calendar calendarBefore30S = Calendar.getInstance(); - calendarBefore30S.setTime(new Date()); - calendarBefore30S.set(Calendar.SECOND, calendarBefore30S.get(Calendar.SECOND) - 30); + Instant instantBefore5S = Instant.now().minusMillis(TimeUnit.SECONDS.toMillis(5)); + Instant instantBefore30S = Instant.now().minusMillis(TimeUnit.SECONDS.toMillis(30)); + for (String deviceId : keys) { CatalogData catalogData = data.get(deviceId); - if ( catalogData.getLastTime().before(calendarBefore5S.getTime())) { // 超过五秒收不到消息任务超时， 只更新这一部分数据 + 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()) { @@ -124,7 +117,7 @@ public class CatalogDataCatch { } catalogData.setStatus(CatalogData.CatalogDataStatus.end); } - if (catalogData.getStatus().equals(CatalogData.CatalogDataStatus.end) && catalogData.getLastTime().before(calendarBefore30S.getTime())) { // 超过三十秒，如果标记为end则删除 + if (catalogData.getStatus().equals(CatalogData.CatalogDataStatus.end) && catalogData.getLastTime().isBefore(instantBefore30S)) { // 超过三十秒，如果标记为end则删除 data.remove(deviceId); } } diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/session/RecordDataCatch.java b/src/main/java/com/genersoft/iot/vmp/gb28181/session/RecordDataCatch.java index 0d166c12..dd5b8dfe 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/session/RecordDataCatch.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/session/RecordDataCatch.java @@ -3,16 +3,15 @@ package com.genersoft.iot.vmp.gb28181.session; import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder; import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage; -import com.genersoft.iot.vmp.storager.IVideoManagerStorage; -import com.genersoft.iot.vmp.utils.DateUtil; import com.genersoft.iot.vmp.vmanager.bean.WVPResult; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; -import java.text.SimpleDateFormat; +import java.time.Instant; import java.util.*; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; /** * @author lin @@ -35,7 +34,7 @@ public class RecordDataCatch { recordInfo.setSn(sn.trim()); recordInfo.setSumNum(sumNum); recordInfo.setRecordList(Collections.synchronizedList(new ArrayList<>())); - recordInfo.setLastTime(new Date(System.currentTimeMillis())); + recordInfo.setLastTime(Instant.now()); recordInfo.getRecordList().addAll(recordItems); data.put(key, recordInfo); }else { @@ -44,7 +43,7 @@ public class RecordDataCatch { return 0; } recordInfo.getRecordList().addAll(recordItems); - recordInfo.setLastTime(new Date(System.currentTimeMillis())); + recordInfo.setLastTime(Instant.now()); } return recordInfo.getRecordList().size(); } @@ -52,14 +51,12 @@ public class RecordDataCatch { @Scheduled(fixedRate = 5 * 1000) //每5秒执行一次, 发现数据5秒未更新则移除数据并认为数据接收超时 private void timerTask(){ Set keys = data.keySet(); - Calendar calendarBefore5S = Calendar.getInstance(); - calendarBefore5S.setTime(new Date()); - calendarBefore5S.set(Calendar.SECOND, calendarBefore5S.get(Calendar.SECOND) - 5); - + // 获取五秒前的时刻 + Instant instantBefore5S = Instant.now().minusMillis(TimeUnit.SECONDS.toMillis(5)); for (String key : keys) { RecordInfo recordInfo = data.get(key); // 超过五秒收不到消息任务超时， 只更新这一部分数据 - if ( recordInfo.getLastTime().before(calendarBefore5S.getTime())) { + if ( recordInfo.getLastTime().isBefore(instantBefore5S)) { // 处理录像数据， 返回给前端 String msgKey = DeferredResultHolder.CALLBACK_CMD_RECORDINFO + recordInfo.getDeviceId() + recordInfo.getSn(); diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java index 7c63279d..2723da47 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/InviteRequestProcessor.java @@ -40,7 +40,7 @@ import javax.sip.header.CallIdHeader; import javax.sip.message.Request; import javax.sip.message.Response; import java.text.ParseException; -import java.util.Date; +import java.time.Instant; import java.util.Vector; /** @@ -180,16 +180,16 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements Long startTime = null; Long stopTime = null; - Date start = null; - Date end = null; + Instant start = null; + Instant end = null; if (sdp.getTimeDescriptions(false) != null && sdp.getTimeDescriptions(false).size() > 0) { TimeDescriptionImpl timeDescription = (TimeDescriptionImpl)(sdp.getTimeDescriptions(false).get(0)); TimeField startTimeFiled = (TimeField)timeDescription.getTime(); startTime = startTimeFiled.getStartTime(); stopTime = startTimeFiled.getStopTime(); - start = new Date(startTime*1000); - end = new Date(stopTime*1000); + start = Instant.ofEpochMilli(startTime*1000); + end = Instant.ofEpochMilli(stopTime*1000); } // 获取支持的格式 Vector mediaDescriptions = sdp.getMediaDescriptions(true); @@ -335,8 +335,8 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements sendRtpItem.setStreamId(ssrcInfo.getStream()); // 写入redis， 超时时回复 redisCatchStorage.updateSendRTPSever(sendRtpItem); - playService.playBack(mediaServerItem, ssrcInfo, device.getDeviceId(), channelId, DateUtil.format.format(start), - DateUtil.format.format(end), null, result -> { + playService.playBack(mediaServerItem, ssrcInfo, device.getDeviceId(), channelId, DateUtil.formatter.format(start), + DateUtil.formatter.format(end), null, result -> { if (result.getCode() != 0){ logger.warn("录像回放失败"); if (result.getEvent() != null) { diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/KeepaliveNotifyMessageHandler.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/KeepaliveNotifyMessageHandler.java index a6280831..d5d4c1d0 100644 --- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/KeepaliveNotifyMessageHandler.java +++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/event/request/impl/message/notify/cmd/KeepaliveNotifyMessageHandler.java @@ -24,8 +24,6 @@ import javax.sip.SipException; import javax.sip.header.ViaHeader; import javax.sip.message.Response; import java.text.ParseException; -import java.util.Calendar; -import java.util.Date; @Component public class KeepaliveNotifyMessageHandler extends SIPRequestProcessorParent implements InitializingBean, IMessageHandler { diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java index 5cfa8525..0144e83b 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java @@ -11,24 +11,18 @@ import com.genersoft.iot.vmp.gb28181.task.impl.CatalogSubscribeTask; import com.genersoft.iot.vmp.gb28181.task.impl.MobilePositionSubscribeTask; import com.genersoft.iot.vmp.gb28181.bean.SyncStatus; import com.genersoft.iot.vmp.service.IMediaServerService; -import com.genersoft.iot.vmp.service.IMediaService; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.dao.DeviceMapper; import com.genersoft.iot.vmp.utils.DateUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Service; import org.springframework.util.StringUtils; -import javax.sip.DialogState; -import javax.sip.TimeoutEvent; -import java.text.ParseException; -import java.util.Calendar; -import java.util.Date; +import java.time.Instant; import java.util.List; +import java.util.concurrent.TimeUnit; /** * 设备业务（目录订阅） @@ -101,9 +95,7 @@ public class DeviceServiceImpl implements IDeviceService { // 刷新过期任务 String registerExpireTaskKey = registerExpireTaskKeyPrefix + device.getDeviceId(); dynamicTask.stop(registerExpireTaskKey); - dynamicTask.startDelay(registerExpireTaskKey, ()->{ - offline(device.getDeviceId()); - }, device.getExpires() * 1000); + dynamicTask.startDelay(registerExpireTaskKey, ()-> offline(device.getDeviceId()), device.getExpires() * 1000); } @Override @@ -217,18 +209,9 @@ public class DeviceServiceImpl implements IDeviceService { @Override public boolean expire(Device device) { - Date registerTimeDate; - try { - registerTimeDate = DateUtil.format.parse(device.getRegisterTime()); - } catch (ParseException e) { - logger.error("设备时间格式化失败：{}->{} ", device.getDeviceId(), device.getRegisterTime() ); - return false; - } - int expires = device.getExpires(); - Calendar calendarForExpire = Calendar.getInstance(); - calendarForExpire.setTime(registerTimeDate); - calendarForExpire.set(Calendar.SECOND, calendarForExpire.get(Calendar.SECOND) + expires); - return calendarForExpire.before(DateUtil.getNow()); + Instant registerTimeDate = Instant.from(DateUtil.formatter.parse(device.getRegisterTime())); + Instant expireInstant = registerTimeDate.plusMillis(TimeUnit.SECONDS.toMillis(device.getExpires())); + return expireInstant.isBefore(Instant.now()); } @Override