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

Merge branch '648540858:master' into develop-add-api-key

pull/1389/head
ancienter 2024-04-09 09:52:23 +08:00 committed by GitHub
parent 24b4f02a70 a890b62fa1
commit 25ff2fc4ef
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
57 changed files with 741 additions and 414 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
README.md
View File

@ -112,6 +112,7 @@ https://gitee.com/pan648540858/wvp-GB28181-pro.git
# 非开源的内容
- [X] ONVIF设备的接入，支持点播，云台控制，国标级联点播，自动点播。在[知识星球](https://t.zsxq.com/10WAnH2MP)放了试用安装包以及使用教程，没有使用时间限制，需要源码可以星球私信我或者邮箱联系。
- [X] 支持国标28181-2022协议，支持巡航轨迹查询，PTZ精准控制，存储卡格式化，设备软件升级，OSD配置，h265+aac，支持辅码流，录像倒放等。具体的功能列表可在[知识星球](https://t.zsxq.com/18GXkpkqs)查看，需要源码和测试可以在星球私信联系或者发邮件给我
# 授权协议
@ -119,7 +120,7 @@ https://gitee.com/pan648540858/wvp-GB28181-pro.git
# 技术支持
[知识星球](https://t.zsxq.com/0d8VAD3Dm)专栏列表：
[知识星球](https://t.zsxq.com/0d8VAD3Dm)专栏列表：，
- [使用入门系列一：WVP-PRO能做什么](https://t.zsxq.com/0dLguVoSp)
有偿技术支持，请发送邮件到648540858@qq.com
@ -135,12 +136,6 @@ https://gitee.com/pan648540858/wvp-GB28181-pro.git
[ydpd](https://github.com/ydpd) [szy833](https://github.com/szy833) [ydwxb](https://github.com/ydwxb) [Albertzhu666](https://github.com/Albertzhu666)
[mk1990](https://github.com/mk1990) [SaltFish001](https://github.com/SaltFish001)
同时感谢JetBrains对开源项目的支持，本项目使用IntelliJ IDEA开发与调试：
ffmpeg -re -i 123.mp3 -acodec pcm_alaw -ar 8000 -ac 1 -f rtsp rtsp://192.168.1.3:30554/broadcast/34020000001320000101_34020000001310000001
ffmpeg -re -i 123.mp3 -acodec pcm_alaw -ar 8000 -ac 1 -f rtsp rtsp://192.168.1.3:30554/talk/34020000001320000011_34020000001370000001
ffmpeg -re -i 123.mp3 -acodec pcm_alaw -ar 8000 -ac 1 -f rtsp rtsp://192.168.1.3:30554/talk/34020000001320000101_34020000001310000001
![JetBrains](https://resources.jetbrains.com/storage/products/company/brand/logos/IntelliJ_IDEA_icon.svg?_ga=2.143694769.529214288.1712023294-439039083.1711422571&_gl=1*102dv9n*_ga*NDM5MDM5MDgzLjE3MTE0MjI1NzE.*_ga_9J976DJZ68*MTcxMjEyNjg4NC45LjEuMTcxMjEyNzc2My4zMy4wLjA.)

4
src/main/java/com/genersoft/iot/vmp/conf/DynamicTask.java
View File

@ -9,6 +9,7 @@ import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.time.Instant;
import java.util.Date;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@ -59,7 +60,8 @@ public class DynamicTask {
}
}
// scheduleWithFixedDelay 必须等待上一个任务结束才开始计时period， cycleForCatalog表示执行的间隔
future = threadPoolTaskScheduler.scheduleAtFixedRate(task, cycleForCatalog);
future = threadPoolTaskScheduler.scheduleAtFixedRate(task, new Date(System.currentTimeMillis() + cycleForCatalog), cycleForCatalog);
if (future != null){
futureMap.put(key, future);
runnableMap.put(key, task);

0
src/main/java/com/genersoft/iot/vmp/conf/redis/RedisConfig.java
View File

32
src/main/java/com/genersoft/iot/vmp/gb28181/bean/DeviceChannel.java
View File

@ -142,13 +142,13 @@ public class DeviceChannel {
* 云台类型
*/
@Schema(description = "云台类型")
private int PTZType;
private int ptzType;
/**
* 云台类型描述字符串
*/
@Schema(description = "云台类型描述字符串")
private String PTZTypeText;
private String ptzTypeText;
/**
* 创建时间
@ -266,23 +266,23 @@ public class DeviceChannel {
this.deviceId = deviceId;
}
public void setPTZType(int PTZType) {
this.PTZType = PTZType;
switch (PTZType) {
public void setPtzType(int ptzType) {
this.ptzType = ptzType;
switch (ptzType) {
case 0:
this.PTZTypeText = "未知";
this.ptzTypeText = "未知";
break;
case 1:
this.PTZTypeText = "球机";
this.ptzTypeText = "球机";
break;
case 2:
this.PTZTypeText = "半球";
this.ptzTypeText = "半球";
break;
case 3:
this.PTZTypeText = "固定枪机";
this.ptzTypeText = "固定枪机";
break;
case 4:
this.PTZTypeText = "遥控枪机";
this.ptzTypeText = "遥控枪机";
break;
}
}
@ -447,16 +447,16 @@ public class DeviceChannel {
this.password = password;
}
public int getPTZType() {
return PTZType;
public int getPtzType() {
return ptzType;
}
public String getPTZTypeText() {
return PTZTypeText;
public String getPtzTypeText() {
return ptzTypeText;
}
public void setPTZTypeText(String PTZTypeText) {
this.PTZTypeText = PTZTypeText;
public void setPtzTypeText(String ptzTypeText) {
this.ptzTypeText = ptzTypeText;
}
public boolean isStatus() {

11
src/main/java/com/genersoft/iot/vmp/gb28181/bean/ParentPlatform.java
View File

@ -189,6 +189,9 @@ public class ParentPlatform {
@Schema(description = "是否作为消息通道")
private boolean autoPushChannel;
@Schema(description = "点播回复200OK使用次IP")
private String sendStreamIp;
public Integer getId() {
return id;
}
@ -436,4 +439,12 @@ public class ParentPlatform {
public void setAutoPushChannel(boolean autoPushChannel) {
this.autoPushChannel = autoPushChannel;
}
public String getSendStreamIp() {
return sendStreamIp;
}
public void setSendStreamIp(String sendStreamIp) {
this.sendStreamIp = sendStreamIp;
}
}

29
src/main/java/com/genersoft/iot/vmp/gb28181/bean/SendRtpItem.java
View File

@ -305,4 +305,33 @@ public class SendRtpItem {
public void setReceiveStream(String receiveStream) {
this.receiveStream = receiveStream;
}
@Override
public String toString() {
return "SendRtpItem{" +
"ip='" + ip + '\'' +
", port=" + port +
", ssrc='" + ssrc + '\'' +
", platformId='" + platformId + '\'' +
", deviceId='" + deviceId + '\'' +
", app='" + app + '\'' +
", channelId='" + channelId + '\'' +
", status=" + status +
", stream='" + stream + '\'' +
", tcp=" + tcp +
", tcpActive=" + tcpActive +
", localPort=" + localPort +
", mediaServerId='" + mediaServerId + '\'' +
", serverId='" + serverId + '\'' +
", CallId='" + CallId + '\'' +
", fromTag='" + fromTag + '\'' +
", toTag='" + toTag + '\'' +
", pt=" + pt +
", usePs=" + usePs +
", onlyAudio=" + onlyAudio +
", rtcp=" + rtcp +
", playType=" + playType +
", receiveStream='" + receiveStream + '\'' +
'}';
}
}

10
src/main/java/com/genersoft/iot/vmp/gb28181/bean/SubscribeHolder.java
View File

@ -103,6 +103,16 @@ public class SubscribeHolder {
return platforms;
}
public List<String> getAllMobilePositionSubscribePlatform() {
List<String> platforms = new ArrayList<>();
if(!mobilePositionMap.isEmpty()) {
for (String key : mobilePositionMap.keySet()) {
platforms.add(mobilePositionMap.get(key).getId());
}
}
return platforms;
}
public void removeAllSubscribe(String platformId) {
removeMobilePositionSubscribe(platformId);
removeCatalogSubscribe(platformId);

8
src/main/java/com/genersoft/iot/vmp/gb28181/event/EventPublisher.java
View File

@ -4,6 +4,7 @@ import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.device.RequestTimeoutEvent;
import com.genersoft.iot.vmp.gb28181.event.record.RecordEndEvent;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.gb28181.event.subscribe.mobilePosition.MobilePositionEvent;
import com.genersoft.iot.vmp.media.zlm.event.ZLMOfflineEvent;
import com.genersoft.iot.vmp.media.zlm.event.ZLMOnlineEvent;
import org.springframework.beans.factory.annotation.Autowired;
@ -94,6 +95,13 @@ public class EventPublisher {
}
public void mobilePositionEventPublish(MobilePosition mobilePosition) {
MobilePositionEvent event = new MobilePositionEvent(this);
event.setMobilePosition(mobilePosition);
applicationEventPublisher.publishEvent(event);
}
public void catalogEventPublishForStream(String platformId, List<GbStream> gbStreams, String type) {
CatalogEvent outEvent = new CatalogEvent(this);
outEvent.setGbStreams(gbStreams);

20
src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/mobilePosition/MobilePositionEvent.java Executable file
View File

@ -0,0 +1,20 @@
package com.genersoft.iot.vmp.gb28181.event.subscribe.mobilePosition;
import com.genersoft.iot.vmp.gb28181.bean.MobilePosition;
import org.springframework.context.ApplicationEvent;
public class MobilePositionEvent extends ApplicationEvent {
public MobilePositionEvent(Object source) {
super(source);
}
private MobilePosition mobilePosition;
public MobilePosition getMobilePosition() {
return mobilePosition;
}
public void setMobilePosition(MobilePosition mobilePosition) {
this.mobilePosition = mobilePosition;
}
}

61
src/main/java/com/genersoft/iot/vmp/gb28181/event/subscribe/mobilePosition/MobilePositionEventLister.java Executable file
View File

@ -0,0 +1,61 @@
package com.genersoft.iot.vmp.gb28181.event.subscribe.mobilePosition;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.SubscribeHolder;
import com.genersoft.iot.vmp.gb28181.bean.SubscribeInfo;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommanderFroPlatform;
import com.genersoft.iot.vmp.service.bean.GPSMsgInfo;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;
import javax.sip.InvalidArgumentException;
import javax.sip.SipException;
import java.text.ParseException;
import java.util.List;
/**
* 移动位置通知消息转发
*/
@Component
public class MobilePositionEventLister implements ApplicationListener<MobilePositionEvent> {
private final static Logger logger = LoggerFactory.getLogger(MobilePositionEventLister.class);
@Autowired
private IVideoManagerStorage storager;
@Autowired
private SIPCommanderFroPlatform sipCommanderFroPlatform;
@Autowired
private SubscribeHolder subscribeHolder;
@Override
public void onApplicationEvent(MobilePositionEvent event) {
// 获取所用订阅
List<String> platforms = subscribeHolder.getAllMobilePositionSubscribePlatform();
if (platforms.isEmpty()) {
return;
}
List<ParentPlatform> parentPlatformsForGB = storager.queryPlatFormListForGBWithGBId(event.getMobilePosition().getChannelId(), platforms);
for (ParentPlatform platform : parentPlatformsForGB) {
logger.info("[向上级发送MobilePosition] 通道：{}，平台：{}， 位置： {}:{}", event.getMobilePosition().getChannelId(),
platform.getServerGBId(), event.getMobilePosition().getLongitude(), event.getMobilePosition().getLatitude());
SubscribeInfo subscribe = subscribeHolder.getMobilePositionSubscribe(platform.getServerGBId());
try {
sipCommanderFroPlatform.sendNotifyMobilePosition(platform, GPSMsgInfo.getInstance(event.getMobilePosition()),
subscribe);
} catch (InvalidArgumentException | ParseException | NoSuchFieldException | SipException |
IllegalAccessException e) {
logger.error("[命令发送失败] 国标级联 Catalog通知: {}", e.getMessage());
}
}
}
}

18
src/main/java/com/genersoft/iot/vmp/gb28181/session/VideoStreamSessionManager.java
View File

@ -50,7 +50,7 @@ public class VideoStreamSessionManager {
ssrcTransaction.setType(type);
redisTemplate.opsForValue().set(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetting.getServerId()
+ "_" + deviceId + "_" + channelId + "_" + callId + "_" + stream, ssrcTransaction);
+ ":" + deviceId + ":" + channelId + ":" + callId + ":" + stream, ssrcTransaction);
}
public SsrcTransaction getSsrcTransaction(String deviceId, String channelId, String callId, String stream){
@ -67,7 +67,7 @@ public class VideoStreamSessionManager {
if (ObjectUtils.isEmpty(stream)) {
stream ="*";
}
String key = VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetting.getServerId() + "_" + deviceId + "_" + channelId + "_" + callId+ "_" + stream;
String key = VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetting.getServerId() + ":" + deviceId + ":" + channelId + ":" + callId+ ":" + stream;
List<Object> scanResult = RedisUtil.scan(redisTemplate, key);
if (scanResult.size() == 0) {
return null;
@ -80,12 +80,12 @@ public class VideoStreamSessionManager {
if (ObjectUtils.isEmpty(callId)) {
return null;
}
String key = VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetting.getServerId() + "_*_*_" + callId+ "_*";
String key = VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetting.getServerId() + ":*:*:" + callId+ ":*";
List<Object> scanResult = RedisUtil.scan(redisTemplate, key);
if (!scanResult.isEmpty()) {
return (SsrcTransaction)redisTemplate.opsForValue().get(scanResult.get(0));
}else {
key = VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetting.getServerId() + "_*_*_play_*";
key = VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetting.getServerId() + ":*:*:play:*";
scanResult = RedisUtil.scan(redisTemplate, key);
if (scanResult.isEmpty()) {
return null;
@ -115,7 +115,7 @@ public class VideoStreamSessionManager {
if (ObjectUtils.isEmpty(stream)) {
stream ="*";
}
String key = VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetting.getServerId() + "_" + deviceId + "_" + channelId + "_" + callId+ "_" + stream;
String key = VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetting.getServerId() + ":" + deviceId + ":" + channelId + ":" + callId+ ":" + stream;
List<Object> scanResult = RedisUtil.scan(redisTemplate, key);
if (scanResult.size() == 0) {
return null;
@ -149,8 +149,8 @@ public class VideoStreamSessionManager {
return;
}
for (SsrcTransaction ssrcTransaction : ssrcTransactionList) {
redisTemplate.delete(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetting.getServerId() + "_"
+ deviceId + "_" + channelId + "_" + ssrcTransaction.getCallId() + "_" + ssrcTransaction.getStream());
redisTemplate.delete(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetting.getServerId() + ":"
+ deviceId + ":" + channelId + ":" + ssrcTransaction.getCallId() + ":" + ssrcTransaction.getStream());
}
}
@ -159,8 +159,8 @@ public class VideoStreamSessionManager {
if (ssrcTransaction == null ) {
return;
}
redisTemplate.delete(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetting.getServerId() + "_"
+ deviceId + "_" + channelId + "_" + ssrcTransaction.getCallId() + "_" + ssrcTransaction.getStream());
redisTemplate.delete(VideoManagerConstants.MEDIA_TRANSACTION_USED_PREFIX + userSetting.getServerId() + ":"
+ deviceId + ":" + channelId + ":" + ssrcTransaction.getCallId() + ":" + ssrcTransaction.getStream());
}

7
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/cmd/ISIPCommander.java
View File

@ -220,13 +220,8 @@ public interface ISIPCommander {
/**
* 看守位控制命令
*
* @param device 视频设备
* @param channelId 通道id，非通道则是设备本身
* @param enabled 看守位使能：1 = 开启，0 = 关闭
* @param resetTime 自动归位时间间隔，开启看守位时使用，单位:秒(s)
* @param presetIndex 调用预置位编号，开启看守位时使用，取值范围0~255
*/
void homePositionCmd(Device device, String channelId, String enabled, String resetTime, String presetIndex, SipSubscribe.Event errorEvent,SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException;
void homePositionCmd(Device device, String channelId, Boolean enabled, Integer resetTime, Integer presetIndex, SipSubscribe.Event errorEvent,SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException;
/**
* 设备配置命令

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

@ -880,7 +880,7 @@ public class SIPCommander implements ISIPCommander {
* @param presetIndex 调用预置位编号，开启看守位时使用，取值范围0~255
*/
@Override
public void homePositionCmd(Device device, String channelId, String enabled, String resetTime, String presetIndex, SipSubscribe.Event errorEvent,SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException {
public void homePositionCmd(Device device, String channelId, Boolean enabled, Integer resetTime, Integer presetIndex, SipSubscribe.Event errorEvent,SipSubscribe.Event okEvent) throws InvalidArgumentException, SipException, ParseException {
StringBuffer cmdXml = new StringBuffer(200);
String charset = device.getCharset();
@ -894,18 +894,10 @@ public class SIPCommander implements ISIPCommander {
cmdXml.append("<DeviceID>" + channelId + "</DeviceID>\r\n");
}
cmdXml.append("<HomePosition>\r\n");
if (NumericUtil.isInteger(enabled) && (!enabled.equals("0"))) {
if (enabled) {
cmdXml.append("<Enabled>1</Enabled>\r\n");
if (NumericUtil.isInteger(resetTime)) {
cmdXml.append("<ResetTime>" + resetTime + "</ResetTime>\r\n");
} else {
cmdXml.append("<ResetTime>0</ResetTime>\r\n");
}
if (NumericUtil.isInteger(presetIndex)) {
cmdXml.append("<PresetIndex>" + presetIndex + "</PresetIndex>\r\n");
} else {
cmdXml.append("<PresetIndex>0</PresetIndex>\r\n");
}
cmdXml.append("<ResetTime>" + resetTime + "</ResetTime>\r\n");
cmdXml.append("<PresetIndex>" + presetIndex + "</PresetIndex>\r\n");
} else {
cmdXml.append("<Enabled>0</Enabled>\r\n");
}

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

@ -358,8 +358,8 @@ public class SIPCommanderFroPlatform implements ISIPCommanderForPlatform {
}else {
catalogXml.append("<Password></Password>\r\n");
}
if (!ObjectUtils.isEmpty(channel.getPTZType())) {
catalogXml.append("<PTZType>" + channel.getPTZType() + "</PTZType>\r\n");
if (!ObjectUtils.isEmpty(channel.getPtzType())) {
catalogXml.append("<PTZType>" + channel.getPtzType() + "</PTZType>\r\n");
}else {
catalogXml.append("<PTZType></PTZType>\r\n");
}

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

@ -116,7 +116,7 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In
if (parentPlatform != null) {
Map<String, Object> param = getSendRtpParam(sendRtpItem);
if (mediaInfo == null) {
if (!userSetting.getServerId().equals(sendRtpItem.getServerId())) {
RequestPushStreamMsg requestPushStreamMsg = RequestPushStreamMsg.getInstance(
sendRtpItem.getMediaServerId(), sendRtpItem.getApp(), sendRtpItem.getStream(),
sendRtpItem.getIp(), sendRtpItem.getPort(), sendRtpItem.getSsrc(), sendRtpItem.isTcp(),

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

@ -15,8 +15,11 @@ import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor
import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
import com.genersoft.iot.vmp.service.*;
import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
import com.genersoft.iot.vmp.service.bean.RequestStopPushStreamMsg;
import com.genersoft.iot.vmp.service.redisMsg.RedisGbPlayMsgListener;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import gov.nist.javax.sip.message.SIPRequest;
@ -92,6 +95,12 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
@Autowired
private UserSetting userSetting;
@Autowired
private IStreamPushService pushService;
@Autowired
private RedisGbPlayMsgListener redisGbPlayMsgListener;
@Override
public void afterPropertiesSet() throws Exception {
// 添加消息处理的订阅
@ -115,7 +124,7 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
// 收流端发送的停止
if (sendRtpItem != null){
logger.info("[收到bye] 来自{}，停止通道：{}, 类型： {}", sendRtpItem.getPlatformId(), sendRtpItem.getChannelId(), sendRtpItem.getPlayType());
logger.info("[收到bye] 来自{}，停止通道：{}, 类型： {}, callId: {}", sendRtpItem.getPlatformId(), sendRtpItem.getChannelId(), sendRtpItem.getPlayType(), callIdHeader.getCallId());
String streamId = sendRtpItem.getStream();
Map<String, Object> param = new HashMap<>();
@ -123,59 +132,82 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
param.put("app",sendRtpItem.getApp());
param.put("stream",streamId);
param.put("ssrc",sendRtpItem.getSsrc());
logger.info("[收到bye] 停止推流：{}", streamId);
MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
redisCatchStorage.deleteSendRTPServer(sendRtpItem.getPlatformId(), sendRtpItem.getChannelId(),
callIdHeader.getCallId(), null);
zlmServerFactory.stopSendRtpStream(mediaInfo, param);
if (userSetting.getUseCustomSsrcForParentInvite()) {
mediaServerService.releaseSsrc(mediaInfo.getId(), sendRtpItem.getSsrc());
}
logger.info("[收到bye] 停止推流：{}, 媒体节点： {}", streamId, sendRtpItem.getMediaServerId());
if (sendRtpItem.getPlayType().equals(InviteStreamType.PUSH)) {
ParentPlatform platform = platformService.queryPlatformByServerGBId(sendRtpItem.getPlatformId());
if (platform != null) {
MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,
sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getChannelId(),
sendRtpItem.getPlatformId(), platform.getName(), userSetting.getServerId(), sendRtpItem.getMediaServerId());
messageForPushChannel.setPlatFormIndex(platform.getId());
redisCatchStorage.sendPlatformStopPlayMsg(messageForPushChannel);
// 查询这路流是否是本平台的
StreamPushItem push = pushService.getPush(sendRtpItem.getApp(), sendRtpItem.getStream());
if (push!= null && !push.isSelf()) {
// 不是本平台的就发送redis消息让其他wvp停止发流
ParentPlatform platform = platformService.queryPlatformByServerGBId(sendRtpItem.getPlatformId());
if (platform != null) {
RequestStopPushStreamMsg streamMsg = RequestStopPushStreamMsg.getInstance(sendRtpItem, platform.getName(), platform.getId());
redisGbPlayMsgListener.sendMsgForStopSendRtpStream(sendRtpItem.getServerId(), streamMsg);
}
}else {
logger.info("[上级平台停止观看] 未找到平台{}的信息，发送redis消息失败", sendRtpItem.getPlatformId());
MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
redisCatchStorage.deleteSendRTPServer(sendRtpItem.getPlatformId(), sendRtpItem.getChannelId(),
callIdHeader.getCallId(), null);
zlmServerFactory.stopSendRtpStream(mediaInfo, param);
if (userSetting.getUseCustomSsrcForParentInvite()) {
mediaServerService.releaseSsrc(mediaInfo.getId(), sendRtpItem.getSsrc());
}
ParentPlatform platform = platformService.queryPlatformByServerGBId(sendRtpItem.getPlatformId());
if (platform != null) {
MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,
sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getChannelId(),
sendRtpItem.getPlatformId(), platform.getName(), userSetting.getServerId(), sendRtpItem.getMediaServerId());
messageForPushChannel.setPlatFormIndex(platform.getId());
redisCatchStorage.sendPlatformStopPlayMsg(messageForPushChannel);
}else {
logger.info("[上级平台停止观看] 未找到平台{}的信息，发送redis消息失败", sendRtpItem.getPlatformId());
}
}
}else {
MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
redisCatchStorage.deleteSendRTPServer(sendRtpItem.getPlatformId(), sendRtpItem.getChannelId(),
callIdHeader.getCallId(), null);
zlmServerFactory.stopSendRtpStream(mediaInfo, param);
if (userSetting.getUseCustomSsrcForParentInvite()) {
mediaServerService.releaseSsrc(mediaInfo.getId(), sendRtpItem.getSsrc());
}
}
MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
if (mediaInfo != null) {
AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
if (audioBroadcastCatch != null && audioBroadcastCatch.getSipTransactionInfo().getCallId().equals(callIdHeader.getCallId())) {
// 来自上级平台的停止对讲
logger.info("[停止对讲] 来自上级，平台：{}, 通道：{}", sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
audioBroadcastManager.del(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
}
AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
if (audioBroadcastCatch != null && audioBroadcastCatch.getSipTransactionInfo().getCallId().equals(callIdHeader.getCallId())) {
// 来自上级平台的停止对讲
logger.info("[停止对讲] 来自上级，平台：{}, 通道：{}", sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
audioBroadcastManager.del(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
}
int totalReaderCount = zlmServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId);
if (totalReaderCount <= 0) {
logger.info("[收到bye] {} 无其它观看者，通知设备停止推流", streamId);
if (sendRtpItem.getPlayType().equals(InviteStreamType.PLAY)) {
Device device = deviceService.getDevice(sendRtpItem.getDeviceId());
if (device == null) {
logger.info("[收到bye] {} 通知设备停止推流时未找到设备信息", streamId);
}
try {
logger.info("[停止点播] {}/{}", sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
cmder.streamByeCmd(device, sendRtpItem.getChannelId(), streamId, null);
} catch (InvalidArgumentException | ParseException | SipException |
SsrcTransactionNotFoundException e) {
logger.error("[收到bye] {} 无其它观看者，通知设备停止推流， 发送BYE失败 {}",streamId, e.getMessage());
int totalReaderCount = zlmServerFactory.totalReaderCount(mediaInfo, sendRtpItem.getApp(), streamId);
if (totalReaderCount <= 0) {
logger.info("[收到bye] {} 无其它观看者，通知设备停止推流", streamId);
if (sendRtpItem.getPlayType().equals(InviteStreamType.PLAY)) {
Device device = deviceService.getDevice(sendRtpItem.getDeviceId());
if (device == null) {
logger.info("[收到bye] {} 通知设备停止推流时未找到设备信息", streamId);
}
try {
logger.info("[停止点播] {}/{}", sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
cmder.streamByeCmd(device, sendRtpItem.getChannelId(), streamId, null);
} catch (InvalidArgumentException | ParseException | SipException |
SsrcTransactionNotFoundException e) {
logger.error("[收到bye] {} 无其它观看者，通知设备停止推流， 发送BYE失败 {}",streamId, e.getMessage());
}
}
}
}
}
// 可能是设备发送的停止
SsrcTransaction ssrcTransaction = streamSession.getSsrcTransactionByCallId(callIdHeader.getCallId());
if (ssrcTransaction == null) {
return;
}
logger.info("[收到bye] 来自设备：{}, 通道已停止推流: {}", ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId());
// 可能是设备发送的停止
SsrcTransaction ssrcTransaction = streamSession.getSsrcTransactionByCallId(callIdHeader.getCallId());
if (ssrcTransaction == null) {
return;
}
logger.info("[收到bye] 来自设备：{}, 通道已停止推流: {}", ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId());
ParentPlatform platform = platformService.queryPlatformByServerGBId(ssrcTransaction.getDeviceId());
if (platform != null ) {

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

@ -38,6 +38,7 @@ import gov.nist.javax.sdp.fields.TimeField;
import gov.nist.javax.sdp.fields.URIField;
import gov.nist.javax.sip.message.SIPRequest;
import gov.nist.javax.sip.message.SIPResponse;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -404,12 +405,15 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
// * 2 推流中
sendRtpItem.setStatus(1);
redisCatchStorage.updateSendRTPSever(sendRtpItem);
String sdpIp = mediaServerItemInUSe.getSdpIp();
if (!ObjectUtils.isEmpty(platform.getSendStreamIp())) {
sdpIp = platform.getSendStreamIp();
}
StringBuffer content = new StringBuffer(200);
content.append("v=0\r\n");
content.append("o=" + channelId + " 0 0 IN IP4 " + mediaServerItemInUSe.getSdpIp() + "\r\n");
content.append("o=" + channelId + " 0 0 IN IP4 " + sdpIp + "\r\n");
content.append("s=" + sessionName + "\r\n");
content.append("c=IN IP4 " + mediaServerItemInUSe.getSdpIp() + "\r\n");
content.append("c=IN IP4 " + sdpIp + "\r\n");
if ("Playback".equalsIgnoreCase(sessionName)) {
content.append("t=" + finalStartTime + " " + finalStopTime + "\r\n");
} else {
@ -498,6 +502,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
String endTimeStr = DateUtil.urlFormatter.format(end);
String stream = device.getDeviceId() + "_" + channelId + "_" + startTimeStr + "_" + endTimeStr;
SSRCInfo ssrcInfo = mediaServerService.openRTPServer(mediaServerItem, stream, null, device.isSsrcCheck(), true, 0,false, false, device.getStreamModeForParam());
sendRtpItem.setStream(stream);
// 写入redis， 超时时回复
redisCatchStorage.updateSendRTPSever(sendRtpItem);
playService.playBack(mediaServerItem, ssrcInfo, device.getDeviceId(), channelId, DateUtil.formatter.format(start),
@ -574,14 +579,20 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
}
if ("push".equals(gbStream.getStreamType())) {
if (streamPushItem != null && streamPushItem.isPushIng()) {
// 推流状态
pushStream(evt, request, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
} else {
// 未推流 拉起
notifyStreamOnline(evt, request, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
if (streamPushItem != null) {
// 从redis查询是否正在接收这个推流
OnStreamChangedHookParam pushListItem = redisCatchStorage.getPushListItem(gbStream.getApp(), gbStream.getStream());
if (pushListItem != null) {
StreamPushItem transform = streamPushService.transform(pushListItem);
transform.setSelf(userSetting.getServerId().equals(pushListItem.getSeverId()));
// 推流状态
pushStream(evt, request, gbStream, transform, platform, callIdHeader, mediaServerItem, port, tcpActive,
mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
}else {
// 未推流 拉起
notifyStreamOnline(evt, request, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
}
}
} else if ("proxy".equals(gbStream.getStreamType())) {
if (null != proxyByAppAndStream) {
@ -900,11 +911,15 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
public SIPResponse sendStreamAck(MediaServerItem mediaServerItem, SIPRequest request, SendRtpItem sendRtpItem, ParentPlatform platform, RequestEvent evt) {
String sdpIp = mediaServerItem.getSdpIp();
if (!ObjectUtils.isEmpty(platform.getSendStreamIp())) {
sdpIp = platform.getSendStreamIp();
}
StringBuffer content = new StringBuffer(200);
content.append("v=0\r\n");
content.append("o=" + sendRtpItem.getChannelId() + " 0 0 IN IP4 " + mediaServerItem.getSdpIp() + "\r\n");
content.append("o=" + sendRtpItem.getChannelId() + " 0 0 IN IP4 " + sdpIp + "\r\n");
content.append("s=Play\r\n");
content.append("c=IN IP4 " + mediaServerItem.getSdpIp() + "\r\n");
content.append("c=IN IP4 " + sdpIp + "\r\n");
content.append("t=0 0\r\n");
// 非严格模式端口不统一, 增加兼容性，修改为一个不为0的端口
int localPort = sendRtpItem.getLocalPort();
@ -1006,7 +1021,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
Media media = mediaDescription.getMedia();
Vector mediaFormats = media.getMediaFormats(false);
if (mediaFormats.contains("8")) {
// if (mediaFormats.contains("8")) {
port = media.getMediaPort();
String protocol = media.getProtocol();
// 区分TCP发流还是udp， 当前默认udp
@ -1022,7 +1037,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
}
}
break;
}
// }
}
if (port == -1) {
logger.info("不支持的媒体格式，返回415");

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

@ -1,7 +1,5 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.conf.CivilCodeFileConf;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.*;
@ -78,9 +76,6 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
@Autowired
private NotifyRequestForCatalogProcessor notifyRequestForCatalogProcessor;
@Autowired
private CivilCodeFileConf civilCodeFileConf;
private ConcurrentLinkedQueue<HandlerCatchData> taskQueue = new ConcurrentLinkedQueue<>();
@Qualifier("taskExecutor")
@ -98,7 +93,6 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
@Override
public void process(RequestEvent evt) {
try {
if (taskQueue.size() >= userSetting.getMaxNotifyCountQueue()) {
responseAck((SIPRequest) evt.getRequest(), Response.BUSY_HERE, null, null);
logger.error("[notify] 待处理消息队列已满 {}，返回486 BUSY_HERE，消息不做处理", userSetting.getMaxNotifyCountQueue());
@ -234,25 +228,8 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
mobilePosition.setLongitudeGcj02(deviceChannel.getLongitudeGcj02());
mobilePosition.setLatitudeGcj02(deviceChannel.getLatitudeGcj02());
if (userSetting.getSavePositionHistory()) {
storager.insertMobilePosition(mobilePosition);
}
deviceChannelService.updateChannelGPS(device, deviceChannel, mobilePosition);
storager.updateChannelPosition(deviceChannel);
// 向关联了该通道并且开启移动位置订阅的上级平台发送移动位置订阅消息
// 发送redis消息。 通知位置信息的变化
JSONObject jsonObject = new JSONObject();
jsonObject.put("time", DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(mobilePosition.getTime()));
jsonObject.put("serial", deviceId);
jsonObject.put("code", channelId);
jsonObject.put("longitude", mobilePosition.getLongitude());
jsonObject.put("latitude", mobilePosition.getLatitude());
jsonObject.put("altitude", mobilePosition.getAltitude());
jsonObject.put("direction", mobilePosition.getDirection());
jsonObject.put("speed", mobilePosition.getSpeed());
redisCatchStorage.sendMobilePositionMsg(jsonObject);
} catch (DocumentException e) {
logger.error("未处理的异常 ", e);
}
@ -340,25 +317,8 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
mobilePosition.setLongitudeGcj02(deviceChannel.getLongitudeGcj02());
mobilePosition.setLatitudeGcj02(deviceChannel.getLatitudeGcj02());
if (userSetting.getSavePositionHistory()) {
storager.insertMobilePosition(mobilePosition);
}
storager.updateChannelPosition(deviceChannel);
// 发送redis消息。 通知位置信息的变化
JSONObject jsonObject = new JSONObject();
jsonObject.put("time", DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(mobilePosition.getTime()));
jsonObject.put("serial", deviceChannel.getDeviceId());
jsonObject.put("code", deviceChannel.getChannelId());
jsonObject.put("longitude", mobilePosition.getLongitude());
jsonObject.put("latitude", mobilePosition.getLatitude());
jsonObject.put("altitude", mobilePosition.getAltitude());
jsonObject.put("direction", mobilePosition.getDirection());
jsonObject.put("speed", mobilePosition.getSpeed());
redisCatchStorage.sendMobilePositionMsg(jsonObject);
deviceChannelService.updateChannelGPS(device, deviceChannel, mobilePosition);
}
// TODO: 需要实现存储报警信息、报警分类
// 回复200 OK
if (redisCatchStorage.deviceIsOnline(deviceId)) {

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

@ -248,7 +248,7 @@ public class DeviceControlQueryMessageHandler extends SIPRequestProcessorParent
HomePositionRequest homePosition = loadElement(rootElement, HomePositionRequest.class);
//获取整个消息主体，我们只需要修改请求头即可
HomePositionRequest.HomePosition info = homePosition.getHomePosition();
cmder.homePositionCmd(device, channelId, info.getEnabled(), info.getResetTime(), info.getPresetIndex(),
cmder.homePositionCmd(device, channelId, !"0".equals(info.getEnabled()), Integer.parseInt(info.getResetTime()), Integer.parseInt(info.getPresetIndex()),
errorResult -> onError(request, errorResult),
okResult -> onOk(request, okResult));
} catch (Exception e) {

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

@ -75,6 +75,9 @@ public class AlarmNotifyMessageHandler extends SIPRequestProcessorParent impleme
@Autowired
private ThreadPoolTaskExecutor taskExecutor;
@Autowired
private EventPublisher eventPublisher;
@Override
public void afterPropertiesSet() throws Exception {
@ -158,22 +161,7 @@ public class AlarmNotifyMessageHandler extends SIPRequestProcessorParent impleme
mobilePosition.setLongitudeGcj02(deviceChannel.getLongitudeGcj02());
mobilePosition.setLatitudeGcj02(deviceChannel.getLatitudeGcj02());
if (userSetting.getSavePositionHistory()) {
storager.insertMobilePosition(mobilePosition);
}
storager.updateChannelPosition(deviceChannel);
// 发送redis消息。 通知位置信息的变化
JSONObject jsonObject = new JSONObject();
jsonObject.put("time", DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(mobilePosition.getTime()));
jsonObject.put("serial", deviceChannel.getDeviceId());
jsonObject.put("code", deviceChannel.getChannelId());
jsonObject.put("longitude", mobilePosition.getLongitude());
jsonObject.put("latitude", mobilePosition.getLatitude());
jsonObject.put("altitude", mobilePosition.getAltitude());
jsonObject.put("direction", mobilePosition.getDirection());
jsonObject.put("speed", mobilePosition.getSpeed());
redisCatchStorage.sendMobilePositionMsg(jsonObject);
deviceChannelService.updateChannelGPS(device, deviceChannel, mobilePosition);
}
}
if (!ObjectUtils.isEmpty(deviceAlarm.getDeviceId())) {

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

@ -1,8 +1,8 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.notify.cmd;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
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.notify.NotifyMessageHandler;
@ -57,6 +57,9 @@ public class MobilePositionNotifyMessageHandler extends SIPRequestProcessorParen
@Autowired
private IDeviceChannelService deviceChannelService;
@Autowired
private EventPublisher eventPublisher;
private ConcurrentLinkedQueue<SipMsgInfo> taskQueue = new ConcurrentLinkedQueue<>();
@Qualifier("taskExecutor")
@ -137,22 +140,7 @@ public class MobilePositionNotifyMessageHandler extends SIPRequestProcessorParen
mobilePosition.setLongitudeGcj02(deviceChannel.getLongitudeGcj02());
mobilePosition.setLatitudeGcj02(deviceChannel.getLatitudeGcj02());
if (userSetting.getSavePositionHistory()) {
storager.insertMobilePosition(mobilePosition);
}
storager.updateChannelPosition(deviceChannel);
// 发送redis消息。 通知位置信息的变化
JSONObject jsonObject = new JSONObject();
jsonObject.put("time", DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(mobilePosition.getTime()));
jsonObject.put("serial", deviceChannel.getDeviceId());
jsonObject.put("code", deviceChannel.getChannelId());
jsonObject.put("longitude", mobilePosition.getLongitude());
jsonObject.put("latitude", mobilePosition.getLatitude());
jsonObject.put("altitude", mobilePosition.getAltitude());
jsonObject.put("direction", mobilePosition.getDirection());
jsonObject.put("speed", mobilePosition.getSpeed());
redisCatchStorage.sendMobilePositionMsg(jsonObject);
deviceChannelService.updateChannelGPS(device, deviceChannel, mobilePosition);
} catch (DocumentException e) {
logger.error("未处理的异常 ", e);

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

@ -1,6 +1,5 @@
package com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
@ -131,11 +130,7 @@ public class MobilePositionResponseMessageHandler extends SIPRequestProcessorPar
mobilePosition.setLongitudeGcj02(deviceChannel.getLongitudeGcj02());
mobilePosition.setLatitudeGcj02(deviceChannel.getLatitudeGcj02());
if (userSetting.getSavePositionHistory()) {
storager.insertMobilePosition(mobilePosition);
}
storager.updateChannelPosition(deviceChannel);
deviceChannelService.updateChannelGPS(device, deviceChannel, mobilePosition);
String key = DeferredResultHolder.CALLBACK_CMD_MOBILE_POSITION + device.getDeviceId();
RequestMessage msg = new RequestMessage();
@ -143,17 +138,6 @@ public class MobilePositionResponseMessageHandler extends SIPRequestProcessorPar
msg.setData(mobilePosition);
resultHolder.invokeAllResult(msg);
// 发送redis消息。 通知位置信息的变化
JSONObject jsonObject = new JSONObject();
jsonObject.put("time", DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(mobilePosition.getTime()));
jsonObject.put("serial", deviceChannel.getDeviceId());
jsonObject.put("code", deviceChannel.getChannelId());
jsonObject.put("longitude", mobilePosition.getLongitude());
jsonObject.put("latitude", mobilePosition.getLatitude());
jsonObject.put("altitude", mobilePosition.getAltitude());
jsonObject.put("direction", mobilePosition.getDirection());
jsonObject.put("speed", mobilePosition.getSpeed());
redisCatchStorage.sendMobilePositionMsg(jsonObject);
//回复 200 OK
try {
responseAck(request, Response.OK);

4
src/main/java/com/genersoft/iot/vmp/gb28181/utils/XmlUtil.java
View File

@ -568,14 +568,14 @@ public class XmlUtil {
String ptzTypeFromInfo = XmlUtil.getText(info, "PTZType");
if(!ObjectUtils.isEmpty(ptzTypeFromInfo)){
try {
deviceChannel.setPTZType(Integer.parseInt(ptzTypeFromInfo));
deviceChannel.setPtzType(Integer.parseInt(ptzTypeFromInfo));
}catch (NumberFormatException e){
logger.warn("[xml解析] 从通道数据info中获取PTZType失败： {}", ptzTypeFromInfo);
}
}
} else {
try {
deviceChannel.setPTZType(Integer.parseInt(ptzType));
deviceChannel.setPtzType(Integer.parseInt(ptzType));
}catch (NumberFormatException e){
logger.warn("[xml解析] 从通道数据中获取PTZType失败： {}", ptzType);
}

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

@ -72,9 +72,6 @@ public class ZLMHttpHookListener {
@Autowired
private AudioBroadcastManager audioBroadcastManager;
@Autowired
private ZLMServerFactory zlmServerFactory;
@Autowired
private IPlayService playService;
@ -123,9 +120,6 @@ public class ZLMHttpHookListener {
@Autowired
private VideoStreamSessionManager sessionManager;
@Autowired
private AssistRESTfulUtils assistRESTfulUtils;
@Autowired
private SSRCFactory ssrcFactory;
@ -147,7 +141,7 @@ public class ZLMHttpHookListener {
taskExecutor.execute(() -> {
List<ZlmHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(HookType.on_server_keepalive);
if (subscribes != null && subscribes.size() > 0) {
if (subscribes != null && !subscribes.isEmpty()) {
for (ZlmHttpHookSubscribe.Event subscribe : subscribes) {
subscribe.response(null, param);
}
@ -166,7 +160,7 @@ public class ZLMHttpHookListener {
@PostMapping(value = "/on_play", produces = "application/json;charset=UTF-8")
public HookResult onPlay(@RequestBody OnPlayHookParam param) {
if (logger.isDebugEnabled()) {
logger.debug("[ZLM HOOK] 播放鉴权：{}->{}" + param.getMediaServerId(), param);
logger.debug("[ZLM HOOK] 播放鉴权：{}->{}", param.getMediaServerId(), param);
}
String mediaServerId = param.getMediaServerId();
@ -252,11 +246,7 @@ public class ZLMHttpHookListener {
taskExecutor.execute(() -> {
ZlmHttpHookSubscribe.Event subscribe = this.subscribe.sendNotify(HookType.on_publish, json);
if (subscribe != null) {
if (mediaInfo != null) {
subscribe.response(mediaInfo, param);
} else {
new HookResultForOnPublish(1, "zlm not register");
}
subscribe.response(mediaInfo, param);
}
});
@ -519,32 +509,46 @@ public class ZLMHttpHookListener {
List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByStream(param.getStream());
if (!sendRtpItems.isEmpty()) {
for (SendRtpItem sendRtpItem : sendRtpItems) {
if (sendRtpItem != null && sendRtpItem.getApp().equals(param.getApp())) {
String platformId = sendRtpItem.getPlatformId();
ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId);
Device device = deviceService.getDevice(platformId);
if (sendRtpItem == null) {
continue;
}
try {
if (platform != null) {
commanderFroPlatform.streamByeCmd(platform, sendRtpItem);
redisCatchStorage.deleteSendRTPServer(platformId, sendRtpItem.getChannelId(),
sendRtpItem.getCallId(), sendRtpItem.getStream());
} else {
cmder.streamByeCmd(device, sendRtpItem.getChannelId(), param.getStream(), sendRtpItem.getCallId());
if (sendRtpItem.getPlayType().equals(InviteStreamType.BROADCAST)
|| sendRtpItem.getPlayType().equals(InviteStreamType.TALK)) {
AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
if (audioBroadcastCatch != null) {
// 来自上级平台的停止对讲
logger.info("[停止对讲] 来自上级，平台：{}, 通道：{}", sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
audioBroadcastManager.del(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
if (sendRtpItem.getApp().equals(param.getApp())) {
logger.info(sendRtpItem.toString());
if (userSetting.getServerId().equals(sendRtpItem.getServerId())) {
MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,
sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getChannelId(),
sendRtpItem.getPlatformId(), null, userSetting.getServerId(), param.getMediaServerId());
// 通知其他wvp停止发流
redisCatchStorage.sendPushStreamClose(messageForPushChannel);
}else {
String platformId = sendRtpItem.getPlatformId();
ParentPlatform platform = storager.queryParentPlatByServerGBId(platformId);
Device device = deviceService.getDevice(platformId);
try {
if (platform != null) {
commanderFroPlatform.streamByeCmd(platform, sendRtpItem);
redisCatchStorage.deleteSendRTPServer(platformId, sendRtpItem.getChannelId(),
sendRtpItem.getCallId(), sendRtpItem.getStream());
} else {
cmder.streamByeCmd(device, sendRtpItem.getChannelId(), param.getStream(), sendRtpItem.getCallId());
if (sendRtpItem.getPlayType().equals(InviteStreamType.BROADCAST)
|| sendRtpItem.getPlayType().equals(InviteStreamType.TALK)) {
AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
if (audioBroadcastCatch != null) {
// 来自上级平台的停止对讲
logger.info("[停止对讲] 来自上级，平台：{}, 通道：{}", sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
audioBroadcastManager.del(sendRtpItem.getDeviceId(), sendRtpItem.getChannelId());
}
}
}
} catch (SipException | InvalidArgumentException | ParseException |
SsrcTransactionNotFoundException e) {
logger.error("[命令发送失败] 发送BYE: {}", e.getMessage());
}
} catch (SipException | InvalidArgumentException | ParseException |
SsrcTransactionNotFoundException e) {
logger.error("[命令发送失败] 发送BYE: {}", e.getMessage());
}
}
}
}
@ -579,9 +583,9 @@ public class ZLMHttpHookListener {
}
// 收到无人观看说明流也没有在往上级推送
if (redisCatchStorage.isChannelSendingRTP(inviteInfo.getChannelId())) {
List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId(
List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChannelId(
inviteInfo.getChannelId());
if (sendRtpItems.size() > 0) {
if (!sendRtpItems.isEmpty()) {
for (SendRtpItem sendRtpItem : sendRtpItems) {
ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(sendRtpItem.getPlatformId());
try {
@ -798,7 +802,7 @@ public class ZLMHttpHookListener {
logger.info("[ZLM HOOK] zlm 启动 " + zlmServerConfig.getGeneralMediaServerId());
taskExecutor.execute(() -> {
List<ZlmHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(HookType.on_server_started);
if (subscribes != null && subscribes.size() > 0) {
if (subscribes != null && !subscribes.isEmpty()) {
for (ZlmHttpHookSubscribe.Event subscribe : subscribes) {
subscribe.response(null, zlmServerConfig);
}
@ -847,12 +851,11 @@ public class ZLMHttpHookListener {
*/
@ResponseBody
@PostMapping(value = "/on_rtp_server_timeout", produces = "application/json;charset=UTF-8")
public HookResult onRtpServerTimeout(HttpServletRequest request, @RequestBody OnRtpServerTimeoutHookParam
public HookResult onRtpServerTimeout(@RequestBody OnRtpServerTimeoutHookParam
param) {
logger.info("[ZLM HOOK] rtpServer收流超时：{}->{}({})", param.getMediaServerId(), param.getStream_id(), param.getSsrc());
taskExecutor.execute(() -> {
JSONObject json = (JSONObject) JSON.toJSON(param);
List<ZlmHttpHookSubscribe.Event> subscribes = this.subscribe.getSubscribes(HookType.on_rtp_server_timeout);
if (subscribes != null && !subscribes.isEmpty()) {
for (ZlmHttpHookSubscribe.Event subscribe : subscribes) {

4
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMServerFactory.java
View File

@ -289,6 +289,10 @@ public class ZLMServerFactory {
* 调用zlm RESTful API —— stopSendRtp
*/
public Boolean stopSendRtpStream(MediaServerItem mediaServerItem, Map<String, Object>param) {
if (mediaServerItem == null) {
logger.error("[停止RTP推流] 失败: 媒体节点为NULL");
return false;
}
Boolean result = false;
JSONObject jsonObject = zlmresTfulUtils.stopSendRtp(mediaServerItem, param);
if (jsonObject == null) {

7
src/main/java/com/genersoft/iot/vmp/service/IDeviceChannelService.java
View File

@ -2,6 +2,7 @@ package com.genersoft.iot.vmp.service;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.MobilePosition;
import com.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo;
import com.genersoft.iot.vmp.vmanager.gb28181.platform.bean.ChannelReduce;
@ -92,4 +93,10 @@ public interface IDeviceChannelService {
* 修改通道的码流类型
*/
void updateChannelStreamIdentification(DeviceChannel channel);
List<DeviceChannel> queryChaneListByDeviceId(String deviceId);
void updateChannelGPS(Device device, DeviceChannel deviceChannel, MobilePosition mobilePosition);
void stopPlay(String deviceId, String channelId);
}

1
src/main/java/com/genersoft/iot/vmp/service/IPlayService.java
View File

@ -68,4 +68,5 @@ public interface IPlayService {
void getSnap(String deviceId, String channelId, String fileName, ErrorCallback errorCallback);
void stopPlay(Device device, String channelId);
}

15
src/main/java/com/genersoft/iot/vmp/service/bean/GPSMsgInfo.java
View File

@ -1,5 +1,8 @@
package com.genersoft.iot.vmp.service.bean;
import com.genersoft.iot.vmp.gb28181.bean.MobilePosition;
import com.genersoft.iot.vmp.utils.DateUtil;
public class GPSMsgInfo {
/**
@ -39,6 +42,18 @@ public class GPSMsgInfo {
private boolean stored;
public static GPSMsgInfo getInstance(MobilePosition mobilePosition) {
GPSMsgInfo gpsMsgInfo = new GPSMsgInfo();
gpsMsgInfo.setId(mobilePosition.getChannelId());
gpsMsgInfo.setAltitude(mobilePosition.getAltitude() + "");
gpsMsgInfo.setLng(mobilePosition.getLongitude());
gpsMsgInfo.setLat(mobilePosition.getLatitude());
gpsMsgInfo.setSpeed(mobilePosition.getSpeed());
gpsMsgInfo.setDirection(mobilePosition.getDirection() + "");
gpsMsgInfo.setTime(DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(mobilePosition.getTime()));
return gpsMsgInfo;
}
public String getId() {
return id;

49
src/main/java/com/genersoft/iot/vmp/service/bean/RequestStopPushStreamMsg.java Executable file
View File

@ -0,0 +1,49 @@
package com.genersoft.iot.vmp.service.bean;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
public class RequestStopPushStreamMsg {
private SendRtpItem sendRtpItem;
private String platformName;
private int platFormIndex;
public SendRtpItem getSendRtpItem() {
return sendRtpItem;
}
public void setSendRtpItem(SendRtpItem sendRtpItem) {
this.sendRtpItem = sendRtpItem;
}
public String getPlatformName() {
return platformName;
}
public void setPlatformName(String platformName) {
this.platformName = platformName;
}
public int getPlatFormIndex() {
return platFormIndex;
}
public void setPlatFormIndex(int platFormIndex) {
this.platFormIndex = platFormIndex;
}
public static RequestStopPushStreamMsg getInstance(SendRtpItem sendRtpItem, String platformName, int platFormIndex) {
RequestStopPushStreamMsg streamMsg = new RequestStopPushStreamMsg();
streamMsg.setSendRtpItem(sendRtpItem);
streamMsg.setPlatformName(platformName);
streamMsg.setPlatFormIndex(platFormIndex);
return streamMsg;
}
}

10
src/main/java/com/genersoft/iot/vmp/service/bean/WvpRedisMsgCmd.java
View File

@ -6,7 +6,17 @@ package com.genersoft.iot.vmp.service.bean;
public class WvpRedisMsgCmd {
/**
* 请求获取推流信息
*/
public static final String GET_SEND_ITEM = "GetSendItem";
/**
* 请求推流的请求
*/
public static final String REQUEST_PUSH_STREAM = "RequestPushStream";
/**
* 停止推流的请求
*/
public static final String REQUEST_STOP_PUSH_STREAM = "RequestStopPushStream";
}

82
src/main/java/com/genersoft/iot/vmp/service/impl/DeviceChannelServiceImpl.java
View File

@ -1,16 +1,21 @@
package com.genersoft.iot.vmp.service.impl;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
import com.genersoft.iot.vmp.gb28181.bean.MobilePosition;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.utils.Coordtransform;
import com.genersoft.iot.vmp.service.IDeviceChannelService;
import com.genersoft.iot.vmp.service.IInviteStreamService;
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.storager.dao.DeviceMobilePositionMapper;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo;
import com.genersoft.iot.vmp.vmanager.gb28181.platform.bean.ChannelReduce;
@ -35,7 +40,7 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
private final static Logger logger = LoggerFactory.getLogger(DeviceChannelServiceImpl.class);
@Autowired
private IRedisCatchStorage redisCatchStorage;
private EventPublisher eventPublisher;
@Autowired
private IInviteStreamService inviteStreamService;
@ -46,6 +51,15 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
@Autowired
private DeviceMapper deviceMapper;
@Autowired
private DeviceMobilePositionMapper deviceMobilePositionMapper;
@Autowired
private UserSetting userSetting;
@Autowired
private IRedisCatchStorage redisCatchStorage;
@Override
public DeviceChannel updateGps(DeviceChannel deviceChannel, Device device) {
if (deviceChannel.getLongitude()*deviceChannel.getLatitude() > 0) {
@ -84,7 +98,6 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
public void updateChannel(String deviceId, DeviceChannel channel) {
String channelId = channel.getChannelId();
channel.setDeviceId(deviceId);
// StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId);
if (inviteInfo != null && inviteInfo.getStreamInfo() != null) {
channel.setStreamId(inviteInfo.getStreamInfo().getStream());
@ -280,4 +293,69 @@ public class DeviceChannelServiceImpl implements IDeviceChannelService {
}
channelMapper.updateChannelStreamIdentification(channel);
}
@Override
public List<DeviceChannel> queryChaneListByDeviceId(String deviceId) {
return channelMapper.queryAllChannels(deviceId);
}
@Override
public void updateChannelGPS(Device device, DeviceChannel deviceChannel, MobilePosition mobilePosition) {
if (userSetting.getSavePositionHistory()) {
deviceMobilePositionMapper.insertNewPosition(mobilePosition);
}
if (deviceChannel.getChannelId().equals(deviceChannel.getDeviceId())) {
deviceChannel.setChannelId(null);
}
if (deviceChannel.getGpsTime() == null) {
deviceChannel.setGpsTime(DateUtil.getNow());
}
int updated = channelMapper.updatePosition(deviceChannel);
if (updated == 0) {
return;
}
List<DeviceChannel> deviceChannels = new ArrayList<>();
if (deviceChannel.getChannelId() == null) {
// 有的设备这里上报的deviceId与通道Id是一样，这种情况更新设备下的全部通道
List<DeviceChannel> deviceChannelsInDb = queryChaneListByDeviceId(device.getDeviceId());
deviceChannels.addAll(deviceChannelsInDb);
}else {
deviceChannels.add(deviceChannel);
}
if (deviceChannels.isEmpty()) {
return;
}
if (deviceChannels.size() > 100) {
logger.warn("[更新通道位置信息后发送通知] 设备可能是平台，上报的位置信息未标明通道编号，" +
"导致所有通道被更新位置， deviceId:{}", device.getDeviceId());
}
for (DeviceChannel channel : deviceChannels) {
// 向关联了该通道并且开启移动位置订阅的上级平台发送移动位置订阅消息
mobilePosition.setChannelId(channel.getChannelId());
try {
eventPublisher.mobilePositionEventPublish(mobilePosition);
}catch (Exception e) {
logger.error("[向上级转发移动位置失败] ", e);
}
// 发送redis消息。 通知位置信息的变化
JSONObject jsonObject = new JSONObject();
jsonObject.put("time", DateUtil.yyyy_MM_dd_HH_mm_ssToISO8601(mobilePosition.getTime()));
jsonObject.put("serial", mobilePosition.getDeviceId());
jsonObject.put("code", mobilePosition.getChannelId());
jsonObject.put("longitude", mobilePosition.getLongitude());
jsonObject.put("latitude", mobilePosition.getLatitude());
jsonObject.put("altitude", mobilePosition.getAltitude());
jsonObject.put("direction", mobilePosition.getDirection());
jsonObject.put("speed", mobilePosition.getSpeed());
redisCatchStorage.sendMobilePositionMsg(jsonObject);
}
}
@Override
public void stopPlay(String deviceId, String channelId) {
channelMapper.stopPlay(deviceId, channelId);
}
}

3
src/main/java/com/genersoft/iot/vmp/service/impl/DeviceServiceImpl.java
View File

@ -269,6 +269,8 @@ public class DeviceServiceImpl implements IDeviceService {
int subscribeCycleForCatalog = Math.max(device.getSubscribeCycleForCatalog(),30);
// 设置最小值为30
dynamicTask.startCron(device.getDeviceId() + "catalog", catalogSubscribeTask, (subscribeCycleForCatalog -1) * 1000);
catalogSubscribeTask.run();
return true;
}
@ -302,6 +304,7 @@ public class DeviceServiceImpl implements IDeviceService {
int subscribeCycleForCatalog = Math.max(device.getSubscribeCycleForMobilePosition(),30);
// 刷新订阅
dynamicTask.startCron(device.getDeviceId() + "mobile_position" , mobilePositionSubscribeTask, subscribeCycleForCatalog * 1000);
mobilePositionSubscribeTask.run();
return true;
}

36
src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
View File

@ -34,7 +34,6 @@ import com.genersoft.iot.vmp.service.bean.*;
import com.genersoft.iot.vmp.service.redisMsg.RedisGbPlayMsgListener;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.storager.dao.CloudRecordServiceMapper;
import com.genersoft.iot.vmp.utils.CloudRecordUtils;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.AudioBroadcastResult;
@ -123,9 +122,6 @@ public class PlayServiceImpl implements IPlayService {
@Autowired
private DynamicTask dynamicTask;
@Autowired
private CloudRecordServiceMapper cloudRecordServiceMapper;
@Autowired
private ISIPCommanderForPlatform commanderForPlatform;
@ -1170,7 +1166,7 @@ public class PlayServiceImpl implements IPlayService {
dynamicTask.startDelay(key, ()->{
logger.info("[语音广播]等待invite消息超时：{}/{}", device.getDeviceId(), channelId);
stopAudioBroadcast(device.getDeviceId(), channelId);
}, 2000);
}, 10*1000);
}, eventResultForError -> {
// 发送失败
logger.error("语音广播发送失败： {}:{}", channelId, eventResultForError.msg);
@ -1409,6 +1405,14 @@ public class PlayServiceImpl implements IPlayService {
logger.info("调用ZLM推流接口, 结果： {}", jsonObject);
logger.info("RTP推流成功[ {}/{} ]，{}->{}, ", param.get("app"), param.get("stream"), jsonObject.getString("local_port"),
sendRtpItem.isTcpActive()?"被动发流": param.get("dst_url") + ":" + param.get("dst_port"));
if (sendRtpItem.getPlayType() == InviteStreamType.PUSH && correlationInfo instanceof ParentPlatform) {
ParentPlatform platform = (ParentPlatform)correlationInfo;
MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0, sendRtpItem.getApp(), sendRtpItem.getStream(),
sendRtpItem.getChannelId(), platform.getServerGBId(), platform.getName(), userSetting.getServerId(),
sendRtpItem.getMediaServerId());
messageForPushChannel.setPlatFormIndex(platform.getId());
redisCatchStorage.sendPlatformStartPlayMsg(messageForPushChannel);
}
} else {
logger.error("RTP推流失败: {}, 参数：{}", jsonObject.getString("msg"), JSONObject.toJSONString(param));
if (sendRtpItem.isOnlyAudio()) {
@ -1584,4 +1588,26 @@ public class PlayServiceImpl implements IPlayService {
});
}
@Override
public void stopPlay(Device device, String channelId) {
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channelId);
if (inviteInfo == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "点播未找到");
}
if (InviteSessionStatus.ok == inviteInfo.getStatus()) {
try {
logger.info("[停止点播] {}/{}", device.getDeviceId(), channelId);
cmder.streamByeCmd(device, channelId, inviteInfo.getStream(), null, null);
} catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) {
logger.error("[命令发送失败] 停止点播， 发送BYE: {}", e.getMessage());
throw new ControllerException(ErrorCode.ERROR100.getCode(), "命令发送失败: " + e.getMessage());
}
}
inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, device.getDeviceId(), channelId);
storager.stopPlay(device.getDeviceId(), channelId);
channelService.stopPlay(device.getDeviceId(), channelId);
if (inviteInfo.getStreamInfo() != null) {
mediaServerService.closeRTPServer(inviteInfo.getStreamInfo().getMediaServerId(), inviteInfo.getStream());
}
}
}

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

@ -8,7 +8,6 @@ import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent;
import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
@ -25,7 +24,6 @@ import com.genersoft.iot.vmp.service.IStreamProxyService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.storager.dao.GbStreamMapper;
import com.genersoft.iot.vmp.storager.dao.ParentPlatformMapper;
import com.genersoft.iot.vmp.storager.dao.PlatformGbStreamMapper;
import com.genersoft.iot.vmp.storager.dao.StreamProxyMapper;
import com.genersoft.iot.vmp.utils.DateUtil;
@ -333,8 +331,6 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
result = zlmresTfulUtils.addStreamProxy(mediaServerItem, param.getApp(), param.getStream(), param.getUrl().trim(),
param.isEnableAudio(), param.isEnableMp4(), param.getRtpType());
}
System.out.println("addStreamProxyToZlm====");
System.out.println(result);
if (result != null && result.getInteger("code") == 0) {
JSONObject data = result.getJSONObject("data");
if (data == null) {

50
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisGbPlayMsgListener.java
View File

@ -133,7 +133,10 @@ public class RedisGbPlayMsgListener implements MessageListener {
case WvpRedisMsgCmd.REQUEST_PUSH_STREAM:
RequestPushStreamMsg param = JSON.to(RequestPushStreamMsg.class, wvpRedisMsg.getContent());
requestPushStreamMsgHand(param, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
break;
case WvpRedisMsgCmd.REQUEST_STOP_PUSH_STREAM:
RequestStopPushStreamMsg streamMsg = JSON.to(RequestStopPushStreamMsg.class, wvpRedisMsg.getContent());
requestStopPushStreamMsgHand(streamMsg, wvpRedisMsg.getFromId(), wvpRedisMsg.getSerial());
break;
default:
break;
@ -397,6 +400,19 @@ public class RedisGbPlayMsgListener implements MessageListener {
redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
}
/**
* 发送请求推流的消息
*/
public void sendMsgForStopSendRtpStream(String serverId, RequestStopPushStreamMsg streamMsg) {
String key = UUID.randomUUID().toString();
WvpRedisMsg redisMsg = WvpRedisMsg.getRequestInstance(userSetting.getServerId(), serverId,
WvpRedisMsgCmd.REQUEST_STOP_PUSH_STREAM, key, JSON.toJSONString(streamMsg));
JSONObject jsonObject = (JSONObject)JSON.toJSON(redisMsg);
logger.info("[REDIS 请求其他平台停止推流] {}: {}", serverId, jsonObject);
redisTemplate.convertAndSend(WVP_PUSH_STREAM_KEY, jsonObject);
}
private SendRtpItem querySendRTPServer(String platformGbId, String channelId, String streamId, String callId) {
if (platformGbId == null) {
platformGbId = "*";
@ -423,4 +439,36 @@ public class RedisGbPlayMsgListener implements MessageListener {
return null;
}
}
/**
* 处理收到的请求推流的请求
*/
private void requestStopPushStreamMsgHand(RequestStopPushStreamMsg streamMsg, String fromId, String serial) {
SendRtpItem sendRtpItem = streamMsg.getSendRtpItem();
if (sendRtpItem == null) {
logger.info("[REDIS 执行其他平台的请求停止推流] 失败： sendRtpItem为NULL");
return;
}
MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
if (mediaInfo == null) {
// TODO 回复错误
return;
}
Map<String, Object> param = new HashMap<>();
param.put("vhost","__defaultVhost__");
param.put("app",sendRtpItem.getApp());
param.put("stream",sendRtpItem.getStream());
param.put("ssrc", sendRtpItem.getSsrc());
if (zlmServerFactory.stopSendRtpStream(mediaInfo, param)) {
logger.info("[REDIS 执行其他平台的请求停止推流] 成功： {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream());
// 发送redis消息
MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,
sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getChannelId(),
sendRtpItem.getPlatformId(), streamMsg.getPlatformName(), userSetting.getServerId(), sendRtpItem.getMediaServerId());
messageForPushChannel.setPlatFormIndex(streamMsg.getPlatFormIndex());
redisCatchStorage.sendPlatformStopPlayMsg(messageForPushChannel);
}
}
}

25
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamCloseResponseListener.java
View File

@ -2,12 +2,10 @@ package com.genersoft.iot.vmp.service.redisMsg;
import com.alibaba.fastjson2.JSON;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.InviteStreamType;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IStreamPushService;
@ -25,7 +23,6 @@ import org.springframework.stereotype.Component;
import javax.sip.InvalidArgumentException;
import javax.sip.SipException;
import java.text.ParseException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@ -73,7 +70,7 @@ public class RedisPushStreamCloseResponseListener implements MessageListener {
MessageForPushChannel pushChannel = JSON.parseObject(message.getBody(), MessageForPushChannel.class);
StreamPushItem push = streamPushService.getPush(pushChannel.getApp(), pushChannel.getStream());
if (push != null) {
List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChnnelId(
List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServerByChannelId(
push.getGbId());
if (!sendRtpItems.isEmpty()) {
for (SendRtpItem sendRtpItem : sendRtpItems) {
@ -86,26 +83,6 @@ public class RedisPushStreamCloseResponseListener implements MessageListener {
logger.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage());
}
}
if (push.isSelf()) {
// 停止向上级推流
String streamId = sendRtpItem.getStream();
Map<String, Object> param = new HashMap<>();
param.put("vhost","__defaultVhost__");
param.put("app",sendRtpItem.getApp());
param.put("stream",streamId);
param.put("ssrc",sendRtpItem.getSsrc());
logger.info("[REDIS消息-推流结束] 停止向上级推流：{}", streamId);
MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
redisCatchStorage.deleteSendRTPServer(sendRtpItem.getPlatformId(), sendRtpItem.getChannelId(), sendRtpItem.getCallId(), sendRtpItem.getStream());
zlmServerFactory.stopSendRtpStream(mediaInfo, param);
if (InviteStreamType.PUSH == sendRtpItem.getPlayType()) {
MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(0,
sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getChannelId(),
sendRtpItem.getPlatformId(), parentPlatform.getName(), userSetting.getServerId(), sendRtpItem.getMediaServerId());
messageForPushChannel.setPlatFormIndex(parentPlatform.getId());
redisCatchStorage.sendPlatformStopPlayMsg(messageForPushChannel);
}
}
}
}
}

3
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamStatusListMsgListener.java
View File

@ -88,7 +88,8 @@ public class RedisPushStreamStatusListMsgListener implements MessageListener {
streamPushItemForSave.add(streamPushItem);
allGBId.put(streamPushItem.getGbId(), streamPushItem);
} else {
if (allGBId.containsKey(streamPushItem.getGbId())) {
if (allGBId.containsKey(streamPushItem.getGbId())
&& (!allGBId.get(streamPushItem.getGbId()).getApp().equals(streamPushItem.getApp()) || !allGBId.get(streamPushItem.getGbId()).getStream().equals(streamPushItem.getStream()))) {
GbStream gbStream = allGBId.get(streamPushItem.getGbId());
logger.warn("[REDIS消息-推流设备列表更新-UPDATE] 国标编号重复: {}, 已分配给{}/{}",
streamPushItem.getGbId(), gbStream.getApp(), gbStream.getStream());

98
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisStreamMsgListener.java
View File

@ -1,11 +1,7 @@
package com.genersoft.iot.vmp.service.redisMsg;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@ -41,52 +37,52 @@ public class RedisStreamMsgListener implements MessageListener {
@Override
public void onMessage(Message message, byte[] bytes) {
boolean isEmpty = taskQueue.isEmpty();
taskQueue.offer(message);
if (isEmpty) {
taskExecutor.execute(() -> {
while (!taskQueue.isEmpty()) {
Message msg = taskQueue.poll();
try {
JSONObject steamMsgJson = JSON.parseObject(msg.getBody(), JSONObject.class);
if (steamMsgJson == null) {
logger.warn("[收到redis 流变化]消息解析失败");
continue;
}
String serverId = steamMsgJson.getString("serverId");
if (userSetting.getServerId().equals(serverId)) {
// 自己发送的消息忽略即可
continue;
}
logger.info("[收到redis 流变化]： {}", new String(message.getBody()));
String app = steamMsgJson.getString("app");
String stream = steamMsgJson.getString("stream");
boolean register = steamMsgJson.getBoolean("register");
String mediaServerId = steamMsgJson.getString("mediaServerId");
OnStreamChangedHookParam onStreamChangedHookParam = new OnStreamChangedHookParam();
onStreamChangedHookParam.setSeverId(serverId);
onStreamChangedHookParam.setApp(app);
onStreamChangedHookParam.setStream(stream);
onStreamChangedHookParam.setRegist(register);
onStreamChangedHookParam.setMediaServerId(mediaServerId);
onStreamChangedHookParam.setCreateStamp(System.currentTimeMillis()/1000);
onStreamChangedHookParam.setAliveSecond(0L);
onStreamChangedHookParam.setTotalReaderCount("0");
onStreamChangedHookParam.setOriginType(0);
onStreamChangedHookParam.setOriginTypeStr("0");
onStreamChangedHookParam.setOriginTypeStr("unknown");
if (register) {
zlmMediaListManager.addPush(onStreamChangedHookParam);
}else {
zlmMediaListManager.removeMedia(app, stream);
}
}catch (Exception e) {
logger.warn("[REDIS消息-流变化] 发现未处理的异常, \r\n{}", JSON.toJSONString(message));
logger.error("[REDIS消息-流变化] 异常内容： ", e);
}
}
});
}
// boolean isEmpty = taskQueue.isEmpty();
// taskQueue.offer(message);
// if (isEmpty) {
// taskExecutor.execute(() -> {
// while (!taskQueue.isEmpty()) {
// Message msg = taskQueue.poll();
// try {
// JSONObject steamMsgJson = JSON.parseObject(msg.getBody(), JSONObject.class);
// if (steamMsgJson == null) {
// logger.warn("[收到redis 流变化]消息解析失败");
// continue;
// }
// String serverId = steamMsgJson.getString("serverId");
//
// if (userSetting.getServerId().equals(serverId)) {
// // 自己发送的消息忽略即可
// continue;
// }
// logger.info("[收到redis 流变化]： {}", new String(message.getBody()));
// String app = steamMsgJson.getString("app");
// String stream = steamMsgJson.getString("stream");
// boolean register = steamMsgJson.getBoolean("register");
// String mediaServerId = steamMsgJson.getString("mediaServerId");
// OnStreamChangedHookParam onStreamChangedHookParam = new OnStreamChangedHookParam();
// onStreamChangedHookParam.setSeverId(serverId);
// onStreamChangedHookParam.setApp(app);
// onStreamChangedHookParam.setStream(stream);
// onStreamChangedHookParam.setRegist(register);
// onStreamChangedHookParam.setMediaServerId(mediaServerId);
// onStreamChangedHookParam.setCreateStamp(System.currentTimeMillis()/1000);
// onStreamChangedHookParam.setAliveSecond(0L);
// onStreamChangedHookParam.setTotalReaderCount("0");
// onStreamChangedHookParam.setOriginType(0);
// onStreamChangedHookParam.setOriginTypeStr("0");
// onStreamChangedHookParam.setOriginTypeStr("unknown");
// if (register) {
// zlmMediaListManager.addPush(onStreamChangedHookParam);
// }else {
// zlmMediaListManager.removeMedia(app, stream);
// }
// }catch (Exception e) {
// logger.warn("[REDIS消息-流变化] 发现未处理的异常, \r\n{}", JSON.toJSONString(message));
// logger.error("[REDIS消息-流变化] 异常内容： ", e);
// }
// }
// });
// }
}
}

7
src/main/java/com/genersoft/iot/vmp/storager/IRedisCatchStorage.java
View File

@ -181,7 +181,7 @@ public interface IRedisCatchStorage {
*/
void sendStreamPushRequestedMsgForStatus();
List<SendRtpItem> querySendRTPServerByChnnelId(String channelId);
List<SendRtpItem> querySendRTPServerByChannelId(String channelId);
List<SendRtpItem> querySendRTPServerByStream(String stream);
@ -211,5 +211,10 @@ public interface IRedisCatchStorage {
void addPushListItem(String app, String stream, OnStreamChangedHookParam param);
OnStreamChangedHookParam getPushListItem(String app, String stream);
void removePushListItem(String app, String stream, String mediaServerId);
void sendPushStreamClose(MessageForPushChannel messageForPushChannel);
}

12
src/main/java/com/genersoft/iot/vmp/storager/dao/DeviceChannelMapper.java
View File

@ -23,7 +23,7 @@ public interface DeviceChannelMapper {
"longitude_wgs84, latitude_wgs84, has_audio, create_time, update_time, business_group_id, gps_time, stream_identification) " +
"VALUES (#{channelId}, #{deviceId}, #{name}, #{manufacture}, #{model}, #{owner}, #{civilCode}, #{block}," +
"#{address}, #{parental}, #{parentId}, #{safetyWay}, #{registerWay}, #{certNum}, #{certifiable}, #{errCode}, #{secrecy}, " +
"#{ipAddress}, #{port}, #{password}, #{PTZType}, #{status}, #{streamId}, #{longitude}, #{latitude}, #{longitudeGcj02}, " +
"#{ipAddress}, #{port}, #{password}, #{ptzType}, #{status}, #{streamId}, #{longitude}, #{latitude}, #{longitudeGcj02}, " +
"#{latitudeGcj02}, #{longitudeWgs84}, #{latitudeWgs84}, #{hasAudio}, #{createTime}, #{updateTime}, #{businessGroupId}, #{gpsTime}, #{streamIdentification})")
int add(DeviceChannel channel);
@ -48,7 +48,7 @@ public interface DeviceChannelMapper {
"<if test='ipAddress != null'>, ip_address=#{ipAddress}</if>" +
"<if test='port != null'>, port=#{port}</if>" +
"<if test='password != null'>, password=#{password}</if>" +
"<if test='PTZType != null'>, custom_ptz_type=#{PTZType}</if>" +
"<if test='ptzType != null'>, custom_ptz_type=#{ptzType}</if>" +
"<if test='status != null'>, status=#{status}</if>" +
"<if test='streamId != null'>, stream_id=#{streamId}</if>" +
"<if test='hasAudio != null'>, has_audio=#{hasAudio}</if>" +
@ -250,7 +250,7 @@ public interface DeviceChannelMapper {
"#{item.owner}, #{item.civilCode}, #{item.block},#{item.subCount}," +
"#{item.address}, #{item.parental}, #{item.parentId}, #{item.safetyWay}, #{item.registerWay}, " +
"#{item.certNum}, #{item.certifiable}, #{item.errCode}, #{item.secrecy}, " +
"#{item.ipAddress}, #{item.port}, #{item.password}, #{item.PTZType}, #{item.status}, " +
"#{item.ipAddress}, #{item.port}, #{item.password}, #{item.ptzType}, #{item.status}, " +
"#{item.streamId}, #{item.longitude}, #{item.latitude},#{item.longitudeGcj02}, " +
"#{item.latitudeGcj02},#{item.longitudeWgs84}, #{item.latitudeWgs84}, #{item.hasAudio}, now(), now(), " +
"#{item.businessGroupId}, #{item.gpsTime}, #{item.streamIdentification}) " +
@ -271,7 +271,7 @@ public interface DeviceChannelMapper {
"#{item.owner}, #{item.civilCode}, #{item.block},#{item.subCount}," +
"#{item.address}, #{item.parental}, #{item.parentId}, #{item.safetyWay}, #{item.registerWay}, " +
"#{item.certNum}, #{item.certifiable}, #{item.errCode}, #{item.secrecy}, " +
"#{item.ipAddress}, #{item.port}, #{item.password}, #{item.PTZType}, #{item.status}, " +
"#{item.ipAddress}, #{item.port}, #{item.password}, #{item.ptzType}, #{item.status}, " +
"#{item.streamId}, #{item.longitude}, #{item.latitude},#{item.longitudeGcj02}, " +
"#{item.latitudeGcj02},#{item.longitudeWgs84}, #{item.latitudeWgs84}, #{item.hasAudio}, now(), now(), " +
"#{item.businessGroupId}, #{item.gpsTime}) " +
@ -339,7 +339,7 @@ public interface DeviceChannelMapper {
"<if test='item.ipAddress != null'>, ip_address=#{item.ipAddress}</if>" +
"<if test='item.port != null'>, port=#{item.port}</if>" +
"<if test='item.password != null'>, password=#{item.password}</if>" +
"<if test='item.PTZType != null'>, ptz_type=#{item.PTZType}</if>" +
"<if test='item.ptzType != null'>, ptz_type=#{item.ptzType}</if>" +
"<if test='item.status != null'>, status=#{item.status}</if>" +
"<if test='item.streamId != null'>, stream_id=#{item.streamId}</if>" +
"<if test='item.hasAudio != null'>, has_audio=#{item.hasAudio}</if>" +
@ -395,7 +395,7 @@ public interface DeviceChannelMapper {
"WHERE device_id=#{deviceId} " +
" <if test='channelId != null' > AND channel_id=#{channelId}</if>" +
" </script>"})
void updatePosition(DeviceChannel deviceChannel);
int updatePosition(DeviceChannel deviceChannel);
@Select("SELECT * FROM wvp_device_channel WHERE length(trim(stream_id)) > 0")
List<DeviceChannel> getAllChannelInPlay();

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

@ -269,7 +269,7 @@ public interface DeviceMapper {
"charset,"+
"ssrc_check,"+
"as_message_channel,"+
"broadcastPushAfterAck,"+
"broadcast_push_after_ack,"+
"geo_coord_sys,"+
"on_line,"+
"media_server_id"+

5
src/main/java/com/genersoft/iot/vmp/storager/dao/ParentPlatformMapper.java
View File

@ -17,10 +17,10 @@ public interface ParentPlatformMapper {
@Insert("INSERT INTO wvp_platform (enable, name, server_gb_id, server_gb_domain, server_ip, server_port,device_gb_id,device_ip,"+
"device_port,username,password,expires,keep_timeout,transport,character_set,ptz,rtcp,as_message_channel,auto_push_channel,"+
"status,start_offline_push,catalog_id,administrative_division,catalog_group,create_time,update_time) " +
"status,start_offline_push,catalog_id,administrative_division,catalog_group,create_time,update_time,send_stream_ip) " +
" VALUES (#{enable}, #{name}, #{serverGBId}, #{serverGBDomain}, #{serverIP}, #{serverPort}, #{deviceGBId}, #{deviceIp}, " +
" #{devicePort}, #{username}, #{password}, #{expires}, #{keepTimeout}, #{transport}, #{characterSet}, #{ptz}, #{rtcp}, #{asMessageChannel}, #{autoPushChannel}, " +
" #{status}, #{startOfflinePush}, #{catalogId}, #{administrativeDivision}, #{catalogGroup}, #{createTime}, #{updateTime})")
" #{status}, #{startOfflinePush}, #{catalogId}, #{administrativeDivision}, #{catalogGroup}, #{createTime}, #{updateTime}, #{sendStreamIp})")
int addParentPlatform(ParentPlatform parentPlatform);
@Update("UPDATE wvp_platform " +
@ -49,6 +49,7 @@ public interface ParentPlatformMapper {
"administrative_division=#{administrativeDivision}, " +
"create_time=#{createTime}, " +
"update_time=#{updateTime}, " +
"send_stream_ip=#{sendStreamIp}, " +
"catalog_id=#{catalogId} " +
"WHERE id=#{id}")
int updateParentPlatform(ParentPlatform parentPlatform);

15
src/main/java/com/genersoft/iot/vmp/storager/impl/RedisCatchStorageImpl.java
View File

@ -184,7 +184,7 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
}
@Override
public List<SendRtpItem> querySendRTPServerByChnnelId(String channelId) {
public List<SendRtpItem> querySendRTPServerByChannelId(String channelId) {
if (channelId == null) {
return null;
}
@ -656,6 +656,12 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
redisTemplate.opsForValue().set(key, param);
}
@Override
public OnStreamChangedHookParam getPushListItem(String app, String stream) {
String key = VideoManagerConstants.PUSH_STREAM_LIST + app + "_" + stream;
return (OnStreamChangedHookParam)redisTemplate.opsForValue().get(key);
}
@Override
public void removePushListItem(String app, String stream, String mediaServerId) {
String key = VideoManagerConstants.PUSH_STREAM_LIST + app + "_" + stream;
@ -665,4 +671,11 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
}
}
@Override
public void sendPushStreamClose(MessageForPushChannel msg) {
String key = VideoManagerConstants.VM_MSG_STREAM_PUSH_CLOSE_REQUESTED;
logger.info("[redis发送通知] 发送 停止向上级推流 {}: {}/{}->{}", key, msg.getApp(), msg.getStream(), msg.getPlatFormId());
redisTemplate.convertAndSend(key, JSON.toJSON(msg));
}
}

5
src/main/java/com/genersoft/iot/vmp/vmanager/cloudRecord/CloudRecordController.java
View File

@ -1,12 +1,8 @@
package com.genersoft.iot.vmp.vmanager.cloudRecord;
import com.alibaba.fastjson2.JSONArray;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.conf.security.JwtUtils;
import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager;
import com.genersoft.iot.vmp.media.zlm.ZLMServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.ICloudRecordService;
import com.genersoft.iot.vmp.service.IMediaServerService;
@ -22,7 +18,6 @@ import org.apache.commons.lang3.ObjectUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.*;
import javax.servlet.http.HttpServletRequest;

18
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/device/DeviceControl.java
View File

@ -13,7 +13,7 @@ import com.genersoft.iot.vmp.conf.security.JwtUtils;
import com.genersoft.iot.vmp.gb28181.bean.Device;
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.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import io.swagger.v3.oas.annotations.Operation;
@ -45,7 +45,7 @@ public class DeviceControl {
private IVideoManagerStorage storager;
@Autowired
private SIPCommander cmder;
private ISIPCommander cmder;
@Autowired
private DeferredResultHolder resultHolder;
@ -254,15 +254,13 @@ public class DeviceControl {
@Operation(summary = "看守位控制", security = @SecurityRequirement(name = JwtUtils.HEADER))
@Parameter(name = "deviceId", description = "设备国标编号", required = true)
@Parameter(name = "channelId", description = "通道国标编号", required = true)
@Parameter(name = "enabled", description = "是否开启看守位 1:开启,0:关闭", required = true)
@Parameter(name = "enabled", description = "是否开启看守位", required = true)
@Parameter(name = "presetIndex", description = "调用预置位编号")
@Parameter(name = "resetTime", description = "自动归位时间间隔")
@GetMapping("/home_position/{deviceId}/{enabled}")
public DeferredResult<String> homePositionApi(@PathVariable String deviceId,
@PathVariable String enabled,
@RequestParam(required = false) String resetTime,
@RequestParam(required = false) String presetIndex,
String channelId) {
@Parameter(name = "resetTime", description = "自动归位时间间隔 单位：秒")
@GetMapping("/home_position")
public DeferredResult<String> homePositionApi(String deviceId, String channelId, Boolean enabled,
@RequestParam(required = false) Integer resetTime,
@RequestParam(required = false) Integer presetIndex) {
if (logger.isDebugEnabled()) {
logger.debug("报警复位API调用");
}

31
src/main/java/com/genersoft/iot/vmp/vmanager/gb28181/play/PlayController.java
View File

@ -3,12 +3,10 @@ package com.genersoft.iot.vmp.vmanager.gb28181.play;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.genersoft.iot.vmp.common.InviteInfo;
import com.genersoft.iot.vmp.common.InviteSessionStatus;
import com.genersoft.iot.vmp.common.InviteSessionType;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
import com.genersoft.iot.vmp.conf.security.JwtUtils;
import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.SsrcTransaction;
@ -26,7 +24,7 @@ import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.utils.DateUtil;
import com.genersoft.iot.vmp.vmanager.bean.*;
import com.genersoft.iot.vmp.vmanager.bean.AudioBroadcastResult;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.StreamContent;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
@ -41,11 +39,8 @@ import org.springframework.web.bind.annotation.*;
import org.springframework.web.context.request.async.DeferredResult;
import javax.servlet.http.HttpServletRequest;
import javax.sip.InvalidArgumentException;
import javax.sip.SipException;
import java.net.MalformedURLException;
import java.net.URL;
import java.text.ParseException;
import java.util.List;
import java.util.UUID;
@ -157,7 +152,8 @@ public class PlayController {
wvpResult.setMsg(msg);
}
requestMessage.setData(wvpResult);
resultHolder.invokeResult(requestMessage);
// 此处必须释放所有请求
resultHolder.invokeAllResult(requestMessage);
});
return result;
}
@ -165,9 +161,8 @@ public class PlayController {
@Operation(summary = "停止点播", security = @SecurityRequirement(name = JwtUtils.HEADER))
@Parameter(name = "deviceId", description = "设备国标编号", required = true)
@Parameter(name = "channelId", description = "通道国标编号", required = true)
@Parameter(name = "isSubStream", description = "是否子码流（true-子码流，false-主码流），默认为false", required = true)
@GetMapping("/stop/{deviceId}/{channelId}")
public JSONObject playStop(@PathVariable String deviceId, @PathVariable String channelId,boolean isSubStream) {
public JSONObject playStop(@PathVariable String deviceId, @PathVariable String channelId) {
logger.debug(String.format("设备预览/回放停止API调用，streamId：%s_%s", deviceId, channelId ));
@ -180,26 +175,10 @@ public class PlayController {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "设备[" + deviceId + "]不存在");
}
InviteInfo inviteInfo = inviteStreamService.getInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId);
if (inviteInfo == null) {
throw new ControllerException(ErrorCode.ERROR100.getCode(), "点播未找到");
}
if (InviteSessionStatus.ok == inviteInfo.getStatus()) {
try {
logger.info("[停止点播] {}/{}", device.getDeviceId(), channelId);
cmder.streamByeCmd(device, channelId, inviteInfo.getStream(), null, null);
} catch (InvalidArgumentException | SipException | ParseException | SsrcTransactionNotFoundException e) {
logger.error("[命令发送失败] 停止点播， 发送BYE: {}", e.getMessage());
throw new ControllerException(ErrorCode.ERROR100.getCode(), "命令发送失败: " + e.getMessage());
}
}
inviteStreamService.removeInviteInfoByDeviceAndChannel(InviteSessionType.PLAY, deviceId, channelId);
storager.stopPlay(deviceId, channelId);
playService.stopPlay(device, channelId);
JSONObject json = new JSONObject();
json.put("deviceId", deviceId);
json.put("channelId", channelId);
json.put("isSubStream", isSubStream);
return json;
}

2
src/main/resources/application-dev.yml
View File

@ -114,5 +114,5 @@ user-settings:
device-status-notify: true
# [可选] 日志配置, 一般不需要改
logging:
config: classpath:logback-spring-local.xml
config: classpath:logback-spring.xml

42
web_src/src/components/channelList.vue
View File

@ -100,9 +100,9 @@
<span v-show="!scope.row.edit">{{ scope.row.location }}</span>
</template>
</el-table-column>
<el-table-column prop="ptztype" label="云台类型" min-width="100">
<el-table-column prop="ptzType" label="云台类型" min-width="100">
<template v-slot:default="scope">
<el-select v-show="scope.row.edit" v-model="scope.row.ptztype"
<el-select v-show="scope.row.edit" v-model="scope.row.ptzType"
placeholder="云台类型" filterable>
<el-option
v-for="(value, key) in ptzTypes"
@ -111,7 +111,7 @@
:value="key"
/>
</el-select>
<div v-show="!scope.row.edit">{{ scope.row.ptztypeText }}</div>
<div v-show="!scope.row.edit">{{ scope.row.ptzTypeText }}</div>
</template>
</el-table-column>
<el-table-column label="开启音频" min-width="100">
@ -178,13 +178,24 @@
@click="changeSubchannel(scope.row)">查看
</el-button>
<el-divider v-if="scope.row.subCount > 0 || scope.row.parental === 1" direction="vertical"></el-divider>
<el-button size="medium" v-bind:disabled="device == null || device.online === 0"
icon="el-icon-video-camera"
type="text" @click="queryRecords(scope.row)">设备录像
</el-button>
<el-button size="medium" v-bind:disabled="device == null || device.online === 0" icon="el-icon-cloudy"
type="text" @click="queryCloudRecords(scope.row)">云端录像
</el-button>
<!-- <el-button size="medium" v-bind:disabled="device == null || device.online === 0"-->
<!-- icon="el-icon-video-camera"-->
<!-- type="text" @click="queryRecords(scope.row)">设备录像-->
<!-- </el-button>-->
<!-- <el-button size="medium" v-bind:disabled="device == null || device.online === 0" icon="el-icon-cloudy"-->
<!-- type="text" @click="queryCloudRecords(scope.row)">云端录像-->
<!-- </el-button>-->
<el-dropdown @command="(command)=>{moreClick(command, scope.row)}">
<el-button size="medium" type="text" >
更多功能<i class="el-icon-arrow-down el-icon--right"></i>
</el-button>
<el-dropdown-menu slot="dropdown">
<el-dropdown-item command="records" v-bind:disabled="device == null || device.online === 0">
设备录像</el-dropdown-item>
<el-dropdown-item command="cloudRecords" v-bind:disabled="device == null || device.online === 0" >
云端录像</el-dropdown-item>
</el-dropdown-menu>
</el-dropdown>
</template>
</el-table-column>
</el-table>
@ -312,7 +323,7 @@ export default {
that.total = res.data.data.total;
that.deviceChannelList = res.data.data.list;
that.deviceChannelList.forEach(e => {
e.ptztype = e.ptztype + "";
e.ptzType = e.ptzType + "";
that.$set(e, "edit", false);
that.$set(e, "location", "");
if (e.longitude && e.latitude) {
@ -372,6 +383,13 @@ export default {
// that.$message.error("请求超时");
});
},
moreClick: function (command, itemData) {
if (command === "records") {
this.queryRecords(itemData)
}else if (command === "cloudRecords") {
this.queryCloudRecords(itemData)
}
},
queryRecords: function (itemData) {
let deviceId = this.deviceId;
let channelId = itemData.channelId;
@ -460,7 +478,7 @@ export default {
this.total = res.data.data.total;
this.deviceChannelList = res.data.data.list;
this.deviceChannelList.forEach(e => {
e.ptztype = e.ptztype + "";
e.ptzType = e.ptzType + "";
this.$set(e, "edit", false);
this.$set(e, "location", "");
if (e.longitude && e.latitude) {

6
web_src/src/components/common/DeviceTree.vue
View File

@ -131,11 +131,11 @@ export default {
type = 2;
}
console.log(type)
if (item.basicData.ptztype === 1 ) { // 1-球机;2-半球;3-固定枪机;4-遥控枪机
if (item.basicData.ptzType === 1 ) { // 1-球机;2-半球;3-固定枪机;4-遥控枪机
type = 4;
}else if (item.basicData.ptztype === 2) {
}else if (item.basicData.ptzType === 2) {
type = 5;
}else if (item.basicData.ptztype === 3 || item.basicData.ptztype === 4) {
}else if (item.basicData.ptzType === 3 || item.basicData.ptzType === 4) {
type = 6;
}
}else {

2
web_src/src/components/dialog/channelMapInfobox.vue
View File

@ -7,7 +7,7 @@
<el-descriptions-item label="设备归属" >{{channel.owner}}</el-descriptions-item>
<el-descriptions-item label="行政区域" >{{channel.civilCode}}</el-descriptions-item>
<el-descriptions-item label="安装地址" >{{channel.address}}</el-descriptions-item>
<el-descriptions-item label="云台类型" >{{channel.ptztypeText}}</el-descriptions-item>
<el-descriptions-item label="云台类型" >{{channel.ptzTypeText}}</el-descriptions-item>
<el-descriptions-item label="经纬度" >{{channel.longitude}},{{channel.latitude}}</el-descriptions-item>
<el-descriptions-item label="状态">
<el-tag size="small" v-if="channel.status === 1">在线</el-tag>

12
web_src/src/components/dialog/platformEdit.vue
View File

@ -37,8 +37,8 @@
<el-form-item label="本地端口" prop="devicePort">
<el-input v-model="platform.devicePort" :disabled="true" type="number"></el-input>
</el-form-item>
<el-form-item label="SIP认证用户名" prop="username">
<el-input v-model="platform.username"></el-input>
<el-form-item label="SDP发流IP" prop="sendStreamIp">
<el-input v-model="platform.sendStreamIp"></el-input>
</el-form-item>
</el-form>
</el-col>
@ -47,6 +47,9 @@
<el-form-item label="行政区划" prop="administrativeDivision">
<el-input v-model="platform.administrativeDivision" clearable></el-input>
</el-form-item>
<el-form-item label="SIP认证用户名" prop="username">
<el-input v-model="platform.username"></el-input>
</el-form-item>
<el-form-item label="SIP认证密码" prop="password">
<el-input v-model="platform.password" ></el-input>
</el-form-item>
@ -159,7 +162,8 @@ export default {
characterSet: "GB2312",
startOfflinePush: false,
catalogGroup: 1,
administrativeDivision: null,
administrativeDivision: "",
sendStreamIp: null,
},
rules: {
name: [{ required: true, message: "请输入平台名称", trigger: "blur" }],
@ -198,6 +202,7 @@ export default {
that.platform.devicePort = res.data.data.devicePort;
that.platform.username = res.data.data.username;
that.platform.password = res.data.data.password;
that.platform.sendStreamIp = res.data.data.sendStreamIp;
that.platform.administrativeDivision = res.data.data.username.substr(0, 6);
}
@ -228,6 +233,7 @@ export default {
this.platform.catalogId = platform.catalogId;
this.platform.startOfflinePush = platform.startOfflinePush;
this.platform.catalogGroup = platform.catalogGroup;
this.platform.sendStreamIp = platform.sendStreamIp;
this.platform.administrativeDivision = platform.administrativeDivision;
this.onSubmit_text = "保存";
this.saveUrl = "/api/platform/save";

4
web_src/src/components/dialog/rtcPlayer.vue
View File

@ -41,8 +41,8 @@ export default {
zlmsdpUrl: url,//流地址
simulecast: false,
useCamera: false,
audioEnable: false,
videoEnable: false,
audioEnable: true,
videoEnable: true,
recvOnly: true,
})
webrtcPlayer.on(ZLMRTCClient.Events.WEBRTC_ICE_CANDIDATE_ERROR,(e)=>{// ICE 协商出错

2
web_src/src/components/map.vue
View File

@ -243,7 +243,7 @@ export default {
},
getImageByChannel: function (channel) {
let src = "static/images/gis/camera.png"
switch (channel.ptztype) {
switch (channel.ptzType) {
case 1:
if (channel.status === 1) {
src = "static/images/gis/camera1.png"

1
数据库/2.7.0/初始化-mysql-2.7.0.sql
View File

@ -198,6 +198,7 @@ create table wvp_platform (
update_time character varying(50),
as_message_channel bool default false,
auto_push_channel bool default false,
send_stream_ip character varying(50),
constraint uk_platform_unique_server_gb_id unique (server_gb_id)
);

1
数据库/2.7.0/初始化-postgresql-kingbase-2.7.0.sql
View File

@ -198,6 +198,7 @@ create table wvp_platform (
update_time character varying(50),
as_message_channel bool default false,
auto_push_channel bool default false,
send_stream_ip character varying(50),
constraint uk_platform_unique_server_gb_id unique (server_gb_id)
);

5
数据库/2.7.0/更新-mysql-2.7.0.sql
View File

@ -2,4 +2,7 @@ alter table wvp_device_channel
add stream_identification character varying(50);
alter table wvp_device
drop switch_primary_sub_stream;
drop switch_primary_sub_stream;
alter table wvp_platform
add send_stream_ip character varying(50);

5
数据库/2.7.0/更新-postgresql-kingbase-2.7.0.sql
View File

@ -2,4 +2,7 @@ alter table wvp_device_channel
add stream_identification character varying(50);
alter table wvp_device
drop switch_primary_sub_stream;
drop switch_primary_sub_stream;
alter table wvp_platform
add send_stream_ip character varying(50);