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

优化语音广播流程

结构优化
648540858 2022-05-10 13:08:32 +08:00
parent 65fa75fb44
commit f8f65d473b
8 changed files with 157 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
src/main/java/com/genersoft/iot/vmp/gb28181/bean/AudioBroadcastCatch.java
View File

@ -1,6 +1,11 @@
package com.genersoft.iot.vmp.gb28181.bean;
import gov.nist.javax.sip.message.SIPRequest;
import gov.nist.javax.sip.stack.SIPDialog;
import javax.sip.Dialog;
/**
* 缓存语音广播的状态
* @author lin
@ -32,6 +37,16 @@ public class AudioBroadcastCatch {
*/
private AudioBroadcastCatchStatus status;
/**
* 请求信息
*/
private SIPRequest request;
/**
* 会话信息
*/
private SIPDialog dialog;
public String getDeviceId() {
return deviceId;
@ -56,4 +71,20 @@ public class AudioBroadcastCatch {
public void setStatus(AudioBroadcastCatchStatus status) {
this.status = status;
}
public void setDialog(SIPDialog dialog) {
this.dialog = dialog;
}
public SIPDialog getDialog() {
return dialog;
}
public SIPRequest getRequest() {
return request;
}
public void setRequest(SIPRequest request) {
this.request = request;
}
}

24
src/main/java/com/genersoft/iot/vmp/gb28181/session/AudioBroadcastManager.java
View File

@ -1,13 +1,14 @@
package com.genersoft.iot.vmp.gb28181.session;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.gb28181.bean.AudioBroadcastCatch;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import java.util.stream.Stream;
/**
* 语音广播消息管理类
@ -16,6 +17,9 @@ import java.util.concurrent.ConcurrentHashMap;
@Component
public class AudioBroadcastManager {
@Autowired
private SipConfig config;
public static Map<String, AudioBroadcastCatch> data = new ConcurrentHashMap<>();
public void add(AudioBroadcastCatch audioBroadcastCatch) {
@ -54,6 +58,16 @@ public class AudioBroadcastManager {
}
public AudioBroadcastCatch get(String deviceId, String channelId) {
return data.get(deviceId + channelId);
AudioBroadcastCatch audioBroadcastCatch = data.get(deviceId + channelId);
if (audioBroadcastCatch == null) {
Stream<AudioBroadcastCatch> allAudioBroadcastCatchStreamForDevice = data.values().stream().filter(
audioBroadcastCatchItem -> Objects.equals(audioBroadcastCatchItem.getDeviceId(), deviceId));
List<AudioBroadcastCatch> audioBroadcastCatchList = allAudioBroadcastCatchStreamForDevice.collect(Collectors.toList());
if (audioBroadcastCatchList.size() == 1 && Objects.equals(config.getId(), channelId)) {
audioBroadcastCatch = audioBroadcastCatchList.get(0);
}
}
return audioBroadcastCatch;
}
}

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

@ -18,6 +18,8 @@ import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
import gov.nist.javax.sip.message.SIPRequest;
import gov.nist.javax.sip.stack.SIPDialog;
import org.ehcache.shadow.org.terracotta.offheapstore.storage.IntegerStorageEngine;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -28,15 +30,18 @@ import org.springframework.stereotype.Component;
import javax.sip.Dialog;
import javax.sip.DialogState;
import javax.sip.RequestEvent;
import javax.sip.SipException;
import javax.sip.address.SipURI;
import javax.sip.header.CallIdHeader;
import javax.sip.header.FromHeader;
import javax.sip.header.HeaderAddress;
import javax.sip.header.ToHeader;
import java.text.ParseException;
import java.util.*;
/**
* SIP命令类型： ACK请求
* @author lin
*/
@Component
public class AckRequestProcessor extends SIPRequestProcessorParent implements InitializingBean, ISIPRequestProcessor {
@ -96,8 +101,8 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In
ParentPlatform parentPlatform = storager.queryParentPlatByServerGBId(platformGbId);
// 取消设置的超时任务
dynamicTask.stop(callIdHeader.getCallId());
String channelId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser();
SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(platformGbId, channelId, null, callIdHeader.getCallId());
// String channelId = ((SipURI) ((HeaderAddress) evt.getRequest().getHeader(ToHeader.NAME)).getAddress().getURI()).getUser();
SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(platformGbId, null, null, callIdHeader.getCallId());
String is_Udp = sendRtpItem.isTcp() ? "0" : "1";
MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
logger.info("收到ACK，开始向上级推流 rtp/{}", sendRtpItem.getStreamId());
@ -121,7 +126,14 @@ public class AckRequestProcessor extends SIPRequestProcessorParent implements In
} else {
logger.error("RTP推流失败: {}, 参数：{}",jsonObject.getString("msg"),JSONObject.toJSON(param));
if (sendRtpItem.isOnlyAudio()) {
// TODO 可能是语音对讲
// 语音对讲
try {
cmder.streamByeCmd((SIPDialog) evt.getDialog(), (SIPRequest)evt.getRequest(), null);
} catch (SipException e) {
throw new RuntimeException(e);
} catch (ParseException e) {
throw new RuntimeException(e);
}
}else {
// 向上级平台
commanderForPlatform.streamByeCmd(parentPlatform, callIdHeader.getCallId());

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

@ -13,6 +13,7 @@ import com.genersoft.iot.vmp.gb28181.transmit.event.request.SIPRequestProcessorP
import com.genersoft.iot.vmp.media.zlm.ZLMRTPServerFactory;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IPlayService;
import com.genersoft.iot.vmp.service.bean.MessageForPushChannel;
import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
import com.genersoft.iot.vmp.storager.IVideoManagerStorage;
@ -65,6 +66,9 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
@Autowired
private VideoStreamSessionManager streamSession;
@Autowired
private IPlayService playService;
@Override
public void afterPropertiesSet() throws Exception {
// 添加消息处理的订阅
@ -106,6 +110,9 @@ public class ByeRequestProcessor extends SIPRequestProcessorParent implements In
if (sendRtpItem.getPlayType().equals(InviteStreamType.PLAY)) {
cmder.streamByeCmd(sendRtpItem.getDeviceId(), channelId, streamId, null);
}
if (sendRtpItem.isOnlyAudio()) {
playService.stopAudioBroadcast(sendRtpItem.getDeviceId(), channelId);
}
if (sendRtpItem.getPlayType().equals(InviteStreamType.PUSH)) {
MessageForPushChannel messageForPushChannel = new MessageForPushChannel();
messageForPushChannel.setType(0);

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

@ -114,6 +114,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
private SipConfig config;
@Override
public void afterPropertiesSet() throws Exception {
// 添加消息处理的订阅
@ -492,7 +493,6 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
gbStream.getApp(), gbStream.getStream(), channelId,
mediaTransmissionTCP);
if (sendRtpItem == null) {
logger.warn("服务器端口资源不足");
responseAck(evt, Response.BUSY_HERE);
@ -562,25 +562,16 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
}
}
public void inviteFromDeviceHandle(RequestEvent evt, String requesterId, String channelId) throws InvalidArgumentException, ParseException, SipException, SdpException {
public void inviteFromDeviceHandle(RequestEvent evt, String requesterId, String channelId1) throws InvalidArgumentException, ParseException, SipException, SdpException {
// 兼容奇葩的海康这里使用的不是通道编号而是本平台编号
// if (channelId.equals(config.getId())) {
// List<AudioBroadcastCatch> all = audioBroadcastManager.getAll();
// for (AudioBroadcastCatch audioBroadcastCatch : all) {
// if (audioBroadcastCatch.getDeviceId().equals(requesterId)) {
// channelId = audioBroadcastCatch.getChannelId();
// }
// }
// }
// // 兼容失败
// if (channelId.equals(config.getId())) {
// responseAck(evt, Response.BAD_REQUEST);
// return;
// }
// 非上级平台请求，查询是否设备请求（通常为接收语音广播的设备）
Device device = redisCatchStorage.getDevice(requesterId);
AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(requesterId, channelId1);
if (audioBroadcastCatch == null) {
logger.warn("来自设备的Invite请求非语音广播，已忽略");
responseAck(evt, Response.FORBIDDEN);
return;
}
Request request = evt.getRequest();
if (device != null) {
logger.info("收到设备" + requesterId + "的语音广播Invite请求");
@ -606,7 +597,6 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
// 查看是否支持PS 负载96
int port = -1;
//boolean recvonly = false;
boolean mediaTransmissionTCP = false;
Boolean tcpActive = null;
for (int i = 0; i < mediaDescriptions.size(); i++) {
@ -638,7 +628,6 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
responseAck(evt, Response.UNSUPPORTED_MEDIA_TYPE); // 不支持的格式，发415
return;
}
String sessionName = sdp.getSessionName().getValue();
String addressStr = sdp.getOrigin().getAddress();
logger.info("设备{}请求语音流，地址：{}:{}，ssrc：{}", requesterId, addressStr, port, ssrc);
@ -649,20 +638,19 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
return;
}
SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, port, ssrc, requesterId,
device.getDeviceId(), channelId,
device.getDeviceId(), audioBroadcastCatch.getChannelId(),
mediaTransmissionTCP);
sendRtpItem.setTcp(mediaTransmissionTCP);
if (tcpActive != null) {
sendRtpItem.setTcpActive(tcpActive);
}
if (sendRtpItem == null) {
logger.warn("服务器端口资源不足");
responseAck(evt, Response.BUSY_HERE);
return;
}
sendRtpItem.setTcp(mediaTransmissionTCP);
if (tcpActive != null) {
sendRtpItem.setTcpActive(tcpActive);
}
String app = "broadcast";
String stream = device.getDeviceId() + "_" + channelId;
String stream = device.getDeviceId() + "_" + audioBroadcastCatch.getChannelId();
CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME);
sendRtpItem.setPlayType(InviteStreamType.PLAY);
@ -685,12 +673,9 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
subscribeKey.put("schema", "rtmp");
subscribeKey.put("mediaServerId", mediaServerItem.getId());
String finalSsrc = ssrc;
String waiteStreamTimeoutTaskKey = "waite-stream-" + device.getDeviceId() + channelId;
// 流已经存在时直接推流
if (zlmrtpServerFactory.isStreamReady(mediaServerItem, app, stream)) {
logger.info("发现已经在推流");
dynamicTask.stop(waiteStreamTimeoutTaskKey);
sendRtpItem.setStatus(2);
redisCatchStorage.updateSendRTPSever(sendRtpItem);
StringBuffer content = new StringBuffer(200);
@ -711,6 +696,10 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
parentPlatform.setServerGBId(device.getDeviceId());
try {
responseSdpAck(evt, content.toString(), parentPlatform);
Dialog dialog = evt.getDialog();
audioBroadcastCatch.setDialog((SIPDialog) dialog);
audioBroadcastCatch.setRequest((SIPRequest) request);
audioBroadcastManager.update(audioBroadcastCatch);
} catch (SipException e) {
throw new RuntimeException(e);
} catch (InvalidArgumentException e) {
@ -721,20 +710,17 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
}else {
// 流不存在时监听流上线
// 设置等待推流的超时; 默认20s
String waiteStreamTimeoutTaskKey = "waite-stream-" + device.getDeviceId() + audioBroadcastCatch.getChannelId();
dynamicTask.startDelay(waiteStreamTimeoutTaskKey, ()->{
logger.info("等待推流超时: {}/{}", app, stream);
if (audioBroadcastManager.exit(device.getDeviceId(), channelId)) {
audioBroadcastManager.del(device.getDeviceId(), channelId);
}else {
// 兼容海康使用了错误的通道ID的情况
audioBroadcastManager.delByDeviceId(device.getDeviceId());
}
playService.stopAudioBroadcast(device.getDeviceId(), audioBroadcastCatch.getChannelId());
// 发送bye
try {
cmder.streamByeCmd((SIPDialog)evt.getServerTransaction().getDialog(), (SIPRequest) evt.getRequest(), null);
responseAck(evt, Response.BUSY_HERE);
} catch (SipException e) {
throw new RuntimeException(e);
} catch (InvalidArgumentException e) {
throw new RuntimeException(e);
} catch (ParseException e) {
throw new RuntimeException(e);
}
@ -743,10 +729,11 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_stream_changed, subscribeKey,
(MediaServerItem mediaServerItemInUse, JSONObject json)->{
sendRtpItem.setStatus(2);
dynamicTask.stop(waiteStreamTimeoutTaskKey);
redisCatchStorage.updateSendRTPSever(sendRtpItem);
StringBuffer content = new StringBuffer(200);
content.append("v=0\r\n");
content.append("o="+ channelId +" 0 0 IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
content.append("o="+ audioBroadcastCatch.getChannelId() +" 0 0 IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
content.append("s=Play\r\n");
content.append("c=IN IP4 "+mediaServerItem.getSdpIp()+"\r\n");
content.append("t=0 0\r\n");
@ -771,8 +758,6 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
}
});
}
String timeOutTaskKey = "audio-broadcast-" + device.getDeviceId() + channelId;
dynamicTask.stop(timeOutTaskKey);
String key = DeferredResultHolder.CALLBACK_CMD_BROADCAST + device.getDeviceId();
WVPResult<AudioBroadcastResult> wvpResult = new WVPResult<>();
wvpResult.setCode(0);

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

@ -43,4 +43,5 @@ public interface IPlayService {
StreamInfo getDownLoadInfo(String deviceId, String channelId, String stream);
void audioBroadcast(Device device, String channelId, int timeout, AudioBroadcastEvent event);
void stopAudioBroadcast(String deviceId, String channelId);
}

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

@ -5,6 +5,7 @@ import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.common.StreamInfo;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.conf.SipConfig;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.*;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
@ -44,9 +45,13 @@ import org.springframework.util.ResourceUtils;
import org.springframework.web.context.request.async.DeferredResult;
import javax.sip.ResponseEvent;
import javax.sip.SipException;
import java.io.FileNotFoundException;
import java.math.BigDecimal;
import java.text.ParseException;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@SuppressWarnings(value = {"rawtypes", "unchecked"})
@Service
@ -93,6 +98,9 @@ public class PlayServiceImpl implements IPlayService {
@Autowired
private UserSetting userSetting;
@Autowired
private SipConfig sipConfig;
@Autowired
private DynamicTask dynamicTask;
@ -641,16 +649,13 @@ public class PlayServiceImpl implements IPlayService {
}
// 查询通道使用状态
if (audioBroadcastManager.exit(device.getDeviceId(), channelId)) {
logger.warn("语音广播已经开启： {}", channelId);
event.call("语音广播已经开启");
return;
SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(device.getDeviceId(), channelId, null, null);
if (sendRtpItem != null && sendRtpItem.isOnlyAudio()) {
logger.warn("语音广播已经开启： {}", channelId);
event.call("语音广播已经开启");
return;
}
}
String timeOutTaskKey = "audio-broadcast-" + device.getDeviceId() + channelId;
dynamicTask.startDelay(timeOutTaskKey, ()->{
logger.error("语音广播发送超时： {}:{}", device.getDeviceId(), channelId);
event.call("语音广播发送超时");
audioBroadcastManager.del(device.getDeviceId(), channelId);
}, timeout * 1000);
// 发送通知
cmder.audioBroadcastCmd(device, channelId, eventResultForOk -> {
@ -658,11 +663,38 @@ public class PlayServiceImpl implements IPlayService {
AudioBroadcastCatch audioBroadcastCatch = new AudioBroadcastCatch(device.getDeviceId(), channelId, AudioBroadcastCatchStatus.Ready);
audioBroadcastManager.add(audioBroadcastCatch);
}, eventResultForError -> {
dynamicTask.stop(timeOutTaskKey);
// 发送失败
logger.error("语音广播发送失败： {}:{}", channelId, eventResultForError.msg);
event.call("语音广播发送失败");
audioBroadcastManager.del(device.getDeviceId(), channelId);
stopAudioBroadcast(device.getDeviceId(), channelId);
});
}
@Override
public void stopAudioBroadcast(String deviceId, String channelId){
AudioBroadcastCatch audioBroadcastCatch = audioBroadcastManager.get(deviceId, channelId);
if (audioBroadcastCatch != null) {
audioBroadcastManager.del(deviceId, audioBroadcastCatch.getChannelId());
}
try {
SendRtpItem sendRtpItem = redisCatchStorage.querySendRTPServer(deviceId, channelId, null, null);
if (sendRtpItem != null) {
redisCatchStorage.deleteSendRTPServer(deviceId, sendRtpItem.getChannelId(), null, null);
MediaServerItem mediaInfo = mediaServerService.getOne(sendRtpItem.getMediaServerId());
Map<String, Object> param = new HashMap<>();
param.put("vhost", "__defaultVhost__");
param.put("app", sendRtpItem.getApp());
param.put("stream", sendRtpItem.getStreamId());
zlmresTfulUtils.stopSendRtp(mediaInfo, param);
}
if (audioBroadcastCatch.getStatus() == AudioBroadcastCatchStatus.Ok) {
cmder.streamByeCmd(audioBroadcastCatch.getDialog(), audioBroadcastCatch.getRequest(), null);
}
} catch (SipException e) {
throw new RuntimeException(e);
} catch (ParseException e) {
throw new RuntimeException(e);
}
}
}

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

@ -319,6 +319,22 @@ public class PlayController {
return result;
}
@ApiOperation("停止语音广播")
@ApiImplicitParams({
@ApiImplicitParam(name = "deviceId", value = "设备Id", dataTypeClass = String.class),
@ApiImplicitParam(name = "channelId", value = "通道Id", dataTypeClass = String.class),
})
@GetMapping("/broadcast/stop/{deviceId}/{channelId}")
@PostMapping("/broadcast/stop/{deviceId}/{channelId}")
public WVPResult<String> stopBroadcastA(@PathVariable String deviceId, @PathVariable String channelId) {
if (logger.isDebugEnabled()) {
logger.debug("停止语音广播API调用");
}
playService.stopAudioBroadcast(deviceId, channelId);
return new WVPResult<>(0, "success", null);
}
@ApiOperation("获取所有的ssrc")
@GetMapping("/ssrc")
public WVPResult<JSONObject> getSsrc() {