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

Merge branch 'master' into dev/zlm 

# Conflicts:
#	src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMHttpHookListener.java
#	src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
#	src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamCloseResponseListener.java
#	src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisStreamMsgListener.java
#	src/main/java/com/genersoft/iot/vmp/vmanager/cloudRecord/CloudRecordController.java
pull/1411/head
648540858 2024-04-09 10:35:11 +08:00
parent 0b2f635176 a890b62fa1
commit b1b6fae22c
18 changed files with 161 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
README.md
View File

@ -136,12 +136,6 @@ https://gitee.com/pan648540858/wvp-GB28181-pro.git
[ydpd](https://github.com/ydpd) [szy833](https://github.com/szy833) [ydwxb](https://github.com/ydwxb) [Albertzhu666](https://github.com/Albertzhu666)
[mk1990](https://github.com/mk1990) [SaltFish001](https://github.com/SaltFish001)
同时感谢JetBrains对开源项目的支持，本项目使用IntelliJ IDEA开发与调试：
ffmpeg -re -i 123.mp3 -acodec pcm_alaw -ar 8000 -ac 1 -f rtsp rtsp://192.168.1.3:30554/broadcast/34020000001320000101_34020000001310000001
ffmpeg -re -i 123.mp3 -acodec pcm_alaw -ar 8000 -ac 1 -f rtsp rtsp://192.168.1.3:30554/talk/34020000001320000011_34020000001370000001
ffmpeg -re -i 123.mp3 -acodec pcm_alaw -ar 8000 -ac 1 -f rtsp rtsp://192.168.1.3:30554/talk/34020000001320000101_34020000001310000001
![JetBrains](https://resources.jetbrains.com/storage/products/company/brand/logos/IntelliJ_IDEA_icon.svg?_ga=2.143694769.529214288.1712023294-439039083.1711422571&_gl=1*102dv9n*_ga*NDM5MDM5MDgzLjE3MTE0MjI1NzE.*_ga_9J976DJZ68*MTcxMjEyNjg4NC45LjEuMTcxMjEyNzc2My4zMy4wLjA.)

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

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

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

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

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

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

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

@ -41,6 +41,7 @@ import gov.nist.javax.sdp.fields.TimeField;
import gov.nist.javax.sdp.fields.URIField;
import gov.nist.javax.sip.message.SIPRequest;
import gov.nist.javax.sip.message.SIPResponse;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -407,12 +408,15 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
// * 2 推流中
sendRtpItem.setStatus(1);
redisCatchStorage.updateSendRTPSever(sendRtpItem);
String sdpIp = mediaServerItemInUSe.getSdpIp();
if (!ObjectUtils.isEmpty(platform.getSendStreamIp())) {
sdpIp = platform.getSendStreamIp();
}
StringBuffer content = new StringBuffer(200);
content.append("v=0\r\n");
content.append("o=" + channelId + " 0 0 IN IP4 " + mediaServerItemInUSe.getSdpIp() + "\r\n");
content.append("o=" + channelId + " 0 0 IN IP4 " + sdpIp + "\r\n");
content.append("s=" + sessionName + "\r\n");
content.append("c=IN IP4 " + mediaServerItemInUSe.getSdpIp() + "\r\n");
content.append("c=IN IP4 " + sdpIp + "\r\n");
if ("Playback".equalsIgnoreCase(sessionName)) {
content.append("t=" + finalStartTime + " " + finalStopTime + "\r\n");
} else {
@ -578,14 +582,20 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
}
if ("push".equals(gbStream.getStreamType())) {
if (streamPushItem != null && streamPushItem.isPushIng()) {
// 推流状态
pushStream(evt, request, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
} else {
// 未推流 拉起
notifyStreamOnline(evt, request, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
if (streamPushItem != null) {
// 从redis查询是否正在接收这个推流
OnStreamChangedHookParam pushListItem = redisCatchStorage.getPushListItem(gbStream.getApp(), gbStream.getStream());
if (pushListItem != null) {
StreamPushItem transform = streamPushService.transform(pushListItem);
transform.setSelf(userSetting.getServerId().equals(pushListItem.getSeverId()));
// 推流状态
pushStream(evt, request, gbStream, transform, platform, callIdHeader, mediaServerItem, port, tcpActive,
mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
}else {
// 未推流 拉起
notifyStreamOnline(evt, request, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
}
}
} else if ("proxy".equals(gbStream.getStreamType())) {
if (null != proxyByAppAndStream) {
@ -903,11 +913,15 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
public SIPResponse sendStreamAck(MediaServer mediaServerItem, SIPRequest request, SendRtpItem sendRtpItem, ParentPlatform platform, RequestEvent evt) {
String sdpIp = mediaServerItem.getSdpIp();
if (!ObjectUtils.isEmpty(platform.getSendStreamIp())) {
sdpIp = platform.getSendStreamIp();
}
StringBuffer content = new StringBuffer(200);
content.append("v=0\r\n");
content.append("o=" + sendRtpItem.getChannelId() + " 0 0 IN IP4 " + mediaServerItem.getSdpIp() + "\r\n");
content.append("o=" + sendRtpItem.getChannelId() + " 0 0 IN IP4 " + sdpIp + "\r\n");
content.append("s=Play\r\n");
content.append("c=IN IP4 " + mediaServerItem.getSdpIp() + "\r\n");
content.append("c=IN IP4 " + sdpIp + "\r\n");
content.append("t=0 0\r\n");
// 非严格模式端口不统一, 增加兼容性，修改为一个不为0的端口
int localPort = sendRtpItem.getLocalPort();

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

@ -1275,7 +1275,7 @@ public class PlayServiceImpl implements IPlayService {
dynamicTask.startDelay(key, ()->{
logger.info("[语音广播]等待invite消息超时：{}/{}", device.getDeviceId(), channelId);
stopAudioBroadcast(device.getDeviceId(), channelId);
}, 2000);
}, 10*1000);
}, eventResultForError -> {
// 发送失败
logger.error("语音广播发送失败： {}:{}", channelId, eventResultForError.msg);

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

@ -2,7 +2,6 @@ package com.genersoft.iot.vmp.service.redisMsg;
import com.alibaba.fastjson2.JSON;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.InviteStreamType;
import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
import com.genersoft.iot.vmp.gb28181.bean.SendRtpItem;
import com.genersoft.iot.vmp.gb28181.transmit.cmd.ISIPCommanderForPlatform;
@ -25,7 +24,6 @@ import org.springframework.stereotype.Component;
import javax.sip.InvalidArgumentException;
import javax.sip.SipException;
import java.text.ParseException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

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

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

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

@ -212,5 +212,10 @@ public interface IRedisCatchStorage {
void addPushListItem(String app, String stream, MediaArrivalEvent param);
OnStreamChangedHookParam getPushListItem(String app, String stream);
void removePushListItem(String app, String stream, String mediaServerId);
void sendPushStreamClose(MessageForPushChannel messageForPushChannel);
}

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

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

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

@ -658,6 +658,12 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
redisTemplate.opsForValue().set(key, event);
}
@Override
public OnStreamChangedHookParam getPushListItem(String app, String stream) {
String key = VideoManagerConstants.PUSH_STREAM_LIST + app + "_" + stream;
return (OnStreamChangedHookParam)redisTemplate.opsForValue().get(key);
}
@Override
public void removePushListItem(String app, String stream, String mediaServerId) {
String key = VideoManagerConstants.PUSH_STREAM_LIST + app + "_" + stream;
@ -667,4 +673,11 @@ public class RedisCatchStorageImpl implements IRedisCatchStorage {
}
}
@Override
public void sendPushStreamClose(MessageForPushChannel msg) {
String key = VideoManagerConstants.VM_MSG_STREAM_PUSH_CLOSE_REQUESTED;
logger.info("[redis发送通知] 发送 停止向上级推流 {}: {}/{}->{}", key, msg.getApp(), msg.getStream(), msg.getPlatFormId());
redisTemplate.convertAndSend(key, JSON.toJSON(msg));
}
}

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

@ -4,6 +4,7 @@ import com.alibaba.fastjson2.JSONArray;
import com.genersoft.iot.vmp.conf.exception.ControllerException;
import com.genersoft.iot.vmp.conf.security.JwtUtils;
import com.genersoft.iot.vmp.media.bean.MediaServer;
import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem;
import com.genersoft.iot.vmp.service.ICloudRecordService;
import com.genersoft.iot.vmp.media.service.IMediaServerService;
import com.genersoft.iot.vmp.service.bean.CloudRecordItem;

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

@ -139,7 +139,8 @@ public class PlayController {
wvpResult.setMsg(msg);
}
requestMessage.setData(wvpResult);
resultHolder.invokeResult(requestMessage);
// 此处必须释放所有请求
resultHolder.invokeAllResult(requestMessage);
});
return result;
}

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

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

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

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

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

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

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

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

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

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