From 764d04b497356ba6bcbb75fd42b51eca750f7223 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期三, 29 五月 2024 15:02:51 +0800
Subject: [PATCH] 调整上级观看消息的发送

---
 src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java |  355 +++++++++++++++++++++++++++++++++++++++++-----------------
 1 files changed, 248 insertions(+), 107 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
old mode 100644
new mode 100755
index ed59230..a64b71a
--- 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,38 +1,49 @@
 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.alibaba.fastjson2.JSONObject;
+import com.baomidou.dynamic.datasource.annotation.DS;
+import com.genersoft.iot.vmp.common.StreamInfo;
 import com.genersoft.iot.vmp.conf.MediaConfig;
 import com.genersoft.iot.vmp.conf.UserSetting;
-import com.genersoft.iot.vmp.gb28181.bean.*;
+import com.genersoft.iot.vmp.gb28181.bean.GbStream;
+import com.genersoft.iot.vmp.gb28181.bean.ParentPlatform;
+import com.genersoft.iot.vmp.gb28181.bean.PlatformCatalog;
 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.*;
+import com.genersoft.iot.vmp.media.bean.MediaInfo;
+import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;
+import com.genersoft.iot.vmp.media.event.media.MediaDepartureEvent;
+import com.genersoft.iot.vmp.media.service.IMediaServerService;
+import com.genersoft.iot.vmp.media.bean.MediaServer;
+import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
+import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
+import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
+import com.genersoft.iot.vmp.media.zlm.dto.hook.OriginType;
 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.service.bean.StreamPushItemFromRedis;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.storager.dao.*;
 import com.genersoft.iot.vmp.utils.DateUtil;
+import com.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo;
 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.context.event.EventListener;
 import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.TransactionDefinition;
 import org.springframework.transaction.TransactionStatus;
-import org.springframework.util.StringUtils;
+import org.springframework.util.ObjectUtils;
 
 import java.util.*;
 import java.util.stream.Collectors;
 
 @Service
