Shikong/wvp-gb28181-project
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

bug修复

Browse Source
This commit is contained in:
648540858 2022-07-26 11:43:56 +08:00
parent 3d058b506e
commit 14270c619e
12 changed files with 71 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/main/java/com/genersoft/iot/vmp/gb28181/task/impl/MobilePositionSubscribeTask.java
View File

@ -39,9 +39,9 @@ public class MobilePositionSubscribeTask implements ISubscribeTask {
dynamicTask.stop(taskKey);
}
sipCommander.mobilePositionSubscribe(device, dialog, eventResult -> {
// if (eventResult.dialog != null || eventResult.dialog.getState().equals(DialogState.CONFIRMED)) {
// dialog = eventResult.dialog;
// }
if (eventResult.dialog != null || eventResult.dialog.getState().equals(DialogState.CONFIRMED)) {
dialog = eventResult.dialog;
}
ResponseEvent event = (ResponseEvent) eventResult.event;
if (event.getResponse().getRawContent() != null) {
// 成功

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

@ -419,12 +419,12 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
}
}
} else if (gbStream != null) {
if (streamPushItem != null && streamPushItem.isStatus()) {
// 在线状态
if (streamPushItem != null && streamPushItem.isPushIng()) {
// 推流状态
pushStream(evt, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
} else {
// 不在线 拉起
// 未推流 拉起
notifyStreamOnline(evt, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);
}
@ -451,7 +451,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
int port, Boolean tcpActive, boolean mediaTransmissionTCP,
String channelId, String addressStr, String ssrc, String requesterId) throws InvalidArgumentException, ParseException, SipException {
// 推流
if (streamPushItem.getServerId().equals(userSetting.getServerId())) {
if (streamPushItem.isSelf()) {
Boolean streamReady = zlmrtpServerFactory.isStreamReady(mediaServerItem, gbStream.getApp(), gbStream.getStream());
if (streamReady) {
// 自平台内容
@ -500,7 +500,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
String channelId, String addressStr, String ssrc, String requesterId) throws InvalidArgumentException, ParseException, SipException {
if ("proxy".equals(gbStream.getStreamType())) {
// TODO 控制启用以使设备上线
logger.info("[ app={}, stream={} ]通道离线，启用流后开始推流", gbStream.getApp(), gbStream.getStream());
logger.info("[ app={}, stream={} ]通道未推流，启用流后开始推流", gbStream.getApp(), gbStream.getStream());
responseAck(evt, Response.BAD_REQUEST, "channel [" + gbStream.getGbId() + "] offline");
} else if ("push".equals(gbStream.getStreamType())) {
if (!platform.isStartOfflinePush()) {
@ -508,7 +508,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
return;
}
// 发送redis消息以使设备上线
logger.info("[ app={}, stream={} ]通道离线，发送redis信息控制设备开始推流", gbStream.getApp(), gbStream.getStream());
logger.info("[ app={}, stream={} ]通道未推流，发送redis信息控制设备开始推流", gbStream.getApp(), gbStream.getStream());
MessageForPushChannel messageForPushChannel = MessageForPushChannel.getInstance(1,
gbStream.getApp(), gbStream.getStream(), gbStream.getGbId(), gbStream.getPlatformId(),
@ -518,7 +518,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
dynamicTask.startDelay(callIdHeader.getCallId(), () -> {
logger.info("[ app={}, stream={} ] 等待设备开始推流超时", gbStream.getApp(), gbStream.getStream());
try {
mediaListManager.removedChannelOnlineEventLister(gbStream.getGbId());
mediaListManager.removedChannelOnlineEventLister(gbStream.getApp(), gbStream.getStream());
responseAck(evt, Response.REQUEST_TIMEOUT); // 超时
} catch (SipException e) {
e.printStackTrace();
@ -533,7 +533,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
Boolean finalTcpActive = tcpActive;
// 添加在本机上线的通知
mediaListManager.addChannelOnlineEventLister(gbStream.getGbId(), (app, stream, serverId) -> {
mediaListManager.addChannelOnlineEventLister(gbStream.getApp(), gbStream.getStream(), (app, stream, serverId) -> {
dynamicTask.stop(callIdHeader.getCallId());
if (serverId.equals(userSetting.getServerId())) {
SendRtpItem sendRtpItem = zlmrtpServerFactory.createSendRtpItem(mediaServerItem, addressStr, finalPort, ssrc, requesterId,
@ -621,7 +621,7 @@ public class InviteRequestProcessor extends SIPRequestProcessorParent implements
// 离线
// 查询是否在本机上线了
StreamPushItem currentStreamPushItem = streamPushService.getPush(streamPushItem.getApp(), streamPushItem.getStream());
if (currentStreamPushItem.isStatus()) {
if (currentStreamPushItem.isPushIng()) {
// 在线状态
pushStream(evt, gbStream, streamPushItem, platform, callIdHeader, mediaServerItem, port, tcpActive,
mediaTransmissionTCP, channelId, addressStr, ssrc, requesterId);

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

@ -350,17 +350,17 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
switch (event) {
case CatalogEvent.ON:
// 上线
logger.info("收到来自设备【{}】的通道【{}】上线通知", device.getDeviceId(), channel.getChannelId());
logger.info("[收到通道上线通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
storager.deviceChannelOnline(deviceId, channel.getChannelId());
break;
case CatalogEvent.OFF :
// 离线
logger.info("收到来自设备【{}】的通道【{}】离线通知", device.getDeviceId(), channel.getChannelId());
logger.info("[收到通道离线通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
storager.deviceChannelOffline(deviceId, channel.getChannelId());
break;
case CatalogEvent.VLOST:
// 视频丢失
logger.info("收到来自设备【{}】的通道【{}】视频丢失通知", device.getDeviceId(), channel.getChannelId());
logger.info("[收到通道视频丢失通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
storager.deviceChannelOffline(deviceId, channel.getChannelId());
break;
case CatalogEvent.DEFECT:
@ -368,17 +368,17 @@ public class NotifyRequestProcessor extends SIPRequestProcessorParent implements
break;
case CatalogEvent.ADD:
// 增加
logger.info("收到来自设备【{}】的增加通道【{}】通知", device.getDeviceId(), channel.getChannelId());
logger.info("[收到增加通道通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
deviceChannelService.updateChannel(deviceId, channel);
break;
case CatalogEvent.DEL:
// 删除
logger.info("收到来自设备【{}】的删除通道【{}】通知", device.getDeviceId(), channel.getChannelId());
logger.info("[收到删除通道通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
storager.delChannel(deviceId, channel.getChannelId());
break;
case CatalogEvent.UPDATE:
// 更新
logger.info("收到来自设备【{}】的更新通道【{}】通知", device.getDeviceId(), channel.getChannelId());
logger.info("[收到更新通道通知] 来自设备: {}, 通道 {}", device.getDeviceId(), channel.getChannelId());
deviceChannelService.updateChannel(deviceId, channel);
break;
default:

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

@ -69,7 +69,7 @@ public class AlarmNotifyMessageHandler extends SIPRequestProcessorParent impleme
@Override
public void handForDevice(RequestEvent evt, Device device, Element rootElement) {
logger.info("收到来自设备[{}]的报警通知", device.getDeviceId());
logger.info("[收到报警通知]设备：{}", device.getDeviceId());
// 回复200 OK
try {
responseAck(evt, Response.OK);

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

@ -111,7 +111,7 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp
int sumNum = Integer.parseInt(sumNumElement.getText());
if (sumNum == 0) {
logger.info("收到来自设备【{}】的通道: 0个", take.getDevice().getDeviceId());
logger.info("[收到通道]设备:{}的: 0个", take.getDevice().getDeviceId());
// 数据已经完整接收
storager.cleanChannelsForDevice(take.getDevice().getDeviceId());
catalogDataCatch.setChannelSyncEnd(take.getDevice().getDeviceId(), null);
@ -133,7 +133,7 @@ public class CatalogResponseMessageHandler extends SIPRequestProcessorParent imp
}
int sn = Integer.parseInt(snElement.getText());
catalogDataCatch.put(take.getDevice().getDeviceId(), sn, sumNum, take.getDevice(), channelList);
logger.info("收到来自设备【{}】的通道: {}个，{}/{}", take.getDevice().getDeviceId(), channelList.size(), catalogDataCatch.get(take.getDevice().getDeviceId()) == null ? 0 :catalogDataCatch.get(take.getDevice().getDeviceId()).size(), sumNum);
logger.info("[收到通道]设备: {} -> {}个，{}/{}", take.getDevice().getDeviceId(), channelList.size(), catalogDataCatch.get(take.getDevice().getDeviceId()) == null ? 0 :catalogDataCatch.get(take.getDevice().getDeviceId()).size(), sumNum);
if (catalogDataCatch.get(take.getDevice().getDeviceId()).size() == sumNum) {
// 数据已经完整接收
boolean resetChannelsResult = storager.resetChannels(take.getDevice().getDeviceId(), catalogDataCatch.get(take.getDevice().getDeviceId()));

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

@ -240,6 +240,8 @@ public class ZLMHttpHookListener {
if (mediaInfo != null) {
assistRESTfulUtils.addStreamCallInfo(mediaInfo, param.getApp(), param.getStream(), callId, null);
}
}else {
zlmMediaListManager.sendStreamEvent(param.getApp(),param.getStream(), param.getMediaServerId());
}
ret.put("code", 0);

67
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaListManager.java
View File

@ -4,6 +4,7 @@ import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.conf.UserSetting;
import com.genersoft.iot.vmp.gb28181.bean.GbStream;
import com.genersoft.iot.vmp.media.zlm.dto.*;
import com.genersoft.iot.vmp.service.IMediaServerService;
import com.genersoft.iot.vmp.service.IStreamProxyService;
import com.genersoft.iot.vmp.service.IStreamPushService;
import com.genersoft.iot.vmp.service.bean.ThirdPartyGB;
@ -63,7 +64,13 @@ public class ZLMMediaListManager {
@Autowired
private UserSetting userSetting;
private Map<String, ChannelOnlineEvent> channelOnlineEvents = new ConcurrentHashMap<>();
@Autowired
private ZLMRTPServerFactory zlmrtpServerFactory;
@Autowired
private IMediaServerService mediaServerService;
private Map<String, ChannelOnlineEvent> channelOnPublishEvents = new ConcurrentHashMap<>();
public StreamPushItem addPush(MediaItem mediaItem) {
// 查找此直播流是否存在redis预设gbId
@ -79,9 +86,26 @@ public class ZLMMediaListManager {
}else {
streamPushMapper.update(transform);
}
if (transform != null) {
if (getChannelOnlineEventLister(transform.getApp(), transform.getStream()) != null) {
getChannelOnlineEventLister(transform.getApp(), transform.getStream()).run(transform.getApp(), transform.getStream(), transform.getServerId());
removedChannelOnlineEventLister(transform.getApp(), transform.getStream());
}
}
return transform;
}
public void sendStreamEvent(String app, String stream, String mediaServerId) {
MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
// 查看推流状态
if (zlmrtpServerFactory.isStreamReady(mediaServerItem, app, stream)) {
if (getChannelOnlineEventLister(app, stream) != null) {
getChannelOnlineEventLister(app, stream).run(app, stream, mediaServerId);
removedChannelOnlineEventLister(app, stream);
}
}
}
public int removeMedia(String app, String streamId) {
// 查找是否关联了国标， 关联了不删除， 置为离线
GbStream gbStream = gbStreamMapper.selectOne(app, streamId);
@ -89,48 +113,21 @@ public class ZLMMediaListManager {
if (gbStream == null) {
result = storager.removeMedia(app, streamId);
}else {
// TODO 暂不设置为离线
result =storager.mediaOffline(app, streamId);
}
return result;
}
public void addChannelOnlineEventLister(String key, ChannelOnlineEvent callback) {
this.channelOnlineEvents.put(key,callback);
public void addChannelOnlineEventLister(String app, String stream, ChannelOnlineEvent callback) {
this.channelOnPublishEvents.put(app + "_" + stream, callback);
}
public void removedChannelOnlineEventLister(String key) {
this.channelOnlineEvents.remove(key);
public void removedChannelOnlineEventLister(String app, String stream) {
this.channelOnPublishEvents.remove(app + "_" + stream);
}
public ChannelOnlineEvent getChannelOnlineEventLister(String app, String stream) {
return this.channelOnPublishEvents.get(app + "_" + stream);
}
// public void clearAllSessions() {
// logger.info("清空所有国标相关的session");
// JSONObject allSessionJSON = zlmresTfulUtils.getAllSession();
// ZLMServerConfig mediaInfo = redisCatchStorage.getMediaInfo();
// HashSet<String> allLocalPorts = new HashSet();
// if (allSessionJSON.getInteger("code") == 0) {
// JSONArray data = allSessionJSON.getJSONArray("data");
// if (data.size() > 0) {
// for (int i = 0; i < data.size(); i++) {
// JSONObject sessionJOSN = data.getJSONObject(i);
// Integer local_port = sessionJOSN.getInteger("local_port");
// if (!local_port.equals(Integer.valueOf(mediaInfo.getHttpPort())) &&
// !local_port.equals(Integer.valueOf(mediaInfo.getHttpSSLport())) &&
// !local_port.equals(Integer.valueOf(mediaInfo.getRtmpPort())) &&
// !local_port.equals(Integer.valueOf(mediaInfo.getRtspPort())) &&
// !local_port.equals(Integer.valueOf(mediaInfo.getRtspSSlport())) &&
// !local_port.equals(Integer.valueOf(mediaInfo.getHookOnFlowReport()))){
// allLocalPorts.add(sessionJOSN.getInteger("local_port") + "");
// }
// }
// }
// }
// if (allLocalPorts.size() > 0) {
// List<String> result = new ArrayList<>(allLocalPorts);
// String localPortSStr = String.join(",", result);
// zlmresTfulUtils.kickSessions(localPortSStr);
// }
// }
}

6
src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMRTPServerFactory.java
View File

@ -273,8 +273,10 @@ public class ZLMRTPServerFactory {
* 查询待转推的流是否就绪
*/
public Boolean isStreamReady(MediaServerItem mediaServerItem, String app, String streamId) {
JSONObject mediaInfo = zlmresTfulUtils.getMediaInfo(mediaServerItem, app, "rtmp", streamId);
return (mediaInfo.getInteger("code") == 0 && mediaInfo.getBoolean("online"));
JSONObject mediaInfo = zlmresTfulUtils.getMediaList(mediaServerItem, app, streamId);
return (mediaInfo.getInteger("code") == 0
&& mediaInfo.getJSONArray("data") != null
&& mediaInfo.getJSONArray("data").size() > 0);
}
/**

9
src/main/java/com/genersoft/iot/vmp/service/impl/RedisStreamMsgListener.java
View File

@ -72,9 +72,10 @@ public class RedisStreamMsgListener implements MessageListener {
mediaItem.setOriginType(0);
mediaItem.setOriginTypeStr("0");
mediaItem.setOriginTypeStr("unknown");
zlmMediaListManager.addPush(mediaItem);
if (register) {
zlmMediaListManager.addPush(mediaItem);
}else {
zlmMediaListManager.removeMedia(app, stream);
}
}
}

9
src/main/java/com/genersoft/iot/vmp/vmanager/streamPush/StreamPushController.java
View File

@ -264,9 +264,8 @@ public class StreamPushController {
})
@GetMapping(value = "/getPlayUrl")
@ResponseBody
public WVPResult<StreamInfo> getPlayUrl(HttpServletRequest request, @RequestParam String app,
@RequestParam String stream,
@RequestParam(required = false) String mediaServerId){
public WVPResult<StreamInfo> getPlayUrl(@RequestParam String app,@RequestParam String stream,
@RequestParam(required = false) String mediaServerId){
boolean authority = false;
// 是否登陆用户, 登陆用户返回完整信息
LoginUser userInfo = SecurityUtils.getUserInfo();
@ -275,7 +274,7 @@ public class StreamPushController {
}
WVPResult<StreamInfo> result = new WVPResult<>();
StreamPushItem push = streamPushService.getPush(app, stream);
if (!userSetting.getServerId().equals(push.getServerId()) ) {
if (push != null && !push.isSelf()) {
result.setCode(-1);
result.setMsg("来自其他平台的推流信息");
return result;
@ -283,7 +282,7 @@ public class StreamPushController {
StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, authority);
if (streamInfo != null){
result.setCode(0);
result.setMsg("scccess");
result.setMsg("success");
result.setData(streamInfo);
}else {
result.setCode(-1);

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

@ -76,7 +76,7 @@
<el-table-column label="操作" min-width="360" fixed="right">
<template slot-scope="scope">
<el-button size="medium" icon="el-icon-video-play"
v-if="(scope.row.status == false && scope.row.gbId == null) || scope.row.status"
v-if="scope.row.pushIng === true"
@click="playPush(scope.row)" type="text">播放
</el-button>
<el-divider direction="vertical"></el-divider>

5
web_src/src/components/dialog/catalogEdit.vue
View File

@ -70,12 +70,11 @@ export default {
console.log(catalogType)
// 216 为虚拟组织 215 为业务分组；目录第一级必须为业务分组， 业务分组下为虚拟组织，虚拟组织下可以有其他虚拟组织
if (this.level === 1 && catalogType !== "215") {
return callback(new Error('业务分组模式下第一层目录的编号10到13位必须为215'));
return callback(new Error('业务分组模式下第一层目录的编号11到13位必须为215'));
}
if (this.level > 1 && catalogType !== "216") {
return callback(new Error('业务分组模式下第一层以下目录的编号10到13位必须为216'));
return callback(new Error('业务分组模式下第一层以下目录的编号11到13位必须为216'));
}
}
callback();
}