临时提交

src/main/java/com/genersoft/iot/vmp/streamPush/bean/StreamPush.java

@@ -0,0 +1,119 @@
+package com.genersoft.iot.vmp.streamPush.bean;
+
+import com.genersoft.iot.vmp.common.StreamInfo;
+import com.genersoft.iot.vmp.gb28181.bean.CommonGBChannel;
+import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;
+import com.genersoft.iot.vmp.utils.DateUtil;
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import org.jetbrains.annotations.NotNull;
+import org.springframework.util.ObjectUtils;
+
+
+@Data
+@Schema(description = "推流信息")
+@EqualsAndHashCode(callSuper = true)
+public class StreamPush extends CommonGBChannel implements Comparable<StreamPush>{
+
+    /**
+     * id
+     */
+    @Schema(description = "id")
+    private Integer id;
+
+    /**
+     * 应用名
+     */
+    @Schema(description = "应用名")
+    private String app;
+
+    /**
+     * 流id
+     */
+    @Schema(description = "流id")
+    private String stream;
+
+    /**
+     * 使用的流媒体ID
+     */
+    @Schema(description = "使用的流媒体ID")
+    private String mediaServerId;
+
+    /**
+     * 使用的服务ID
+     */
+    @Schema(description = "使用的服务ID")
+    private String serverId;
+
+    /**
+     * 推流时间
+     */
+    @Schema(description = "推流时间")
+    private String pushTime;
+
+    /**
+     * 更新时间
+     */
+    @Schema(description = "更新时间")
+    private String updateTime;
+
+    /**
+     * 创建时间
+     */
+    @Schema(description = "创建时间")
+    private String createTime;
+
+    /**
+     * 是否正在推流
+     */
+    @Schema(description = "是否正在推流")
+    private boolean pushIng;
+
+    /**
+     * 是否自己平台的推流
+     */
+    @Schema(description = "是否自己平台的推流")
+    private boolean self;
+
+    @Override
+    public int compareTo(@NotNull StreamPush streamPushItem) {
+        return Long.valueOf(DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(this.createTime)
+                - DateUtil.yyyy_MM_dd_HH_mm_ssToTimestamp(streamPushItem.getCreateTime())).intValue();
+    }
+
+    public StreamPush getInstance(StreamInfo streamInfo) {
+        StreamPush streamPushItem = new StreamPush();
+        streamPushItem.setApp(streamInfo.getApp());
+        streamPushItem.setMediaServerId(streamInfo.getMediaServerId());
+        streamPushItem.setStream(streamInfo.getStream());
+        streamPushItem.setCreateTime(DateUtil.getNow());
+        streamPushItem.setServerId(streamInfo.getMediaServerId());
+        return streamPushItem;
+
+    }
+
+    public static StreamPush getInstance(MediaArrivalEvent event, String serverId){
+        StreamPush streamPushItem = new StreamPush();
+        streamPushItem.setApp(event.getApp());
+        streamPushItem.setMediaServerId(event.getMediaServer().getId());
+        streamPushItem.setStream(event.getStream());
+        streamPushItem.setCreateTime(DateUtil.getNow());
+        streamPushItem.setServerId(serverId);
+        return streamPushItem;
+    }
+
+    public CommonGBChannel getCommonGBChannel() {
+        if (ObjectUtils.isEmpty(this.getGbDeviceId())) {
+            return null;
+        }
+        if (ObjectUtils.isEmpty(this.getGbName())) {
+            this.setGbName( app+ "-" +stream);
+        }
+        this.setStreamPushId(this.getId());
+        return this;
+    }
+
+
+}
+

src/main/java/com/genersoft/iot/vmp/streamPush/bean/StreamPushExcelDto.java

@@ -0,0 +1,29 @@
+package com.genersoft.iot.vmp.streamPush.bean;
+
+import com.alibaba.excel.annotation.ExcelProperty;
+import lombok.Data;
+
+@Data
+public class StreamPushExcelDto {
+
+    @ExcelProperty("名称")
+    private String name;
+
+    @ExcelProperty("应用名")
+    private String app;
+
+    @ExcelProperty("流ID")
+    private String stream;
+
+    @ExcelProperty("国标ID")
+    private String gbId;
+
+    @ExcelProperty("平台ID")
+    private String platformId;
+
+    @ExcelProperty("目录ID")
+    private String catalogId;
+
+    @ExcelProperty("在线状态")
+    private boolean status;
+}

src/main/java/com/genersoft/iot/vmp/streamPush/bean/StreamPushInfoForUpdateLoad.java

@@ -0,0 +1,16 @@
+package com.genersoft.iot.vmp.streamPush.bean;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+/**
+ * 用来关联推流数据关联的平台和目录
+ */
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class StreamPushInfoForUpdateLoad extends StreamPush{
+
+    private String platformId;
+
+    private String catalogId;
+}

src/main/java/com/genersoft/iot/vmp/streamPush/controller/StreamPushController.java