+@DS("master")
 public class StreamPushServiceImpl implements IStreamPushService {
 
     private final static Logger logger = LoggerFactory.getLogger(StreamPushServiceImpl.class);
@@ -62,9 +73,6 @@
     private EventPublisher eventPublisher;
 
     @Autowired
-    private ZLMRESTfulUtils zlmresTfulUtils;
-
-    @Autowired
     private IRedisCatchStorage redisCatchStorage;
 
     @Autowired
@@ -82,35 +90,127 @@
     @Autowired
     private MediaConfig mediaConfig;
 
-
-    @Override
-    public List<StreamPushItem> handleJSON(String jsonData, MediaServerItem mediaServerItem) {
-        if (jsonData == null) {
-            return null;
+    /**
+     * 娴佸埌鏉ョ殑澶勭悊
+     */
+    @Async("taskExecutor")
+    @EventListener
+    public void onApplicationEvent(MediaArrivalEvent event) {
+        MediaInfo mediaInfo = event.getMediaInfo();
+        if (mediaInfo == null) {
+            return;
+        }
+        if (mediaInfo.getOriginType() != OriginType.RTMP_PUSH.ordinal()
+                && mediaInfo.getOriginType() != OriginType.RTSP_PUSH.ordinal()
+                && mediaInfo.getOriginType() != OriginType.RTC_PUSH.ordinal()) {
+            return;
         }
 
+        StreamAuthorityInfo streamAuthorityInfo = redisCatchStorage.getStreamAuthorityInfo(event.getApp(), event.getStream());
+        if (streamAuthorityInfo == null) {
+            streamAuthorityInfo = StreamAuthorityInfo.getInstanceByHook(event);
+        } else {
+            streamAuthorityInfo.setOriginType(mediaInfo.getOriginType());
+        }
+        redisCatchStorage.updateStreamAuthorityInfo(event.getApp(), event.getStream(), streamAuthorityInfo);
+        StreamPushItem transform = StreamPushItem.getInstance(event, userSetting.getServerId());
+        transform.setPushIng(true);
+        transform.setUpdateTime(DateUtil.getNow());
+        transform.setPushTime(DateUtil.getNow());
+        transform.setSelf(true);
+        StreamPushItem pushInDb = getPush(event.getApp(), event.getStream());
+        if (pushInDb == null) {
+            transform.setCreateTime(DateUtil.getNow());
+            streamPushMapper.add(transform);
+        }else {
+            streamPushMapper.update(transform);
+            gbStreamMapper.updateMediaServer(event.getApp(), event.getStream(), event.getMediaServer().getId());
+        }
+        // TODO 鐩稿叧鐨勪簨浠惰嚜琛岀鐞嗭紝涓嶉渶瑕佸啓鍏LMMediaListManager
+//        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());
+//        }
+        // 鍐椾綑鏁版嵁锛岃嚜宸辩郴缁熶腑鑷敤
+        redisCatchStorage.addPushListItem(event.getApp(), event.getStream(), event);
+
+        // 鍙戦�佹祦鍙樺寲redis娑堟伅
+        JSONObject jsonObject = new JSONObject();
+        jsonObject.put("serverId", userSetting.getServerId());
+        jsonObject.put("app", event.getApp());
+        jsonObject.put("stream", event.getStream());
+        jsonObject.put("register", true);
+        jsonObject.put("mediaServerId", event.getMediaServer().getId());
+        redisCatchStorage.sendStreamChangeMsg(OriginType.values()[event.getMediaInfo().getOriginType()].getType(), jsonObject);
+    }
+
+    /**
+     * 娴佺寮�鐨勫鐞�
+     */
+    @Async("taskExecutor")
+    @EventListener
+    public void onApplicationEvent(MediaDepartureEvent event) {
+        // 鍏煎娴佹敞閿�鏃剁被鍨嬩粠redis璁板綍鑾峰彇
+        MediaInfo mediaInfo = redisCatchStorage.getStreamInfo(
+                event.getApp(), event.getStream(), event.getMediaServer().getId());
+        if (mediaInfo != null) {
+            String type = OriginType.values()[mediaInfo.getOriginType()].getType();
+            redisCatchStorage.removeStream(event.getMediaServer().getId(), type, event.getApp(), event.getStream());
+            if ("PUSH".equalsIgnoreCase(type)) {
+                // 鍐椾綑鏁版嵁锛岃嚜宸辩郴缁熶腑鑷敤
+                redisCatchStorage.removePushListItem(event.getApp(), event.getStream(), event.getMediaServer().getId());
+            }
+            if (type != null) {
+                // 鍙戦�佹祦鍙樺寲redis娑堟伅
+                JSONObject jsonObject = new JSONObject();
+                jsonObject.put("serverId", userSetting.getServerId());
+                jsonObject.put("app", event.getApp());
+                jsonObject.put("stream", event.getStream());
+                jsonObject.put("register", false);
+                jsonObject.put("mediaServerId", event.getMediaServer().getId());
+                redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
+            }
+        }
+        GbStream gbStream = gbStreamMapper.selectOne(event.getApp(), event.getStream());
+        if (gbStream != null) {
+            if (userSetting.isUsePushingAsStatus()) {
+                streamPushMapper.updatePushStatus(event.getApp(), event.getStream(), false);
+                eventPublisher.catalogEventPublishForStream(null, gbStream, CatalogEvent.OFF);
+            }
+        }else {
+            streamPushMapper.del(event.getApp(), event.getStream());
+        }
+    }
+
+
+    private List<StreamPushItem> handleJSON(List<StreamInfo> streamInfoList) {
+        if (streamInfoList == null || streamInfoList.isEmpty()) {
+            return null;
+        }
         Map<String, StreamPushItem> result = new HashMap<>();
-
-        List<MediaItem> mediaItems = JSON.parseObject(jsonData, new TypeReference<List<MediaItem>>() {});
-        for (MediaItem item : mediaItems) {
-
+        for (StreamInfo streamInfo : streamInfoList) {
             // 涓嶄繚瀛樺浗鏍囨帹鐞嗕互鍙婃媺娴佷唬鐞嗙殑娴�
-            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();
+            if (streamInfo.getOriginType() == OriginType.RTSP_PUSH.ordinal()
+                    || streamInfo.getOriginType() == OriginType.RTMP_PUSH.ordinal()
+                    || streamInfo.getOriginType() == OriginType.RTC_PUSH.ordinal() ) {
+                String key = streamInfo.getApp() + "_" + streamInfo.getStream();
                 StreamPushItem streamPushItem = result.get(key);
                 if (streamPushItem == null) {
-                    streamPushItem = transform(item);
+                    streamPushItem = streamPushItem.getInstance(streamInfo);
                     result.put(key, streamPushItem);
                 }
             }
         }
-
         return new ArrayList<>(result.values());
     }
+
     @Override
-    public StreamPushItem transform(MediaItem item) {
+    public StreamPushItem transform(OnStreamChangedHookParam item) {
         StreamPushItem streamPushItem = new StreamPushItem();
         streamPushItem.setApp(item.getApp());
         streamPushItem.setMediaServerId(item.getMediaServerId());
@@ -159,15 +259,10 @@
         gbStreamService.sendCatalogMsg(stream, CatalogEvent.DEL);
         platformGbStreamMapper.delByAppAndStream(stream.getApp(), stream.getStream());
         int del = gbStreamMapper.del(stream.getApp(), stream.getStream());
-        MediaServerItem mediaInfo = mediaServerService.getOne(stream.getMediaServerId());
-        JSONObject mediaList = zlmresTfulUtils.getMediaList(mediaInfo, 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());
-                }
-            }
+        MediaServer mediaInfo = mediaServerService.getOne(stream.getMediaServerId());
+        List<StreamInfo> mediaList = mediaServerService.getMediaList(mediaInfo, stream.getApp(), stream.getStream(), null);
+        if (mediaList != null && mediaList.isEmpty()) {
+            streamPushMapper.del(stream.getApp(), stream.getStream());
         }
         return del > 0;
     }
@@ -180,15 +275,18 @@
 
     @Override
     public boolean stop(String app, String streamId) {
+        logger.info("[鎺ㄦ祦 ] 鍋滄娴侊細 {}/{}", app, streamId);
         StreamPushItem streamPushItem = streamPushMapper.selectOne(app, streamId);
-        gbStreamService.sendCatalogMsg(streamPushItem, CatalogEvent.DEL);
+        if (streamPushItem != null) {
+            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);
+            MediaServer mediaServerItem = mediaServerService.getOne(streamPushItem.getMediaServerId());
+            mediaServerService.closeStreams(mediaServerItem,app, streamId);
         }
         return true;
     }
@@ -196,7 +294,7 @@
     @Override
     public void zlmServerOnline(String mediaServerId) {
         // 鍚屾zlm鎺ㄦ祦淇℃伅
-        MediaServerItem mediaServerItem = mediaServerService.getOne(mediaServerId);
+        MediaServer mediaServerItem = mediaServerService.getOne(mediaServerId);
         if (mediaServerItem == null) {
             return;
         }
@@ -204,74 +302,81 @@
         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<>();
+        List<MediaInfo> mediaInfoList = redisCatchStorage.getStreams(mediaServerId, "PUSH");
+        Map<String, MediaInfo> streamInfoPushItemMap = new HashMap<>();
         if (pushList.size() > 0) {
             for (StreamPushItem streamPushItem : pushList) {
-                if (StringUtils.isEmpty(streamPushItem.getGbId())) {
+                if (ObjectUtils.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);
+        if (mediaInfoList.size() > 0) {
+            for (MediaInfo mediaInfo : mediaInfoList) {
+                streamInfoPushItemMap.put(mediaInfo.getApp() + mediaInfo.getStream(), mediaInfo);
             }
         }
-        zlmresTfulUtils.getMediaList(mediaServerItem, (mediaList ->{
-            if (mediaList == null) {
-                return;
+        // 鑾峰彇鎵�鏈夋帹娴侀壌鏉冧俊鎭紝娓呯悊杩囨湡鐨�
+        List<StreamAuthorityInfo> allStreamAuthorityInfo = redisCatchStorage.getAllStreamAuthorityInfo();
+        Map<String, StreamAuthorityInfo> streamAuthorityInfoInfoMap = new HashMap<>();
+        for (StreamAuthorityInfo streamAuthorityInfo : allStreamAuthorityInfo) {
+            streamAuthorityInfoInfoMap.put(streamAuthorityInfo.getApp() + streamAuthorityInfo.getStream(), streamAuthorityInfo);
+        }
+        List<StreamInfo> mediaList = mediaServerService.getMediaList(mediaServerItem, null, null, null);
+        if (mediaList == null) {
+            return;
+        }
+        List<StreamPushItem> streamPushItems = handleJSON(mediaList);
+        if (streamPushItems != null) {
+            for (StreamPushItem streamPushItem : streamPushItems) {
+                pushItemMap.remove(streamPushItem.getApp() + streamPushItem.getStream());
+                streamInfoPushItemMap.remove(streamPushItem.getApp() + streamPushItem.getStream());
+                streamAuthorityInfoInfoMap.remove(streamPushItem.getApp() + streamPushItem.getStream());
             }
-            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);
+        }
+        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();
                     }
-                }else {
-                    streamPushMapper.delAll(offlinePushItems);
+                    List<StreamPushItem> streamPushItemsSub = offlinePushItems.subList(i, toIndex);
+                    streamPushMapper.delAll(streamPushItemsSub);
                 }
+            }else {
+                streamPushMapper.delAll(offlinePushItems);
+            }
 
+        }
+        Collection<MediaInfo> mediaInfos = streamInfoPushItemMap.values();
+        if (mediaInfos.size() > 0) {
+            String type = "PUSH";
+            for (MediaInfo mediaInfo : mediaInfos) {
+                JSONObject jsonObject = new JSONObject();
+                jsonObject.put("serverId", userSetting.getServerId());
+                jsonObject.put("app", mediaInfo.getApp());
+                jsonObject.put("stream", mediaInfo.getStream());
+                jsonObject.put("register", false);
+                jsonObject.put("mediaServerId", mediaServerId);
+                redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
+                // 绉婚櫎redis鍐呮祦鐨勪俊鎭�
+                redisCatchStorage.removeStream(mediaServerItem.getId(), "PUSH", mediaInfo.getApp(), mediaInfo.getStream());
+                // 鍐椾綑鏁版嵁锛岃嚜宸辩郴缁熶腑鑷敤
+                redisCatchStorage.removePushListItem(mediaInfo.getApp(), mediaInfo.getStream(), mediaServerItem.getId());
             }
-            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());
-                }
+        }
+
+        Collection<StreamAuthorityInfo> streamAuthorityInfos = streamAuthorityInfoInfoMap.values();
+        if (streamAuthorityInfos.size() > 0) {
+            for (StreamAuthorityInfo streamAuthorityInfo : streamAuthorityInfos) {
+                // 绉婚櫎redis鍐呮祦鐨勪俊鎭�
+                redisCatchStorage.removeStreamAuthorityInfo(streamAuthorityInfo.getApp(), streamAuthorityInfo.getStream());
             }
-        }));
+        }
     }
 
     @Override
