增加推流转发到国标，尚不完善

2021-04-02 19:04:01 +08:00
parent f83a192bfd
commit b17280522b
42 changed files with 720 additions and 509 deletions
--- a/src/main/java/com/genersoft/iot/vmp/service/IGbStreamService.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/IGbStreamService.java
@@ -0,0 +1,40 @@
+package com.genersoft.iot.vmp.service;
+
+import com.genersoft.iot.vmp.gb28181.bean.GbStream;
+import com.github.pagehelper.PageInfo;
+
+import java.util.List;
+
+/**
+ * 级联国标平台关联流业务接口
+ */
+public interface IGbStreamService {
+
+    /**
+     * 分页获取所有
+     * @param page
+     * @param count
+     * @return
+     */
+    PageInfo<GbStream> getAll(Integer page, Integer count);
+
+
+    /**
+     * 移除
+     * @param app
+     * @param stream
+     */
+    void del(String app, String stream);
+
+    /**
+     * 保存国标关联
+     * @param gbStreams
+     */
+    boolean addPlatformInfo(List<GbStream> gbStreams, String platformId);
+
+    /**
+     * 移除国标关联
+     * @param gbStreams
+     */
+    boolean delPlatformInfo(List<GbStream> gbStreams);
+}
--- a/src/main/java/com/genersoft/iot/vmp/service/IMediaService.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/IMediaService.java
@@ -0,0 +1,28 @@
+package com.genersoft.iot.vmp.service;
+
+import com.genersoft.iot.vmp.common.StreamInfo;
+import com.genersoft.iot.vmp.gb28181.bean.GbStream;
+import com.genersoft.iot.vmp.gb28181.bean.PlatformGbStream;
+
+/**
+ * 媒体信息业务
+ */
+public interface IMediaService {
+
+    /**
+     * 根据应用名和流ID获取播放地址, 通过zlm接口检查是否存在
+     * @param app
+     * @param stream
+     * @return
+     */
+    StreamInfo getStreamInfoByAppAndStreamWithCheck(String app, String stream);
+
+    /**
+     * 根据应用名和流ID获取播放地址, 只是地址拼接
+     * @param app
+     * @param stream
+     * @return
+     */
+    StreamInfo getStreamInfoByAppAndStream(String app, String stream);
+
+}
--- a/src/main/java/com/genersoft/iot/vmp/service/IPlayService.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/IPlayService.java
@@ -0,0 +1,17 @@
+package com.genersoft.iot.vmp.service;
+
+import com.alibaba.fastjson.JSONObject;
+import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
+import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
+import com.genersoft.iot.vmp.vmanager.play.bean.PlayResult;
+
+/**
+ * 点播处理
+ */
+public interface IPlayService {
+
+    void onPublishHandlerForPlayBack(JSONObject resonse, String deviceId, String channelId, String uuid);
+    void onPublishHandlerForPlay(JSONObject resonse, String deviceId, String channelId, String uuid);
+
+    PlayResult play(String deviceId, String channelId, ZLMHttpHookSubscribe.Event event, SipSubscribe.Event errorEvent);
+}
--- a/src/main/java/com/genersoft/iot/vmp/service/IStreamProxyService.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/IStreamProxyService.java
@@ -0,0 +1,59 @@
+package com.genersoft.iot.vmp.service;
+
+import com.alibaba.fastjson.JSONObject;
+import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
+import com.github.pagehelper.PageInfo;
+
+public interface IStreamProxyService {
+
+    /**
+     * 保存视频代理
+     * @param param
+     */
+    void save(StreamProxyItem param);
+
+    /**
+     * 添加视频代理到zlm
+     * @param param
+     * @return
+     */
+    JSONObject addStreamProxyToZlm(StreamProxyItem param);
+
+    /**
+     * 从zlm移除视频代理
+     * @param param
+     * @return
+     */
+    JSONObject removeStreamProxyFromZlm(StreamProxyItem param);
+
+    /**
+     * 分页查询
+     * @param page
+     * @param count
+     * @return
+     */
+    PageInfo<StreamProxyItem> getAll(Integer page, Integer count);
+
+    /**
+     * 删除视频代理
+     * @param app
+     * @param stream
+     */
+    void del(String app, String stream);
+
+    /**
+     * 启用视频代理
+     * @param app
+     * @param stream
+     * @return
+     */
+    boolean start(String app, String stream);
+
+    /**
+     * 停用用视频代理
+     * @param app
+     * @param stream
+     * @return
+     */
+    boolean stop(String app, String stream);
+}
--- a/src/main/java/com/genersoft/iot/vmp/service/IStreamPushService.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/IStreamPushService.java
@@ -0,0 +1,36 @@
+package com.genersoft.iot.vmp.service;
+
+import com.alibaba.fastjson.JSONObject;
+import com.genersoft.iot.vmp.gb28181.bean.GbStream;
+import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
+import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
+import com.github.pagehelper.PageInfo;
+
+import java.util.List;
+
+public interface IStreamPushService {
+
+    List<StreamPushItem> handleJSON(String json);
+
+    /**
+     * 将应用名和流ID加入国标关联
+     * @param stream
+     * @return
+     */
+    boolean saveToGB(GbStream stream);
+
+    /**
+     * 将应用名和流ID移出国标关联
+     * @param stream
+     * @return
+     */
+    boolean removeFromGB(GbStream stream);
+
+    /**
+     * 获取
+     * @param page
+     * @param count
+     * @return
+     */
+    PageInfo<StreamPushItem> getPushList(Integer page, Integer count);
+}
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/GbStreamServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/impl/GbStreamServiceImpl.java
@@ -0,0 +1,86 @@
+package com.genersoft.iot.vmp.service.impl;
+
+import com.genersoft.iot.vmp.gb28181.bean.GbStream;
+import com.genersoft.iot.vmp.storager.dao.GbStreamMapper;
+import com.genersoft.iot.vmp.storager.dao.PlatformGbStreamMapper;
+import com.genersoft.iot.vmp.service.IGbStreamService;
+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.jdbc.datasource.DataSourceTransactionManager;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.TransactionStatus;
+
+import java.util.List;
+
+@Service
+public class GbStreamServiceImpl implements IGbStreamService {
+
+    private final static Logger logger = LoggerFactory.getLogger(GbStreamServiceImpl.class);
+
+    @Autowired
+    DataSourceTransactionManager dataSourceTransactionManager;
+
+    @Autowired
+    TransactionDefinition transactionDefinition;
+
+    @Autowired
+    private GbStreamMapper gbStreamMapper;
+
+    @Autowired
+    private PlatformGbStreamMapper platformGbStreamMapper;
+
+    @Override
+    public PageInfo<GbStream> getAll(Integer page, Integer count) {
+        PageHelper.startPage(page, count);
+        List<GbStream> all = gbStreamMapper.selectAll();
+        return new PageInfo<>(all);
+    }
+
+    @Override
+    public void del(String app, String stream) {
+        gbStreamMapper.del(app, stream);
+    }
+
+
+    @Override
+    public boolean addPlatformInfo(List<GbStream> gbStreams, String platformId) {
+        // 放在事务内执行
+        boolean result = false;
+        TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
+        try {
+            for (GbStream gbStream : gbStreams) {
+                gbStream.setPlatformId(platformId);
+                platformGbStreamMapper.add(gbStream);
+            }
+            dataSourceTransactionManager.commit(transactionStatus);     //手动提交
+            result = true;
+        }catch (Exception e) {
+            logger.error("批量保存流与平台的关系时错误", e);
+            dataSourceTransactionManager.rollback(transactionStatus);
+        }
+        return result;
+
+    }
+
+    @Override
+    public boolean delPlatformInfo(List<GbStream> gbStreams) {
+        // 放在事务内执行
+        boolean result = false;
+        TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
+        try {
+            for (GbStream gbStream : gbStreams) {
+                platformGbStreamMapper.delByAppAndStream(gbStream.getApp(), gbStream.getStream());
+            }
+            dataSourceTransactionManager.commit(transactionStatus);     //手动提交
+            result = true;
+        }catch (Exception e) {
+            logger.error("批量移除流与平台的关系时错误", e);
+            dataSourceTransactionManager.rollback(transactionStatus);
+        }
+        return result;
+    }
+}
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/impl/MediaServiceImpl.java
@@ -0,0 +1,60 @@
+package com.genersoft.iot.vmp.service.impl;
+
+import com.alibaba.fastjson.JSONObject;
+import com.genersoft.iot.vmp.common.StreamInfo;
+import com.genersoft.iot.vmp.conf.MediaServerConfig;
+import com.genersoft.iot.vmp.gb28181.bean.GbStream;
+import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
+import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
+import com.genersoft.iot.vmp.service.IMediaService;
+import com.genersoft.iot.vmp.storager.dao.GbStreamMapper;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+@Service
+public class MediaServiceImpl implements IMediaService {
+
+    @Autowired
+    private IRedisCatchStorage redisCatchStorage;
+
+    @Autowired
+    private IVideoManagerStorager storager;
+
+    @Autowired
+    private ZLMRESTfulUtils zlmresTfulUtils;
+
+
+
+    @Override
+    public StreamInfo getStreamInfoByAppAndStream(String app, String stream) {
+        MediaServerConfig mediaInfo = redisCatchStorage.getMediaInfo();
+        StreamInfo streamInfoResult = new StreamInfo();
+        streamInfoResult.setStreamId(stream);
+        streamInfoResult.setApp(app);
+        streamInfoResult.setRtmp(String.format("rtmp://%s:%s/%s/%s", mediaInfo.getWanIp(), mediaInfo.getRtmpPort(), app,  stream));
+        streamInfoResult.setRtsp(String.format("rtsp://%s:%s/%s/%s", mediaInfo.getWanIp(), mediaInfo.getRtspPort(), app,  stream));
+        streamInfoResult.setFlv(String.format("http://%s:%s/%s/%s.flv", mediaInfo.getWanIp(), mediaInfo.getHttpPort(), app,  stream));
+        streamInfoResult.setWs_flv(String.format("ws://%s:%s/%s/%s.flv", mediaInfo.getWanIp(), mediaInfo.getHttpPort(), app,  stream));
+        streamInfoResult.setHls(String.format("http://%s:%s/%s/%s/hls.m3u8", mediaInfo.getWanIp(), mediaInfo.getHttpPort(), app,  stream));
+        streamInfoResult.setWs_hls(String.format("ws://%s:%s/%s/%s/hls.m3u8", mediaInfo.getWanIp(), mediaInfo.getHttpPort(), app,  stream));
+        streamInfoResult.setFmp4(String.format("http://%s:%s/%s/%s.live.mp4", mediaInfo.getWanIp(), mediaInfo.getHttpPort(), app,  stream));
+        streamInfoResult.setWs_fmp4(String.format("ws://%s:%s/%s/%s.live.mp4", mediaInfo.getWanIp(), mediaInfo.getHttpPort(), app,  stream));
+        streamInfoResult.setTs(String.format("http://%s:%s/%s/%s.live.ts", mediaInfo.getWanIp(), mediaInfo.getHttpPort(), app,  stream));
+        streamInfoResult.setWs_ts(String.format("ws://%s:%s/%s/%s.live.ts", mediaInfo.getWanIp(), mediaInfo.getHttpPort(), app,  stream));
+        return streamInfoResult;
+    }
+
+    @Override
+    public StreamInfo getStreamInfoByAppAndStreamWithCheck(String app, String stream) {
+        StreamInfo streamInfo = null;
+        JSONObject mediaList = zlmresTfulUtils.getMediaList(app, stream);
+        if (mediaList != null) {
+            streamInfo = getStreamInfoByAppAndStream(app, stream);
+        }
+        return streamInfo;
+    }
+
+
+
+}
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/impl/PlayServiceImpl.java
@@ -0,0 +1,160 @@
+package com.genersoft.iot.vmp.service.impl;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import com.genersoft.iot.vmp.common.StreamInfo;
+import com.genersoft.iot.vmp.gb28181.bean.Device;
+import com.genersoft.iot.vmp.gb28181.bean.DeviceChannel;
+import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
+import com.genersoft.iot.vmp.gb28181.transmit.callback.DeferredResultHolder;
+import com.genersoft.iot.vmp.gb28181.transmit.callback.RequestMessage;
+import com.genersoft.iot.vmp.gb28181.transmit.cmd.impl.SIPCommander;
+import com.genersoft.iot.vmp.media.zlm.ZLMHttpHookSubscribe;
+import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
+import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
+import com.genersoft.iot.vmp.vmanager.play.bean.PlayResult;
+import com.genersoft.iot.vmp.service.IMediaService;
+import com.genersoft.iot.vmp.service.IPlayService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.ResponseEntity;
+import org.springframework.stereotype.Service;
+import org.springframework.web.context.request.async.DeferredResult;
+
+import javax.sip.message.Response;
+import java.util.UUID;
+
+@Service
+public class PlayServiceImpl implements IPlayService {
+
+    private final static Logger logger = LoggerFactory.getLogger(PlayServiceImpl.class);
+
+    @Autowired
+    private IVideoManagerStorager storager;
+
+    @Autowired
+    private SIPCommander cmder;
+
+    @Autowired
+    private IRedisCatchStorage redisCatchStorage;
+
+    @Autowired
+    private DeferredResultHolder resultHolder;
+
+    @Autowired
+    private ZLMRESTfulUtils zlmresTfulUtils;
+
+    @Autowired
+    private IMediaService mediaService;
+
+
+    @Override
+    public PlayResult play(String deviceId, String channelId, ZLMHttpHookSubscribe.Event hookEvent, SipSubscribe.Event errorEvent) {
+        PlayResult playResult = new PlayResult();
+        Device device = storager.queryVideoDevice(deviceId);
+        StreamInfo streamInfo = redisCatchStorage.queryPlayByDevice(deviceId, channelId);
+        playResult.setDevice(device);
+        UUID uuid = UUID.randomUUID();
+        playResult.setUuid(uuid.toString());
+        DeferredResult<ResponseEntity<String>> result = new DeferredResult<ResponseEntity<String>>();
+        playResult.setResult(result);
+        // 录像查询以channelId作为deviceId查询
+        resultHolder.put(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid, result);
+
+        if (streamInfo == null) {
+            // 发送点播消息
+            cmder.playStreamCmd(device, channelId, (JSONObject response) -> {
+                logger.info("收到订阅消息： " + response.toJSONString());
+                onPublishHandlerForPlay(response, deviceId, channelId, uuid.toString());
+                if (hookEvent != null) {
+                    hookEvent.response(response);
+                }
+            }, event -> {
+                RequestMessage msg = new RequestMessage();
+                msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid);
+                Response response = event.getResponse();
+                msg.setData(String.format("点播失败， 错误码： %s, %s", response.getStatusCode(), response.getReasonPhrase()));
+                resultHolder.invokeResult(msg);
+                if (errorEvent != null) {
+                    errorEvent.response(event);
+                }
+            });
+        } else {
+            String streamId = streamInfo.getStreamId();
+            JSONObject rtpInfo = zlmresTfulUtils.getRtpInfo(streamId);
+            if (rtpInfo.getBoolean("exist")) {
+                RequestMessage msg = new RequestMessage();
+                msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid);
+                msg.setData(JSON.toJSONString(streamInfo));
+                resultHolder.invokeResult(msg);
+                if (hookEvent != null) {
+                    hookEvent.response(JSONObject.parseObject(JSON.toJSONString(streamInfo)));
+                }
+            } else {
+                redisCatchStorage.stopPlay(streamInfo);
+                storager.stopPlay(streamInfo.getDeviceID(), streamInfo.getChannelId());
+                cmder.playStreamCmd(device, channelId, (JSONObject response) -> {
+                    logger.info("收到订阅消息： " + response.toJSONString());
+                    onPublishHandlerForPlay(response, deviceId, channelId, uuid.toString());
+                }, event -> {
+                    RequestMessage msg = new RequestMessage();
+                    msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid);
+                    Response response = event.getResponse();
+                    msg.setData(String.format("点播失败， 错误码： %s, %s", response.getStatusCode(), response.getReasonPhrase()));
+                    resultHolder.invokeResult(msg);
+                });
+            }
+        }
+
+        return playResult;
+    }
+
+    @Override
+    public void onPublishHandlerForPlay(JSONObject resonse, String deviceId, String channelId, String uuid) {
+        RequestMessage msg = new RequestMessage();
+        msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid);
+        StreamInfo streamInfo = onPublishHandler(resonse, deviceId, channelId, uuid);
+        if (streamInfo != null) {
+            DeviceChannel deviceChannel = storager.queryChannel(deviceId, channelId);
+            if (deviceChannel != null) {
+                deviceChannel.setStreamId(streamInfo.getStreamId());
+                storager.startPlay(deviceId, channelId, streamInfo.getStreamId());
+            }
+
+            redisCatchStorage.startPlay(streamInfo);
+            msg.setData(JSON.toJSONString(streamInfo));
+            resultHolder.invokeResult(msg);
+        } else {
+            logger.warn("设备预览API调用失败！");
+            msg.setData("设备预览API调用失败！");
+            resultHolder.invokeResult(msg);
+        }
+    }
+
+    @Override
+    public void onPublishHandlerForPlayBack(JSONObject resonse, String deviceId, String channelId, String uuid) {
+        RequestMessage msg = new RequestMessage();
+        msg.setId(DeferredResultHolder.CALLBACK_CMD_PlAY + uuid);
+        StreamInfo streamInfo = onPublishHandler(resonse, deviceId, channelId, uuid);
+        if (streamInfo != null) {
+            redisCatchStorage.startPlayback(streamInfo);
+            msg.setData(JSON.toJSONString(streamInfo));
+            resultHolder.invokeResult(msg);
+        } else {
+            logger.warn("设备预览API调用失败！");
+            msg.setData("设备预览API调用失败！");
+            resultHolder.invokeResult(msg);
+        }
+    }
+
+    public StreamInfo onPublishHandler(JSONObject resonse, String deviceId, String channelId, String uuid) {
+        String streamId = resonse.getString("id");
+        StreamInfo streamInfo = mediaService.getStreamInfoByAppAndStream("rtp", streamId);
+        streamInfo.setDeviceID(deviceId);
+        streamInfo.setChannelId(channelId);
+        return streamInfo;
+    }
+
+}
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamProxyServiceImpl.java
@@ -0,0 +1,131 @@
+package com.genersoft.iot.vmp.service.impl;
+
+import com.alibaba.fastjson.JSONObject;
+import com.genersoft.iot.vmp.conf.MediaServerConfig;
+import com.genersoft.iot.vmp.media.zlm.ZLMRESTfulUtils;
+import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
+import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
+import com.genersoft.iot.vmp.storager.IVideoManagerStorager;
+import com.genersoft.iot.vmp.storager.dao.GbStreamMapper;
+import com.genersoft.iot.vmp.storager.dao.PlatformGbStreamMapper;
+import com.genersoft.iot.vmp.storager.dao.StreamProxyMapper;
+import com.genersoft.iot.vmp.service.IStreamProxyService;
+import com.github.pagehelper.PageInfo;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+/**
+ * 视频代理业务
+ */
+@Service
+public class StreamProxyServiceImpl implements IStreamProxyService {
+
+    @Autowired
+    private IVideoManagerStorager videoManagerStorager;
+
+    @Autowired
+    private IRedisCatchStorage redisCatchStorage;
+
+    @Autowired
+    private ZLMRESTfulUtils zlmresTfulUtils;
+
+    @Autowired
+    private StreamProxyMapper streamProxyMapper;
+
+    @Autowired
+    private GbStreamMapper gbStreamMapper;
+
+    @Autowired
+    private PlatformGbStreamMapper platformGbStreamMapper;
+
+
+    @Override
+    public void save(StreamProxyItem param) {
+        MediaServerConfig mediaInfo = redisCatchStorage.getMediaInfo();
+        String dstUrl = String.format("rtmp://%s:%s/%s/%s", "127.0.0.1", mediaInfo.getRtmpPort(), param.getApp(),
+                param.getStream() );
+        param.setDst_url(dstUrl);
+        // 更新
+        if (videoManagerStorager.queryStreamProxy(param.getApp(), param.getStream()) != null) {
+            boolean result = videoManagerStorager.updateStreamProxy(param);
+            if (result && param.isEnable()) {
+                addStreamProxyToZlm(param);
+            }
+        }else { // 新增
+            boolean result = videoManagerStorager.addStreamProxy(param);
+            if (result  && param.isEnable()) {
+                addStreamProxyToZlm(param);
+            }
+        }
+    }
+
+    @Override
+    public JSONObject addStreamProxyToZlm(StreamProxyItem param) {
+        JSONObject result = null;
+        if ("default".equals(param.getType())){
+            result = zlmresTfulUtils.addStreamProxy(param.getApp(), param.getStream(), param.getUrl(),
+                    param.isEnable_hls(), param.isEnable_mp4(), param.getRtp_type());
+        }else if ("ffmpeg".equals(param.getType())) {
+            result = zlmresTfulUtils.addFFmpegSource(param.getSrc_url(), param.getDst_url(),
+                    param.getTimeout_ms() + "");
+        }
+        return result;
+    }
+
+    @Override
+    public JSONObject removeStreamProxyFromZlm(StreamProxyItem param) {
+        JSONObject result = zlmresTfulUtils.closeStreams(param.getApp(), param.getStream());
+
+        return result;
+    }
+
+    @Override
+    public PageInfo<StreamProxyItem> getAll(Integer page, Integer count) {
+        return videoManagerStorager.queryStreamProxyList(page, count);
+    }
+
+    @Override
+    public void del(String app, String stream) {
+        StreamProxyItem streamProxyItem = new StreamProxyItem();
+        streamProxyItem.setApp(app);
+        streamProxyItem.setStream(stream);
+        JSONObject jsonObject = removeStreamProxyFromZlm(streamProxyItem);
+        if (jsonObject.getInteger("code") == 0) {
+            videoManagerStorager.deleteStreamProxy(app, stream);
+            // 如果关联了国标那么移除关联
+            gbStreamMapper.del(app, stream);
+            platformGbStreamMapper.delByAppAndStream(app, stream);
+            // TODO 如果关联的推流， 那么状态设置为离线
+        }
+    }
+
+    @Override
+    public boolean start(String app, String stream) {
+        boolean result = false;
+        StreamProxyItem streamProxy = videoManagerStorager.queryStreamProxy(app, stream);
+        if (!streamProxy.isEnable() &&  streamProxy != null) {
+            JSONObject jsonObject = addStreamProxyToZlm(streamProxy);
+            if (jsonObject.getInteger("code") == 0) {
+                result = true;
+                streamProxy.setEnable(true);
+                videoManagerStorager.updateStreamProxy(streamProxy);
+            }
+        }
+        return result;
+    }
+
+    @Override
+    public boolean stop(String app, String stream) {
+        boolean result = false;
+        StreamProxyItem streamProxyDto = videoManagerStorager.queryStreamProxy(app, stream);
+        if (streamProxyDto.isEnable() &&  streamProxyDto != null) {
+            JSONObject jsonObject = removeStreamProxyFromZlm(streamProxyDto);
+            if (jsonObject.getInteger("code") == 0) {
+                result = true;
+                streamProxyDto.setEnable(false);
+                videoManagerStorager.updateStreamProxy(streamProxyDto);
+            }
+        }
+        return result;
+    }
+}
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java
@@ -0,0 +1,89 @@
+package com.genersoft.iot.vmp.service.impl;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import com.alibaba.fastjson.TypeReference;
+import com.genersoft.iot.vmp.gb28181.bean.GbStream;
+import com.genersoft.iot.vmp.media.zlm.dto.MediaItem;
+import com.genersoft.iot.vmp.media.zlm.dto.StreamProxyItem;
+import com.genersoft.iot.vmp.media.zlm.dto.StreamPushItem;
+import com.genersoft.iot.vmp.service.IStreamPushService;
+import com.genersoft.iot.vmp.storager.dao.GbStreamMapper;
+import com.genersoft.iot.vmp.storager.dao.StreamPushMapper;
+import com.github.pagehelper.PageHelper;
+import com.github.pagehelper.PageInfo;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+@Service
+public class StreamPushServiceImpl implements IStreamPushService {
+
+    @Autowired
+    private GbStreamMapper gbStreamMapper;
+
+    @Autowired
+    private StreamPushMapper streamPushMapper;
+
+    @Override
+    public List<StreamPushItem> handleJSON(String jsonData) {
+        if (jsonData == null) return null;
+
+        Map<String, StreamPushItem> result = new HashMap<>();
+
+        List<MediaItem> mediaItems = JSON.parseObject(jsonData, new TypeReference<List<MediaItem>>() {});
+        for (MediaItem item : mediaItems) {
+
+            // 不保存国标推理以及拉流代理的流
+            if (item.getOriginType() == 3 || item.getOriginType() == 4 || item.getOriginType() == 5) {
+                continue;
+            }
+            String key = item.getApp() + "_" + item.getStream();
+            StreamPushItem streamPushItem = result.get(key);
+            if (streamPushItem == null) {
+                streamPushItem = new StreamPushItem();
+                streamPushItem.setApp(item.getApp());
+                streamPushItem.setStream(item.getStream());
+                streamPushItem.setAliveSecond(item.getAliveSecond());
+                streamPushItem.setCreateStamp(item.getCreateStamp());
+                streamPushItem.setOriginSock(item.getOriginSock());
+                streamPushItem.setTotalReaderCount(item.getTotalReaderCount());
+                streamPushItem.setOriginType(item.getOriginType());
+                streamPushItem.setOriginTypeStr(item.getOriginTypeStr());
+                streamPushItem.setOriginUrl(item.getOriginUrl());
+                streamPushItem.setCreateStamp(item.getCreateStamp());
+                streamPushItem.setAliveSecond(item.getAliveSecond());
+                streamPushItem.setStatus(true);
+                streamPushItem.setVhost(item.getVhost());
+                result.put(key, streamPushItem);
+            }
+        }
+
+        return new ArrayList<>(result.values());
+    }
+
+    @Override
+    public PageInfo<StreamPushItem> getPushList(Integer page, Integer count) {
+        PageHelper.startPage(page, count);
+        List<StreamPushItem> all = streamPushMapper.selectAll();
+        return new PageInfo<>(all);
+    }
+
+    @Override
+    public boolean saveToGB(GbStream stream) {
+        stream.setStreamType("push");
+        stream.setStatus(true);
+        int add = gbStreamMapper.add(stream);
+        return add > 0;
+    }
+
+    @Override
+    public boolean removeFromGB(GbStream stream) {
+        int del = gbStreamMapper.del(stream.getApp(), stream.getStream());
+        return del > 0;
+    }
+}