@@ -0,0 +1,263 @@
+package com.genersoft.iot.vmp.streamPush.controller;
+
+import com.alibaba.excel.EasyExcel;
+import com.alibaba.excel.ExcelReader;
+import com.alibaba.excel.read.metadata.ReadSheet;
+import com.genersoft.iot.vmp.common.StreamInfo;
+import com.genersoft.iot.vmp.conf.UserSetting;
+import com.genersoft.iot.vmp.conf.exception.ControllerException;
+import com.genersoft.iot.vmp.conf.security.JwtUtils;
+import com.genersoft.iot.vmp.conf.security.SecurityUtils;
+import com.genersoft.iot.vmp.conf.security.dto.LoginUser;
+import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
+import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
+import com.genersoft.iot.vmp.streamPush.bean.StreamPush;
+import com.genersoft.iot.vmp.media.service.IMediaServerService;
+import com.genersoft.iot.vmp.service.IMediaService;
+import com.genersoft.iot.vmp.streamPush.bean.StreamPushExcelDto;
+import com.genersoft.iot.vmp.streamPush.service.IStreamPushService;
+import com.genersoft.iot.vmp.streamPush.enent.StreamPushUploadFileHandler;
+import com.genersoft.iot.vmp.vmanager.bean.*;
+import com.github.pagehelper.PageInfo;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.Parameter;
+import io.swagger.v3.oas.annotations.security.SecurityRequirement;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.stereotype.Controller;
+import org.springframework.util.ObjectUtils;
+import org.springframework.web.bind.annotation.*;
+import org.springframework.web.context.request.async.DeferredResult;
+import org.springframework.web.multipart.MultipartFile;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+@Tag(name  = "推流信息管理")
+@Controller
+
+@RequestMapping(value = "/api/push")
+public class StreamPushController {
+
+    private final static Logger logger = LoggerFactory.getLogger(StreamPushController.class);
+
+    @Autowired
+    private IStreamPushService streamPushService;
+
+    @Autowired
+    private IMediaServerService mediaServerService;
+
+    @Autowired
+    private DeferredResultHolder resultHolder;
+
+    @Autowired
+    private IMediaService mediaService;
+
+    @Autowired
+    private UserSetting userSetting;
+
+    @GetMapping(value = "/list")
+    @ResponseBody
+    @Operation(summary = "推流列表查询", security = @SecurityRequirement(name = JwtUtils.HEADER))
+    @Parameter(name = "page", description = "当前页")
+    @Parameter(name = "count", description = "每页查询数量")
+    @Parameter(name = "query", description = "查询内容")
+    @Parameter(name = "pushing", description = "是否正在推流")
+    @Parameter(name = "mediaServerId", description = "流媒体ID")
+    public PageInfo<StreamPush> list(@RequestParam(required = false)Integer page,
+                                     @RequestParam(required = false)Integer count,
+                                     @RequestParam(required = false)String query,
+                                     @RequestParam(required = false)Boolean pushing,
+                                     @RequestParam(required = false)String mediaServerId ){
+
+        if (ObjectUtils.isEmpty(query)) {
+            query = null;
+        }
+        if (ObjectUtils.isEmpty(mediaServerId)) {
+            mediaServerId = null;
+        }
+        PageInfo<StreamPush> pushList = streamPushService.getPushList(page, count, query, pushing, mediaServerId);
+        return pushList;
+    }
+
+
+    @PostMapping(value = "/stop")
+    @ResponseBody
+    @Operation(summary = "中止一个推流", security = @SecurityRequirement(name = JwtUtils.HEADER))
+    @Parameter(name = "app", description = "应用名", required = true)
+    @Parameter(name = "stream", description = "流id", required = true)
+    public void stop(String app, String stream){
+        if (!streamPushService.stopByAppAndStream(app, stream)){
+            throw new ControllerException(ErrorCode.ERROR100);
+        }
+    }
+
+    @DeleteMapping(value = "/batchStop")
+    @ResponseBody
+    @Operation(summary = "中止多个推流", security = @SecurityRequirement(name = JwtUtils.HEADER))
+    public void batchStop(@RequestBody BatchGBStreamParam batchGBStreamParam){
+        if (batchGBStreamParam.getGbStreams().size() == 0) {
+            throw new ControllerException(ErrorCode.ERROR100);
+        }
+        if (!streamPushService.batchStop(batchGBStreamParam.getGbStreams())){
+            throw new ControllerException(ErrorCode.ERROR100);
+        }
+    }
+
+    @PostMapping(value = "upload")
+    @ResponseBody
+    public DeferredResult<ResponseEntity<WVPResult<Object>>> uploadChannelFile(@RequestParam(value = "file") MultipartFile file){
+
+        // 最多处理文件一个小时
+        DeferredResult<ResponseEntity<WVPResult<Object>>> result = new DeferredResult<>(60*60*1000L);
+        // 录像查询以channelId作为deviceId查询
+        String key = DeferredResultHolder.UPLOAD_FILE_CHANNEL;
+        String uuid = UUID.randomUUID().toString();
+        logger.info("通道导入文件类型: {}",file.getContentType() );
+        if (file.isEmpty()) {
+            logger.warn("通道导入文件为空");
+            WVPResult<Object> wvpResult = new WVPResult<>();
+            wvpResult.setCode(-1);
+            wvpResult.setMsg("文件为空");
+            result.setResult(ResponseEntity.status(HttpStatus.BAD_REQUEST).body(wvpResult));
+            return result;
+        }
+        if (file.getContentType() == null) {
+            WVPResult<Object> wvpResult = new WVPResult<>();
+            wvpResult.setCode(-1);
+            wvpResult.setMsg("无法识别文件类型");
+            result.setResult(ResponseEntity.status(HttpStatus.BAD_REQUEST).body(wvpResult));
+            return result;
+        }
+        // 同时只处理一个文件
+        if (resultHolder.exist(key, null)) {
+            logger.warn("已有导入任务正在执行");
+            WVPResult<Object> wvpResult = new WVPResult<>();
+            wvpResult.setCode(-1);
+            wvpResult.setMsg("已有导入任务正在执行");
+            result.setResult(ResponseEntity.status(HttpStatus.TOO_MANY_REQUESTS).body(wvpResult));
+            return result;
+        }
+
+        resultHolder.put(key, uuid, result);
+        result.onTimeout(()->{
+            logger.warn("通道导入超时，可能文件过大");
+            RequestMessage msg = new RequestMessage();
+            msg.setKey(key);
+            WVPResult<Object> wvpResult = new WVPResult<>();
+            wvpResult.setCode(-1);
+            wvpResult.setMsg("导入超时，可能文件过大");
+            msg.setData(wvpResult);
+            resultHolder.invokeAllResult(msg);
+        });
+        //获取文件流
+        InputStream inputStream = null;
+        try {
+            String name = file.getName();
+            inputStream = file.getInputStream();
+        } catch (IOException e) {
+            logger.error("未处理的异常 ", e);
+        }
+        try {
+            //传入参数
+            ExcelReader excelReader = EasyExcel.read(inputStream, StreamPushExcelDto.class,
+                    new StreamPushUploadFileHandler(streamPushService, mediaServerService.getDefaultMediaServer().getId(), (errorStreams, errorGBs)->{
+                        logger.info("通道导入成功，存在重复App+Stream为{}个，存在国标ID为{}个", errorStreams.size(), errorGBs.size());
+                        RequestMessage msg = new RequestMessage();
+                        msg.setKey(key);
+                        WVPResult<Map<String, List<String>>> wvpResult = new WVPResult<>();
+                        if (errorStreams.isEmpty() && errorGBs.isEmpty()) {
+                            wvpResult.setCode(0);
+                            wvpResult.setMsg("成功");
+                        }else {
+                            wvpResult.setCode(1);
+                            wvpResult.setMsg("导入成功。但是存在重复数据");
+                            Map<String, List<String>> errorData = new HashMap<>();
+                            errorData.put("gbId", errorGBs);
+                            errorData.put("stream", errorStreams);
+                            wvpResult.setData(errorData);
+                        }
+                        msg.setData(wvpResult);
+                        resultHolder.invokeAllResult(msg);
+                    })).build();
+            ReadSheet readSheet = EasyExcel.readSheet(0).build();
+            excelReader.read(readSheet);
+            excelReader.finish();
+        }catch (Exception e) {
+            logger.warn("通道导入失败：", e);
+            RequestMessage msg = new RequestMessage();
+            msg.setKey(key);
+            WVPResult<Object> wvpResult = new WVPResult<>();
+            wvpResult.setCode(-1);
+            wvpResult.setMsg("通道导入失败: " + e.getMessage() );
+            msg.setData(wvpResult);
+            resultHolder.invokeAllResult(msg);
+        }
+
+
+        return result;
+    }
+
+    /**
+     * 获取推流播放地址
+     * @param app 应用名
+     * @param stream 流id
+     * @return
+     */
+    @GetMapping(value = "/getPlayUrl")
+    @ResponseBody
+    @Operation(summary = "获取推流播放地址", security = @SecurityRequirement(name = JwtUtils.HEADER))
+    @Parameter(name = "app", description = "应用名", required = true)
+    @Parameter(name = "stream", description = "流id", required = true)
+    @Parameter(name = "mediaServerId", description = "媒体服务器id")
+    public StreamContent getPlayUrl(@RequestParam String app, @RequestParam String stream,
+                                    @RequestParam(required = false) String mediaServerId){
+        boolean authority = false;
+        // 是否登陆用户, 登陆用户返回完整信息
+        LoginUser userInfo = SecurityUtils.getUserInfo();
+        if (userInfo!= null) {
+            authority = true;
+        }
+        StreamPush push = streamPushService.getPush(app, stream);
+        if (push != null && !push.isSelf()) {
+            throw new ControllerException(ErrorCode.ERROR100.getCode(), "来自其他平台的推流信息");
+        }
+        StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, authority);
+        if (streamInfo == null){
+            throw new ControllerException(ErrorCode.ERROR100.getCode(), "获取播放地址失败");
+        }
+        return new StreamContent(streamInfo);
+    }
+
+    /**
+     * 添加推流信息
+     * @param stream 推流信息
+     * @return
+     */
+    @PostMapping(value = "/add")
+    @ResponseBody
+    @Operation(summary = "添加推流信息", security = @SecurityRequirement(name = JwtUtils.HEADER))
+    public void add(@RequestBody StreamPush stream){
+        if (ObjectUtils.isEmpty(stream.getGbId())) {
+            throw new ControllerException(ErrorCode.ERROR400.getCode(), "国标ID不可为空");
+        }
+        if (ObjectUtils.isEmpty(stream.getApp()) && ObjectUtils.isEmpty(stream.getStream())) {
+            throw new ControllerException(ErrorCode.ERROR400.getCode(), "app或stream不可为空");
+        }
+        stream.setStatus(false);
+        stream.setPushIng(false);
+        stream.setAliveSecond(0L);
+        stream.setTotalReaderCount(0);
+        if (!streamPushService.add(stream)) {
+            throw new ControllerException(ErrorCode.ERROR100);
+        }
+    }
+}