@@ -286,18 +391,21 @@
         // 鍙戦�佹祦鍋滄娑堟伅
         String type = "PUSH";
         // 鍙戦�乺edis娑堟伅
-        List<MediaItem> streamInfoList = redisCatchStorage.getStreams(mediaServerId, type);
-        if (streamInfoList.size() > 0) {
-            for (MediaItem mediaItem : streamInfoList) {
+        List<MediaInfo> mediaInfoList = redisCatchStorage.getStreams(mediaServerId, type);
+        if (mediaInfoList.size() > 0) {
+            for (MediaInfo mediaInfo : mediaInfoList) {
                 // 绉婚櫎redis鍐呮祦鐨勪俊鎭�
-                redisCatchStorage.removeStream(mediaServerId, type, mediaItem.getApp(), mediaItem.getStream());
+                redisCatchStorage.removeStream(mediaServerId, type, mediaInfo.getApp(), mediaInfo.getStream());
                 JSONObject jsonObject = new JSONObject();
                 jsonObject.put("serverId", userSetting.getServerId());
-                jsonObject.put("app", mediaItem.getApp());
-                jsonObject.put("stream", mediaItem.getStream());
+                jsonObject.put("app", mediaInfo.getApp());
+                jsonObject.put("stream", mediaInfo.getStream());
                 jsonObject.put("register", false);
                 jsonObject.put("mediaServerId", mediaServerId);
                 redisCatchStorage.sendStreamChangeMsg(type, jsonObject);
+
+                // 鍐椾綑鏁版嵁锛岃嚜宸辩郴缁熶腑鑷敤
+                redisCatchStorage.removePushListItem(mediaInfo.getApp(), mediaInfo.getStream(), mediaServerId);
             }
         }
     }
@@ -346,7 +454,7 @@
         // 瀛樺偍鏁版嵁鍒皊tream_push琛�
         streamPushMapper.addAll(streamPushItems);
         List<StreamPushItem> streamPushItemForGbStream = streamPushItems.stream()
-                .filter(streamPushItem-> streamPushItem.getId() != null)
+                .filter(streamPushItem-> streamPushItem.getGbId() != null)
                 .collect(Collectors.toList());
         // 瀛樺偍鏁版嵁鍒癵b_stream琛紝 id浼氳繑鍥炲埌streamPushItemForGbStream閲�
         if (streamPushItemForGbStream.size() > 0) {
@@ -419,7 +527,7 @@
 
                 }
             }
