From 2ab8b942bdd20daa733ea40376034f3080c0ca9b Mon Sep 17 00:00:00 2001
From: panlinlin <648540858@qq.com>
Date: Thu, 31 Dec 2020 13:15:50 +0800
Subject: [PATCH] =?UTF-8?q?=E5=AD=98=E5=82=A8=E9=83=A8=E5=88=86=E4=BD=BF?=
=?UTF-8?q?=E7=94=A8sqlite=E4=BB=A3=E6=9B=BFredis?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
pom.xml | 7 +
.../genersoft/iot/vmp/gb28181/SipLayer.java | 13 +-
.../iot/vmp/gb28181/bean/Device.java | 34 +-
.../iot/vmp/gb28181/bean/DeviceChannel.java | 27 +-
.../iot/vmp/gb28181/event/SipSubscribe.java | 27 +-
.../gb28181/transmit/SIPProcessorFactory.java | 9 +-
.../callback/DeferredResultHolder.java | 2 +
.../gb28181/transmit/cmd/ISIPCommander.java | 3 +-
.../transmit/cmd/impl/SIPCommander.java | 68 ++-
.../request/impl/MessageRequestProcessor.java | 14 +-
.../impl/RegisterRequestProcessor.java | 8 +-
.../vmp/media/zlm/ZLMHTTPProxyController.java | 8 +-
.../vmp/media/zlm/ZLMHttpHookListener.java | 24 +-
.../iot/vmp/media/zlm/ZLMRunner.java | 6 +-
.../iot/vmp/storager/IRedisCatchStorage.java | 58 ++
.../vmp/storager/IVideoManagerStorager.java | 54 +-
.../storager/VideoManagerStoragerFactory.java | 36 --
.../iot/vmp/storager/VodeoMannagerTask.java | 4 +-
.../vmp/storager/dao/DeviceChannelMapper.java | 20 +
.../iot/vmp/storager/dao/DeviceMapper.java | 24 +
.../storager/impl/RedisCatchStorageImpl.java | 172 ++++++
.../impl/VideoManagerStoragerImpl.java | 401 +++++++++++++
.../jdbc/VideoManagerJdbcStoragerImpl.java | 217 -------
.../redis/VideoManagerRedisStoragerImpl.java | 561 ------------------
.../iot/vmp/vmanager/play/PlayController.java | 70 ++-
.../service/impl/PlayServiceImpl.java | 10 +-
.../iot/vmp/web/ApiDeviceController.java | 2 +-
.../iot/vmp/web/ApiStreamController.java | 8 +-
src/main/resources/wvp.sqlite | Bin 0 -> 32768 bytes
web_src/src/components/channelList.vue | 2 +-
30 files changed, 933 insertions(+), 956 deletions(-)
create mode 100644 src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java
delete mode 100644 src/main/java/com/genersoft/iot/vmp/storager/VideoManagerStoragerFactory.java
create mode 100644 src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java
create mode 100644 src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceMapper.java
create mode 100644 src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java
create mode 100644 src/main/java/com/genersoft/iot/vmp/storager/impl/VideoManagerStoragerImpl.java
delete mode 100644 src/main/java/com/genersoft/iot/vmp/storager/jdbc/VideoManagerJdbcStoragerImpl.java
delete mode 100644 src/main/java/com/genersoft/iot/vmp/storager/redis/VideoManagerRedisStoragerImpl.java
create mode 100644 src/main/resources/wvp.sqlite
diff --git a/pom.xml b/pom.xml
index 2ac45ec2..ae4d6666 100644
--- a/pom.xml
+++ b/pom.xml
@@ -99,6 +99,13 @@
8.0.22
+
+
+ org.xerial
+ sqlite-jdbc
+ 3.32.3.2
+
+
org.mybatis
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 92ba204a..e171297c 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/SipLayer.java
@@ -138,16 +138,25 @@ public class SipLayer implements SipListener {
// TODO Auto-generated catch block
e.printStackTrace();
}
+ if (evt.getResponse() != null && sipSubscribe.getOkSubscribesSize() > 0 ) {
+ CallIdHeader callIdHeader = (CallIdHeader)evt.getResponse().getHeader(CallIdHeader.NAME);
+ if (callIdHeader != null) {
+ SipSubscribe.Event subscribe = sipSubscribe.getOkSubscribe(callIdHeader.getCallId());
+ if (subscribe != null) {
+ subscribe.response(evt);
+ }
+ }
+ }
// } else if (status == Response.TRYING) {
// trying不会回复
} else if ((status >= 100) && (status < 200)) {
// 增加其它无需回复的响应,如101、180等
} else {
logger.warn("接收到失败的response响应!status:" + status + ",message:" + response.getReasonPhrase()/* .getContent().toString()*/);
- if (evt.getResponse() != null && sipSubscribe.getSize() > 0 ) {
+ if (evt.getResponse() != null && sipSubscribe.getErrorSubscribesSize() > 0 ) {
CallIdHeader callIdHeader = (CallIdHeader)evt.getResponse().getHeader(CallIdHeader.NAME);
if (callIdHeader != null) {
- SipSubscribe.Event subscribe = sipSubscribe.getSubscribe(callIdHeader.getCallId());
+ SipSubscribe.Event subscribe = sipSubscribe.getErrorSubscribe(callIdHeader.getCallId());
if (subscribe != null) {
subscribe.response(evt);
}
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/Device.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/Device.java
index de52ac69..12b8a007 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/Device.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/Device.java
@@ -1,11 +1,17 @@
package com.genersoft.iot.vmp.gb28181.bean;
+import java.util.Date;
import java.util.List;
import java.util.Map;
public class Device {
+ /**
+ * 数据库存储ID
+ */
+ private int id;
+
/**
* 设备Id
*/
@@ -55,14 +61,24 @@ public class Device {
*/
private int online;
- /**
- * 通道列表
- */
-// private Map channelMap;
+ /**
+ * 注册时间
+ */
+ private Long registerTimeMillis;
+
+ /**
+ * 通道个数
+ */
private int channelCount;
- private List channelList;
+ public int getId() {
+ return id;
+ }
+
+ public void setId(int id) {
+ this.id = id;
+ }
public String getDeviceId() {
return deviceId;
@@ -144,11 +160,11 @@ public class Device {
this.channelCount = channelCount;
}
- public List getChannelList() {
- return channelList;
+ public Long getRegisterTimeMillis() {
+ return registerTimeMillis;
}
- public void setChannelList(List channelList) {
- this.channelList = channelList;
+ public void setRegisterTimeMillis(Long registerTimeMillis) {
+ this.registerTimeMillis = registerTimeMillis;
}
}
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/DeviceChannel.java b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/DeviceChannel.java
index 19e9eda9..ca6ef60f 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/bean/DeviceChannel.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/bean/DeviceChannel.java
@@ -2,10 +2,17 @@ package com.genersoft.iot.vmp.gb28181.bean;
public class DeviceChannel {
+
+
/**
* 通道id
*/
private String channelId;
+
+ /**
+ * 设备id
+ */
+ private String deviceId;
/**
* 通道名
@@ -146,13 +153,15 @@ public class DeviceChannel {
/**
* 是否含有音频
*/
- private boolean hasAudio;
+ private boolean hasAudio;
- /**
- * 是否正在播放
- */
- private boolean play;
+ public String getDeviceId() {
+ return deviceId;
+ }
+ public void setDeviceId(String deviceId) {
+ this.deviceId = deviceId;
+ }
public void setPTZType(int PTZType) {
this.PTZType = PTZType;
@@ -387,14 +396,6 @@ public class DeviceChannel {
this.hasAudio = hasAudio;
}
- public boolean isPlay() {
- return play;
- }
-
- public void setPlay(boolean play) {
- this.play = play;
- }
-
public String getStreamId() {
return streamId;
}
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 1f78df44..176a435e 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
@@ -17,21 +17,34 @@ public class SipSubscribe {
private final static Logger logger = LoggerFactory.getLogger(SipSubscribe.class);
- private Map allSubscribes = new ConcurrentHashMap<>();
+ private Map errorSubscribes = new ConcurrentHashMap<>();
+
+ private Map okSubscribes = new ConcurrentHashMap<>();
public interface Event {
void response(ResponseEvent event);
}
- public void addSubscribe(String key, SipSubscribe.Event event) {
- allSubscribes.put(key, event);
+ public void addErrorSubscribe(String key, SipSubscribe.Event event) {
+ errorSubscribes.put(key, event);
}
- public SipSubscribe.Event getSubscribe(String key) {
- return allSubscribes.get(key);
+ public void addOkSubscribe(String key, SipSubscribe.Event event) {
+ okSubscribes.put(key, event);
}
- public int getSize(){
- return allSubscribes.size();
+ public SipSubscribe.Event getErrorSubscribe(String key) {
+ return errorSubscribes.get(key);
+ }
+
+ public SipSubscribe.Event getOkSubscribe(String key) {
+ return okSubscribes.get(key);
+ }
+
+ public int getErrorSubscribesSize(){
+ return errorSubscribes.size();
+ }
+ public int getOkSubscribesSize(){
+ return okSubscribes.size();
}
}
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorFactory.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorFactory.java
index d37259ac..b50cc957 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorFactory.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorFactory.java
@@ -4,13 +4,10 @@ import javax.sip.RequestEvent;
import javax.sip.ResponseEvent;
import javax.sip.SipProvider;
import javax.sip.header.CSeqHeader;
-import javax.sip.header.CallIdHeader;
-import javax.sip.header.Header;
import javax.sip.message.Request;
import javax.sip.message.Response;
-import com.alibaba.fastjson.JSON;
-import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -59,6 +56,9 @@ public class SIPProcessorFactory {
@Autowired
private IVideoManagerStorager storager;
+
+ @Autowired
+ private IRedisCatchStorage redisCatchStorage;
@Autowired
private EventPublisher publisher;
@@ -143,6 +143,7 @@ public class SIPProcessorFactory {
processor.setOffLineDetector(offLineDetector);
processor.setCmder(cmder);
processor.setStorager(storager);
+ processor.setRedisCatchStorage(redisCatchStorage);
return processor;
} else {
return new OtherRequestProcessor();
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/callback/DeferredResultHolder.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/callback/DeferredResultHolder.java
index 692e31e5..5fd8cbc9 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/callback/DeferredResultHolder.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/callback/DeferredResultHolder.java
@@ -25,6 +25,8 @@ public class DeferredResultHolder {
public static final String CALLBACK_CMD_PlAY = "CALLBACK_PLAY";
+ public static final String CALLBACK_CMD_STOP = "CALLBACK_STOP";
+
private Map map = new ConcurrentHashMap();
public void put(String key, DeferredResult result) {
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java
index 732b2cdc..67fd9967 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java
@@ -101,8 +101,9 @@ public interface ISIPCommander {
*
* @param ssrc ssrc
*/
+ void streamByeCmd(String ssrc, SipSubscribe.Event okEvent);
void streamByeCmd(String ssrc);
-
+
/**
* 语音广播
*
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 3f0adfd9..af9030b9 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
@@ -1,6 +1,7 @@
package com.genersoft.iot.vmp.gb28181.transmit.cmd.impl;
import java.text.ParseException;
+import java.util.UUID;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -12,11 +13,13 @@ import javax.sip.header.ViaHeader;
import javax.sip.message.Request;
import com.alibaba.fastjson.JSONObject;
+import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.MediaServerConfig;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -53,6 +56,9 @@ public class SIPCommander implements ISIPCommander {
@Autowired
private IVideoManagerStorager storager;
+
+ @Autowired
+ private IRedisCatchStorage redisCatchStorage;
@Autowired
@Qualifier(value="tcpSipProvider")
@@ -229,7 +235,7 @@ public class SIPCommander implements ISIPCommander {
Request request = headerProvider.createMessageRequest(device, ptzXml.toString(), "ViaPtzBranch", "FromPtzTag", "ToPtzTag");
- transmitRequest(device, request, null);
+ transmitRequest(device, request);
return true;
} catch (SipException | ParseException | InvalidArgumentException e) {
e.printStackTrace();
@@ -264,7 +270,7 @@ public class SIPCommander implements ISIPCommander {
ptzXml.append("\r\n");
Request request = headerProvider.createMessageRequest(device, ptzXml.toString(), "ViaPtzBranch", "FromPtzTag", "ToPtzTag");
- transmitRequest(device, request, null);
+ transmitRequest(device, request);
return true;
} catch (SipException | ParseException | InvalidArgumentException e) {
e.printStackTrace();
@@ -291,7 +297,7 @@ public class SIPCommander implements ISIPCommander {
streamId = String.format("%08x", Integer.parseInt(ssrc)).toUpperCase();
}
String streamMode = device.getStreamMode().toUpperCase();
- MediaServerConfig mediaInfo = storager.getMediaInfo();
+ MediaServerConfig mediaInfo = redisCatchStorage.getMediaInfo();
if (mediaInfo == null) {
logger.warn("点播时发现ZLM尚未连接...");
return;
@@ -344,6 +350,9 @@ public class SIPCommander implements ISIPCommander {
}
content.append("y="+ssrc+"\r\n");//ssrc
+// String fromTag = UUID.randomUUID().toString();
+// Request request = headerProvider.createInviteRequest(device, channelId, content.toString(), null, fromTag, null, ssrc);
+
Request request = headerProvider.createInviteRequest(device, channelId, content.toString(), null, "live", null, ssrc);
ClientTransaction transaction = transmitRequest(device, request, errorEvent);
@@ -372,7 +381,7 @@ public class SIPCommander implements ISIPCommander {
public void playbackStreamCmd(Device device, String channelId, String startTime, String endTime, ZLMHttpHookSubscribe.Event event
, SipSubscribe.Event errorEvent) {
try {
- MediaServerConfig mediaInfo = storager.getMediaInfo();
+ MediaServerConfig mediaInfo = redisCatchStorage.getMediaInfo();
String ssrc = streamSession.createPlayBackSsrc();
String streamId = String.format("%08x", Integer.parseInt(ssrc)).toUpperCase();
// 添加订阅
@@ -457,17 +466,28 @@ public class SIPCommander implements ISIPCommander {
e.printStackTrace();
}
}
-
+
+
+
/**
* 视频流停止
*
*/
@Override
- public void streamByeCmd(String streamId) {
+ public void streamByeCmd(String ssrc) {
+ streamByeCmd(ssrc, null);
+ }
+ @Override
+ public void streamByeCmd(String streamId, SipSubscribe.Event okEvent) {
try {
ClientTransaction transaction = streamSession.get(streamId);
+ // 服务重启后
if (transaction == null) {
+ StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(streamId);
+ if (streamInfo != null) {
+
+ }
return;
}
@@ -475,6 +495,9 @@ public class SIPCommander implements ISIPCommander {
if (dialog == null) {
return;
}
+
+
+
Request byeRequest = dialog.createRequest(Request.BYE);
SipURI byeURI = (SipURI) byeRequest.getRequestURI();
String vh = transaction.getRequest().getHeader(ViaHeader.NAME).toString();
@@ -491,7 +514,14 @@ public class SIPCommander implements ISIPCommander {
} else if("UDP".equals(protocol)) {
clientTransaction = udpSipProvider.getNewClientTransaction(byeRequest);
}
+
+ CallIdHeader callIdHeader = (CallIdHeader) byeRequest.getHeader(CallIdHeader.NAME);
+ if (okEvent != null) {
+ sipSubscribe.addOkSubscribe(callIdHeader.getCallId(), okEvent);
+ }
+
dialog.sendRequest(clientTransaction);
+
streamSession.remove(streamId);
zlmrtpServerFactory.closeRTPServer(streamId);
} catch (TransactionDoesNotExistException e) {
@@ -612,7 +642,7 @@ public class SIPCommander implements ISIPCommander {
Request request = headerProvider.createMessageRequest(device, catalogXml.toString(), "ViaDeviceInfoBranch", "FromDeviceInfoTag", "ToDeviceInfoTag");
- transmitRequest(device, request, null);
+ transmitRequest(device, request);
} catch (SipException | ParseException | InvalidArgumentException e) {
e.printStackTrace();
@@ -676,7 +706,7 @@ public class SIPCommander implements ISIPCommander {
Request request = headerProvider.createMessageRequest(device, recordInfoXml.toString(), "ViaRecordInfoBranch", "FromRecordInfoTag", null);
- transmitRequest(device, request, null);
+ transmitRequest(device, request);
} catch (SipException | ParseException | InvalidArgumentException e) {
e.printStackTrace();
return false;
@@ -727,8 +757,16 @@ public class SIPCommander implements ISIPCommander {
// TODO Auto-generated method stub
return false;
}
-
+
+ private ClientTransaction transmitRequest(Device device, Request request) throws SipException {
+ return transmitRequest(device, request, null, null);
+ }
+
private ClientTransaction transmitRequest(Device device, Request request, SipSubscribe.Event errorEvent) throws SipException {
+ return transmitRequest(device, request, errorEvent, null);
+ }
+
+ private ClientTransaction transmitRequest(Device device, Request request, SipSubscribe.Event errorEvent , SipSubscribe.Event okEvent) throws SipException {
ClientTransaction clientTransaction = null;
if("TCP".equals(device.getTransport())) {
clientTransaction = tcpSipProvider.getNewClientTransaction(request);
@@ -736,10 +774,14 @@ public class SIPCommander implements ISIPCommander {
clientTransaction = udpSipProvider.getNewClientTransaction(request);
}
- // 添加订阅
+ CallIdHeader callIdHeader = (CallIdHeader)request.getHeader(CallIdHeader.NAME);
+ // 添加错误订阅
if (errorEvent != null) {
- CallIdHeader callIdHeader = (CallIdHeader)request.getHeader(CallIdHeader.NAME);
- sipSubscribe.addSubscribe(callIdHeader.getCallId(), errorEvent);
+ sipSubscribe.addErrorSubscribe(callIdHeader.getCallId(), errorEvent);
+ }
+ // 添加订阅
+ if (okEvent != null) {
+ sipSubscribe.addOkSubscribe(callIdHeader.getCallId(), okEvent);
}
clientTransaction.sendRequest();
@@ -747,6 +789,8 @@ public class SIPCommander implements ISIPCommander {
}
+
+
@Override
public void closeRTPServer(Device device, String channelId) {
if (rtpEnable) {
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java
index 7b793c09..cab4a9bb 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/MessageRequestProcessor.java
@@ -10,6 +10,7 @@ import javax.sip.SipException;
import javax.sip.message.Request;
import javax.sip.message.Response;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import org.dom4j.Document;
import org.dom4j.DocumentException;
import org.dom4j.Element;
@@ -48,6 +49,8 @@ public class MessageRequestProcessor extends SIPRequestAbstractProcessor {
private IVideoManagerStorager storager;
+ private IRedisCatchStorage redisCatchStorage;
+
private EventPublisher publisher;
private RedisUtil redis;
@@ -451,9 +454,9 @@ public class MessageRequestProcessor extends SIPRequestAbstractProcessor {
String NotifyType =XmlUtil.getText(rootElement, "NotifyType");
if (NotifyType.equals("121")){
logger.info("媒体播放完毕,通知关流");
- StreamInfo streamInfo = storager.queryPlaybackByDevice(deviceId, "*");
+ StreamInfo streamInfo = redisCatchStorage.queryPlaybackByDevice(deviceId, "*");
if (streamInfo != null) {
- storager.stopPlayback(streamInfo);
+ redisCatchStorage.stopPlayback(streamInfo);
cmder.streamByeCmd(streamInfo.getStreamId());
}
}
@@ -507,4 +510,11 @@ public class MessageRequestProcessor extends SIPRequestAbstractProcessor {
this.offLineDetector = offLineDetector;
}
+ public IRedisCatchStorage getRedisCatchStorage() {
+ return redisCatchStorage;
+ }
+
+ public void setRedisCatchStorage(IRedisCatchStorage redisCatchStorage) {
+ this.redisCatchStorage = redisCatchStorage;
+ }
}
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/RegisterRequestProcessor.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/RegisterRequestProcessor.java
index be076bd8..bcd44825 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/RegisterRequestProcessor.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/request/impl/RegisterRequestProcessor.java
@@ -141,9 +141,15 @@ public class RegisterRequestProcessor extends SIPRequestAbstractProcessor {
// 下发catelog查询目录
if (registerFlag == 1 && device != null) {
logger.info("注册成功! deviceId:" + device.getDeviceId());
+ boolean exists = storager.exists(device.getDeviceId());
+ device.setRegisterTimeMillis(System.currentTimeMillis());
storager.updateDevice(device);
publisher.onlineEventPublish(device.getDeviceId(), VideoManagerConstants.EVENT_ONLINE_REGISTER);
- handler.onRegister(device);
+
+ // 只有第一次注册才更新通道
+ if (!exists) {
+ handler.onRegister(device);
+ }
} else if (registerFlag == 2) {
logger.info("注销成功! deviceId:" + device.getDeviceId());
publisher.outlineEventPublish(device.getDeviceId(), VideoManagerConstants.EVENT_OUTLINE_UNREGISTER);
diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHTTPProxyController.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHTTPProxyController.java
index f76cdd90..9daef230 100644
--- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHTTPProxyController.java
+++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHTTPProxyController.java
@@ -2,6 +2,7 @@ package com.genersoft.iot.vmp.media.zlm;
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.conf.MediaServerConfig;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -29,6 +30,9 @@ public class ZLMHTTPProxyController {
@Autowired
private IVideoManagerStorager storager;
+ @Autowired
+ private IRedisCatchStorage redisCatchStorage;
+
@Value("${media.port}")
private int mediaHttpPort;
@@ -36,10 +40,10 @@ public class ZLMHTTPProxyController {
@RequestMapping(value = "/**/**/**", produces = "application/json;charset=UTF-8")
public Object proxy(HttpServletRequest request, HttpServletResponse response){
- if (storager.getMediaInfo() == null) {
+ if (redisCatchStorage.getMediaInfo() == null) {
return "未接入流媒体";
}
- MediaServerConfig mediaInfo = storager.getMediaInfo();
+ MediaServerConfig mediaInfo = redisCatchStorage.getMediaInfo();
String requestURI = String.format("http://%s:%s%s?%s&%s",
mediaInfo.getLocalIP(),
mediaHttpPort,
diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
index 1116ae5a..cb8ad055 100644
--- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
+++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
@@ -11,6 +11,7 @@ import com.alibaba.fastjson.JSONArray;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.MediaServerConfig;
import com.genersoft.iot.vmp.gb28181.bean.Device;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
import com.genersoft.iot.vmp.utils.IpUtil;
import com.genersoft.iot.vmp.vmanager.service.IPlayService;
@@ -52,6 +53,9 @@ public class ZLMHttpHookListener {
@Autowired
private IVideoManagerStorager storager;
+ @Autowired
+ private IRedisCatchStorage redisCatchStorage;
+
@Autowired
private ZLMRESTfulUtils zlmresTfulUtils;
@@ -249,13 +253,13 @@ public class ZLMHttpHookListener {
String app = json.getString("app");
String streamId = json.getString("stream");
boolean regist = json.getBoolean("regist");
- StreamInfo streamInfo = storager.queryPlayByStreamId(streamId);
+ StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(streamId);
if ("rtp".equals(app) && !regist ) {
if (streamInfo!=null){
- storager.stopPlay(streamInfo);
+ redisCatchStorage.stopPlay(streamInfo);
}else{
- streamInfo = storager.queryPlaybackByStreamId(streamId);
- storager.stopPlayback(streamInfo);
+ streamInfo = redisCatchStorage.queryPlaybackByStreamId(streamId);
+ redisCatchStorage.stopPlayback(streamInfo);
}
}
@@ -281,12 +285,12 @@ public class ZLMHttpHookListener {
String streamId = json.getString("stream");
cmder.streamByeCmd(streamId);
- StreamInfo streamInfo = storager.queryPlayByStreamId(streamId);
+ StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(streamId);
if (streamInfo!=null){
- storager.stopPlay(streamInfo);
+ redisCatchStorage.stopPlay(streamInfo);
}else{
- streamInfo = storager.queryPlaybackByStreamId(streamId);
- storager.stopPlayback(streamInfo);
+ streamInfo = redisCatchStorage.queryPlaybackByStreamId(streamId);
+ redisCatchStorage.stopPlayback(streamInfo);
}
JSONObject ret = new JSONObject();
@@ -311,7 +315,7 @@ public class ZLMHttpHookListener {
if (autoApplyPlay) {
String app = json.getString("app");
String streamId = json.getString("stream");
- StreamInfo streamInfo = storager.queryPlayByStreamId(streamId);
+ StreamInfo streamInfo = redisCatchStorage.queryPlayByStreamId(streamId);
if ("rtp".equals(app) && streamId.indexOf("gb_play") > -1 && streamInfo == null) {
String[] s = streamId.split("_");
if (s.length == 4) {
@@ -355,7 +359,7 @@ public class ZLMHttpHookListener {
// MediaServerConfig mediaServerConfig = mediaServerConfigs.get(0);
MediaServerConfig mediaServerConfig = JSON.toJavaObject(json, MediaServerConfig.class);
mediaServerConfig.setLocalIP(mediaIp);
- storager.updateMediaInfo(mediaServerConfig);
+ redisCatchStorage.updateMediaInfo(mediaServerConfig);
// TODO Auto-generated method stub
JSONObject ret = new JSONObject();
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 3f88b2a3..282699f9 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
@@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.conf.MediaServerConfig;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
import okhttp3.*;
import org.slf4j.Logger;
@@ -30,6 +31,9 @@ public class ZLMRunner implements CommandLineRunner {
@Autowired
private IVideoManagerStorager storager;
+ @Autowired
+ private IRedisCatchStorage redisCatchStorage;
+
@Value("${media.ip}")
private String mediaIp;
@@ -69,7 +73,7 @@ public class ZLMRunner implements CommandLineRunner {
logger.info("zlm接入成功...");
if (autoConfig) saveZLMConfig();
mediaServerConfig = getMediaServerConfig();
- storager.updateMediaInfo(mediaServerConfig);
+ redisCatchStorage.updateMediaInfo(mediaServerConfig);
}
}
diff --git a/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java b/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java
new file mode 100644
index 00000000..8bc78b93
--- /dev/null
+++ b/src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java
@@ -0,0 +1,58 @@
+package com.genersoft.iot.vmp.storager;
+
+import com.genersoft.iot.vmp.common.StreamInfo;
+import com.genersoft.iot.vmp.conf.MediaServerConfig;
+
+import java.util.Map;
+
+public interface IRedisCatchStorage {
+
+ /**
+ * 开始播放时将流存入
+ *
+ * @param stream 流信息
+ * @return
+ */
+ boolean startPlay(StreamInfo stream);
+
+
+ /**
+ * 停止播放时删除
+ *
+ * @return
+ */
+ boolean stopPlay(StreamInfo streamInfo);
+
+ /**
+ * 查询播放列表
+ * @return
+ */
+ StreamInfo queryPlay(StreamInfo streamInfo);
+
+ StreamInfo queryPlayByStreamId(String steamId);
+
+ StreamInfo queryPlaybackByStreamId(String steamId);
+
+ StreamInfo queryPlayByDevice(String deviceId, String code);
+
+ /**
+ * 更新流媒体信息
+ * @param mediaServerConfig
+ * @return
+ */
+ boolean updateMediaInfo(MediaServerConfig mediaServerConfig);
+
+ /**
+ * 获取流媒体信息
+ * @return
+ */
+ MediaServerConfig getMediaInfo();
+
+ Map queryPlayByDeviceId(String deviceId);
+
+ boolean startPlayback(StreamInfo stream);
+
+ boolean stopPlayback(StreamInfo streamInfo);
+
+ StreamInfo queryPlaybackByDevice(String deviceId, String code);
+}
diff --git a/src/main/java/com/genersoft/iot/vmp/storager/IVideoManagerStorager.java b/src/main/java/com/genersoft/iot/vmp/storager/IVideoManagerStorager.java
index e85d1ff8..4174507b 100644
--- a/src/main/java/com/genersoft/iot/vmp/storager/IVideoManagerStorager.java
+++ b/src/main/java/com/genersoft/iot/vmp/storager/IVideoManagerStorager.java
@@ -17,19 +17,6 @@ import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
*/
public interface IVideoManagerStorager {
- /**
- * 更新流媒体信息
- * @param mediaServerConfig
- * @return
- */
- public boolean updateMediaInfo(MediaServerConfig mediaServerConfig);
-
- /**
- * 获取流媒体信息
- * @return
- */
- public MediaServerConfig getMediaInfo();
-
/**
* 根据设备ID判断设备是否存在
*
@@ -106,10 +93,9 @@ public interface IVideoManagerStorager {
/**
* 获取多个设备
*
- * @param deviceIds 设备ID数组
* @return List 设备对象数组
*/
- public List queryVideoDeviceList(String[] deviceIds);
+ public List queryVideoDeviceList();
/**
* 删除设备
@@ -135,27 +121,6 @@ public interface IVideoManagerStorager {
*/
public boolean outline(String deviceId);
- /**
- * 开始播放时将流存入
- *
- * @param stream 流信息
- * @return
- */
- public boolean startPlay(StreamInfo stream);
-
- /**
- * 停止播放时删除
- *
- * @return
- */
- public boolean stopPlay(StreamInfo streamInfo);
-
- /**
- * 查找视频流
- *
- * @return
- */
- public StreamInfo queryPlay(StreamInfo streamInfo);
/**
* 查询子设备
@@ -168,10 +133,6 @@ public interface IVideoManagerStorager {
*/
PageResult querySubChannels(String deviceId, String channelId, String query, Boolean hasSubChannel, String online, int page, int count);
- /**
- * 更新缓存
- */
- public void updateCatch();
/**
* 清空通道
@@ -179,17 +140,4 @@ public interface IVideoManagerStorager {
*/
void cleanChannelsForDevice(String deviceId);
- StreamInfo queryPlayByStreamId(String streamId);
-
- StreamInfo queryPlayByDevice(String deviceId, String code);
-
- Map queryPlayByDeviceId(String deviceId);
-
- boolean startPlayback(StreamInfo streamInfo);
-
- boolean stopPlayback(StreamInfo streamInfo);
-
- StreamInfo queryPlaybackByDevice(String deviceId, String channelId);
-
- StreamInfo queryPlaybackByStreamId(String streamId);
}
diff --git a/src/main/java/com/genersoft/iot/vmp/storager/VideoManagerStoragerFactory.java b/src/main/java/com/genersoft/iot/vmp/storager/VideoManagerStoragerFactory.java
deleted file mode 100644
index 70bdad76..00000000
--- a/src/main/java/com/genersoft/iot/vmp/storager/VideoManagerStoragerFactory.java
+++ /dev/null
@@ -1,36 +0,0 @@
-package com.genersoft.iot.vmp.storager;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.context.annotation.Bean;
-import org.springframework.stereotype.Component;
-
-import com.genersoft.iot.vmp.conf.VManagerConfig;
-
-/**
- * @Description:视频设备数据存储工厂,根据存储策略,返回对应的存储器
- * @author: swwheihei
- * @date: 2020年5月6日 下午2:15:16
- */
-@Component
-public class VideoManagerStoragerFactory {
-
- @Autowired
- private VManagerConfig vmConfig;
-
- @Autowired
- private IVideoManagerStorager jdbcStorager;
-
- @Autowired
- private IVideoManagerStorager redisStorager;
-
- @Bean("storager")
- public IVideoManagerStorager getStorager() {
- if ("redis".equals(vmConfig.getDatabase().toLowerCase())) {
- return redisStorager;
- } else if ("jdbc".equals(vmConfig.getDatabase().toLowerCase())) {
- return jdbcStorager;
- }
- return redisStorager;
- }
-
-}
diff --git a/src/main/java/com/genersoft/iot/vmp/storager/VodeoMannagerTask.java b/src/main/java/com/genersoft/iot/vmp/storager/VodeoMannagerTask.java
index c96e4bb0..c2074843 100644
--- a/src/main/java/com/genersoft/iot/vmp/storager/VodeoMannagerTask.java
+++ b/src/main/java/com/genersoft/iot/vmp/storager/VodeoMannagerTask.java
@@ -8,10 +8,10 @@ import org.springframework.stereotype.Component;
public class VodeoMannagerTask implements CommandLineRunner {
@Autowired
- private IVideoManagerStorager storager;
+ private IVideoManagerStorager redisStorager;
@Override
public void run(String... strings) throws Exception {
- storager.updateCatch();
+ redisStorager.updateCatch();
}
}
diff --git a/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java b/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java
new file mode 100644
index 00000000..bf67095f
--- /dev/null
+++ b/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java
@@ -0,0 +1,20 @@
+package com.genersoft.iot.vmp.storager.dao;
+
+import com.genersoft.iot.vmp.common.PageResult;
+import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
+import org.apache.ibatis.annotations.Mapper;
+
+import java.util.List;
+
+@Mapper
+public interface DeviceChannelMapper {
+ int update(DeviceChannel channel);
+
+ List queryChannelsByDeviceId(String deviceId);
+
+ List queryChannelsByDeviceId(String deviceId, String parentChannelId);
+
+ DeviceChannel queryChannel(String deviceId, String channelId);
+
+ int cleanChannelsByDeviceId(String deviceId);
+}
diff --git a/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceMapper.java b/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceMapper.java
new file mode 100644
index 00000000..da455fbe
--- /dev/null
+++ b/src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceMapper.java
@@ -0,0 +1,24 @@
+package com.genersoft.iot.vmp.storager.dao;
+
+import com.genersoft.iot.vmp.gb28181.bean.Device;
+import org.apache.ibatis.annotations.Insert;
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Select;
+
+import java.util.List;
+
+@Mapper
+public interface DeviceMapper {
+
+ @Select("SELECT * FROM device WHERE deviceId = #{deviceId}")
+ Device getDeviceByDeviceId(String deviceId);
+
+ @Insert("SELECT * FROM device WHERE deviceId = #{deviceId}")
+ int add(Device device);
+
+ int update(Device device);
+
+ List getDevices();
+
+ int del(String deviceId);
+}
diff --git a/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java b/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java
new file mode 100644
index 00000000..8eaaf68a
--- /dev/null
+++ b/src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java
@@ -0,0 +1,172 @@
+package com.genersoft.iot.vmp.storager.impl;
+
+import com.genersoft.iot.vmp.common.StreamInfo;
+import com.genersoft.iot.vmp.common.VideoManagerConstants;
+import com.genersoft.iot.vmp.conf.MediaServerConfig;
+import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
+import com.genersoft.iot.vmp.storager.dao.DeviceChannelMapper;
+import com.genersoft.iot.vmp.storager.dao.DeviceMapper;
+import com.genersoft.iot.vmp.utils.redis.RedisUtil;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+
+@Component
+public class RedisCatchStorageImpl implements IRedisCatchStorage {
+
+ @Autowired
+ private RedisUtil redis;
+
+ @Autowired
+ private DeviceMapper deviceMapper;
+
+ @Autowired
+ private DeviceChannelMapper deviceChannelMapper;
+
+
+ /**
+ * 开始播放时将流存入redis
+ *
+ * @return
+ */
+ @Override
+ public boolean startPlay(StreamInfo stream) {
+ return redis.set(String.format("%S_%s_%s_%s", VideoManagerConstants.PLAYER_PREFIX, stream.getStreamId(),stream.getDeviceID(), stream.getCahnnelId()),
+ stream);
+ }
+
+ /**
+ * 停止播放时从redis删除
+ *
+ * @return
+ */
+ @Override
+ public boolean stopPlay(StreamInfo streamInfo) {
+ if (streamInfo == null) return false;
+ DeviceChannel deviceChannel = deviceChannelMapper.queryChannel(streamInfo.getDeviceID(), streamInfo.getCahnnelId());
+ if (deviceChannel != null) {
+ deviceChannel.setStreamId(null);
+ deviceChannel.setPlay(false);
+ deviceChannel.setDeviceId(streamInfo.getDeviceID());
+ deviceChannelMapper.update(deviceChannel);
+ }
+ return redis.del(String.format("%S_%s_%s_%s", VideoManagerConstants.PLAYER_PREFIX,
+ streamInfo.getStreamId(),
+ streamInfo.getDeviceID(),
+ streamInfo.getCahnnelId()));
+ }
+
+ /**
+ * 查询播放列表
+ * @return
+ */
+ @Override
+ public StreamInfo queryPlay(StreamInfo streamInfo) {
+ return (StreamInfo)redis.get(String.format("%S_%s_%s_%s",
+ VideoManagerConstants.PLAYER_PREFIX,
+ streamInfo.getStreamId(),
+ streamInfo.getDeviceID(),
+ streamInfo.getCahnnelId()));
+ }
+ @Override
+ public StreamInfo queryPlayByStreamId(String steamId) {
+ List