src/main/java/com/genersoft/iot/vmp/streamPush/dao/StreamPushMapper.java

@@ -0,0 +1,213 @@
+package com.genersoft.iot.vmp.streamPush.dao;
+
+import com.genersoft.iot.vmp.gb28181.bean.GbStream;
+import com.genersoft.iot.vmp.streamPush.bean.StreamPush;
+import com.genersoft.iot.vmp.service.bean.StreamPushItemFromRedis;
+import org.apache.ibatis.annotations.*;
+import org.springframework.stereotype.Repository;
+
+import java.util.List;
+import java.util.Map;
+
+@Mapper
+@Repository
+public interface StreamPushMapper {
+
+    @Insert("INSERT INTO wvp_stream_push (app, stream, total_reader_count, origin_type, origin_type_str, " +
+            "push_time, alive_second, media_server_id, server_id, update_time, create_time, push_ing, self) VALUES" +
+            "(#{app}, #{stream}, #{totalReaderCount}, #{originType}, #{originTypeStr}, " +
+            "#{pushTime}, #{aliveSecond}, #{mediaServerId} , #{serverId} , #{updateTime} , #{createTime}, " +
+            "#{pushIng}, #{self} )")
+    int add(StreamPush streamPushItem);
+
+
+    @Update(value = {" <script>" +
+            "UPDATE wvp_stream_push " +
+            "SET update_time=#{updateTime}" +
+            "<if test=\"mediaServerId != null\">, media_server_id=#{mediaServerId}</if>" +
+            "<if test=\"serverId != null\">, server_id=#{serverId}</if>" +
+            "<if test=\"totalReaderCount != null\">, total_reader_count=#{totalReaderCount}</if>" +
+            "<if test=\"originType != null\">, origin_type=#{originType}</if>" +
+            "<if test=\"originTypeStr != null\">, origin_type_str=#{originTypeStr}</if>" +
+            "<if test=\"pushTime != null\">, push_time=#{pushTime}</if>" +
+            "<if test=\"aliveSecond != null\">, alive_second=#{aliveSecond}</if>" +
+            "<if test=\"pushIng != null\">, push_ing=#{pushIng}</if>" +
+            "<if test=\"self != null\">, self=#{self}</if>" +
+            "WHERE app=#{app} AND stream=#{stream}"+
+            " </script>"})
+    int update(StreamPush streamPushItem);
+
+    @Delete("DELETE FROM wvp_stream_push WHERE app=#{app} AND stream=#{stream}")
+    int del(String app, String stream);
+
+    @Delete("<script> "+
+            "DELETE sp FROM wvp_stream_push sp left join wvp_gb_stream gs on sp.app = gs.app AND sp.stream = gs.stream where " +
+            "<foreach collection='streamPushItems' item='item' separator='or'>" +
+            "(sp.app=#{item.app} and sp.stream=#{item.stream} and gs.gb_id is null) " +
+            "</foreach>" +
+            "</script>")
+    int delAllWithoutGBId(List<StreamPush> streamPushItems);
+
+    @Delete("<script> "+
+            "DELETE FROM wvp_stream_push where " +
+            "<foreach collection='streamPushItems' item='item' separator='or'>" +
+            "(app=#{item.app} and stream=#{item.stream}) " +
+            "</foreach>" +
+            "</script>")
+    int delAll(List<StreamPush> streamPushItems);
+
+    @Delete("<script> "+
+            "DELETE FROM wvp_stream_push where " +
+            "<foreach collection='gbStreams' item='item' separator='or'>" +
+            "(app=#{item.app} and stream=#{item.stream}) " +
+            "</foreach>" +
+            "</script>")
+    int delAllForGbStream(List<GbStream> gbStreams);
+
+
+    @Select(value = {" <script>" +
+            "SELECT " +
+            "st.*, " +
+            "gs.gb_id, gs.name, gs.longitude, gs.latitude, gs.gb_stream_id " +
+            "from " +
+            "wvp_stream_push st " +
+            "LEFT join wvp_gb_stream gs " +
+            "on st.app = gs.app AND st.stream = gs.stream " +
+            "WHERE " +
+            "1=1 " +
+            " <if test='query != null'> AND (st.app LIKE concat('%',#{query},'%') OR st.stream LIKE concat('%',#{query},'%') OR gs.gb_id LIKE concat('%',#{query},'%') OR gs.name LIKE concat('%',#{query},'%'))</if> " +
+            " <if test='pushing == true' > AND (gs.gb_id is null OR st.push_ing=1)</if>" +
+            " <if test='pushing == false' > AND (st.push_ing is null OR st.push_ing=0) </if>" +
+            " <if test='mediaServerId != null' > AND st.media_server_id=#{mediaServerId} </if>" +
+            "order by st.create_time desc" +
+            " </script>"})
+    List<StreamPush> selectAllForList(@Param("query") String query, @Param("pushing") Boolean pushing, @Param("mediaServerId") String mediaServerId);
+
+    @Select("SELECT st.*, gs.gb_id, gs.name, gs.longitude, gs.latitude FROM wvp_stream_push st LEFT join wvp_gb_stream gs on st.app = gs.app AND st.stream = gs.stream order by st.create_time desc")
+    List<StreamPush> selectAll();
+
+    @Select("SELECT st.*, gs.gb_id, gs.name, gs.longitude, gs.latitude FROM wvp_stream_push st LEFT join wvp_gb_stream gs on st.app = gs.app AND st.stream = gs.stream WHERE st.app=#{app} AND st.stream=#{stream}")
+    StreamPush selectByAppAndStream(@Param("app") String app, @Param("stream") String stream);
+
+    @Insert("<script>"  +
+            "Insert INTO wvp_stream_push (app, stream, total_reader_count, origin_type, origin_type_str, " +
+            "create_time, alive_second, media_server_id, server_id, status, push_ing) " +
+            "VALUES <foreach collection='streamPushItems' item='item' index='index' separator=','>" +
+            "( #{item.app}, #{item.stream}, #{item.totalReaderCount}, #{item.originType}, " +
+            "#{item.originTypeStr},#{item.createTime}, #{item.aliveSecond}, #{item.mediaServerId},#{item.serverId}, #{item.status} ," +
+            " #{item.pushIng} )" +
+            " </foreach>" +
+            "</script>")
+    @Options(useGeneratedKeys = true, keyProperty = "id", keyColumn = "id")
+    int addAll(List<StreamPush> streamPushItems);
+
+    @Delete("DELETE FROM wvp_stream_push")
+    void clear();
+
+    @Delete("delete" +
+            " from wvp_stream_push " +
+            " where id in " +
+            "   (select temp.id from " +
+            "       (select wgs.gb_stream_id as id " +
+            "          from wvp_gb_stream wgs" +
+            "               left join wvp_stream_push sp on sp.id = wgs.gb_stream_id" +
+            "           where wgs.gb_id is null and wgs.media_server_id = #{mediaServerId}) temp)"
+            )
+    void deleteWithoutGBId(String mediaServerId);
+
+    @Select("SELECT * FROM wvp_stream_push WHERE media_server_id=#{mediaServerId}")
+    List<StreamPush> selectAllByMediaServerId(String mediaServerId);
+
+    @Select("SELECT sp.* FROM wvp_stream_push sp left join wvp_gb_stream gs on gs.app = sp.app and gs.stream= sp.stream WHERE sp.media_server_id=#{mediaServerId} and gs.gb_id is null")
+    List<StreamPush> selectAllByMediaServerIdWithOutGbID(String mediaServerId);
+
+    @Update("UPDATE wvp_stream_push " +
+            "SET status=#{status} " +
+            "WHERE app=#{app} AND stream=#{stream}")
+    int updateStatus(@Param("app") String app, @Param("stream") String stream, @Param("status") boolean status);
+
+    @Update("UPDATE wvp_stream_push " +
+            "SET push_ing=#{pushIng} " +
+            "WHERE app=#{app} AND stream=#{stream}")
+    int updatePushStatus(@Param("app") String app, @Param("stream") String stream, @Param("pushIng") boolean pushIng);
+
+    @Update("UPDATE wvp_stream_push " +
+            "SET status=#{status} " +
+            "WHERE media_server_id=#{mediaServerId}")
+    void updateStatusByMediaServerId(@Param("mediaServerId") String mediaServerId, @Param("status") boolean status);
+
+
+    @Select("<script> "+
+            "SELECT gs.* FROM wvp_stream_push sp left join wvp_gb_stream gs on sp.app = gs.app AND sp.stream = gs.stream " +
+            "where sp.status = true and (gs.app, gs.stream) in (" +
+            "<foreach collection='offlineStreams' item='item' separator=','>" +
+            "(#{item.app}, #{item.stream}) " +
+            "</foreach>" +
+            ")</script>")
+    List<GbStream> getOnlinePusherForGbInList(List<StreamPushItemFromRedis> offlineStreams);
+
+    @Update("<script> "+
+            "UPDATE wvp_stream_push SET status=0  where (app, stream) in (" +
+            "<foreach collection='offlineStreams' item='item' separator=','>" +
+            "(#{item.app}, #{item.stream}) " +
+            "</foreach>" +
+            ")</script>")
+    void offline(List<StreamPushItemFromRedis> offlineStreams);
+
+    @Select("<script> "+
+            "SELECT * FROM wvp_stream_push sp left join wvp_gb_stream gs on sp.app = gs.app AND sp.stream = gs.stream " +
+            "where sp.status = 0 and (gs.app, gs.stream) in (" +
+            "<foreach collection='onlineStreams' item='item' separator=','>" +
+            "(#{item.app}, #{item.stream}) " +
+            "</foreach>" +
+            ") </script>")
+    List<GbStream> getOfflinePusherForGbInList(List<StreamPushItemFromRedis> onlineStreams);
+
+    @Update("<script> "+
+            "UPDATE wvp_stream_push SET status=1  where (app, stream) in (" +
+            "<foreach collection='onlineStreams' item='item' separator=','>" +
+            "(#{item.app}, #{item.stream}) " +
+            "</foreach>" +
+            ")</script>")
+    void online(List<StreamPushItemFromRedis> onlineStreams);
+
+    @Select("SELECT gs.* FROM wvp_stream_push sp left join wvp_gb_stream gs on sp.app = gs.app AND sp.stream = gs.stream where sp.status = true")
+    List<GbStream> getOnlinePusherForGb();
+
+    @Update("UPDATE wvp_stream_push SET status=0")
+    void setAllStreamOffline();
+
+    @Select("SELECT CONCAT(app,stream) from wvp_gb_stream")
+    List<String> getAllAppAndStream();
+
+    @Select("select count(1) from wvp_stream_push ")
+    int getAllCount();
+
+    @Select(value = {" <script>" +
+            " <if test='pushIngAsOnline == true'> select count(1) from wvp_stream_push where push_ing = true </if>" +
+            " <if test='pushIngAsOnline == false'> select count(1)from wvp_stream_push where status = true  </if>" +
+            " </script>"})
+    int getAllOnline(Boolean usePushingAsStatus);
+
+    @Select("<script> " +
+            "select app, stream from wvp_stream_push where (app, stream) in " +
+            "<foreach collection='streamPushItems' item='item' separator=','>" +
+            "(#{item.app}, #{item.stream}) " +
+            "</foreach>" +
+            "</script>")
+    List<StreamPush> getListIn(List<StreamPush> streamPushItems);
+
+    @MapKey("vhost")
+    @Select("SELECT CONCAT(wsp.app, wsp.stream) as vhost, wsp.app, wsp.stream, wgs.gb_id, wgs.name " +
+            " from wvp_stream_push wsp " +
+            " left join wvp_gb_stream  wgs  on wgs.app = wsp.app and wgs.stream = wsp.stream")
+    Map<String, StreamPush> getAllAppAndStreamMap();
+
+
+    @MapKey("gb_id")
+    @Select("SELECT wgs.gb_id, wsp.app, wsp.stream, wgs.gb_id, wgs.name " +
+            " from wvp_stream_push wsp " +
+            " left join wvp_gb_stream  wgs  on wgs.app = wsp.app and wgs.stream = wsp.stream")
+    Map<String, StreamPush> getAllGBId();
+
+}