-            if (streamPushItemListFroPlatform.size() > 0) {
+            if (!streamPushItemListFroPlatform.isEmpty()) {
                 platformGbStreamMapper.batchAdd(streamPushItemListFroPlatform);
                 // 鍙戦�侀�氱煡
                 for (String platformId : platformForEvent.keySet()) {
@@ -427,7 +535,6 @@
                             platformId, platformForEvent.get(platformId), CatalogEvent.ADD);
                 }
             }
-
         }
     }
 
@@ -443,8 +550,8 @@
         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());
+                MediaServer mediaServerItem = mediaServerService.getOne(gbStream.getMediaServerId());
+                mediaServerService.closeStreams(mediaServerItem, gbStream.getApp(), gbStream.getStream());
             }
 
         }
@@ -486,13 +593,16 @@
         stream.setUpdateTime(DateUtil.getNow());
         stream.setCreateTime(DateUtil.getNow());
         stream.setServerId(userSetting.getServerId());
+        stream.setMediaServerId(mediaConfig.getId());
+        stream.setSelf(true);
+        stream.setPushIng(true);
 
         // 鏀惧湪浜嬪姟鍐呮墽琛�
         boolean result = false;
         TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
         try {
             int addStreamResult = streamPushMapper.add(stream);
-            if (!StringUtils.isEmpty(stream.getGbId())) {
+            if (!ObjectUtils.isEmpty(stream.getGbId())) {
                 stream.setStreamType("push");
                 gbStreamMapper.add(stream);
             }
@@ -507,6 +617,37 @@
 
     @Override
     public List<String> getAllAppAndStream() {
+
         return streamPushMapper.getAllAppAndStream();
     }
+
+    @Override
+    public ResourceBaseInfo getOverview() {
+        int total = streamPushMapper.getAllCount();
+        int online = streamPushMapper.getAllOnline(userSetting.isUsePushingAsStatus());
+
+        return new ResourceBaseInfo(total, online);
+    }
+
+    @Override
+    public Map<String, StreamPushItem> getAllAppAndStreamMap() {
+        return streamPushMapper.getAllAppAndStreamMap();
+    }
+
+    @Override
+    public void updatePush(OnStreamChangedHookParam param) {
+        StreamPushItem transform = transform(param);
+        StreamPushItem pushInDb = getPush(param.getApp(), param.getStream());
+        transform.setPushIng(param.isRegist());
+        transform.setUpdateTime(DateUtil.getNow());
+        transform.setPushTime(DateUtil.getNow());
+        transform.setSelf(userSetting.getServerId().equals(param.getSeverId()));
+        if (pushInDb == null) {
+            transform.setCreateTime(DateUtil.getNow());
+            streamPushMapper.add(transform);
+        }else {
+            streamPushMapper.update(transform);
+            gbStreamMapper.updateMediaServer(param.getApp(), param.getStream(), param.getMediaServerId());
+        }
+    }
 }

--
Gitblit v1.8.0