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

临时提交

pull/1642/head
648540858 2024-09-09 11:16:02 +08:00
parent 489fbe31a5
commit 2e90344c89
16 changed files with 271 additions and 265 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
src/main/java/com/genersoft/iot/vmp/gb28181/bean/SendRtpInfo.java
View File

@ -2,8 +2,6 @@ package com.genersoft.iot.vmp.gb28181.bean;
import com.genersoft.iot.vmp.media.bean.MediaServer; import com.genersoft.iot.vmp.media.bean.MediaServer;
import com.genersoft.iot.vmp.service.bean.RequestPushStreamMsg; import com.genersoft.iot.vmp.service.bean.RequestPushStreamMsg;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import lombok.Data; import lombok.Data;
@Data @Data
@ -34,6 +32,11 @@ public class SendRtpInfo {
*/ */
private String targetName; private String targetName;
/**
* 是否是发送给上级平台
*/
private boolean sendToPlatform;
/** /**
* 直播流的应用名 * 直播流的应用名
*/ */
@ -182,8 +185,13 @@ public class SendRtpInfo {
sendRtpItem.setIp(ip); sendRtpItem.setIp(ip);
sendRtpItem.setPort(port); sendRtpItem.setPort(port);
sendRtpItem.setSsrc(ssrc); sendRtpItem.setSsrc(ssrc);
sendRtpItem.setDeviceId(deviceId); if (deviceId != null) {
sendRtpItem.setPlatformId(platformId); sendRtpItem.setTargetId(deviceId);
sendRtpItem.setSendToPlatform(false);
}else {
sendRtpItem.setTargetId(platformId);
sendRtpItem.setSendToPlatform(true);
}
sendRtpItem.setChannelId(channelId); sendRtpItem.setChannelId(channelId);
sendRtpItem.setTcp(isTcp); sendRtpItem.setTcp(isTcp);
sendRtpItem.setRtcp(rtcp); sendRtpItem.setRtcp(rtcp);
@ -200,9 +208,7 @@ public class SendRtpInfo {
"ip='" + ip + '\'' + "ip='" + ip + '\'' +
", port=" + port + ", port=" + port +
", ssrc='" + ssrc + '\'' + ", ssrc='" + ssrc + '\'' +
", platformId='" + platformId + '\'' + ", targetId='" + targetId + '\'' +
", platformName='" + platformName + '\'' +
", deviceId='" + deviceId + '\'' +
", app='" + app + '\'' + ", app='" + app + '\'' +
", channelId='" + channelId + '\'' + ", channelId='" + channelId + '\'' +
", status=" + status + ", status=" + status +

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

@ -10,6 +10,9 @@ import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.dao.DeviceChannelMapper; import com.genersoft.iot.vmp.gb28181.dao.DeviceChannelMapper;
import com.genersoft.iot.vmp.gb28181.dao.DeviceMapper; import com.genersoft.iot.vmp.gb28181.dao.DeviceMapper;
import com.genersoft.iot.vmp.gb28181.dao.PlatformChannelMapper; import com.genersoft.iot.vmp.gb28181.dao.PlatformChannelMapper;
import com.genersoft.iot.vmp.gb28181.service.IDeviceChannelService;
import com.genersoft.iot.vmp.gb28181.service.IDeviceService;
import com.genersoft.iot.vmp.gb28181.service.IInviteStreamService;
import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager; import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
import com.genersoft.iot.vmp.gb28181.session.SipInviteSessionManager; import com.genersoft.iot.vmp.gb28181.session.SipInviteSessionManager;
import com.genersoft.iot.vmp.gb28181.task.ISubscribeTask; import com.genersoft.iot.vmp.gb28181.task.ISubscribeTask;
@ -20,9 +23,6 @@ import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd.CatalogResponseMessageHandler; import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.response.cmd.CatalogResponseMessageHandler;
import com.genersoft.iot.vmp.media.bean.MediaServer; import com.genersoft.iot.vmp.media.bean.MediaServer;
import com.genersoft.iot.vmp.media.service.IMediaServerService; import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.gb28181.service.IDeviceChannelService;
import com.genersoft.iot.vmp.gb28181.service.IDeviceService;
import com.genersoft.iot.vmp.gb28181.service.IInviteStreamService;
import com.genersoft.iot.vmp.service.ISendRtpServerService; import com.genersoft.iot.vmp.service.ISendRtpServerService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.utils.DateUtil; import com.genersoft.iot.vmp.utils.DateUtil;
@ -32,9 +32,7 @@ import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo; import com.github.pagehelper.PageInfo;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.ObjectUtils; import org.springframework.util.ObjectUtils;
@ -234,10 +232,10 @@ public class DeviceServiceImpl implements IDeviceService {
removeMobilePositionSubscribe(device, null); removeMobilePositionSubscribe(device, null);
List<AudioBroadcastCatch> audioBroadcastCatches = audioBroadcastManager.getByDeviceId(deviceId); List<AudioBroadcastCatch> audioBroadcastCatches = audioBroadcastManager.getByDeviceId(deviceId);
if (audioBroadcastCatches.size() > 0) { if (!audioBroadcastCatches.isEmpty()) {
for (AudioBroadcastCatch audioBroadcastCatch : audioBroadcastCatches) { for (AudioBroadcastCatch audioBroadcastCatch : audioBroadcastCatches) {
SendRtpInfo sendRtpItem = sendRtpServerService.queryByChannelId(audioBroadcastCatch.getChannelId()); SendRtpInfo sendRtpItem = sendRtpServerService.queryByChannelId(audioBroadcastCatch.getChannelId(), deviceId);
if (sendRtpItem != null) { if (sendRtpItem != null) {
sendRtpServerService.delete(sendRtpItem); sendRtpServerService.delete(sendRtpItem);
MediaServer mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); MediaServer mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());

53
src/main/java/com/genersoft/iot/vmp/gb28181/service/impl/PlatformServiceImpl.java
View File

@ -9,7 +9,6 @@ import com.genersoft.iot.vmp.conf.exception.SsrcTransactionNotFoundException;
import com.genersoft.iot.vmp.gb28181.bean.*; import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.dao.PlatformChannelMapper; import com.genersoft.iot.vmp.gb28181.dao.PlatformChannelMapper;
import com.genersoft.iot.vmp.gb28181.dao.PlatformMapper; import com.genersoft.iot.vmp.gb28181.dao.PlatformMapper;
import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe; import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.service.IGbChannelService; import com.genersoft.iot.vmp.gb28181.service.IGbChannelService;
import com.genersoft.iot.vmp.gb28181.service.IInviteStreamService; import com.genersoft.iot.vmp.gb28181.service.IInviteStreamService;
@ -111,20 +110,21 @@ public class PlatformServiceImpl implements IPlatformService {
@Async("taskExecutor") @Async("taskExecutor")
@EventListener @EventListener
public void onApplicationEvent(MediaDepartureEvent event) { public void onApplicationEvent(MediaDepartureEvent event) {
SendRtpInfo sendRtpItems = sendRtpServerService.queryByStream(event.getStream()); List<SendRtpInfo> sendRtpItems = sendRtpServerService.queryByStream(event.getStream());
if (sendRtpItems != null) { if (!sendRtpItems.isEmpty()) {
if (sendRtpItem != null && sendRtpItem.getApp().equals(event.getApp())) { for (SendRtpInfo sendRtpItem : sendRtpItems) {
String platformId = sendRtpItem.getPlatformId(); if (sendRtpItem != null && sendRtpItem.getApp().equals(event.getApp()) && sendRtpItem.isSendToPlatform()) {
Platform platform = platformMapper.getParentPlatByServerGBId(platformId); String platformId = sendRtpItem.getTargetId();
CommonGBChannel channel = channelService.getOne(sendRtpItem.getChannelId()); Platform platform = platformMapper.getParentPlatByServerGBId(platformId);
try { CommonGBChannel channel = channelService.getOne(sendRtpItem.getChannelId());
if (platform != null && channel != null) { try {
commanderForPlatform.streamByeCmd(platform, sendRtpItem, channel); if (platform != null && channel != null) {
redisCatchStorage.deleteSendRTPServer(platformId, channel.getGbDeviceId(), commanderForPlatform.streamByeCmd(platform, sendRtpItem, channel);
sendRtpItem.getCallId(), sendRtpItem.getStream()); sendRtpServerService.delete(sendRtpItem);
}
} catch (SipException | InvalidArgumentException | ParseException e) {
log.error("[命令发送失败] 发送BYE: {}", e.getMessage());
} }
} catch (SipException | InvalidArgumentException | ParseException e) {
log.error("[命令发送失败] 发送BYE: {}", e.getMessage());
} }
} }
} }
@ -137,19 +137,20 @@ public class PlatformServiceImpl implements IPlatformService {
@Async("taskExecutor") @Async("taskExecutor")
@EventListener @EventListener
public void onApplicationEvent(MediaSendRtpStoppedEvent event) { public void onApplicationEvent(MediaSendRtpStoppedEvent event) {
List<SendRtpInfo> sendRtpItems = redisCatchStorage.querySendRTPServerByStream(event.getStream()); List<SendRtpInfo> sendRtpItems = sendRtpServerService.queryByStream(event.getStream());
if (sendRtpItems != null && !sendRtpItems.isEmpty()) { if (sendRtpItems != null && !sendRtpItems.isEmpty()) {
for (SendRtpInfo sendRtpItem : sendRtpItems) { for (SendRtpInfo sendRtpItem : sendRtpItems) {
Platform platform = platformMapper.getParentPlatByServerGBId(sendRtpItem.getPlatformId()); if (sendRtpItem != null && sendRtpItem.getApp().equals(event.getApp()) && sendRtpItem.isSendToPlatform()) {
CommonGBChannel channel = channelService.getOne(sendRtpItem.getChannelId()); Platform platform = platformMapper.getParentPlatByServerGBId(sendRtpItem.getTargetId());
ssrcFactory.releaseSsrc(sendRtpItem.getMediaServerId(), sendRtpItem.getSsrc()); CommonGBChannel channel = channelService.getOne(sendRtpItem.getChannelId());
try { ssrcFactory.releaseSsrc(sendRtpItem.getMediaServerId(), sendRtpItem.getSsrc());
commanderForPlatform.streamByeCmd(platform, sendRtpItem, channel); try {
} catch (SipException | InvalidArgumentException | ParseException e) { commanderForPlatform.streamByeCmd(platform, sendRtpItem, channel);
log.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage()); } catch (SipException | InvalidArgumentException | ParseException e) {
log.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage());
}
sendRtpServerService.delete(sendRtpItem);
} }
redisCatchStorage.deleteSendRTPServer(platform.getServerGBId(), sendRtpItem.getChannelId(),
sendRtpItem.getCallId(), sendRtpItem.getStream());
} }
} }
} }
@ -419,11 +420,11 @@ public class PlatformServiceImpl implements IPlatformService {
} }
private void stopAllPush(String platformId) { private void stopAllPush(String platformId) {
List<SendRtpInfo> sendRtpItems = redisCatchStorage.querySendRTPServer(platformId); List<SendRtpInfo> sendRtpItems = sendRtpServerService.queryForPlatform(platformId);
if (sendRtpItems != null && sendRtpItems.size() > 0) { if (sendRtpItems != null && sendRtpItems.size() > 0) {
for (SendRtpInfo sendRtpItem : sendRtpItems) { for (SendRtpInfo sendRtpItem : sendRtpItems) {
ssrcFactory.releaseSsrc(sendRtpItem.getMediaServerId(), sendRtpItem.getSsrc()); ssrcFactory.releaseSsrc(sendRtpItem.getMediaServerId(), sendRtpItem.getSsrc());
redisCatchStorage.deleteSendRTPServer(platformId, sendRtpItem.getChannelId(), null, null); sendRtpServerService.delete(sendRtpItem);
MediaServer mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); MediaServer mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
mediaServerService.stopSendRtp(mediaInfo, sendRtpItem.getApp(), sendRtpItem.getStream(), null); mediaServerService.stopSendRtp(mediaInfo, sendRtpItem.getApp(), sendRtpItem.getStream(), null);
} }

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

@ -29,6 +29,7 @@ import com.genersoft.iot.vmp.media.event.media.MediaDepartureEvent;
import com.genersoft.iot.vmp.media.event.media.MediaNotFoundEvent; import com.genersoft.iot.vmp.media.event.media.MediaNotFoundEvent;
import com.genersoft.iot.vmp.media.service.IMediaServerService; import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager; import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager;
import com.genersoft.iot.vmp.service.ISendRtpServerService;
import com.genersoft.iot.vmp.service.bean.DownloadFileInfo; import com.genersoft.iot.vmp.service.bean.DownloadFileInfo;
import com.genersoft.iot.vmp.service.bean.ErrorCallback; import com.genersoft.iot.vmp.service.bean.ErrorCallback;
import com.genersoft.iot.vmp.service.bean.InviteErrorCode; import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
@ -126,6 +127,9 @@ public class PlayServiceImpl implements IPlayService {
@Autowired @Autowired
private IGbChannelService channelService; private IGbChannelService channelService;
@Autowired
private ISendRtpServerService sendRtpServerService;
/** /**
* 流到来的处理 * 流到来的处理
*/ */
@ -180,23 +184,23 @@ public class PlayServiceImpl implements IPlayService {
@Async("taskExecutor") @Async("taskExecutor")
@EventListener @EventListener
public void onApplicationEvent(MediaDepartureEvent event) { public void onApplicationEvent(MediaDepartureEvent event) {
List<SendRtpInfo> sendRtpItems = redisCatchStorage.querySendRTPServerByStream(event.getStream()); List<SendRtpInfo> sendRtpInfos = sendRtpServerService.queryByStream(event.getStream());
if (!sendRtpItems.isEmpty()) { if (!sendRtpInfos.isEmpty()) {
for (SendRtpInfo sendRtpItem : sendRtpItems) { for (SendRtpInfo sendRtpInfo : sendRtpInfos) {
if (sendRtpItem != null && sendRtpItem.getApp().equals(event.getApp())) { if (sendRtpInfo != null && sendRtpInfo.isSendToPlatform() && sendRtpInfo.getApp().equals(event.getApp())) {
String platformId = sendRtpItem.getPlatformId(); String platformId = sendRtpInfo.getTargetId();
Device device = deviceService.getDeviceByDeviceId(platformId); Device device = deviceService.getDeviceByDeviceId(platformId);
DeviceChannel channel = deviceChannelService.getOneById(sendRtpItem.getChannelId()); DeviceChannel channel = deviceChannelService.getOneById(sendRtpInfo.getChannelId());
try { try {
if (device != null && channel != null) { if (device != null && channel != null) {
cmder.streamByeCmd(device, channel.getDeviceId(), event.getStream(), sendRtpItem.getCallId()); cmder.streamByeCmd(device, channel.getDeviceId(), event.getStream(), sendRtpInfo.getCallId());
if (sendRtpItem.getPlayType().equals(InviteStreamType.BROADCAST) if (sendRtpInfo.getPlayType().equals(InviteStreamType.BROADCAST)
|| sendRtpItem.getPlayType().equals(InviteStreamType.TALK)) { || sendRtpInfo.getPlayType().equals(InviteStreamType.TALK)) {
AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(channel.getId()); AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(channel.getId());
if (audioBroadcastCatch != null) { if (audioBroadcastCatch != null) {
// 来自上级平台的停止对讲 // 来自上级平台的停止对讲
log.info("[停止对讲] 来自上级，平台：{}, 通道：{}", sendRtpItem.getDeviceId(), sendRtpItem.getChannelId()); log.info("[停止对讲] 来自上级，平台：{}, 通道：{}", sendRtpInfo.getTargetId(), sendRtpInfo.getChannelId());
audioBroadcastManager.del(sendRtpItem.getChannelId()); audioBroadcastManager.del(sendRtpInfo.getChannelId());
} }
} }
} }
@ -362,57 +366,47 @@ public class PlayServiceImpl implements IPlayService {
audioEvent.call("ssrc已经用尽"); audioEvent.call("ssrc已经用尽");
return; return;
} }
SendRtpInfo sendRtpItem = new SendRtpInfo(); SendRtpInfo sendRtpInfo;
sendRtpItem.setApp("talk"); try {
sendRtpItem.setStream(stream); sendRtpInfo = sendRtpServerService.createSendRtpInfo(mediaServerItem, null, null, playSsrc, device.getDeviceId(), "talk", stream,
sendRtpItem.setSsrc(playSsrc); channel.getId(), true, false);
sendRtpItem.setDeviceId(device.getDeviceId()); }catch (PlayException e) {
sendRtpItem.setPlatformId(device.getDeviceId()); log.info("[语音对讲]开始 获取发流端口失败 deviceId: {}, channelId: {},", device.getDeviceId(), channel.getDeviceId());
sendRtpItem.setChannelId(channel.getId());
sendRtpItem.setRtcp(false);
sendRtpItem.setMediaServerId(mediaServerItem.getId());
sendRtpItem.setOnlyAudio(true);
sendRtpItem.setPlayType(InviteStreamType.TALK);
sendRtpItem.setPt(8);
sendRtpItem.setStatus(1);
sendRtpItem.setTcpActive(false);
sendRtpItem.setTcp(true);
sendRtpItem.setUsePs(false);
sendRtpItem.setReceiveStream(stream + "_talk");
String callId = SipUtils.getNewCallId();
int port = sendRtpPortManager.getNextPort(mediaServerItem);
//端口获取失败的ssrcInfo 没有必要发送点播指令
if (port <= 0) {
log.info("[语音对讲] 端口分配异常，deviceId={},channelId={}", device.getDeviceId(), channel.getDeviceId());
audioEvent.call("端口分配异常");
return; return;
} }
sendRtpItem.setLocalPort(port);
sendRtpItem.setPort(port);
log.info("[语音对讲]开始 deviceId: {}, channelId: {},收流端口： {}, 收流模式：{}, SSRC: {}, SSRC校验：{}", device.getDeviceId(), channel.getDeviceId(), sendRtpItem.getLocalPort(), device.getStreamMode(), sendRtpItem.getSsrc(), false); sendRtpInfo.setOnlyAudio(true);
sendRtpInfo.setPt(8);
sendRtpInfo.setStatus(1);
sendRtpInfo.setTcpActive(false);
sendRtpInfo.setUsePs(false);
sendRtpInfo.setReceiveStream(stream + "_talk");
String callId = SipUtils.getNewCallId();
log.info("[语音对讲]开始 deviceId: {}, channelId: {},收流端口： {}, 收流模式：{}, SSRC: {}, SSRC校验：{}", device.getDeviceId(), channel.getDeviceId(), sendRtpInfo.getLocalPort(), device.getStreamMode(), sendRtpInfo.getSsrc(), false);
// 超时处理 // 超时处理
String timeOutTaskKey = UUID.randomUUID().toString(); String timeOutTaskKey = UUID.randomUUID().toString();
dynamicTask.startDelay(timeOutTaskKey, () -> { dynamicTask.startDelay(timeOutTaskKey, () -> {
log.info("[语音对讲] 收流超时 deviceId: {}, channelId: {}，端口：{}, SSRC: {}", device.getDeviceId(), channel.getDeviceId(), sendRtpItem.getPort(), sendRtpItem.getSsrc()); log.info("[语音对讲] 收流超时 deviceId: {}, channelId: {}，端口：{}, SSRC: {}", device.getDeviceId(), channel.getDeviceId(), sendRtpInfo.getPort(), sendRtpInfo.getSsrc());
timeoutCallback.run(); timeoutCallback.run();
// 点播超时回复BYE 同时释放ssrc以及此次点播的资源 // 点播超时回复BYE 同时释放ssrc以及此次点播的资源
try { try {
cmder.streamByeCmd(device, channel.getDeviceId(), sendRtpItem.getStream(), null); cmder.streamByeCmd(device, channel.getDeviceId(), stream, null);
} catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) { } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) {
log.error("[语音对讲]超时， 发送BYE失败 {}", e.getMessage()); log.error("[语音对讲]超时， 发送BYE失败 {}", e.getMessage());
} finally { } finally {
timeoutCallback.run(); timeoutCallback.run();
mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpItem.getSsrc()); mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpInfo.getSsrc());
sessionManager.removeByStream(sendRtpItem.getStream()); sessionManager.removeByStream(sendRtpInfo.getStream());
} }
}, userSetting.getPlayTimeout()); }, userSetting.getPlayTimeout());
try { try {
mediaServerService.startSendRtpPassive(mediaServerItem, sendRtpItem, userSetting.getPlayTimeout() * 1000); mediaServerService.startSendRtpPassive(mediaServerItem, sendRtpInfo, userSetting.getPlayTimeout() * 1000);
}catch (ControllerException e) { }catch (ControllerException e) {
mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpItem.getSsrc()); mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpInfo.getSsrc());
log.info("[语音对讲]失败 deviceId: {}, channelId: {}", device.getDeviceId(), channel.getDeviceId()); log.info("[语音对讲]失败 deviceId: {}, channelId: {}", device.getDeviceId(), channel.getDeviceId());
audioEvent.call("失败, " + e.getMessage()); audioEvent.call("失败, " + e.getMessage());
// 查看是否已经建立了通道，存在则发送bye // 查看是否已经建立了通道，存在则发送bye
@ -422,7 +416,7 @@ public class PlayServiceImpl implements IPlayService {
// 查看设备是否已经在推流 // 查看设备是否已经在推流
try { try {
cmder.talkStreamCmd(mediaServerItem, sendRtpItem, device, channel, callId, (hookData) -> { cmder.talkStreamCmd(mediaServerItem, sendRtpInfo, device, channel, callId, (hookData) -> {
log.info("[语音对讲] 流已生成， 开始推流： " + hookData); log.info("[语音对讲] 流已生成， 开始推流： " + hookData);
dynamicTask.stop(timeOutTaskKey); dynamicTask.stop(timeOutTaskKey);
// TODO 暂不做处理 // TODO 暂不做处理
@ -437,13 +431,13 @@ public class PlayServiceImpl implements IPlayService {
ResponseEvent responseEvent = (ResponseEvent) event.event; ResponseEvent responseEvent = (ResponseEvent) event.event;
if (responseEvent.getResponse() instanceof SIPResponse) { if (responseEvent.getResponse() instanceof SIPResponse) {
SIPResponse response = (SIPResponse) responseEvent.getResponse(); SIPResponse response = (SIPResponse) responseEvent.getResponse();
sendRtpItem.setFromTag(response.getFromTag()); sendRtpInfo.setFromTag(response.getFromTag());
sendRtpItem.setToTag(response.getToTag()); sendRtpInfo.setToTag(response.getToTag());
sendRtpItem.setCallId(response.getCallIdHeader().getCallId()); sendRtpInfo.setCallId(response.getCallIdHeader().getCallId());
redisCatchStorage.updateSendRTPSever(sendRtpItem); sendRtpServerService.update(sendRtpInfo);
SsrcTransaction ssrcTransaction = SsrcTransaction.buildForDevice(device.getDeviceId(), sendRtpItem.getChannelId(), "talk", SsrcTransaction ssrcTransaction = SsrcTransaction.buildForDevice(device.getDeviceId(), sendRtpInfo.getChannelId(), "talk",
sendRtpItem.getStream(), sendRtpItem.getSsrc(), sendRtpItem.getMediaServerId(), sendRtpInfo.getStream(), sendRtpInfo.getSsrc(), sendRtpInfo.getMediaServerId(),
response, InviteSessionType.TALK); response, InviteSessionType.TALK);
sessionManager.put(ssrcTransaction); sessionManager.put(ssrcTransaction);
@ -456,21 +450,21 @@ public class PlayServiceImpl implements IPlayService {
}, (event) -> { }, (event) -> {
dynamicTask.stop(timeOutTaskKey); dynamicTask.stop(timeOutTaskKey);
mediaServerService.closeRTPServer(mediaServerItem, sendRtpItem.getStream()); mediaServerService.closeRTPServer(mediaServerItem, sendRtpInfo.getStream());
// 释放ssrc // 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpItem.getSsrc()); mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpInfo.getSsrc());
sessionManager.removeByStream(sendRtpItem.getStream()); sessionManager.removeByStream(sendRtpInfo.getStream());
errorEvent.response(event); errorEvent.response(event);
}); });
} catch (InvalidArgumentException | SipException | ParseException e) { } catch (InvalidArgumentException | SipException | ParseException e) {
log.error("[命令发送失败] 对讲消息: {}", e.getMessage()); log.error("[命令发送失败] 对讲消息: {}", e.getMessage());
dynamicTask.stop(timeOutTaskKey); dynamicTask.stop(timeOutTaskKey);
mediaServerService.closeRTPServer(mediaServerItem, sendRtpItem.getStream()); mediaServerService.closeRTPServer(mediaServerItem, sendRtpInfo.getStream());
// 释放ssrc // 释放ssrc
mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpItem.getSsrc()); mediaServerService.releaseSsrc(mediaServerItem.getId(), sendRtpInfo.getSsrc());
sessionManager.removeByStream(sendRtpItem.getStream()); sessionManager.removeByStream(sendRtpInfo.getStream());
SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(); SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult();
eventResult.type = SipSubscribe.EventResultType.cmdSendFailEvent; eventResult.type = SipSubscribe.EventResultType.cmdSendFailEvent;
eventResult.statusCode = -1; eventResult.statusCode = -1;
@ -1138,14 +1132,14 @@ public class PlayServiceImpl implements IPlayService {
@Override @Override
public void zlmServerOffline(String mediaServerId) { public void zlmServerOffline(String mediaServerId) {
// 处理正在向上推流的上级平台 // 处理正在向上推流的上级平台
List<SendRtpInfo> sendRtpItems = redisCatchStorage.querySendRTPServer(null); List<SendRtpInfo> sendRtpInfos = sendRtpServerService.queryAll();
if (sendRtpItems.size() > 0) { if (!sendRtpInfos.isEmpty()) {
for (SendRtpInfo sendRtpItem : sendRtpItems) { for (SendRtpInfo sendRtpInfo : sendRtpInfos) {
if (sendRtpItem.getMediaServerId().equals(mediaServerId)) { if (sendRtpInfo.getMediaServerId().equals(mediaServerId) && sendRtpInfo.isSendToPlatform()) {
Platform platform = platformService.queryPlatformByServerGBId(sendRtpItem.getPlatformId()); Platform platform = platformService.queryPlatformByServerGBId(sendRtpInfo.getTargetId());
CommonGBChannel channel = channelService.getOne(sendRtpItem.getChannelId()); CommonGBChannel channel = channelService.getOne(sendRtpInfo.getChannelId());
try { try {
sipCommanderFroPlatform.streamByeCmd(platform, sendRtpItem, channel); sipCommanderFroPlatform.streamByeCmd(platform, sendRtpInfo, channel);
} catch (SipException | InvalidArgumentException | ParseException e) { } catch (SipException | InvalidArgumentException | ParseException e) {
log.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage()); log.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage());
} }
@ -1216,10 +1210,10 @@ public class PlayServiceImpl implements IPlayService {
} }
// 查询通道使用状态 // 查询通道使用状态
if (audioBroadcastManager.exit(deviceChannel.getId())) { if (audioBroadcastManager.exit(deviceChannel.getId())) {
SendRtpInfo sendRtpItem = redisCatchStorage.querySendRTPServer(device.getDeviceId(), channel.getDeviceId(), null, null); SendRtpInfo sendRtpInfo = sendRtpServerService.queryByChannelId(channel.getId(), device.getDeviceId());
if (sendRtpItem != null && sendRtpItem.isOnlyAudio()) { if (sendRtpInfo != null && sendRtpInfo.isOnlyAudio()) {
// 查询流是否存在，不存在则认为是异常状态 // 查询流是否存在，不存在则认为是异常状态
Boolean streamReady = mediaServerService.isStreamReady(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream()); Boolean streamReady = mediaServerService.isStreamReady(mediaServerItem, sendRtpInfo.getApp(), sendRtpInfo.getStream());
if (streamReady) { if (streamReady) {
log.warn("语音广播已经开启： {}", channel.getDeviceId()); log.warn("语音广播已经开启： {}", channel.getDeviceId());
event.call("语音广播已经开启"); event.call("语音广播已经开启");
@ -1256,11 +1250,11 @@ public class PlayServiceImpl implements IPlayService {
@Override @Override
public boolean audioBroadcastInUse(Device device, DeviceChannel channel) { public boolean audioBroadcastInUse(Device device, DeviceChannel channel) {
if (audioBroadcastManager.exit(channel.getId())) { if (audioBroadcastManager.exit(channel.getId())) {
SendRtpInfo sendRtpItem = redisCatchStorage.querySendRTPServer(device.getDeviceId(), channel.getDeviceId(), null, null); SendRtpInfo sendRtpInfo = sendRtpServerService.queryByChannelId(channel.getId(), device.getDeviceId());
if (sendRtpItem != null && sendRtpItem.isOnlyAudio()) { if (sendRtpInfo != null && sendRtpInfo.isOnlyAudio()) {
// 查询流是否存在，不存在则认为是异常状态 // 查询流是否存在，不存在则认为是异常状态
MediaServer mediaServerServiceOne = mediaServerService.getOne(sendRtpItem.getMediaServerId()); MediaServer mediaServerServiceOne = mediaServerService.getOne(sendRtpInfo.getMediaServerId());
Boolean streamReady = mediaServerService.isStreamReady(mediaServerServiceOne, sendRtpItem.getApp(), sendRtpItem.getStream()); Boolean streamReady = mediaServerService.isStreamReady(mediaServerServiceOne, sendRtpInfo.getApp(), sendRtpInfo.getStream());
if (streamReady) { if (streamReady) {
log.warn("语音广播通道使用中： {}", channel.getDeviceId()); log.warn("语音广播通道使用中： {}", channel.getDeviceId());
return true; return true;
@ -1285,11 +1279,11 @@ public class PlayServiceImpl implements IPlayService {
if (audioBroadcastCatch == null) { if (audioBroadcastCatch == null) {
continue; continue;
} }
SendRtpInfo sendRtpItem = redisCatchStorage.querySendRTPServer(device.getDeviceId(), channel.getDeviceId(), null, null); SendRtpInfo sendRtpInfo = sendRtpServerService.queryByChannelId(channel.getId(), device.getDeviceId());
if (sendRtpItem != null) { if (sendRtpInfo != null) {
redisCatchStorage.deleteSendRTPServer(device.getDeviceId(), channel.getDeviceId(), null, null); sendRtpServerService.delete(sendRtpInfo);
MediaServer mediaServer = mediaServerService.getOne(sendRtpItem.getMediaServerId()); MediaServer mediaServer = mediaServerService.getOne(sendRtpInfo.getMediaServerId());
mediaServerService.stopSendRtp(mediaServer, sendRtpItem.getApp(), sendRtpItem.getStream(), null); mediaServerService.stopSendRtp(mediaServer, sendRtpInfo.getApp(), sendRtpInfo.getStream(), null);
try { try {
cmder.streamByeCmdForDeviceInvite(device, channel.getDeviceId(), audioBroadcastCatch.getSipTransactionInfo(), null); cmder.streamByeCmdForDeviceInvite(device, channel.getDeviceId(), audioBroadcastCatch.getSipTransactionInfo(), null);
} catch (InvalidArgumentException | ParseException | SipException | } catch (InvalidArgumentException | ParseException | SipException |
@ -1305,54 +1299,6 @@ public class PlayServiceImpl implements IPlayService {
@Override @Override
public void zlmServerOnline(String mediaServerId) { public void zlmServerOnline(String mediaServerId) {
// TODO 查找之前的点播，流如果不存在则给下级发送bye
// MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
// zlmresTfulUtils.getMediaList(mediaServerItem, (mediaList ->{
// Integer code = mediaList.getInteger("code");
// if (code == 0) {
// JSONArray data = mediaList.getJSONArray("data");
// if (data == null || data.size() == 0) {
// zlmServerOffline(mediaServerId);
// }else {
// Map<String, JSONObject> mediaListMap = new HashMap<>();
// for (int i = 0; i < data.size(); i++) {
// JSONObject json = data.getJSONObject(i);
// String app = json.getString("app");
// if ("rtp".equals(app)) {
// String stream = json.getString("stream");
// if (mediaListMap.get(stream) != null) {
// continue;
// }
// mediaListMap.put(stream, json);
// // 处理正在观看的国标设备
// List<SsrcTransaction> ssrcTransactions = streamSession.getSsrcTransactionForAll(null, null, null, stream);
// if (ssrcTransactions.size() > 0) {
// for (SsrcTransaction ssrcTransaction : ssrcTransactions) {
// if(ssrcTransaction.getMediaServerId().equals(mediaServerId)) {
// cmder.streamByeCmd(ssrcTransaction.getDeviceId(), ssrcTransaction.getChannelId(),
// ssrcTransaction.getStream(), null);
// }
// }
// }
// }
// }
// if (mediaListMap.size() > 0 ) {
// // 处理正在向上推流的上级平台
// List<SendRtpItem> sendRtpItems = redisCatchStorage.querySendRTPServer(null);
// if (sendRtpItems.size() > 0) {
// for (SendRtpItem sendRtpItem : sendRtpItems) {
// if (sendRtpItem.getMediaServerId().equals(mediaServerId)) {
// if (mediaListMap.get(sendRtpItem.getStreamId()) == null) {
// ParentPlatform platform = storager.queryPlatformByServerGBId(sendRtpItem.getPlatformId());
// sipCommanderFroPlatform.streamByeCmd(platform, sendRtpItem.getCallId());
// }
// }
// }
// }
// }
// }
// }
// }));
} }
@Override @Override
@ -1414,36 +1360,36 @@ public class PlayServiceImpl implements IPlayService {
} }
@Override @Override
public void startPushStream(SendRtpInfo sendRtpItem, DeviceChannel channel, SIPResponse sipResponse, Platform platform, CallIdHeader callIdHeader) { public void startPushStream(SendRtpInfo sendRtpInfo, DeviceChannel channel, SIPResponse sipResponse, Platform platform, CallIdHeader callIdHeader) {
// 开始发流 // 开始发流
MediaServer mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); MediaServer mediaInfo = mediaServerService.getOne(sendRtpInfo.getMediaServerId());
if (mediaInfo != null) { if (mediaInfo != null) {
try { try {
if (sendRtpItem.isTcpActive()) { if (sendRtpInfo.isTcpActive()) {
mediaServerService.startSendRtpPassive(mediaInfo, sendRtpItem, null); mediaServerService.startSendRtpPassive(mediaInfo, sendRtpInfo, null);
} else { } else {
mediaServerService.startSendRtp(mediaInfo, sendRtpItem); mediaServerService.startSendRtp(mediaInfo, sendRtpInfo);
} }
redisCatchStorage.sendPlatformStartPlayMsg(sendRtpItem, channel, platform); redisCatchStorage.sendPlatformStartPlayMsg(sendRtpInfo, channel, platform);
}catch (ControllerException e) { }catch (ControllerException e) {
log.error("RTP推流失败: {}", e.getMessage()); log.error("RTP推流失败: {}", e.getMessage());
startSendRtpStreamFailHand(sendRtpItem, platform, callIdHeader); startSendRtpStreamFailHand(sendRtpInfo, platform, callIdHeader);
return; return;
} }
log.info("RTP推流成功[ {}/{} ]，{}, ", sendRtpItem.getApp(), sendRtpItem.getStream(), log.info("RTP推流成功[ {}/{} ]，{}, ", sendRtpInfo.getApp(), sendRtpInfo.getStream(),
sendRtpItem.isTcpActive()?"被动发流": sendRtpItem.getIp() + ":" + sendRtpItem.getPort()); sendRtpInfo.isTcpActive()?"被动发流": sendRtpInfo.getIp() + ":" + sendRtpInfo.getPort());
} }
} }
@Override @Override
public void startSendRtpStreamFailHand(SendRtpInfo sendRtpItem, Platform platform, CallIdHeader callIdHeader) { public void startSendRtpStreamFailHand(SendRtpInfo sendRtpInfo, Platform platform, CallIdHeader callIdHeader) {
if (sendRtpItem.isOnlyAudio()) { if (sendRtpInfo.isOnlyAudio()) {
Device device = deviceService.getDeviceByDeviceId(sendRtpItem.getDeviceId()); Device device = deviceService.getDeviceByDeviceId(sendRtpInfo.getTargetId());
DeviceChannel deviceChannel = deviceChannelService.getOneById(sendRtpItem.getChannelId()); DeviceChannel deviceChannel = deviceChannelService.getOneById(sendRtpInfo.getChannelId());
AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(sendRtpItem.getChannelId()); AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(sendRtpInfo.getChannelId());
if (audioBroadcastCatch != null) { if (audioBroadcastCatch != null) {
try { try {
cmder.streamByeCmd(device, deviceChannel.getDeviceId(), audioBroadcastCatch.getSipTransactionInfo(), null); cmder.streamByeCmd(device, deviceChannel.getDeviceId(), audioBroadcastCatch.getSipTransactionInfo(), null);
@ -1455,9 +1401,9 @@ public class PlayServiceImpl implements IPlayService {
} else { } else {
if (platform != null) { if (platform != null) {
// 向上级平台 // 向上级平台
CommonGBChannel channel = channelService.getOne(sendRtpItem.getChannelId()); CommonGBChannel channel = channelService.getOne(sendRtpInfo.getChannelId());
try { try {
commanderForPlatform.streamByeCmd(platform, sendRtpItem, channel); commanderForPlatform.streamByeCmd(platform, sendRtpInfo, channel);
} catch (SipException | InvalidArgumentException | ParseException e) { } catch (SipException | InvalidArgumentException | ParseException e) {
log.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage()); log.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage());
} }
@ -1475,11 +1421,11 @@ public class PlayServiceImpl implements IPlayService {
log.info("[语音对讲] device： {}, channel: {}", device.getDeviceId(), channel.getDeviceId()); log.info("[语音对讲] device： {}, channel: {}", device.getDeviceId(), channel.getDeviceId());
// 查询通道使用状态 // 查询通道使用状态
if (audioBroadcastManager.exit(channel.getId())) { if (audioBroadcastManager.exit(channel.getId())) {
SendRtpInfo sendRtpItem = redisCatchStorage.querySendRTPServer(device.getDeviceId(), channel.getDeviceId(), null, null); SendRtpInfo sendRtpInfo = sendRtpServerService.queryByChannelId(channel.getId(), device.getDeviceId());
if (sendRtpItem != null && sendRtpItem.isOnlyAudio()) { if (sendRtpInfo != null && sendRtpInfo.isOnlyAudio()) {
// 查询流是否存在，不存在则认为是异常状态 // 查询流是否存在，不存在则认为是异常状态
MediaServer mediaServer = mediaServerService.getOne(sendRtpItem.getMediaServerId()); MediaServer mediaServer = mediaServerService.getOne(sendRtpInfo.getMediaServerId());
Boolean streamReady = mediaServerService.isStreamReady(mediaServer, sendRtpItem.getApp(), sendRtpItem.getStream()); Boolean streamReady = mediaServerService.isStreamReady(mediaServer, sendRtpInfo.getApp(), sendRtpInfo.getStream());
if (streamReady) { if (streamReady) {
log.warn("[语音对讲] 正在语音广播，无法开启语音通话： {}", channel.getDeviceId()); log.warn("[语音对讲] 正在语音广播，无法开启语音通话： {}", channel.getDeviceId());
event.call("正在语音广播"); event.call("正在语音广播");
@ -1490,10 +1436,10 @@ public class PlayServiceImpl implements IPlayService {
} }
} }
SendRtpInfo sendRtpItem = redisCatchStorage.querySendRTPServer(device.getDeviceId(), channel.getDeviceId(), stream, null); SendRtpInfo sendRtpInfo = sendRtpServerService.queryByChannelId(channel.getId(), device.getDeviceId());
if (sendRtpItem != null) { if (sendRtpInfo != null) {
MediaServer mediaServer = mediaServerService.getOne(sendRtpItem.getMediaServerId()); MediaServer mediaServer = mediaServerService.getOne(sendRtpInfo.getMediaServerId());
Boolean streamReady = mediaServerService.isStreamReady(mediaServer, "rtp", sendRtpItem.getReceiveStream()); Boolean streamReady = mediaServerService.isStreamReady(mediaServer, "rtp", sendRtpInfo.getReceiveStream());
if (streamReady) { if (streamReady) {
log.warn("[语音对讲] 进行中： {}", channel.getDeviceId()); log.warn("[语音对讲] 进行中： {}", channel.getDeviceId());
event.call("语音对讲进行中"); event.call("语音对讲进行中");
@ -1526,13 +1472,13 @@ public class PlayServiceImpl implements IPlayService {
@Override @Override
public void stopTalk(Device device, DeviceChannel channel, Boolean streamIsReady) { public void stopTalk(Device device, DeviceChannel channel, Boolean streamIsReady) {
log.info("[语音对讲] 停止， {}/{}", device.getDeviceId(), channel.getDeviceId()); log.info("[语音对讲] 停止， {}/{}", device.getDeviceId(), channel.getDeviceId());
SendRtpInfo sendRtpItem = redisCatchStorage.querySendRTPServer(device.getDeviceId(), channel.getDeviceId(), null, null); SendRtpInfo sendRtpInfo = sendRtpServerService.queryByChannelId(channel.getId(), device.getDeviceId());
if (sendRtpItem == null) { if (sendRtpInfo == null) {
log.info("[语音对讲] 停止失败， 未找到发送信息，可能已经停止"); log.info("[语音对讲] 停止失败， 未找到发送信息，可能已经停止");
return; return;
} }
// 停止向设备推流 // 停止向设备推流
String mediaServerId = sendRtpItem.getMediaServerId(); String mediaServerId = sendRtpInfo.getMediaServerId();
if (mediaServerId == null) { if (mediaServerId == null) {
return; return;
} }
@ -1540,20 +1486,20 @@ public class PlayServiceImpl implements IPlayService {
MediaServer mediaServer = mediaServerService.getOne(mediaServerId); MediaServer mediaServer = mediaServerService.getOne(mediaServerId);
if (streamIsReady == null || streamIsReady) { if (streamIsReady == null || streamIsReady) {
mediaServerService.stopSendRtp(mediaServer, sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getSsrc()); mediaServerService.stopSendRtp(mediaServer, sendRtpInfo.getApp(), sendRtpInfo.getStream(), sendRtpInfo.getSsrc());
} }
ssrcFactory.releaseSsrc(mediaServerId, sendRtpItem.getSsrc()); ssrcFactory.releaseSsrc(mediaServerId, sendRtpInfo.getSsrc());
SsrcTransaction ssrcTransaction = sessionManager.getSsrcTransactionByStream(sendRtpItem.getStream()); SsrcTransaction ssrcTransaction = sessionManager.getSsrcTransactionByStream(sendRtpInfo.getStream());
if (ssrcTransaction != null) { if (ssrcTransaction != null) {
try { try {
cmder.streamByeCmd(device, channel.getDeviceId(), sendRtpItem.getStream(), null); cmder.streamByeCmd(device, channel.getDeviceId(), sendRtpInfo.getStream(), null);
} catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) { } catch (InvalidArgumentException | ParseException | SipException | SsrcTransactionNotFoundException e) {
log.info("[语音对讲] 停止消息发送失败，可能已经停止"); log.info("[语音对讲] 停止消息发送失败，可能已经停止");
} }
} }
redisCatchStorage.deleteSendRTPServer(device.getDeviceId(), channel.getDeviceId(),null, null); sendRtpServerService.deleteByChannel(channel.getId(), device.getDeviceId());
} }
@Override @Override

7
src/main/java/com/genersoft/iot/vmp/gb28181/task/SipRunner.java
View File

@ -4,15 +4,14 @@ import com.genersoft.iot.vmp.gb28181.bean.CommonGBChannel;
import com.genersoft.iot.vmp.gb28181.bean.Device; import com.genersoft.iot.vmp.gb28181.bean.Device;
import com.genersoft.iot.vmp.gb28181.bean.Platform; import com.genersoft.iot.vmp.gb28181.bean.Platform;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpInfo; import com.genersoft.iot.vmp.gb28181.bean.SendRtpInfo;
import com.genersoft.iot.vmp.gb28181.service.IDeviceService;
import com.genersoft.iot.vmp.gb28181.service.IGbChannelService; import com.genersoft.iot.vmp.gb28181.service.IGbChannelService;
import com.genersoft.iot.vmp.gb28181.service.IPlatformService;
import com.genersoft.iot.vmp.gb28181.session.SSRCFactory; import com.genersoft.iot.vmp.gb28181.session.SSRCFactory;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform; import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.media.bean.MediaServer; import com.genersoft.iot.vmp.media.bean.MediaServer;
import com.genersoft.iot.vmp.media.service.IMediaServerService; import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.gb28181.service.IDeviceService;
import com.genersoft.iot.vmp.gb28181.service.IPlatformService;
import com.genersoft.iot.vmp.service.ISendRtpServerService; import com.genersoft.iot.vmp.service.ISendRtpServerService;
import com.genersoft.iot.vmp.service.impl.SendRtpServerServiceImpl;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
@ -109,7 +108,7 @@ public class SipRunner implements CommandLineRunner {
ssrcFactory.releaseSsrc(sendRtpItem.getMediaServerId(), sendRtpItem.getSsrc()); ssrcFactory.releaseSsrc(sendRtpItem.getMediaServerId(), sendRtpItem.getSsrc());
boolean stopResult = mediaServerService.initStopSendRtp(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getSsrc()); boolean stopResult = mediaServerService.initStopSendRtp(mediaServerItem, sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getSsrc());
if (stopResult) { if (stopResult) {
Platform platform = platformService.queryPlatformByServerGBId(sendRtpItem.getPlatformId()); Platform platform = platformService.queryPlatformByServerGBId(sendRtpItem.getTargetId());
if (platform != null) { if (platform != null) {
try { try {

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

@ -10,7 +10,6 @@ import com.genersoft.iot.vmp.gb28181.session.AudioBroadcastManager;
import com.genersoft.iot.vmp.gb28181.session.SipInviteSessionManager; import com.genersoft.iot.vmp.gb28181.session.SipInviteSessionManager;
import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver; import com.genersoft.iot.vmp.gb28181.transmit.SIPProcessorObserver;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander; import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommander;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor; 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.gb28181.transmit.event.request.SIPRequestProcessorParent;
import com.genersoft.iot.vmp.media.bean.MediaInfo; import com.genersoft.iot.vmp.media.bean.MediaInfo;
@ -118,14 +117,14 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
// 收流端发送的停止 // 收流端发送的停止
if (sendRtpItem != null){ if (sendRtpItem != null){
CommonGBChannel channel = channelService.getOne(sendRtpItem.getChannelId()); CommonGBChannel channel = channelService.getOne(sendRtpItem.getChannelId());
log.info("[收到bye] 来自{}，停止通道：{}, 类型： {}, callId: {}", sendRtpItem.getPlatformId(), channel.getGbDeviceId(), sendRtpItem.getPlayType(), callIdHeader.getCallId()); log.info("[收到bye] 来自{}，停止通道：{}, 类型： {}, callId: {}", sendRtpItem.getTargetId(), channel.getGbDeviceId(), sendRtpItem.getPlayType(), callIdHeader.getCallId());
String streamId = sendRtpItem.getStream(); String streamId = sendRtpItem.getStream();
log.info("[收到bye] 停止推流：{}, 媒体节点： {}", streamId, sendRtpItem.getMediaServerId()); log.info("[收到bye] 停止推流：{}, 媒体节点： {}", streamId, sendRtpItem.getMediaServerId());
if (sendRtpItem.getPlayType().equals(InviteStreamType.PUSH)) { if (sendRtpItem.getPlayType().equals(InviteStreamType.PUSH)) {
// 不是本平台的就发送redis消息让其他wvp停止发流 // 不是本平台的就发送redis消息让其他wvp停止发流
Platform platform = platformService.queryPlatformByServerGBId(sendRtpItem.getPlatformId()); Platform platform = platformService.queryPlatformByServerGBId(sendRtpItem.getTargetId());
if (platform != null) { if (platform != null) {
redisCatchStorage.sendPlatformStopPlayMsg(sendRtpItem, platform, channel); redisCatchStorage.sendPlatformStopPlayMsg(sendRtpItem, platform, channel);
if (!userSetting.getServerId().equals(sendRtpItem.getServerId())) { if (!userSetting.getServerId().equals(sendRtpItem.getServerId())) {
@ -140,7 +139,7 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
} }
} }
}else { }else {
log.info("[上级平台停止观看] 未找到平台{}的信息，发送redis消息失败", sendRtpItem.getPlatformId()); log.info("[上级平台停止观看] 未找到平台{}的信息，发送redis消息失败", sendRtpItem.getTargetId());
} }
}else { }else {
MediaServer mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId()); MediaServer mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
@ -155,7 +154,7 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(sendRtpItem.getChannelId()); AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(sendRtpItem.getChannelId());
if (audioBroadcastCatch != null && audioBroadcastCatch.getSipTransactionInfo().getCallId().equals(callIdHeader.getCallId())) { if (audioBroadcastCatch != null && audioBroadcastCatch.getSipTransactionInfo().getCallId().equals(callIdHeader.getCallId())) {
// 来自上级平台的停止对讲 // 来自上级平台的停止对讲
log.info("[停止对讲] 来自上级，平台：{}, 通道：{}", sendRtpItem.getDeviceId(), sendRtpItem.getChannelId()); log.info("[停止对讲] 来自上级，平台：{}, 通道：{}", sendRtpItem.getTargetId(), sendRtpItem.getChannelId());
audioBroadcastManager.del(sendRtpItem.getChannelId()); audioBroadcastManager.del(sendRtpItem.getChannelId());
} }
@ -164,7 +163,7 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
if (mediaInfo.getReaderCount() <= 0) { if (mediaInfo.getReaderCount() <= 0) {
log.info("[收到bye] {} 无其它观看者，通知设备停止推流", streamId); log.info("[收到bye] {} 无其它观看者，通知设备停止推流", streamId);
if (sendRtpItem.getPlayType().equals(InviteStreamType.PLAY)) { if (sendRtpItem.getPlayType().equals(InviteStreamType.PLAY)) {
Device device = deviceService.getDeviceByDeviceId(sendRtpItem.getDeviceId()); Device device = deviceService.getDeviceByDeviceId(sendRtpItem.getTargetId());
if (device == null) { if (device == null) {
log.info("[收到bye] {} 通知设备停止推流时未找到设备信息", streamId); log.info("[收到bye] {} 通知设备停止推流时未找到设备信息", streamId);
return; return;
@ -175,7 +174,7 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
return; return;
} }
try { try {
log.info("[停止点播] {}/{}", sendRtpItem.getDeviceId(), sendRtpItem.getChannelId()); log.info("[停止点播] {}/{}", sendRtpItem.getTargetId(), sendRtpItem.getChannelId());
cmder.streamByeCmd(device, deviceChannel.getDeviceId(), streamId, null); cmder.streamByeCmd(device, deviceChannel.getDeviceId(), streamId, null);
} catch (InvalidArgumentException | ParseException | SipException | } catch (InvalidArgumentException | ParseException | SipException |
SsrcTransactionNotFoundException e) { SsrcTransactionNotFoundException e) {

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

@ -25,7 +25,6 @@ import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
import com.genersoft.iot.vmp.service.redisMsg.RedisPushStreamResponseListener; import com.genersoft.iot.vmp.service.redisMsg.RedisPushStreamResponseListener;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import com.genersoft.iot.vmp.streamProxy.service.IStreamProxyService;
import gov.nist.javax.sdp.TimeDescriptionImpl; import gov.nist.javax.sdp.TimeDescriptionImpl;
import gov.nist.javax.sdp.fields.TimeField; import gov.nist.javax.sdp.fields.TimeField;
import gov.nist.javax.sdp.fields.URIField; import gov.nist.javax.sdp.fields.URIField;
@ -893,7 +892,6 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
sendRtpItem.setPlayType(InviteStreamType.BROADCAST); sendRtpItem.setPlayType(InviteStreamType.BROADCAST);
sendRtpItem.setCallId(callIdHeader.getCallId()); sendRtpItem.setCallId(callIdHeader.getCallId());
sendRtpItem.setPlatformId(requesterId);
sendRtpItem.setStatus(1); sendRtpItem.setStatus(1);
sendRtpItem.setApp(broadcastCatch.getApp()); sendRtpItem.setApp(broadcastCatch.getApp());
sendRtpItem.setStream(broadcastCatch.getStream()); sendRtpItem.setStream(broadcastCatch.getStream());

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

@ -151,7 +151,7 @@ public class BroadcastNotifyMessageHandler extends SIPRequestProcessorParent imp
broadcastCatch.setMediaServerItem(hookData.getMediaServer()); broadcastCatch.setMediaServerItem(hookData.getMediaServer());
audioBroadcastManager.update(broadcastCatch); audioBroadcastManager.update(broadcastCatch);
// 推流到设备 // 推流到设备
SendRtpInfo sendRtpItem = sendRtpServerService.queryByStream(null, targetId, hookData.getStream(), null); SendRtpInfo sendRtpItem = sendRtpServerService.queryByStream(hookData.getStream(), targetId);
if (sendRtpItem == null) { if (sendRtpItem == null) {
log.warn("[国标级联] 语音喊话 异常，未找到发流信息， channelId: {}, stream: {}", targetId, hookData.getStream()); log.warn("[国标级联] 语音喊话 异常，未找到发流信息， channelId: {}, stream: {}", targetId, hookData.getStream());
log.info("[国标级联] 语音喊话 重新开始，channelId: {}, stream: {}", targetId, hookData.getStream()); log.info("[国标级联] 语音喊话 重新开始，channelId: {}, stream: {}", targetId, hookData.getStream());

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

@ -114,11 +114,11 @@ public class MediaStatusNotifyMessageHandler extends SIPRequestProcessorParent i
Hook hook = Hook.getInstance(HookType.on_media_arrival, "rtp", ssrcTransaction.getStream(), ssrcTransaction.getMediaServerId()); Hook hook = Hook.getInstance(HookType.on_media_arrival, "rtp", ssrcTransaction.getStream(), ssrcTransaction.getMediaServerId());
subscribe.removeSubscribe(hook); subscribe.removeSubscribe(hook);
// 如果级联播放，需要给上级发送此通知 TODO 多个上级同时观看一个下级 可能存在停错的问题，需要将点播CallId进行上下级绑定 // 如果级联播放，需要给上级发送此通知 TODO 多个上级同时观看一个下级 可能存在停错的问题，需要将点播CallId进行上下级绑定
SendRtpInfo sendRtpItem = sendRtpServerService.queryByChannelId(ssrcTransaction.getChannelId()); SendRtpInfo sendRtpItem = sendRtpServerService.queryByChannelId(ssrcTransaction.getChannelId(), ssrcTransaction.getPlatformId());
if (sendRtpItem != null) { if (sendRtpItem != null) {
Platform parentPlatform = platformService.queryPlatformByServerGBId(sendRtpItem.getPlatformId()); Platform parentPlatform = platformService.queryPlatformByServerGBId(sendRtpItem.getTargetId());
if (parentPlatform == null) { if (parentPlatform == null) {
log.warn("[级联消息发送]：发送MediaStatus发现上级平台{}不存在", sendRtpItem.getPlatformId()); log.warn("[级联消息发送]：发送MediaStatus发现上级平台{}不存在", sendRtpItem.getTargetId());
return; return;
} }
try { try {

24
src/main/java/com/genersoft/iot/vmp/service/ISendRtpServerService.java
View File

@ -7,29 +7,37 @@ import java.util.List;
public interface ISendRtpServerService { public interface ISendRtpServerService {
SendRtpInfo createSendRtpInfo(MediaServer mediaServer, String ip, int port, String ssrc, String requesterId, SendRtpInfo createSendRtpInfo(MediaServer mediaServer, String ip, Integer port, String ssrc, String requesterId,
String deviceId, Integer channelId, boolean isTcp, boolean rtcp); String deviceId, Integer channelId, Boolean isTcp, Boolean rtcp);
SendRtpInfo createSendRtpInfo(MediaServer mediaServer, String ip, int port, String ssrc, String platformId, SendRtpInfo createSendRtpInfo(MediaServer mediaServer, String ip, Integer port, String ssrc, String platformId,
String app, String stream, Integer channelId, boolean tcp, boolean rtcp); String app, String stream, Integer channelId, Boolean tcp, Boolean rtcp);
void update(SendRtpInfo sendRtpItem); void update(SendRtpInfo sendRtpItem);
SendRtpInfo queryByChannelId(Integer channelId); SendRtpInfo queryByChannelId(Integer channelId, String targetId);
SendRtpInfo queryByCallId(String callId); SendRtpInfo queryByCallId(String callId);
SendRtpInfo queryByStream(String stream); List<SendRtpInfo> queryByStream(String stream);
SendRtpInfo queryByStream(String stream, String targetId);
void delete(SendRtpInfo sendRtpInfo); void delete(SendRtpInfo sendRtpInfo);
void deleteByCallId(String callId); void deleteByCallId(String callId);
void deleteByStream(String Stream); void deleteByStream(String Stream, String targetId);
void deleteByChannel(Integer channelId); void deleteByChannel(Integer channelId, String targetId);
List<SendRtpInfo> queryAll(); List<SendRtpInfo> queryAll();
boolean isChannelSendingRTP(Integer channelId); boolean isChannelSendingRTP(Integer channelId);
List<SendRtpInfo> queryForPlatform(String platformId);
List<SendRtpInfo> queryByChannelId(int id);
void deleteByStream(String stream);
} }

17
src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java
View File

@ -15,6 +15,7 @@ import com.genersoft.iot.vmp.media.bean.MediaServer;
import com.genersoft.iot.vmp.media.bean.ResultForOnPublish; import com.genersoft.iot.vmp.media.bean.ResultForOnPublish;
import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo; import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
import com.genersoft.iot.vmp.service.IMediaService; import com.genersoft.iot.vmp.service.IMediaService;
import com.genersoft.iot.vmp.service.ISendRtpServerService;
import com.genersoft.iot.vmp.service.IUserService; import com.genersoft.iot.vmp.service.IUserService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.streamProxy.bean.StreamProxy; import com.genersoft.iot.vmp.streamProxy.bean.StreamProxy;
@ -78,6 +79,9 @@ public class MediaServiceImpl implements IMediaService {
@Autowired @Autowired
private ISIPCommander commander; private ISIPCommander commander;
@Autowired
private ISendRtpServerService sendRtpServerService;
@Override @Override
public boolean authenticatePlay(String app, String stream, String callId) { public boolean authenticatePlay(String app, String stream, String callId) {
if (app == null || stream == null) { if (app == null || stream == null) {
@ -234,11 +238,11 @@ public class MediaServiceImpl implements IMediaService {
return false; return false;
} }
// 收到无人观看说明流也没有在往上级推送 // 收到无人观看说明流也没有在往上级推送
if (redisCatchStorage.isChannelSendingRTP(deviceChannel.getDeviceId())) { if (sendRtpServerService.isChannelSendingRTP(deviceChannel.getId())) {
List<SendRtpInfo> sendRtpItems = redisCatchStorage.querySendRTPServerByChannelId(deviceChannel.getDeviceId()); List<SendRtpInfo> sendRtpItems = sendRtpServerService.queryByChannelId(deviceChannel.getId());
if (!sendRtpItems.isEmpty()) { if (!sendRtpItems.isEmpty()) {
for (SendRtpInfo sendRtpItem : sendRtpItems) { for (SendRtpInfo sendRtpItem : sendRtpItems) {
Platform parentPlatform = platformService.queryPlatformByServerGBId(sendRtpItem.getPlatformId()); Platform parentPlatform = platformService.queryPlatformByServerGBId(sendRtpItem.getTargetId());
CommonGBChannel channel = channelService.getOne(sendRtpItem.getChannelId()); CommonGBChannel channel = channelService.getOne(sendRtpItem.getChannelId());
if (channel == null) { if (channel == null) {
continue; continue;
@ -248,8 +252,7 @@ public class MediaServiceImpl implements IMediaService {
} catch (SipException | InvalidArgumentException | ParseException e) { } catch (SipException | InvalidArgumentException | ParseException e) {
log.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage()); log.error("[命令发送失败] 国标级联 发送BYE: {}", e.getMessage());
} }
redisCatchStorage.deleteSendRTPServer(parentPlatform.getServerGBId(), channel.getGbDeviceId(), sendRtpServerService.delete(sendRtpItem);
sendRtpItem.getCallId(), sendRtpItem.getStream());
if (InviteStreamType.PUSH == sendRtpItem.getPlayType()) { if (InviteStreamType.PUSH == sendRtpItem.getPlayType()) {
redisCatchStorage.sendPlatformStopPlayMsg(sendRtpItem, parentPlatform, channel); redisCatchStorage.sendPlatformStopPlayMsg(sendRtpItem, parentPlatform, channel);
redisCatchStorage.sendPlatformStopPlayMsg(sendRtpItem, parentPlatform, channel); redisCatchStorage.sendPlatformStopPlayMsg(sendRtpItem, parentPlatform, channel);
@ -280,8 +283,8 @@ public class MediaServiceImpl implements IMediaService {
deviceChannelService.stopPlay(inviteInfo.getChannelId()); deviceChannelService.stopPlay(inviteInfo.getChannelId());
return result; return result;
} }
SendRtpInfo sendRtpItem = redisCatchStorage.querySendRTPServer(null, null, stream, null); List<SendRtpInfo> sendRtpItemList = sendRtpServerService.queryByStream(stream);
if (sendRtpItem != null && "talk".equals(sendRtpItem.getApp())) { if (!sendRtpItemList.isEmpty()) {
return false; return false;
} }
} else if ("talk".equals(app) || "broadcast".equals(app)) { } else if ("talk".equals(app) || "broadcast".equals(app)) {

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

@ -4,6 +4,7 @@ import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.UserSetting; import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.PlayException; import com.genersoft.iot.vmp.gb28181.bean.PlayException;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpInfo; import com.genersoft.iot.vmp.gb28181.bean.SendRtpInfo;
import com.genersoft.iot.vmp.gb28181.conf.StackLoggerImpl;
import com.genersoft.iot.vmp.media.bean.MediaServer; import com.genersoft.iot.vmp.media.bean.MediaServer;
import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager; import com.genersoft.iot.vmp.media.zlm.SendRtpPortManager;
import com.genersoft.iot.vmp.service.ISendRtpServerService; import com.genersoft.iot.vmp.service.ISendRtpServerService;
@ -29,10 +30,12 @@ public class SendRtpServerServiceImpl implements ISendRtpServerService {
@Autowired @Autowired
private RedisTemplate<Object, Object> redisTemplate; private RedisTemplate<Object, Object> redisTemplate;
@Autowired
private StackLoggerImpl stackLoggerImpl;
@Override @Override
public SendRtpInfo createSendRtpInfo(MediaServer mediaServer, String ip, int port, String ssrc, String requesterId, public SendRtpInfo createSendRtpInfo(MediaServer mediaServer, String ip, Integer port, String ssrc, String requesterId,
String deviceId, Integer channelId, boolean isTcp, boolean rtcp) { String deviceId, Integer channelId, Boolean isTcp, Boolean rtcp) {
int localPort = sendRtpPortManager.getNextPort(mediaServer); int localPort = sendRtpPortManager.getNextPort(mediaServer);
if (localPort == 0) { if (localPort == 0) {
return null; return null;
@ -42,8 +45,8 @@ public class SendRtpServerServiceImpl implements ISendRtpServerService {
} }
@Override @Override
public SendRtpInfo createSendRtpInfo(MediaServer mediaServer, String ip, int port, String ssrc, String platformId, public SendRtpInfo createSendRtpInfo(MediaServer mediaServer, String ip, Integer port, String ssrc, String platformId,
String app, String stream, Integer channelId, boolean tcp, boolean rtcp){ String app, String stream, Integer channelId, Boolean tcp, Boolean rtcp){
int localPort = sendRtpPortManager.getNextPort(mediaServer); int localPort = sendRtpPortManager.getNextPort(mediaServer);
if (localPort <= 0) { if (localPort <= 0) {
@ -56,13 +59,13 @@ public class SendRtpServerServiceImpl implements ISendRtpServerService {
@Override @Override
public void update(SendRtpInfo sendRtpItem) { public void update(SendRtpInfo sendRtpItem) {
redisTemplate.opsForValue().set(VideoManagerConstants.SEND_RTP_INFO_CALLID + sendRtpItem.getCallId(), sendRtpItem); redisTemplate.opsForValue().set(VideoManagerConstants.SEND_RTP_INFO_CALLID + sendRtpItem.getCallId(), sendRtpItem);
redisTemplate.opsForValue().set(VideoManagerConstants.SEND_RTP_INFO_STREAM + sendRtpItem.getStream(), sendRtpItem); redisTemplate.opsForValue().set(VideoManagerConstants.SEND_RTP_INFO_STREAM + sendRtpItem.getStream() + ":" + sendRtpItem.getTargetId(), sendRtpItem);
redisTemplate.opsForValue().set(VideoManagerConstants.SEND_RTP_INFO_CHANNEL + sendRtpItem.getChannelId(), sendRtpItem); redisTemplate.opsForValue().set(VideoManagerConstants.SEND_RTP_INFO_CHANNEL + sendRtpItem.getChannelId() + ":" + sendRtpItem.getTargetId(), sendRtpItem);
} }
@Override @Override
public SendRtpInfo queryByChannelId(Integer channelId) { public SendRtpInfo queryByChannelId(Integer channelId, String targetId) {
String key = VideoManagerConstants.SEND_RTP_INFO_CHANNEL + channelId; String key = VideoManagerConstants.SEND_RTP_INFO_CHANNEL + channelId + ":" + targetId;
return JsonUtil.redisJsonToObject(redisTemplate, key, SendRtpInfo.class); return JsonUtil.redisJsonToObject(redisTemplate, key, SendRtpInfo.class);
} }
@ -74,10 +77,24 @@ public class SendRtpServerServiceImpl implements ISendRtpServerService {
@Override @Override
public SendRtpInfo queryByStream(String stream, String targetId) { public SendRtpInfo queryByStream(String stream, String targetId) {
String key = VideoManagerConstants.SEND_RTP_INFO_STREAM + stream; String key = VideoManagerConstants.SEND_RTP_INFO_STREAM + stream + ":" + targetId;
return JsonUtil.redisJsonToObject(redisTemplate, key, SendRtpInfo.class); return JsonUtil.redisJsonToObject(redisTemplate, key, SendRtpInfo.class);
} }
@Override
public List<SendRtpInfo> queryByStream(String stream) {
String key = VideoManagerConstants.SEND_RTP_INFO_STREAM + stream + ":*";
List<Object> queryResult = RedisUtil.scan(redisTemplate, key);
List<SendRtpInfo> result= new ArrayList<>();
for (Object o : queryResult) {
String keyItem = (String) o;
result.add((SendRtpInfo) redisTemplate.opsForValue().get(keyItem));
}
return result;
}
/** /**
* 删除RTP推送信息缓存 * 删除RTP推送信息缓存
*/ */
@ -87,8 +104,8 @@ public class SendRtpServerServiceImpl implements ISendRtpServerService {
return; return;
} }
redisTemplate.delete(VideoManagerConstants.SEND_RTP_INFO_CALLID + sendRtpInfo.getCallId()); redisTemplate.delete(VideoManagerConstants.SEND_RTP_INFO_CALLID + sendRtpInfo.getCallId());
redisTemplate.delete(VideoManagerConstants.SEND_RTP_INFO_STREAM + sendRtpInfo.getStream()); redisTemplate.delete(VideoManagerConstants.SEND_RTP_INFO_STREAM + sendRtpInfo.getStream() + ":" + sendRtpInfo.getTargetId());
redisTemplate.delete(VideoManagerConstants.SEND_RTP_INFO_CHANNEL + sendRtpInfo.getChannelId()); redisTemplate.delete(VideoManagerConstants.SEND_RTP_INFO_CHANNEL + sendRtpInfo.getChannelId() + ":" + sendRtpInfo.getTargetId());
} }
@Override @Override
public void deleteByCallId(String callId) { public void deleteByCallId(String callId) {
@ -99,22 +116,45 @@ public class SendRtpServerServiceImpl implements ISendRtpServerService {
delete(sendRtpInfo); delete(sendRtpInfo);
} }
@Override @Override
public void deleteByStream(String Stream) { public void deleteByStream(String stream, String targetId) {
SendRtpInfo sendRtpInfo = queryByStream(Stream); SendRtpInfo sendRtpInfo = queryByStream(stream, targetId);
if (sendRtpInfo == null) { if (sendRtpInfo == null) {
return; return;
} }
delete(sendRtpInfo); delete(sendRtpInfo);
} }
@Override @Override
public void deleteByChannel(Integer channelId) { public void deleteByStream(String stream) {
SendRtpInfo sendRtpInfo = queryByChannelId(channelId); List<SendRtpInfo> sendRtpInfos = queryByStream(stream);
for (SendRtpInfo sendRtpInfo : sendRtpInfos) {
delete(sendRtpInfo);
}
}
@Override
public void deleteByChannel(Integer channelId, String targetId) {
SendRtpInfo sendRtpInfo = queryByChannelId(channelId, targetId);
if (sendRtpInfo == null) { if (sendRtpInfo == null) {
return; return;
} }
delete(sendRtpInfo); delete(sendRtpInfo);
} }
@Override
public List<SendRtpInfo> queryByChannelId(int channelId) {
String key = VideoManagerConstants.SEND_RTP_INFO_CHANNEL + channelId + ":*";
List<Object> queryResult = RedisUtil.scan(redisTemplate, key);
List<SendRtpInfo> result= new ArrayList<>();
for (Object o : queryResult) {
String keyItem = (String) o;
result.add((SendRtpInfo) redisTemplate.opsForValue().get(keyItem));
}
return result;
}
@Override @Override
public List<SendRtpInfo> queryAll() { public List<SendRtpInfo> queryAll() {
String key = VideoManagerConstants.SEND_RTP_INFO_CALLID String key = VideoManagerConstants.SEND_RTP_INFO_CALLID
@ -135,8 +175,16 @@ public class SendRtpServerServiceImpl implements ISendRtpServerService {
*/ */
@Override @Override
public boolean isChannelSendingRTP(Integer channelId) { public boolean isChannelSendingRTP(Integer channelId) {
SendRtpInfo sendRtpInfo = queryByChannelId(channelId); List<SendRtpInfo> sendRtpInfoList = queryByChannelId(channelId);
return sendRtpInfo != null; return !sendRtpInfoList.isEmpty();
} }
@Override
public List<SendRtpInfo> queryForPlatform(String platformId) {
List<SendRtpInfo> sendRtpInfos = queryAll();
if (!sendRtpInfos.isEmpty()) {
sendRtpInfos.removeIf(sendRtpInfo -> !sendRtpInfo.isSendToPlatform() || !sendRtpInfo.getTargetId().equals(platformId));
}
return sendRtpInfos;
}
} }

2
src/main/java/com/genersoft/iot/vmp/service/redisMsg/IRedisRpcService.java
View File

@ -13,7 +13,7 @@ public interface IRedisRpcService {
WVPResult stopSendRtp(Integer sendRtpChannelId); WVPResult stopSendRtp(Integer sendRtpChannelId);
long waitePushStreamOnline(SendRtpInfo sendRtpItem, CommonCallback<String> callback); long waitePushStreamOnline(SendRtpInfo sendRtpItem, CommonCallback<Integer> callback);
void stopWaitePushStreamOnline(SendRtpInfo sendRtpItem); void stopWaitePushStreamOnline(SendRtpInfo sendRtpItem);

22
src/main/java/com/genersoft/iot/vmp/service/redisMsg/service/RedisRpcServiceImpl.java
View File

@ -13,6 +13,7 @@ import com.genersoft.iot.vmp.media.event.hook.Hook;
import com.genersoft.iot.vmp.media.event.hook.HookSubscribe; import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
import com.genersoft.iot.vmp.media.event.hook.HookType; import com.genersoft.iot.vmp.media.event.hook.HookType;
import com.genersoft.iot.vmp.media.service.IMediaServerService; import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.service.ISendRtpServerService;
import com.genersoft.iot.vmp.service.redisMsg.IRedisRpcService; import com.genersoft.iot.vmp.service.redisMsg.IRedisRpcService;
import com.genersoft.iot.vmp.vmanager.bean.ErrorCode; import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
import com.genersoft.iot.vmp.vmanager.bean.WVPResult; import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
@ -25,6 +26,7 @@ import org.springframework.stereotype.Service;
@Service @Service
public class RedisRpcServiceImpl implements IRedisRpcService { public class RedisRpcServiceImpl implements IRedisRpcService {
@Autowired @Autowired
private RedisRpcConfig redisRpcConfig; private RedisRpcConfig redisRpcConfig;
@ -40,10 +42,12 @@ public class RedisRpcServiceImpl implements IRedisRpcService {
@Autowired @Autowired
private RedisTemplate<Object, Object> redisTemplate; private RedisTemplate<Object, Object> redisTemplate;
@Autowired @Autowired
private IMediaServerService mediaServerService; private IMediaServerService mediaServerService;
@Autowired
private ISendRtpServerService sendRtpServerService;
private RedisRpcRequest buildRequest(String uri, Object param) { private RedisRpcRequest buildRequest(String uri, Object param) {
RedisRpcRequest request = new RedisRpcRequest(); RedisRpcRequest request = new RedisRpcRequest();
request.setFromId(userSetting.getServerId()); request.setFromId(userSetting.getServerId());
@ -53,7 +57,7 @@ public class RedisRpcServiceImpl implements IRedisRpcService {
} }
@Override @Override
public SendRtpInfo getSendRtpItem(String sendRtpItemKey) { public SendRtpInfo getSendRtpItem(Integer sendRtpItemKey) {
RedisRpcRequest request = buildRequest("getSendRtpItem", sendRtpItemKey); RedisRpcRequest request = buildRequest("getSendRtpItem", sendRtpItemKey);
RedisRpcResponse response = redisRpcConfig.request(request, 10); RedisRpcResponse response = redisRpcConfig.request(request, 10);
if (response.getBody() == null) { if (response.getBody() == null) {
@ -63,7 +67,7 @@ public class RedisRpcServiceImpl implements IRedisRpcService {
} }
@Override @Override
public WVPResult startSendRtp(String sendRtpItemKey, SendRtpInfo sendRtpItem) { public WVPResult startSendRtp(Integer sendRtpItemKey, SendRtpInfo sendRtpItem) {
log.info("[请求其他WVP] 开始推流，wvp：{}， {}/{}", sendRtpItem.getServerId(), sendRtpItem.getApp(), sendRtpItem.getStream()); log.info("[请求其他WVP] 开始推流，wvp：{}， {}/{}", sendRtpItem.getServerId(), sendRtpItem.getApp(), sendRtpItem.getStream());
RedisRpcRequest request = buildRequest("startSendRtp", sendRtpItemKey); RedisRpcRequest request = buildRequest("startSendRtp", sendRtpItemKey);
request.setToId(sendRtpItem.getServerId()); request.setToId(sendRtpItem.getServerId());
@ -72,7 +76,7 @@ public class RedisRpcServiceImpl implements IRedisRpcService {
} }
@Override @Override
public WVPResult stopSendRtp(String sendRtpItemKey) { public WVPResult stopSendRtp(Integer sendRtpItemKey) {
SendRtpInfo sendRtpItem = (SendRtpInfo)redisTemplate.opsForValue().get(sendRtpItemKey); SendRtpInfo sendRtpItem = (SendRtpInfo)redisTemplate.opsForValue().get(sendRtpItemKey);
if (sendRtpItem == null) { if (sendRtpItem == null) {
log.info("[请求其他WVP] 停止推流, 未找到redis中的发流信息， key：{}", sendRtpItemKey); log.info("[请求其他WVP] 停止推流, 未找到redis中的发流信息， key：{}", sendRtpItemKey);
@ -86,7 +90,7 @@ public class RedisRpcServiceImpl implements IRedisRpcService {
} }
@Override @Override
public long waitePushStreamOnline(SendRtpInfo sendRtpItem, CommonCallback<String> callback) { public long waitePushStreamOnline(SendRtpInfo sendRtpItem, CommonCallback<Integer> callback) {
log.info("[请求所有WVP监听流上线] {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream()); log.info("[请求所有WVP监听流上线] {}/{}", sendRtpItem.getApp(), sendRtpItem.getStream());
// 监听流上线。 流上线直接发送sendRtpItem消息给实际的信令处理者 // 监听流上线。 流上线直接发送sendRtpItem消息给实际的信令处理者
Hook hook = Hook.getInstance(HookType.on_media_arrival, sendRtpItem.getApp(), sendRtpItem.getStream(), null); Hook hook = Hook.getInstance(HookType.on_media_arrival, sendRtpItem.getApp(), sendRtpItem.getStream(), null);
@ -103,9 +107,9 @@ public class RedisRpcServiceImpl implements IRedisRpcService {
sendRtpItem.setMediaServerId(hookData.getMediaServer().getId()); sendRtpItem.setMediaServerId(hookData.getMediaServer().getId());
sendRtpItem.setLocalIp(hookData.getMediaServer().getSdpIp()); sendRtpItem.setLocalIp(hookData.getMediaServer().getSdpIp());
sendRtpItem.setServerId(userSetting.getServerId()); sendRtpItem.setServerId(userSetting.getServerId());
redisTemplate.opsForValue().set(sendRtpItem.getRedisKey(), sendRtpItem); sendRtpServerService.update(sendRtpItem);
if (callback != null) { if (callback != null) {
callback.run(sendRtpItem.getRedisKey()); callback.run(sendRtpItem.getChannelId());
} }
hookSubscribe.removeSubscribe(hook); hookSubscribe.removeSubscribe(hook);
redisRpcConfig.removeCallback(request.getSn()); redisRpcConfig.removeCallback(request.getSn());
@ -119,7 +123,7 @@ public class RedisRpcServiceImpl implements IRedisRpcService {
log.info("[请求所有WVP监听流上线] 流上线 {}/{}->{}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.toString()); log.info("[请求所有WVP监听流上线] 流上线 {}/{}->{}", sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.toString());
if (callback != null) { if (callback != null) {
callback.run(response.getBody().toString()); callback.run(Integer.parseInt(response.getBody().toString()));
} }
hookSubscribe.removeSubscribe(hook); hookSubscribe.removeSubscribe(hook);
}); });
@ -137,7 +141,7 @@ public class RedisRpcServiceImpl implements IRedisRpcService {
} }
@Override @Override
public void rtpSendStopped(String sendRtpItemKey) { public void rtpSendStopped(Integer sendRtpItemKey) {
SendRtpInfo sendRtpItem = (SendRtpInfo)redisTemplate.opsForValue().get(sendRtpItemKey); SendRtpInfo sendRtpItem = (SendRtpInfo)redisTemplate.opsForValue().get(sendRtpItemKey);
if (sendRtpItem == null) { if (sendRtpItem == null) {
log.info("[停止WVP监听流上线] 未找到redis中的发流信息， key：{}", sendRtpItemKey); log.info("[停止WVP监听流上线] 未找到redis中的发流信息， key：{}", sendRtpItemKey);

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

@ -511,7 +511,7 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
MessageForPushChannel msg = MessageForPushChannel.getInstance(0, MessageForPushChannel msg = MessageForPushChannel.getInstance(0,
sendRtpItem.getApp(), sendRtpItem.getStream(), channel.getGbDeviceId(), sendRtpItem.getApp(), sendRtpItem.getStream(), channel.getGbDeviceId(),
sendRtpItem.getPlatformId(), platform.getName(), userSetting.getServerId(), sendRtpItem.getMediaServerId()); sendRtpItem.getTargetId(), platform.getName(), userSetting.getServerId(), sendRtpItem.getMediaServerId());
msg.setPlatFormIndex(platform.getId()); msg.setPlatFormIndex(platform.getId());
String key = VideoManagerConstants.VM_MSG_STREAM_STOP_PLAY_NOTIFY; String key = VideoManagerConstants.VM_MSG_STREAM_STOP_PLAY_NOTIFY;
@ -562,14 +562,14 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
@Override @Override
public void sendStartSendRtp(SendRtpInfo sendRtpItem) { public void sendStartSendRtp(SendRtpInfo sendRtpItem) {
String key = VideoManagerConstants.START_SEND_PUSH_STREAM + sendRtpItem.getApp() + "_" + sendRtpItem.getStream(); String key = VideoManagerConstants.START_SEND_PUSH_STREAM + sendRtpItem.getApp() + "_" + sendRtpItem.getStream();
log.info("[redis发送通知] 通知其他WVP推流 {}: {}/{}->{}", key, sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getPlatformId()); log.info("[redis发送通知] 通知其他WVP推流 {}: {}/{}->{}", key, sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getTargetId());
redisTemplate.convertAndSend(key, JSON.toJSON(sendRtpItem)); redisTemplate.convertAndSend(key, JSON.toJSON(sendRtpItem));
} }
@Override @Override
public void sendPushStreamOnline(SendRtpInfo sendRtpItem) { public void sendPushStreamOnline(SendRtpInfo sendRtpItem) {
String key = VideoManagerConstants.VM_MSG_STREAM_PUSH_CLOSE_REQUESTED; String key = VideoManagerConstants.VM_MSG_STREAM_PUSH_CLOSE_REQUESTED;
log.info("[redis发送通知] 流上线 {}: {}/{}->{}", key, sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getPlatformId()); log.info("[redis发送通知] 流上线 {}: {}/{}->{}", key, sendRtpItem.getApp(), sendRtpItem.getStream(), sendRtpItem.getTargetId());
redisTemplate.convertAndSend(key, JSON.toJSON(sendRtpItem)); redisTemplate.convertAndSend(key, JSON.toJSON(sendRtpItem));
} }
} }

10
src/main/java/com/genersoft/iot/vmp/streamPush/service/impl/StreamPushServiceImpl.java
View File

@ -16,6 +16,7 @@ import com.genersoft.iot.vmp.media.event.mediaServer.MediaServerOnlineEvent;
import com.genersoft.iot.vmp.media.service.IMediaServerService; import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo; import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
import com.genersoft.iot.vmp.media.zlm.dto.hook.OriginType; import com.genersoft.iot.vmp.media.zlm.dto.hook.OriginType;
import com.genersoft.iot.vmp.service.ISendRtpServerService;
import com.genersoft.iot.vmp.service.bean.StreamPushItemFromRedis; import com.genersoft.iot.vmp.service.bean.StreamPushItemFromRedis;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.streamPush.bean.StreamPush; import com.genersoft.iot.vmp.streamPush.bean.StreamPush;
@ -29,10 +30,8 @@ import com.github.pagehelper.PageInfo;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.event.EventListener; import org.springframework.context.event.EventListener;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.scheduling.annotation.Async; import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert; import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils; import org.springframework.util.ObjectUtils;
@ -57,10 +56,7 @@ public class StreamPushServiceImpl implements IStreamPushService {
private IMediaServerService mediaServerService; private IMediaServerService mediaServerService;
@Autowired @Autowired
DataSourceTransactionManager dataSourceTransactionManager; private ISendRtpServerService sendRtpServerService;
@Autowired
TransactionDefinition transactionDefinition;
@Autowired @Autowired
private IGbChannelService gbChannelService; private IGbChannelService gbChannelService;
@ -290,7 +286,7 @@ public class StreamPushServiceImpl implements IStreamPushService {
if (userSetting.isUsePushingAsStatus()) { if (userSetting.isUsePushingAsStatus()) {
gbChannelService.offline(streamPush.buildCommonGBChannel()); gbChannelService.offline(streamPush.buildCommonGBChannel());
} }
redisCatchStorage.deleteSendRTPServer(null, streamPush.getGbDeviceId(), null, streamPush.getStream()); sendRtpServerService.deleteByStream(streamPush.getStream());
mediaServerService.stopSendRtp(mediaServer, streamPush.getApp(), streamPush.getStream(), null); mediaServerService.stopSendRtp(mediaServer, streamPush.getApp(), streamPush.getStream(), null);
streamPush.setUpdateTime(DateUtil.getNow()); streamPush.setUpdateTime(DateUtil.getNow());
streamPushMapper.update(streamPush); streamPushMapper.update(streamPush);