src/main/java/com/genersoft/iot/vmp/streamPush/enent/StreamPushUploadFileHandler.java

@@ -0,0 +1,184 @@
+package com.genersoft.iot.vmp.streamPush.enent;
+
+import com.alibaba.excel.context.AnalysisContext;
+import com.alibaba.excel.event.AnalysisEventListener;
+import com.genersoft.iot.vmp.streamPush.bean.StreamPushExcelDto;
+import com.genersoft.iot.vmp.streamPush.bean.StreamPushInfoForUpdateLoad;
+import com.genersoft.iot.vmp.streamPush.service.IStreamPushService;
+import com.genersoft.iot.vmp.utils.DateUtil;
+import com.google.common.collect.BiMap;
+import com.google.common.collect.HashBiMap;
+import org.springframework.util.ObjectUtils;
+
+import java.util.*;
+
+public class StreamPushUploadFileHandler extends AnalysisEventListener<StreamPushExcelDto> {
+
+    /**
+     * 错误数据的回调，用于将错误数据发送给页面
+     */
+    private ErrorDataHandler errorDataHandler;
+
+    /**
+     * 推流的业务类用于存储数据
+     */
+    private IStreamPushService pushService;
+
+    /**
+     * 默认流媒体节点ID
+     */
+    private String defaultMediaServerId;
+
+    /**
+     * 用于存储不加过滤的所有数据
+     */
+    private final List<StreamPushInfoForUpdateLoad> streamPushItems = new ArrayList<>();
+
+    /**
+     * 用于存储更具APP+Stream过滤后的数据，可以直接存入stream_push表与gb_stream表
+     */
+    private final Map<String, StreamPushInfoForUpdateLoad> streamPushItemForSave = new HashMap<>();
+
+    /**
+     * 用于存储按照APP+Stream为KEY， 平台ID+目录Id 为value的数据，用于存储到gb_stream表后获取app+Stream对应的平台与目录信息，然后存入关联表
+     */
+    private final Map<String, List<String[]>> streamPushItemsForPlatform = new HashMap<>();
+
+    /**
+     * 用于判断文件是否存在重复的app+Stream+平台ID
+     */
+    private final Set<String> streamPushStreamSet = new HashSet<>();
+
+    /**
+     * 用于存储APP+Stream->国标ID 的数据结构, 数据一一对应，全局判断APP+Stream->国标ID是否存在不对应
+     */
+    private final BiMap<String,String> gBMap = HashBiMap.create();
+
+    /**
+     * 用于存储APP+Stream-> 在数据库中的数据
+     */
+    private final BiMap<String,String> pushMapInDb = HashBiMap.create();
+
+    /**
+     * 记录错误的APP+Stream
+     */
+    private final List<String> errorStreamList = new ArrayList<>();
+
+
+    /**
+     * 记录错误的国标ID
+     */
+    private final List<String> errorInfoList = new ArrayList<>();
+
+    /**
+     * 读取数量计数器
+     */
+    private int loadedSize = 0;
+
+    public StreamPushUploadFileHandler(IStreamPushService pushService, String defaultMediaServerId, ErrorDataHandler errorDataHandler) {
+        this.pushService = pushService;
+        this.defaultMediaServerId = defaultMediaServerId;
+        this.errorDataHandler = errorDataHandler;
+        // 获取数据库已有的数据，已经存在的则忽略
+        List<String> allAppAndStreams = pushService.getAllAppAndStream();
+        if (!allAppAndStreams.isEmpty()) {
+            for (String allAppAndStream : allAppAndStreams) {
+                pushMapInDb.put(allAppAndStream, allAppAndStream);
+            }
+        }
+    }
+
+    public interface ErrorDataHandler{
+        void handle(List<String> streams, List<String> gbId);
+    }
+
+    @Override
+    public void invoke(StreamPushExcelDto streamPushExcelDto, AnalysisContext analysisContext) {
+        if (ObjectUtils.isEmpty(streamPushExcelDto.getApp())
+                || ObjectUtils.isEmpty(streamPushExcelDto.getStream())
+                || ObjectUtils.isEmpty(streamPushExcelDto.getGbId())) {
+            return;
+        }
+        Integer rowIndex = analysisContext.readRowHolder().getRowIndex();
+
+        if (gBMap.get(streamPushExcelDto.getApp() + streamPushExcelDto.getStream()) == null) {
+            try {
+                gBMap.put(streamPushExcelDto.getApp() + streamPushExcelDto.getStream(), streamPushExcelDto.getGbId());
+            }catch (IllegalArgumentException e) {
+                errorInfoList.add("行：" + rowIndex + ", " + streamPushExcelDto.getGbId() + " 国标ID重复使用");
+                return;
+            }
+        }else {
+            if (!gBMap.get(streamPushExcelDto.getApp() + streamPushExcelDto.getStream()).equals(streamPushExcelDto.getGbId())) {
+                errorInfoList.add("行：" + rowIndex + ", " + streamPushExcelDto.getGbId() + " 同样的应用名和流ID使用了不同的国标ID");
+                return;
+            }
+        }
+
+        if (streamPushStreamSet.contains(streamPushExcelDto.getApp() + streamPushExcelDto.getStream() + streamPushExcelDto.getPlatformId())) {
+            errorStreamList.add("行：" + rowIndex + ", " +  streamPushExcelDto.getApp() + "/" + streamPushExcelDto.getStream()+  " 平台信息重复");
+            return;
+        }else {
+            if (pushMapInDb.get(streamPushExcelDto.getApp()+streamPushExcelDto.getStream()) != null) {
+                errorStreamList.add("行：" + rowIndex + ", " +  streamPushExcelDto.getApp() + "/" + streamPushExcelDto.getStream()+  " 数据已存在");
+                return;
+            }
+            streamPushStreamSet.add(streamPushExcelDto.getApp()+streamPushExcelDto.getStream() + streamPushExcelDto.getPlatformId());
+        }
+
+        StreamPushInfoForUpdateLoad streamPushInfoForUpdateLoad = new StreamPushInfoForUpdateLoad();
+        streamPushInfoForUpdateLoad.setApp(streamPushExcelDto.getApp());
+        streamPushInfoForUpdateLoad.setStream(streamPushExcelDto.getStream());
+        streamPushInfoForUpdateLoad.setGbDeviceId(streamPushExcelDto.getGbId());
+        streamPushInfoForUpdateLoad.setGbStatus(streamPushExcelDto.isStatus());
+        streamPushInfoForUpdateLoad.setCreateTime(DateUtil.getNow());
+        streamPushInfoForUpdateLoad.setMediaServerId(defaultMediaServerId);
+        streamPushInfoForUpdateLoad.setGbName(streamPushExcelDto.getName());
+        streamPushInfoForUpdateLoad.setPlatformId(streamPushExcelDto.getPlatformId());
+        streamPushInfoForUpdateLoad.setCatalogId(streamPushExcelDto.getCatalogId());
+
+        // 存入所有的通道信息
+        streamPushItems.add(streamPushInfoForUpdateLoad);
+        streamPushItemForSave.put(streamPushInfoForUpdateLoad.getApp() + streamPushInfoForUpdateLoad.getStream(), streamPushInfoForUpdateLoad);
+
+        if (!ObjectUtils.isEmpty(streamPushExcelDto.getPlatformId())) {
+            List<String[]> platformList = streamPushItemsForPlatform.computeIfAbsent(streamPushInfoForUpdateLoad.getApp() + streamPushInfoForUpdateLoad.getStream(), k -> new ArrayList<>());
+            String platformId = streamPushExcelDto.getPlatformId();
+            String catalogId = streamPushExcelDto.getCatalogId();
+            if (ObjectUtils.isEmpty(streamPushExcelDto.getCatalogId())) {
+                catalogId = null;
+            }
+            String[] platFormInfoArray = new String[]{platformId, catalogId};
+            platformList.add(platFormInfoArray);
+        }
+
+        loadedSize ++;
+        if (loadedSize > 1000) {
+            saveData();
+            streamPushItems.clear();
+            streamPushItemForSave.clear();
+            streamPushItemsForPlatform.clear();
+            loadedSize = 0;
+        }
+
+    }
+
+    @Override
+    public void doAfterAllAnalysed(AnalysisContext analysisContext) {
+        // 这里也要保存数据，确保最后遗留的数据也存储到数据库
+        saveData();
+        streamPushItems.clear();
+        streamPushItemForSave.clear();
+        gBMap.clear();
+        streamPushStreamSet.clear();
+        streamPushItemsForPlatform.clear();
+        errorDataHandler.handle(errorStreamList, errorInfoList);
+    }
+
+    private void saveData(){
+        if (!streamPushItemForSave.isEmpty()) {
+            // 向数据库查询是否存在重复的app
+            pushService.batchAddForUpload(new ArrayList<>(streamPushItemForSave.values()), streamPushItemsForPlatform);
+        }
+    }
+}

