From 381c3bdc2079ece5147cf4cee004e9071edadf7a Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期四, 04 五月 2023 16:04:44 +0800 Subject: [PATCH] 修复国标点播下级平台,ssrc更新的时单端口错误更新rtpserver的问题 --- src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaListManager.java | 159 ++++++++++++++-------------------------------------- 1 files changed, 43 insertions(+), 116 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaListManager.java b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaListManager.java index 3d9b7cb..db2beb0 100644 --- a/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaListManager.java +++ b/src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaListManager.java @@ -1,29 +1,26 @@ package com.genersoft.iot.vmp.media.zlm; -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.media.zlm.dto.hook.OnStreamChangedHookParam; +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; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; import com.genersoft.iot.vmp.storager.IVideoManagerStorage; import com.genersoft.iot.vmp.storager.dao.GbStreamMapper; import com.genersoft.iot.vmp.storager.dao.PlatformGbStreamMapper; import com.genersoft.iot.vmp.storager.dao.StreamPushMapper; import com.genersoft.iot.vmp.utils.DateUtil; -import org.checkerframework.checker.units.qual.C; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; -import org.springframework.util.StringUtils; +import java.text.ParseException; import java.util.*; import java.util.concurrent.ConcurrentHashMap; -import java.util.regex.Matcher; -import java.util.regex.Pattern; /** * @author lin @@ -58,103 +55,60 @@ private StreamPushMapper streamPushMapper; @Autowired - private ZLMHttpHookSubscribe subscribe; + private ZlmHttpHookSubscribe subscribe; @Autowired private UserSetting userSetting; - private Map<String, ChannelOnlineEvent> channelOnlineEvents = new ConcurrentHashMap<>(); + @Autowired + private ZLMRTPServerFactory zlmrtpServerFactory; + @Autowired + private IMediaServerService mediaServerService; - public void updateMediaList(MediaServerItem mediaServerItem) { - storager.clearMediaList(); + private Map<String, ChannelOnlineEvent> channelOnPublishEvents = new ConcurrentHashMap<>(); - // 浣跨敤寮傛鐨勫綋鏃舵洿鏂板獟浣撴祦鍒楄〃 - zlmresTfulUtils.getMediaList(mediaServerItem, (mediaList ->{ - if (mediaList == null) { - return; - } - String dataStr = mediaList.getString("data"); - - Integer code = mediaList.getInteger("code"); - Map<String, StreamPushItem> result = new HashMap<>(); - List<StreamPushItem> streamPushItems = null; - // 鑾峰彇鎵�鏈夌殑鍥芥爣鍏宠仈 -// List<GbStream> gbStreams = gbStreamMapper.selectAllByMediaServerId(mediaServerItem.getId()); - if (code == 0 ) { - if (dataStr != null) { - streamPushItems = streamPushService.handleJSON(dataStr, mediaServerItem); - } - }else { - logger.warn("鏇存柊瑙嗛娴佸け璐ワ紝閿欒code锛� " + code); - } - - if (streamPushItems != null) { - storager.updateMediaList(streamPushItems); - for (StreamPushItem streamPushItem : streamPushItems) { - JSONObject jsonObject = new JSONObject(); - jsonObject.put("app", streamPushItem.getApp()); - jsonObject.put("stream", streamPushItem.getStream()); - jsonObject.put("mediaServerId", mediaServerItem.getId()); - subscribe.addSubscribe(ZLMHttpHookSubscribe.HookType.on_play,jsonObject, - (MediaServerItem mediaServerItemInuse, JSONObject response)->{ - updateMedia(mediaServerItem, response.getString("app"), response.getString("stream")); - } - ); - } - } - })); - - } - - public void addMedia(MediaServerItem mediaServerItem, String app, String streamId) { - //浣跨敤寮傛鏇存柊鎺ㄦ祦 - updateMedia(mediaServerItem, app, streamId); - } - - public StreamPushItem addPush(MediaItem mediaItem) { - // 鏌ユ壘姝ょ洿鎾祦鏄惁瀛樺湪redis棰勮gbId - StreamPushItem transform = streamPushService.transform(mediaItem); - StreamPushItem pushInDb = streamPushService.getPush(mediaItem.getApp(), mediaItem.getStream()); - transform.setPushIng(true); + public StreamPushItem addPush(OnStreamChangedHookParam onStreamChangedHookParam) { + StreamPushItem transform = streamPushService.transform(onStreamChangedHookParam); + StreamPushItem pushInDb = streamPushService.getPush(onStreamChangedHookParam.getApp(), onStreamChangedHookParam.getStream()); + transform.setPushIng(onStreamChangedHookParam.isRegist()); transform.setUpdateTime(DateUtil.getNow()); transform.setPushTime(DateUtil.getNow()); + transform.setSelf(userSetting.getServerId().equals(onStreamChangedHookParam.getSeverId())); if (pushInDb == null) { transform.setCreateTime(DateUtil.getNow()); streamPushMapper.add(transform); }else { streamPushMapper.update(transform); + gbStreamMapper.updateMediaServer(onStreamChangedHookParam.getApp(), onStreamChangedHookParam.getStream(), onStreamChangedHookParam.getMediaServerId()); + } + ChannelOnlineEvent channelOnlineEventLister = getChannelOnlineEventLister(transform.getApp(), transform.getStream()); + if ( channelOnlineEventLister != null) { + try { + channelOnlineEventLister.run(transform.getApp(), transform.getStream(), transform.getServerId());; + } catch (ParseException e) { + logger.error("addPush: ", e); + } + removedChannelOnlineEventLister(transform.getApp(), transform.getStream()); } return transform; } - - public void updateMedia(MediaServerItem mediaServerItem, String app, String streamId) { - //浣跨敤寮傛鏇存柊鎺ㄦ祦 - zlmresTfulUtils.getMediaList(mediaServerItem, app, streamId, "rtmp", json->{ - - if (json == null) { - return; - } - String dataStr = json.getString("data"); - - Integer code = json.getInteger("code"); - Map<String, StreamPushItem> result = new HashMap<>(); - List<StreamPushItem> streamPushItems = null; - if (code == 0 ) { - if (dataStr != null) { - streamPushItems = streamPushService.handleJSON(dataStr, mediaServerItem); + public void sendStreamEvent(String app, String stream, String mediaServerId) { + MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId); + // 鏌ョ湅鎺ㄦ祦鐘舵�� + if (zlmrtpServerFactory.isStreamReady(mediaServerItem, app, stream)) { + ChannelOnlineEvent channelOnlineEventLister = getChannelOnlineEventLister(app, stream); + if (channelOnlineEventLister != null) { + try { + channelOnlineEventLister.run(app, stream, mediaServerId); + } catch (ParseException e) { + logger.error("sendStreamEvent: ", e); } - }else { - logger.warn("鏇存柊瑙嗛娴佸け璐ワ紝閿欒code锛� " + code); + removedChannelOnlineEventLister(app, stream); } - - if (streamPushItems != null && streamPushItems.size() == 1) { - storager.updateMedia(streamPushItems.get(0)); - } - }); + } } - public int removeMedia(String app, String streamId) { // 鏌ユ壘鏄惁鍏宠仈浜嗗浗鏍囷紝 鍏宠仈浜嗕笉鍒犻櫎锛� 缃负绂荤嚎 @@ -163,48 +117,21 @@ 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); -// } -// } } -- Gitblit v1.8.0