From 5d901b5e3f033e8b04e53420d68626cbd87431c8 Mon Sep 17 00:00:00 2001 From: 648540858 <648540858@qq.com> Date: 星期五, 06 五月 2022 10:12:34 +0800 Subject: [PATCH] 使用阿里代码规范。规范代码写法 --- src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java | 450 +++++++++++++++++++++++++++++++++++++++++++++++++++----- 1 files changed, 410 insertions(+), 40 deletions(-) diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java index 3692e54..d710dad 100644 --- a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java +++ b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java @@ -1,31 +1,35 @@ package com.genersoft.iot.vmp.service.impl; import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.JSONArray; import com.alibaba.fastjson.JSONObject; import com.alibaba.fastjson.TypeReference; -import com.genersoft.iot.vmp.gb28181.bean.GbStream; +import com.genersoft.iot.vmp.conf.UserSetting; +import com.genersoft.iot.vmp.gb28181.bean.*; +import com.genersoft.iot.vmp.gb28181.event.EventPublisher; +import com.genersoft.iot.vmp.gb28181.event.subscribe.catalog.CatalogEvent; import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils; -import com.genersoft.iot.vmp.media.zlm.dto.IMediaServerItem; -import com.genersoft.iot.vmp.media.zlm.dto.MediaItem; -import com.genersoft.iot.vmp.media.zlm.dto.MediaServerItem; -import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem; +import com.genersoft.iot.vmp.media.zlm.dto.*; +import com.genersoft.iot.vmp.service.IGbStreamService; import com.genersoft.iot.vmp.service.IMediaServerService; import com.genersoft.iot.vmp.service.IStreamPushService; import com.genersoft.iot.vmp.storager.IRedisCatchStorage; -import com.genersoft.iot.vmp.storager.dao.GbStreamMapper; -import com.genersoft.iot.vmp.storager.dao.StreamPushMapper; +import com.genersoft.iot.vmp.storager.dao.*; import com.github.pagehelper.PageHelper; import com.github.pagehelper.PageInfo; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; +import org.springframework.util.StringUtils; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; +import java.util.*; +import java.util.stream.Collectors; @Service public class StreamPushServiceImpl implements IStreamPushService { + + private final static Logger logger = LoggerFactory.getLogger(StreamPushServiceImpl.class); @Autowired private GbStreamMapper gbStreamMapper; @@ -34,17 +38,37 @@ private StreamPushMapper streamPushMapper; @Autowired + private ParentPlatformMapper parentPlatformMapper; + + @Autowired + private PlatformCatalogMapper platformCatalogMapper; + + @Autowired + private PlatformGbStreamMapper platformGbStreamMapper; + + @Autowired + private IGbStreamService gbStreamService; + + @Autowired + private EventPublisher eventPublisher; + + @Autowired private ZLMRESTfulUtils zlmresTfulUtils; @Autowired private IRedisCatchStorage redisCatchStorage; @Autowired + private UserSetting userSetting; + + @Autowired private IMediaServerService mediaServerService; @Override - public List<StreamPushItem> handleJSON(String jsonData, IMediaServerItem mediaServerItem) { - if (jsonData == null) return null; + public List<StreamPushItem> handleJSON(String jsonData, MediaServerItem mediaServerItem) { + if (jsonData == null) { + return null; + } Map<String, StreamPushItem> result = new HashMap<>(); @@ -52,57 +76,403 @@ for (MediaItem item : mediaItems) { // 涓嶄繚瀛樺浗鏍囨帹鐞嗕互鍙婃媺娴佷唬鐞嗙殑娴� - if (item.getOriginType() == 3 || item.getOriginType() == 4 || item.getOriginType() == 5) { - continue; - } - String key = item.getApp() + "_" + item.getStream(); - StreamPushItem streamPushItem = result.get(key); - if (streamPushItem == null) { - streamPushItem = new StreamPushItem(); - streamPushItem.setApp(item.getApp()); - streamPushItem.setMediaServerId(mediaServerItem.getId()); - streamPushItem.setStream(item.getStream()); - streamPushItem.setAliveSecond(item.getAliveSecond()); - streamPushItem.setCreateStamp(item.getCreateStamp()); - streamPushItem.setOriginSock(item.getOriginSock()); - streamPushItem.setTotalReaderCount(item.getTotalReaderCount()); - streamPushItem.setOriginType(item.getOriginType()); - streamPushItem.setOriginTypeStr(item.getOriginTypeStr()); - streamPushItem.setOriginUrl(item.getOriginUrl()); - streamPushItem.setCreateStamp(item.getCreateStamp()); - streamPushItem.setAliveSecond(item.getAliveSecond()); - streamPushItem.setStatus(true); - streamPushItem.setVhost(item.getVhost()); - result.put(key, streamPushItem); + if (item.getOriginType() == OriginType.RTSP_PUSH.ordinal() + || item.getOriginType() == OriginType.RTMP_PUSH.ordinal() + || item.getOriginType() == OriginType.RTC_PUSH.ordinal() ) { + String key = item.getApp() + "_" + item.getStream(); + StreamPushItem streamPushItem = result.get(key); + if (streamPushItem == null) { + streamPushItem = transform(item); + result.put(key, streamPushItem); + } } } return new ArrayList<>(result.values()); } + @Override + public StreamPushItem transform(MediaItem item) { + StreamPushItem streamPushItem = new StreamPushItem(); + streamPushItem.setApp(item.getApp()); + streamPushItem.setMediaServerId(item.getMediaServerId()); + streamPushItem.setStream(item.getStream()); + streamPushItem.setAliveSecond(item.getAliveSecond()); + streamPushItem.setOriginSock(item.getOriginSock()); + streamPushItem.setTotalReaderCount(item.getTotalReaderCount()); + streamPushItem.setOriginType(item.getOriginType()); + streamPushItem.setOriginTypeStr(item.getOriginTypeStr()); + streamPushItem.setOriginUrl(item.getOriginUrl()); + streamPushItem.setCreateStamp(item.getCreateStamp() * 1000); + streamPushItem.setAliveSecond(item.getAliveSecond()); + streamPushItem.setStatus(true); + streamPushItem.setStreamType("push"); + streamPushItem.setVhost(item.getVhost()); + return streamPushItem; + } @Override - public PageInfo<StreamPushItem> getPushList(Integer page, Integer count) { + public PageInfo<StreamPushItem> getPushList(Integer page, Integer count, String query, Boolean pushing, String mediaServerId) { PageHelper.startPage(page, count); - List<StreamPushItem> all = streamPushMapper.selectAll(); + List<StreamPushItem> all = streamPushMapper.selectAllForList(query, pushing, mediaServerId); return new PageInfo<>(all); + } + + @Override + public List<StreamPushItem> getPushList(String mediaServerId) { + return streamPushMapper.selectAllByMediaServerIdWithOutGbID(mediaServerId); } @Override public boolean saveToGB(GbStream stream) { stream.setStreamType("push"); stream.setStatus(true); + stream.setCreateStamp(System.currentTimeMillis()); int add = gbStreamMapper.add(stream); + + // 鏌ユ壘寮�鍚簡鍏ㄩ儴鐩存挱娴佸叡浜殑涓婄骇骞冲彴 + List<ParentPlatform> parentPlatforms = parentPlatformMapper.selectAllAhareAllLiveStream(); + if (parentPlatforms.size() > 0) { + for (ParentPlatform parentPlatform : parentPlatforms) { + stream.setCatalogId(parentPlatform.getCatalogId()); + stream.setPlatformId(parentPlatform.getServerGBId()); + String streamId = stream.getStream(); + StreamProxyItem streamProxyItem = platformGbStreamMapper.selectOne(stream.getApp(), streamId, parentPlatform.getServerGBId()); + if (streamProxyItem == null) { + platformGbStreamMapper.add(stream); + eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), stream, CatalogEvent.ADD); + }else { + if (!streamProxyItem.getGbId().equals(stream.getGbId())) { + // 姝ゆ祦浣跨敤鍙︿竴涓浗鏍嘔d宸茬粡涓庤骞冲彴鍏宠仈锛岀Щ闄ゆ璁板綍 + platformGbStreamMapper.delByAppAndStreamAndPlatform(stream.getApp(), streamId, parentPlatform.getServerGBId()); + platformGbStreamMapper.add(stream); + eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), stream, CatalogEvent.ADD); + } + } + } + } + return add > 0; } @Override public boolean removeFromGB(GbStream stream) { + // 鍒ゆ柇鏄惁闇�瑕佸彂閫佷簨浠� + gbStreamService.sendCatalogMsg(stream, CatalogEvent.DEL); + platformGbStreamMapper.delByAppAndStream(stream.getApp(), stream.getStream()); int del = gbStreamMapper.del(stream.getApp(), stream.getStream()); - IMediaServerItem mediaInfo = mediaServerService.getOne(stream.getMediaServerId()); + MediaServerItem mediaInfo = mediaServerService.getOne(stream.getMediaServerId()); JSONObject mediaList = zlmresTfulUtils.getMediaList(mediaInfo, stream.getApp(), stream.getStream()); - if (mediaList == null) { - streamPushMapper.del(stream.getApp(), stream.getStream()); + if (mediaList != null) { + if (mediaList.getInteger("code") == 0) { + JSONArray data = mediaList.getJSONArray("data"); + if (data == null) { + streamPushMapper.del(stream.getApp(), stream.getStream()); + } + } } return del > 0; } + + + @Override + public StreamPushItem getPush(String app, String streamId) { + + return streamPushMapper.selectOne(app, streamId); + } + + @Override + public boolean stop(String app, String streamId) { + StreamPushItem streamPushItem = streamPushMapper.selectOne(app, streamId); + gbStreamService.sendCatalogMsg(streamPushItem, CatalogEvent.DEL); + + platformGbStreamMapper.delByAppAndStream(app, streamId); + gbStreamMapper.del(app, streamId); + int delStream = streamPushMapper.del(app, streamId); + if (delStream > 0) { + MediaServerItem mediaServerItem = mediaServerService.getOne(streamPushItem.getMediaServerId()); + zlmresTfulUtils.closeStreams(mediaServerItem,app, streamId); + } + return true; + } + + @Override + public void zlmServerOnline(String mediaServerId) { + // 鍚屾zlm鎺ㄦ祦淇℃伅 + MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId); + if (mediaServerItem == null) { + return; + } + // 鏁版嵁搴撹褰� + List<StreamPushItem> pushList = getPushList(mediaServerId); + Map<String, StreamPushItem> pushItemMap = new HashMap<>(); + // redis璁板綍 + List<MediaItem> mediaItems = redisCatchStorage.getStreams(mediaServerId, "PUSH"); + Map<String, MediaItem> streamInfoPushItemMap = new HashMap<>(); + if (pushList.size() > 0) { + for (StreamPushItem streamPushItem : pushList) { + if (StringUtils.isEmpty(streamPushItem.getGbId())) { + pushItemMap.put(streamPushItem.getApp() + streamPushItem.getStream(), streamPushItem); + } + } + } + if (mediaItems.size() > 0) { + for (MediaItem mediaItem : mediaItems) { + streamInfoPushItemMap.put(mediaItem.getApp() + mediaItem.getStream(), mediaItem); + } + } + zlmresTfulUtils.getMediaList(mediaServerItem, (mediaList ->{ + if (mediaList == null) { + return; + } + String dataStr = mediaList.getString("data"); + + Integer code = mediaList.getInteger("code"); + List<StreamPushItem> streamPushItems = null; + if (code == 0 ) { + if (dataStr != null) { + streamPushItems = handleJSON(dataStr, mediaServerItem); + } + } + + if (streamPushItems != null) { + for (StreamPushItem streamPushItem : streamPushItems) { + pushItemMap.remove(streamPushItem.getApp() + streamPushItem.getStream()); + streamInfoPushItemMap.remove(streamPushItem.getApp() + streamPushItem.getStream()); + } + } + List<StreamPushItem> offlinePushItems = new ArrayList<>(pushItemMap.values()); + if (offlinePushItems.size() > 0) { + String type = "PUSH"; + int runLimit = 300; + if (offlinePushItems.size() > runLimit) { + for (int i = 0; i < offlinePushItems.size(); i += runLimit) { + int toIndex = i + runLimit; + if (i + runLimit > offlinePushItems.size()) { + toIndex = offlinePushItems.size(); + } + List<StreamPushItem> streamPushItemsSub = offlinePushItems.subList(i, toIndex); + streamPushMapper.delAll(streamPushItemsSub); + } + }else { + streamPushMapper.delAll(offlinePushItems); + } + + } + Collection<MediaItem> offlineMediaItemList = streamInfoPushItemMap.values(); + if (offlineMediaItemList.size() > 0) { + String type = "PUSH"; + for (MediaItem offlineMediaItem : offlineMediaItemList) { + JSONObject jsonObject = new JSONObject(); + jsonObject.put("serverId", userSetting.getServerId()); + jsonObject.put("app", offlineMediaItem.getApp()); + jsonObject.put("stream", offlineMediaItem.getStream()); + jsonObject.put("register", false); + jsonObject.put("mediaServerId", mediaServerId); + redisCatchStorage.sendStreamChangeMsg(type, jsonObject); + // 绉婚櫎redis鍐呮祦鐨勪俊鎭� + redisCatchStorage.removeStream(mediaServerItem.getId(), "PUSH", offlineMediaItem.getApp(), offlineMediaItem.getStream()); + } + } + })); + } + + @Override + public void zlmServerOffline(String mediaServerId) { + List<StreamPushItem> streamPushItems = streamPushMapper.selectAllByMediaServerIdWithOutGbID(mediaServerId); + // 绉婚櫎娌℃湁GBId鐨勬帹娴� + streamPushMapper.deleteWithoutGBId(mediaServerId); + gbStreamMapper.deleteWithoutGBId("push", mediaServerId); + // 鍏朵粬鐨勬祦璁剧疆鏈惎鐢� + gbStreamMapper.updateStatusByMediaServerId(mediaServerId, false); + // 鍙戦�佹祦鍋滄娑堟伅 + String type = "PUSH"; + // 鍙戦�乺edis娑堟伅 + List<MediaItem> streamInfoList = redisCatchStorage.getStreams(mediaServerId, type); + if (streamInfoList.size() > 0) { + for (MediaItem mediaItem : streamInfoList) { + // 绉婚櫎redis鍐呮祦鐨勪俊鎭� + redisCatchStorage.removeStream(mediaServerId, type, mediaItem.getApp(), mediaItem.getStream()); + JSONObject jsonObject = new JSONObject(); + jsonObject.put("serverId", userSetting.getServerId()); + jsonObject.put("app", mediaItem.getApp()); + jsonObject.put("stream", mediaItem.getStream()); + jsonObject.put("register", false); + jsonObject.put("mediaServerId", mediaServerId); + redisCatchStorage.sendStreamChangeMsg(type, jsonObject); + } + } + } + + @Override + public void clean() { + + } + + @Override + public boolean saveToRandomGB() { + List<StreamPushItem> streamPushItems = streamPushMapper.selectAll(); + long gbId = 100001; + for (StreamPushItem streamPushItem : streamPushItems) { + streamPushItem.setStreamType("push"); + streamPushItem.setStatus(true); + streamPushItem.setGbId("34020000004111" + gbId); + streamPushItem.setCreateStamp(System.currentTimeMillis()); + gbId ++; + } + int limitCount = 30; + + if (streamPushItems.size() > limitCount) { + for (int i = 0; i < streamPushItems.size(); i += limitCount) { + int toIndex = i + limitCount; + if (i + limitCount > streamPushItems.size()) { + toIndex = streamPushItems.size(); + } + gbStreamMapper.batchAdd(streamPushItems.subList(i, toIndex)); + } + }else { + gbStreamMapper.batchAdd(streamPushItems); + } + return true; + } + + @Override + public void batchAdd(List<StreamPushItem> streamPushItems) { + streamPushMapper.addAll(streamPushItems); + gbStreamMapper.batchAdd(streamPushItems); + // 鏌ユ壘寮�鍚簡鍏ㄩ儴鐩存挱娴佸叡浜殑涓婄骇骞冲彴 + List<ParentPlatform> parentPlatforms = parentPlatformMapper.selectAllAhareAllLiveStream(); + if (parentPlatforms.size() > 0) { + for (StreamPushItem stream : streamPushItems) { + for (ParentPlatform parentPlatform : parentPlatforms) { + stream.setCatalogId(parentPlatform.getCatalogId()); + stream.setPlatformId(parentPlatform.getServerGBId()); + String streamId = stream.getStream(); + StreamProxyItem streamProxyItem = platformGbStreamMapper.selectOne(stream.getApp(), streamId, parentPlatform.getServerGBId()); + if (streamProxyItem == null) { + platformGbStreamMapper.add(stream); + eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), stream, CatalogEvent.ADD); + }else { + if (!streamProxyItem.getGbId().equals(stream.getGbId())) { + // 姝ゆ祦浣跨敤鍙︿竴涓浗鏍嘔d宸茬粡涓庤骞冲彴鍏宠仈锛岀Щ闄ゆ璁板綍 + platformGbStreamMapper.delByAppAndStreamAndPlatform(stream.getApp(), streamId, parentPlatform.getServerGBId()); + platformGbStreamMapper.add(stream); + eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), stream, CatalogEvent.ADD); + stream.setGbId(streamProxyItem.getGbId()); + eventPublisher.catalogEventPublishForStream(parentPlatform.getServerGBId(), stream, CatalogEvent.DEL); + } + } + } + } + } + } + + @Override + public void batchAddForUpload(List<StreamPushItem> streamPushItems, Map<String, List<String[]>> streamPushItemsForAll ) { + // 瀛樺偍鏁版嵁鍒皊tream_push琛� + streamPushMapper.addAll(streamPushItems); + List<StreamPushItem> streamPushItemForGbStream = streamPushItems.stream() + .filter(streamPushItem-> streamPushItem.getId() != null) + .collect(Collectors.toList()); + // 瀛樺偍鏁版嵁鍒癵b_stream琛紝 id浼氳繑鍥炲埌streamPushItemForGbStream閲� + if (streamPushItemForGbStream.size() > 0) { + gbStreamMapper.batchAdd(streamPushItemForGbStream); + } + // 鍘婚櫎娌℃湁ID涔熷氨鏄病鏈夊瓨鍌ㄥ埌鏁版嵁搴撶殑鏁版嵁 + List<StreamPushItem> streamPushItemsForPlatform = streamPushItemForGbStream.stream() + .filter(streamPushItem-> streamPushItem.getGbStreamId() != null) + .collect(Collectors.toList()); + + if (streamPushItemsForPlatform.size() > 0) { + // 鑾峰彇鎵�鏈夊钩鍙帮紝骞冲彴鍜岀洰褰曚俊鎭竴鑸笉浼氱壒鍒ぇ閲忋�� + List<ParentPlatform> parentPlatformList = parentPlatformMapper.getParentPlatformList(); + Map<String, Map<String, PlatformCatalog>> platformInfoMap = new HashMap<>(); + if (parentPlatformList.size() == 0) { + return; + } + for (ParentPlatform platform : parentPlatformList) { + Map<String, PlatformCatalog> catalogMap = new HashMap<>(); + + // 鍒涘缓鏍硅妭鐐� + PlatformCatalog platformCatalog = new PlatformCatalog(); + platformCatalog.setId(platform.getServerGBId()); + catalogMap.put(platform.getServerGBId(), platformCatalog); + + // 鏌ヨ鎵�鏈夎妭鐐逛俊鎭� + List<PlatformCatalog> platformCatalogs = platformCatalogMapper.selectByPlatForm(platform.getServerGBId()); + if (platformCatalogs.size() > 0) { + for (PlatformCatalog catalog : platformCatalogs) { + catalogMap.put(catalog.getId(), catalog); + } + } + platformInfoMap.put(platform.getServerGBId(), catalogMap); + } + List<StreamPushItem> streamPushItemListFroPlatform = new ArrayList<>(); + Map<String, List<GbStream>> platformForEvent = new HashMap<>(); + // 閬嶅巻瀛樺偍缁撴灉锛屾煡鎵綼pp+Stream->platformId+catalogId鐨勫搴斿叧绯伙紝鐒跺悗鎵ц鎵归噺鍐欏叆 + for (StreamPushItem streamPushItem : streamPushItemsForPlatform) { + List<String[]> platFormInfoList = streamPushItemsForAll.get(streamPushItem.getApp() + streamPushItem.getStream()); + if (platFormInfoList != null && platFormInfoList.size() > 0) { + for (String[] platFormInfoArray : platFormInfoList) { + StreamPushItem streamPushItemForPlatform = new StreamPushItem(); + streamPushItemForPlatform.setGbStreamId(streamPushItem.getGbStreamId()); + if (platFormInfoArray.length > 0) { + // 鏁扮粍 platFormInfoArray 0 涓哄钩鍙癐D銆� 1涓虹洰褰旾D + // 涓嶅瓨鍦ㄨ繖涓钩鍙帮紝鍒欏拷鐣ュ鍏ユ鍏宠仈鍏崇郴 + if (platformInfoMap.get(platFormInfoArray[0]) == null + || platformInfoMap.get(platFormInfoArray[0]).get(platFormInfoArray[1]) == null) { + logger.info("瀵煎叆鏁版嵁鏃朵笉瀛樺湪骞冲彴鎴栫洰褰晎}/{},宸插鍏ユ湭鍒嗛厤", platFormInfoArray[0], platFormInfoArray[1] ); + continue; + } + streamPushItemForPlatform.setPlatformId(platFormInfoArray[0]); + List<GbStream> gbStreamList = platformForEvent.get(platFormInfoArray[0]); + if (gbStreamList == null) { + gbStreamList = new ArrayList<>(); + platformForEvent.put(platFormInfoArray[0], gbStreamList); + } + // 涓哄彂閫侀�氱煡鏁寸悊鏁版嵁 + streamPushItemForPlatform.setName(streamPushItem.getName()); + streamPushItemForPlatform.setApp(streamPushItem.getApp()); + streamPushItemForPlatform.setStream(streamPushItem.getStream()); + streamPushItemForPlatform.setGbId(streamPushItem.getGbId()); + gbStreamList.add(streamPushItemForPlatform); + } + if (platFormInfoArray.length > 1) { + streamPushItemForPlatform.setCatalogId(platFormInfoArray[1]); + } + streamPushItemListFroPlatform.add(streamPushItemForPlatform); + } + + } + } + if (streamPushItemListFroPlatform.size() > 0) { + platformGbStreamMapper.batchAdd(streamPushItemListFroPlatform); + // 鍙戦�侀�氱煡 + for (String platformId : platformForEvent.keySet()) { + eventPublisher.catalogEventPublishForStream( + platformId, platformForEvent.get(platformId), CatalogEvent.ADD); + } + } + + } + } + + @Override + public boolean batchStop(List<GbStream> gbStreams) { + if (gbStreams == null || gbStreams.size() == 0) { + return false; + } + gbStreamService.sendCatalogMsgs(gbStreams, CatalogEvent.DEL); + + platformGbStreamMapper.delByGbStreams(gbStreams); + gbStreamMapper.batchDelForGbStream(gbStreams); + int delStream = streamPushMapper.delAllForGbStream(gbStreams); + if (delStream > 0) { + for (GbStream gbStream : gbStreams) { + MediaServerItem mediaServerItem = mediaServerService.getOne(gbStream.getMediaServerId()); + zlmresTfulUtils.closeStreams(mediaServerItem, gbStream.getApp(), gbStream.getStream()); + } + + } + return true; + } } -- Gitblit v1.8.0