src/main/java/com/genersoft/iot/vmp/streamPush/service/IStreamPushService.java

@@ -0,0 +1,136 @@
+package com.genersoft.iot.vmp.streamPush.service;
+
+import com.genersoft.iot.vmp.gb28181.bean.GbStream;
+import com.genersoft.iot.vmp.media.zlm.dto.hook.OnStreamChangedHookParam;
+import com.genersoft.iot.vmp.service.bean.StreamPushItemFromRedis;
+import com.genersoft.iot.vmp.streamPush.bean.StreamPush;
+import com.genersoft.iot.vmp.streamPush.bean.StreamPushInfoForUpdateLoad;
+import com.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo;
+import com.github.pagehelper.PageInfo;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @author lin
+ */
+public interface IStreamPushService {
+
+    /**
+     * 将应用名和流ID加入国标关联
+     * @param stream
+     * @return
+     */
+    boolean saveToGB(GbStream stream);
+
+    /**
+     * 将应用名和流ID移出国标关联
+     * @param stream
+     * @return
+     */
+    boolean removeFromGB(GbStream stream);
+
+    /**
+     * 获取
+     */
+    PageInfo<StreamPush> getPushList(Integer page, Integer count, String query, Boolean pushing, String mediaServerId);
+
+    List<StreamPush> getPushList(String mediaSererId);
+
+    StreamPush transform(OnStreamChangedHookParam item);
+
+    StreamPush getPush(String app, String streamId);
+
+    boolean stop(StreamPush streamPush);
+
+    /**
+     * 停止一路推流
+     * @param app 应用名
+     * @param stream 流ID
+     */
+    boolean stopByAppAndStream(String app, String stream);
+
+    /**
+     * 新的节点加入
+     */
+    void zlmServerOnline(String mediaServerId);
+
+    /**
+     * 节点离线
+     */
+    void zlmServerOffline(String mediaServerId);
+
+    /**
+     * 清空
+     */
+    void clean();
+
+
+    boolean saveToRandomGB();
+
+    /**
+     * 批量添加
+     */
+    void batchAdd(List<StreamPush> streamPushExcelDtoList);
+
+    @Transactional
+    void batchAddForUpload(List<StreamPushInfoForUpdateLoad> streamPushItems, Map<String, List<String[]>> streamPushItemsForAll);
+
+    /**
+     * 中止多个推流
+     */
+    boolean batchStop(List<GbStream> streamPushItems);
+
+    /**
+     * 导入时批量增加
+     */
+    void batchAddForUpload(List<StreamPushInfoForUpdateLoad> streamPushItems, Map<String, List<String[]>> streamPushItemsForAll);
+
+    /**
+     * 全部离线
+     */
+    void allStreamOffline();
+
+    /**
+     * 推流离线
+     */
+    void offline(List<StreamPushItemFromRedis> offlineStreams);
+
+    /**
+     * 推流上线
+     */
+    void online(List<StreamPushItemFromRedis> onlineStreams);
+
+    /**
+     * 增加推流
+     */
+    boolean add(StreamPush stream);
+
+    boolean update(StreamPush stream);
+
+    /**
+     * 获取全部的app+Streanm 用于判断推流列表是新增还是修改
+     * @return
+     */
+    List<String> getAllAppAndStream();
+
+    /**
+     * 获取统计信息
+     * @return
+     */
+    ResourceBaseInfo getOverview();
+
+    Map<String, StreamPush> getAllAppAndStreamMap();
+
+
+    void updatePush(OnStreamChangedHookParam param);
+
+    Map<String, StreamPush> getAllGBId();
+
+    void updateStatus(StreamPush push);
+
+    void deleteByAppAndStream(String app, String stream);
+
+    void updatePushStatus(Integer streamPushId, boolean pushIng);
+}

src/main/java/com/genersoft/iot/vmp/streamPush/service/impl/StreamPushServiceImpl.java

@@ -0,0 +1,679 @@
+package com.genersoft.iot.vmp.streamPush.service.impl;
+
+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.CommonGBChannel;
+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.gb28181.service.IGbChannelService;
+import com.genersoft.iot.vmp.media.bean.MediaInfo;
+import com.genersoft.iot.vmp.media.bean.MediaServer;
+import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;
+import com.genersoft.iot.vmp.media.event.media.MediaDepartureEvent;
+import com.genersoft.iot.vmp.media.event.mediaServer.MediaServerOfflineEvent;
+import com.genersoft.iot.vmp.media.event.mediaServer.MediaServerOnlineEvent;
+import com.genersoft.iot.vmp.media.service.IMediaServerService;
+import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
+import com.genersoft.iot.vmp.media.zlm.dto.hook.OriginType;
+import com.genersoft.iot.vmp.service.bean.StreamPushItemFromRedis;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
+import com.genersoft.iot.vmp.storager.dao.ParentPlatformMapper;
+import com.genersoft.iot.vmp.storager.dao.PlatformCatalogMapper;
+import com.genersoft.iot.vmp.storager.dao.PlatformGbStreamMapper;
+import com.genersoft.iot.vmp.storager.dao.StreamProxyMapper;
+import com.genersoft.iot.vmp.streamPush.bean.StreamPush;
+import com.genersoft.iot.vmp.streamPush.bean.StreamPushInfoForUpdateLoad;
+import com.genersoft.iot.vmp.streamPush.dao.StreamPushMapper;
+import com.genersoft.iot.vmp.streamPush.service.IStreamPushService;
+import com.genersoft.iot.vmp.utils.DateUtil;
+import com.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo;
+import com.genersoft.iot.vmp.vmanager.bean.StreamContent;
+import com.github.pagehelper.PageHelper;
+import com.github.pagehelper.PageInfo;
+import lombok.extern.slf4j.Slf4j;
+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.annotation.Transactional;
+import org.springframework.util.ObjectUtils;
+
+import java.util.*;
+
+@Service
+@Slf4j
+@DS("master")
+public class StreamPushServiceImpl implements IStreamPushService {
+
+    @Autowired
+    private StreamPushMapper streamPushMapper;
+
+    @Autowired
+    private StreamProxyMapper streamProxyMapper;
+
+    @Autowired
+    private ParentPlatformMapper parentPlatformMapper;
+
+    @Autowired
+    private PlatformCatalogMapper platformCatalogMapper;
+
+    @Autowired
+    private PlatformGbStreamMapper platformGbStreamMapper;
+
+    @Autowired
+    private EventPublisher eventPublisher;
+
+    @Autowired
+    private IRedisCatchStorage redisCatchStorage;
+
+    @Autowired
+    private UserSetting userSetting;
+
+    @Autowired
+    private IMediaServerService mediaServerService;
+
+    @Autowired
+    DataSourceTransactionManager dataSourceTransactionManager;
+
+    @Autowired
+    TransactionDefinition transactionDefinition;
+
+    @Autowired
+    private MediaConfig mediaConfig;
+
+    @Autowired
+    private IGbChannelService gbChannelService;
+
+    /**
+     * 流到来的处理
+     */
+    @Async("taskExecutor")
+    @EventListener
+    @Transactional
+    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);
+
+        StreamPush streamPushInDb = getPush(event.getApp(), event.getStream());
+        if (streamPushInDb == null) {
+            StreamPush streamPush = StreamPush.getInstance(event, userSetting.getServerId());
+            streamPush.setPushIng(true);
+            streamPush.setUpdateTime(DateUtil.getNow());
+            streamPush.setPushTime(DateUtil.getNow());
+            streamPush.setSelf(true);
+            add(streamPush);
+        }else {
+            updatePushStatus(streamPushInDb.getId(), true);
+        }
+        // 冗余数据，自己系统中自用
+        if (!"broadcast".equals(event.getApp()) && !"talk".equals(event.getApp())) {
+            StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStream(
+                    event.getMediaServer(), event.getApp(), event.getStream(), event.getMediaInfo(), event.getCallId());
+            event.getHookParam().setStreamInfo(new StreamContent(streamInfo));
+            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
+    @Transactional
+    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);
+            }
+        }
+        StreamPush push = getPush(event.getApp(), event.getStream());
+        push.setPushIng(false);
+        if (push.getGbDeviceId() != null) {
+            if (userSetting.isUsePushingAsStatus()) {
+                push.setGbStatus(false);
+                updateStatus(push);
+//                streamPushMapper.updatePushStatus(event.getApp(), event.getStream(), false);
+//                eventPublisher.catalogEventPublishForStream(null, gbStream, CatalogEvent.OFF);
+            }
+        }else {
+            deleteByAppAndStream(event.getApp(), event.getStream());
+        }
+    }
+
+    /**
+     * 流媒体节点上线
+     */
+    @Async("taskExecutor")
+    @EventListener
+    @Transactional
+    public void onApplicationEvent(MediaServerOnlineEvent event) {
+        zlmServerOnline(event.getMediaServerId());
+    }
+
+    /**
+     * 流媒体节点离线
+     */
+    @Async("taskExecutor")
+    @EventListener
+    @Transactional
+    public void onApplicationEvent(MediaServerOfflineEvent event) {
+        zlmServerOffline(event.getMediaServerId());
+    }
+
+    @Override
+    public PageInfo<StreamPush> getPushList(Integer page, Integer count, String query, Boolean pushing, String mediaServerId) {
+        PageHelper.startPage(page, count);
+        List<StreamPush> all = streamPushMapper.selectAllForList(query, pushing, mediaServerId);
+        return new PageInfo<>(all);
+    }
+
+    @Override
+    public List<StreamPush> getPushList(String mediaServerId) {
+        return streamPushMapper.selectAllByMediaServerIdWithOutGbID(mediaServerId);
+    }
+
+
+    @Override
+    public StreamPush getPush(String app, String stream) {
+        return streamPushMapper.selectByAppAndStream(app, stream);
+    }
+
+    @Override
+    @Transactional
+    public boolean add(StreamPush stream) {
+        log.info("[添加推流] app: {}, stream: {}, 国标编号: {}", stream.getApp(), stream.getStream(), stream.getGbDeviceId());
+        stream.setUpdateTime(DateUtil.getNow());
+        stream.setCreateTime(DateUtil.getNow());
+        int addResult = streamPushMapper.add(stream);
+        if (addResult <= 0) {
+            return false;
+        }
+        if (ObjectUtils.isEmpty(stream.getGbDeviceId())) {
+            return true;
+        }
+        CommonGBChannel channel = gbChannelService.queryByDeviceId(stream.getGbDeviceId());
+        if (channel != null) {
+            log.info("[添加推流]失败，国标编号已存在: {} app: {}, stream: {}, ", stream.getGbDeviceId(), stream.getApp(), stream.getStream());
+        }
+        int addChannelResult = gbChannelService.add(stream.getCommonGBChannel());
+        return addChannelResult > 0;
+    }
+
+    @Override
+    @Transactional
+    public void deleteByAppAndStream(String app, String stream) {
+        log.info("[删除推流] app: {}, stream: {}, ", app, stream);
+        StreamPush streamPush = streamPushMapper.selectByAppAndStream(app, stream);
+        if (streamPush == null) {
+            log.info("[删除推流]失败， 不存在 app: {}, stream: {}, ", app, stream);
+            return;
+        }
+        if (streamPush.isPushIng()) {
+            stop(streamPush);
+        }
+        if (streamPush.getGbId() > 0) {
+            gbChannelService.delete(streamPush.getGbId());
+        }
+    }
+    @Override
+    @Transactional
+    public boolean update(StreamPush streamPush) {
+        log.info("[更新推流]：id: {}, app: {}, stream: {}, ", streamPush.getId(), streamPush.getApp(), streamPush.getStream());
+        assert streamPush.getId() != null;
+        streamPush.setUpdateTime(DateUtil.getNow());
+        streamPushMapper.update(streamPush);
+        if (streamPush.getGbId() > 0) {
+            gbChannelService.update(streamPush.getCommonGBChannel());
+        }
+        return true;
+    }
+
+
+    @Override
+    @Transactional
+    public boolean stop(StreamPush streamPush) {
+        log.info("[主动停止推流] id: {}, app: {}, stream: {}, ", streamPush.getId(), streamPush.getApp(), streamPush.getStream());
+        MediaServer mediaServer = null;
+        if (streamPush.getMediaServerId() == null) {
+            log.info("[主动停止推流]未找到使用MediaServer，开始自动检索 id: {}, app: {}, stream: {}, ", streamPush.getId(), streamPush.getApp(), streamPush.getStream());
+            mediaServer = mediaServerService.getMediaServerByAppAndStream(streamPush.getApp(), streamPush.getStream());
+            if (mediaServer != null) {
+                log.info("[主动停止推流] 检索到MediaServer为{}， id: {}, app: {}, stream: {}, ", mediaServer.getId(), streamPush.getId(), streamPush.getApp(), streamPush.getStream());
+            }else {
+                log.info("[主动停止推流]未找到使用MediaServer id: {}, app: {}, stream: {}, ", streamPush.getId(), streamPush.getApp(), streamPush.getStream());
+            }
+        }else {
+            mediaServer = mediaServerService.getOne(streamPush.getMediaServerId());
+            if (mediaServer == null) {
+                log.info("[主动停止推流]未找到使用的MediaServer： {}，开始自动检索 id: {}, app: {}, stream: {}, ",streamPush.getMediaServerId(),  streamPush.getId(), streamPush.getApp(), streamPush.getStream());
+                mediaServer = mediaServerService.getMediaServerByAppAndStream(streamPush.getApp(), streamPush.getStream());
+                if (mediaServer != null) {
+                    log.info("[主动停止推流] 检索到MediaServer为{}， id: {}, app: {}, stream: {}, ", mediaServer.getId(), streamPush.getId(), streamPush.getApp(), streamPush.getStream());
+                }else {
+                    log.info("[主动停止推流]未找到使用MediaServer id: {}, app: {}, stream: {}, ", streamPush.getId(), streamPush.getApp(), streamPush.getStream());
+                }
+            }
+        }
+        if (mediaServer != null) {
+            mediaServerService.closeStreams(mediaServer, streamPush.getApp(), streamPush.getStream());
+        }
+        streamPush.setPushIng(false);
+        if (userSetting.isUsePushingAsStatus()) {
+            streamPush.setGbStatus(false);
+            gbChannelService.offline(streamPush.getCommonGBChannel());
+        }
+        gbChannelService.closeSend(streamPush.getCommonGBChannel());
+        streamPush.setUpdateTime(DateUtil.getNow());
+        streamPushMapper.update(streamPush);
+        return true;
+    }
+
+    @Override
+    @Transactional
+    public boolean stopByAppAndStream(String app, String stream) {
+        log.info("[主动停止推流] ： app: {}, stream: {}, ", app, stream);
+        StreamPush streamPushItem = streamPushMapper.selectByAppAndStream(app, stream);
+        if (streamPushItem != null) {
+            stop(streamPushItem);
+        }
+        return true;
+    }
+
+    @Override
+    @Transactional
+    public void zlmServerOnline(String mediaServerId) {
+        // 同步zlm推流信息
+        MediaServer mediaServerItem = mediaServerService.getOne(mediaServerId);
+        if (mediaServerItem == null) {
+            return;
+        }
+        // 数据库记录
+        List<StreamPush> pushList = getPushList(mediaServerId);
+        Map<String, StreamPush> pushItemMap = new HashMap<>();
+        // redis记录
+        List<MediaInfo> mediaInfoList = redisCatchStorage.getStreams(mediaServerId, "PUSH");
+        Map<String, MediaInfo> streamInfoPushItemMap = new HashMap<>();
+        if (!pushList.isEmpty()) {
+            for (StreamPush streamPushItem : pushList) {
+                if (ObjectUtils.isEmpty(streamPushItem.getGbId())) {
+                    pushItemMap.put(streamPushItem.getApp() + streamPushItem.getStream(), streamPushItem);
+                }
+            }
+        }
+        if (!mediaInfoList.isEmpty()) {
+            for (MediaInfo mediaInfo : mediaInfoList) {
+                streamInfoPushItemMap.put(mediaInfo.getApp() + mediaInfo.getStream(), mediaInfo);
+            }
+        }
+        // 获取所有推流鉴权信息，清理过期的
+        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<StreamPush> streamPushItems = handleJSON(mediaList);
+        if (streamPushItems != null) {
+            for (StreamPush streamPushItem : streamPushItems) {
+                pushItemMap.remove(streamPushItem.getApp() + streamPushItem.getStream());
+                streamInfoPushItemMap.remove(streamPushItem.getApp() + streamPushItem.getStream());
+                streamAuthorityInfoInfoMap.remove(streamPushItem.getApp() + streamPushItem.getStream());
+            }
+        }
+        List<StreamPush> changedStreamPushList = new ArrayList<>(pushItemMap.values());
+        if (!changedStreamPushList.isEmpty()) {
+            for (StreamPush streamPush : changedStreamPushList) {
+                stop(streamPush);
+            }
+        }
+
+
+//        if (!changedStreamPushList.isEmpty()) {
+//            String type = "PUSH";
+//            int runLimit = 300;
+//            if (changedStreamPushList.size() > runLimit) {
+//                for (int i = 0; i < changedStreamPushList.size(); i += runLimit) {
+//                    int toIndex = i + runLimit;
+//                    if (i + runLimit > changedStreamPushList.size()) {
+//                        toIndex = changedStreamPushList.size();
+//                    }
+//                    List<StreamPush> streamPushItemsSub = changedStreamPushList.subList(i, toIndex);
+//                    streamPushMapper.delAll(streamPushItemsSub);
+//                }
+//            }else {
+//                streamPushMapper.delAll(changedStreamPushList);
+//            }
+//
+//        }
+        Collection<MediaInfo> mediaInfos = streamInfoPushItemMap.values();
+        if (!mediaInfos.isEmpty()) {
+            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<StreamAuthorityInfo> streamAuthorityInfos = streamAuthorityInfoInfoMap.values();
+        if (!streamAuthorityInfos.isEmpty()) {
+            for (StreamAuthorityInfo streamAuthorityInfo : streamAuthorityInfos) {
+                // 移除redis内流的信息
+                redisCatchStorage.removeStreamAuthorityInfo(streamAuthorityInfo.getApp(), streamAuthorityInfo.getStream());
+            }
+        }
+    }
+
+    @Override
+    @Transactional
+    public void zlmServerOffline(String mediaServerId) {
+        List<StreamPush> streamPushItems = streamPushMapper.selectAllByMediaServerId(mediaServerId);
+        if (!streamPushItems.isEmpty()) {
+            for (StreamPush streamPushItem : streamPushItems) {
+                stop(streamPushItem);
+            }
+        }
+//        // 移除没有GBId的推流
+//        streamPushMapper.deleteWithoutGBId(mediaServerId);
+//        // 其他的流设置未启用
+//        streamPushMapper.updateStatusByMediaServerId(mediaServerId, false);
+//        streamProxyMapper.updateStatusByMediaServerId(mediaServerId, false);
+        // 发送流停止消息
+        String type = "PUSH";
+        // 发送redis消息
+        List<MediaInfo> mediaInfoList = redisCatchStorage.getStreams(mediaServerId, type);
+        if (!mediaInfoList.isEmpty()) {
+            for (MediaInfo mediaInfo : mediaInfoList) {
+                // 移除redis内流的信息
+                redisCatchStorage.removeStream(mediaServerId, type, mediaInfo.getApp(), mediaInfo.getStream());
+                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);
+
+                // 冗余数据，自己系统中自用
+                redisCatchStorage.removePushListItem(mediaInfo.getApp(), mediaInfo.getStream(), mediaServerId);
+            }
+        }
+    }
+
+    @Override
+    @Transactional
+    public void batchAdd(List<StreamPush> streamPushItems) {
+        streamPushMapper.addAll(streamPushItems);
+        List<CommonGBChannel> commonGBChannels = new ArrayList<>();
+        for (StreamPush streamPush : streamPushItems) {
+            if (ObjectUtils.isEmpty(streamPush.getGbDeviceId())) {
+                commonGBChannels.add(streamPush.getCommonGBChannel());
+            }
+        }
+        gbChannelService.batchAdd(commonGBChannels);
+    }
+
+
+    @Override
+    @Transactional
+    public void batchAddForUpload(List<StreamPushInfoForUpdateLoad> streamPushItems, Map<String, List<String[]>> streamPushItemsForAll ) {
+        // 存储数据到stream_push表
+        streamPushMapper.addAll(streamPushItems);
+
+        List<CommonGBChannel> channelList = new ArrayList<>();
+        for (StreamPush streamPushItem : streamPushItems) {
+            if (streamPushItem.getGbDeviceId() != null && streamPushItem.getId() > 0) {
+                channelList.add(streamPushItem.getCommonGBChannel());
+            }
+        }
+        // 存储数据到gb_stream表， id会返回到streamPushItemForGbStream里
+        if (!channelList.isEmpty()) {
+            gbChannelService.batchAdd(channelList);
+        }
+
+
+        if (!streamPushItemsForPlatform.isEmpty()) {
+            // 获取所有平台，平台和目录信息一般不会特别大量。
+            List<ParentPlatform> parentPlatformList = parentPlatformMapper.getParentPlatformList();
+            Map<String, Map<String, PlatformCatalog>> platformInfoMap = new HashMap<>();
+            if (parentPlatformList.isEmpty()) {
+                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<StreamPush> streamPushItemListFroPlatform = new ArrayList<>();
+            Map<String, List<GbStream>> platformForEvent = new HashMap<>();
+            // 遍历存储结果，查找app+Stream->platformId+catalogId的对应关系，然后执行批量写入
+            for (StreamPush streamPushItem : streamPushItemsForPlatform) {
+                List<String[]> platFormInfoList = streamPushItemsForAll.get(streamPushItem.getApp() + streamPushItem.getStream());
+                if (platFormInfoList != null && platFormInfoList.size() > 0) {
+                    for (String[] platFormInfoArray : platFormInfoList) {
+                        StreamPush streamPushItemForPlatform = new StreamPush();
+                        streamPushItemForPlatform.setGbStreamId(streamPushItem.getGbStreamId());
+                        if (platFormInfoArray.length > 0) {
+                            // 数组 platFormInfoArray 0 为平台ID。 1为目录ID
+                            // 不存在这个平台，则忽略导入此关联关系
+                            if (platformInfoMap.get(platFormInfoArray[0]) == null
+                                    || platformInfoMap.get(platFormInfoArray[0]).get(platFormInfoArray[1]) == null) {
+                                log.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.isEmpty()) {
+                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) {
+                MediaServer mediaServerItem = mediaServerService.getOne(gbStream.getMediaServerId());
+                mediaServerService.closeStreams(mediaServerItem, gbStream.getApp(), gbStream.getStream());
+            }
+
+        }
+        return true;
+    }
+
+    @Override
+    public void allStreamOffline() {
+        List<GbStream> onlinePushers = streamPushMapper.getOnlinePusherForGb();
+        if (onlinePushers.size() == 0) {
+            return;
+        }
+        streamPushMapper.setAllStreamOffline();
+
+        // 发送通知
+        eventPublisher.catalogEventPublishForStream(null, onlinePushers, CatalogEvent.OFF);
+    }
+
+    @Override
+    public void offline(List<StreamPushItemFromRedis> offlineStreams) {
+        // 更新部分设备离线
+        List<GbStream> onlinePushers = streamPushMapper.getOnlinePusherForGbInList(offlineStreams);
+        streamPushMapper.offline(offlineStreams);
+        // 发送通知
+        eventPublisher.catalogEventPublishForStream(null, onlinePushers, CatalogEvent.OFF);
+    }
+
+    @Override
+    public void online(List<StreamPushItemFromRedis> onlineStreams) {
+        // 更新部分设备上线streamPushService
+        List<GbStream> onlinePushers = streamPushMapper.getOfflinePusherForGbInList(onlineStreams);
+        streamPushMapper.online(onlineStreams);
+        // 发送通知
+        eventPublisher.catalogEventPublishForStream(null, onlinePushers, CatalogEvent.ON);
+    }
+
+    @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, StreamPush> getAllAppAndStreamMap() {
+        return streamPushMapper.getAllAppAndStreamMap();
+    }
+
+    @Override
+    public Map<String, StreamPush> getAllGBId() {
+        return streamPushMapper.getAllGBId();
+    }
+
+    @Override
+    public void updateStatus(StreamPush push) {
+
+    }
+
+
+
+    @Override
+    public void updatePushStatus(Integer streamPushId, boolean pushIng) {
+        streamPushInDb.setPushIng(true);
+        if (userSetting.isUsePushingAsStatus()) {
+            streamPushInDb.setGbStatus(true);
+        }
+        streamPushInDb.setPushTime(DateUtil.getNow());
+    }
+
+    private List<StreamPush> handleJSON(List<StreamInfo> streamInfoList) {
+        if (streamInfoList == null || streamInfoList.isEmpty()) {
+            return null;
+        }
+        Map<String, StreamPush> result = new HashMap<>();
+        for (StreamInfo streamInfo : streamInfoList) {
+            // 不保存国标推理以及拉流代理的流
+            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();
+                StreamPush streamPushItem = result.get(key);
+                if (streamPushItem == null) {
+                    streamPushItem = streamPushItem.getInstance(streamInfo);
+                    result.put(key, streamPushItem);
+                }
+            }
+        }
+        return new ArrayList<>(result.values());
+    }
+
+}