Pre Merge pull request !34 from WangXuewen/master

src/main/java/com/genersoft/iot/vmp/gb28181/controller/MediaController.java

@@ -7,26 +7,19 @@ import com.genersoft.iot.vmp.conf.security.SecurityUtils;
 import com.genersoft.iot.vmp.conf.security.dto.LoginUser;
 import com.genersoft.iot.vmp.media.service.IMediaServerService;
 import com.genersoft.iot.vmp.media.zlm.dto.StreamAuthorityInfo;
-import com.genersoft.iot.vmp.service.bean.ErrorCallback;
-import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.streamProxy.service.IStreamProxyService;
 import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
 import com.genersoft.iot.vmp.vmanager.bean.StreamContent;
-import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
 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 lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.util.ObjectUtils;
 import org.springframework.web.bind.annotation.*;
-import org.springframework.web.context.request.async.DeferredResult;
 
 import javax.servlet.http.HttpServletRequest;
-import java.net.MalformedURLException;
-import java.net.URL;
 
 
 @Tag(name  = "媒体流相关")
@@ -59,12 +52,11 @@ public class MediaController {
     @Parameter(name = "useSourceIpAsStreamIp", description = "是否使用请求IP作为返回的地址IP")
     @GetMapping(value = "/stream_info_by_app_and_stream")
     @ResponseBody
-    public DeferredResult<WVPResult<StreamContent>> getStreamInfoByAppAndStream(HttpServletRequest request, @RequestParam String app,
+    public StreamContent getStreamInfoByAppAndStream(HttpServletRequest request, @RequestParam String app,
-                                                                                @RequestParam String stream,
+                                                     @RequestParam String stream,
-                                                                                @RequestParam(required = false) String mediaServerId,
+                                                     @RequestParam(required = false) String mediaServerId,
-                                                                                @RequestParam(required = false) String callId,
+                                                     @RequestParam(required = false) String callId,
-                                                                                @RequestParam(required = false) Boolean useSourceIpAsStreamIp){
+                                                     @RequestParam(required = false) Boolean useSourceIpAsStreamIp){
-        DeferredResult<WVPResult<StreamContent>> result = new DeferredResult<>();
         boolean authority = false;
         if (callId != null) {
             // 权限校验
@@ -83,7 +75,9 @@ public class MediaController {
                 authority = true;
             }
         }
+
         StreamInfo streamInfo;
+
         if (useSourceIpAsStreamIp != null && useSourceIpAsStreamIp) {
             String host = request.getHeader("Host");
             String localAddr = host.split(":")[0];
@@ -94,37 +88,30 @@ public class MediaController {
         }
 
         if (streamInfo != null){
-            WVPResult<StreamContent> wvpResult = WVPResult.success();
+            return new StreamContent(streamInfo);
-            wvpResult.setData(new StreamContent(streamInfo));
-            result.setResult(wvpResult);
         }else {
-            ErrorCallback<StreamInfo> callback = (code, msg, streamInfoStoStart) -> {
-                if (code == InviteErrorCode.SUCCESS.getCode()) {
-                    WVPResult<StreamContent> wvpResult = WVPResult.success();
-                    if (useSourceIpAsStreamIp != null && useSourceIpAsStreamIp) {
-                        String host;
-                        try {
-                            URL url=new URL(request.getRequestURL().toString());
-                            host=url.getHost();
-                        } catch (MalformedURLException e) {
-                            host=request.getLocalAddr();
-                        }
-                        streamInfoStoStart.changeStreamIp(host);
-                    }
-                    if (!ObjectUtils.isEmpty(streamInfoStoStart.getMediaServer().getTranscodeSuffix())
-                            && !"null".equalsIgnoreCase(streamInfoStoStart.getMediaServer().getTranscodeSuffix())) {
-                        streamInfoStoStart.setStream(streamInfoStoStart.getStream() + "_" + streamInfoStoStart.getMediaServer().getTranscodeSuffix());
-                    }
-                    wvpResult.setData(new StreamContent(streamInfoStoStart));
-                    result.setResult(wvpResult);
-                }else {
-                    result.setResult(WVPResult.fail(code, msg));
-                }
-            };
             //获取流失败，重启拉流后重试一次
-            streamProxyService.startByAppAndStream(app, stream, callback);
+            streamProxyService.stopByAppAndStream(app,stream);
+            boolean start = streamProxyService.startByAppAndStream(app, stream);
+            try {
+                Thread.sleep(1000);
+            } catch (InterruptedException e) {
+                log.error("[线程休眠失败]， {}", e.getMessage());
+            }
+            if (useSourceIpAsStreamIp != null && useSourceIpAsStreamIp) {
+                String host = request.getHeader("Host");
+                String localAddr = host.split(":")[0];
+                log.info("使用{}作为返回流的ip", localAddr);
+                streamInfo = mediaServerService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, localAddr, authority);
+            }else {
+                streamInfo = mediaServerService.getStreamInfoByAppAndStreamWithCheck(app, stream, mediaServerId, authority);
+            }
+            if (streamInfo != null){
+                return new StreamContent(streamInfo);
+            }else {
+                throw new ControllerException(ErrorCode.ERROR100);
+            }
         }
-        return result;
     }
     /**
      * 获取推流播放地址

src/main/java/com/genersoft/iot/vmp/media/service/IMediaNodeServerService.java

@@ -64,7 +64,7 @@ public interface IMediaNodeServerService {
 
     Long updateDownloadProcess(MediaServer mediaServer, String app, String stream);
 
-    void startProxy(MediaServer mediaServer, StreamProxy streamProxy);
+    StreamInfo startProxy(MediaServer mediaServer, StreamProxy streamProxy);
 
     void stopProxy(MediaServer mediaServer, String streamKey);
 

src/main/java/com/genersoft/iot/vmp/media/service/IMediaServerService.java

@@ -150,7 +150,7 @@ public interface IMediaServerService {
 
     Long updateDownloadProcess(MediaServer mediaServerItem, String app, String stream);
 
-    void startProxy(MediaServer mediaServer, StreamProxy streamProxy);
+    StreamInfo startProxy(MediaServer mediaServer, StreamProxy streamProxy);
 
     void stopProxy(MediaServer mediaServer, String streamKey);
 

src/main/java/com/genersoft/iot/vmp/media/service/impl/MediaServerServiceImpl.java

@@ -952,13 +952,13 @@ public class MediaServerServiceImpl implements IMediaServerService {
     }
 
     @Override
-    public void startProxy(MediaServer mediaServer, StreamProxy streamProxy) {
+    public StreamInfo startProxy(MediaServer mediaServer, StreamProxy streamProxy) {
         IMediaNodeServerService mediaNodeServerService = nodeServerServiceMap.get(mediaServer.getType());
         if (mediaNodeServerService == null) {
             log.info("[startProxy] 失败, mediaServer的类型： {}，未找到对应的实现类", mediaServer.getType());
             throw new ControllerException(ErrorCode.ERROR100.getCode(), "未找到mediaServer对应的实现类");
         }
-        mediaNodeServerService.startProxy(mediaServer, streamProxy);
+        return mediaNodeServerService.startProxy(mediaServer, streamProxy);
     }
 
     @Override

src/main/java/com/genersoft/iot/vmp/media/zlm/ZLMMediaNodeServerService.java

@@ -425,7 +425,7 @@ public class ZLMMediaNodeServerService implements IMediaNodeServerService {
     }
 
     @Override
-    public void startProxy(MediaServer mediaServer, StreamProxy streamProxy) {
+    public StreamInfo startProxy(MediaServer mediaServer, StreamProxy streamProxy) {
         String dstUrl;
         if ("ffmpeg".equalsIgnoreCase(streamProxy.getType())) {
 
@@ -463,6 +463,10 @@ public class ZLMMediaNodeServerService implements IMediaNodeServerService {
         MediaInfo mediaInfo = getMediaInfo(mediaServer, streamProxy.getApp(), streamProxy.getStream());
 
         if (mediaInfo != null) {
+            if (mediaInfo.getOriginUrl() != null && mediaInfo.getOriginUrl().equals(streamProxy.getSrcUrl())) {
+                log.info("[启动拉流代理] 已存在， 直接返回， app： {}, stream: {}", mediaInfo.getApp(), streamProxy.getStream());
+                return getStreamInfoByAppAndStream(mediaServer, streamProxy.getApp(), streamProxy.getStream(), mediaInfo, null, true);
+            }
             closeStreams(mediaServer, streamProxy.getApp(), streamProxy.getStream());
         }
 
@@ -486,6 +490,15 @@ public class ZLMMediaNodeServerService implements IMediaNodeServerService {
             JSONObject data = jsonObject.getJSONObject("data");
             if (data == null) {
                 throw new ControllerException(jsonObject.getInteger("code"), "代理结果异常： " + jsonObject);
+            }else {
+                streamProxy.setStreamKey(data.getString("key"));
+                // 由于此时流未注册，手动拼装流信息
+                mediaInfo = new MediaInfo();
+                mediaInfo.setApp(streamProxy.getApp());
+                mediaInfo.setStream(streamProxy.getStream());
+                mediaInfo.setOriginType(4);
+                mediaInfo.setOriginTypeStr("pull");
+                return getStreamInfoByAppAndStream(mediaServer, streamProxy.getApp(), streamProxy.getStream(), mediaInfo, null, true);
             }
         }
     }

src/main/java/com/genersoft/iot/vmp/service/redisMsg/IRedisRpcPlayService.java

@@ -28,7 +28,7 @@ public interface IRedisRpcPlayService {
 
     void playPush(String serverId, Integer id, ErrorCallback<StreamInfo> callback);
 
-    void playProxy(String serverId, int id, ErrorCallback<StreamInfo> callback);
+    StreamInfo playProxy(String serverId, int id);
 
     void stopProxy(String serverId, int id);
 

src/main/java/com/genersoft/iot/vmp/service/redisMsg/control/RedisRpcStreamProxyController.java

@@ -1,6 +1,7 @@
 package com.genersoft.iot.vmp.service.redisMsg.control;
 
 import com.alibaba.fastjson2.JSONObject;
+import com.genersoft.iot.vmp.common.StreamInfo;
 import com.genersoft.iot.vmp.conf.UserSetting;
 import com.genersoft.iot.vmp.conf.redis.RedisRpcConfig;
 import com.genersoft.iot.vmp.conf.redis.bean.RedisRpcMessage;
@@ -62,13 +63,10 @@ public class RedisRpcStreamProxyController extends RpcController {
             response.setBody("param error");
             return response;
         }
-        streamProxyPlayService.startProxy(streamProxy, (code, msg, streamInfo) -> {
+        StreamInfo streamInfo = streamProxyPlayService.startProxy(streamProxy);
-            response.setStatusCode(code);
+        response.setStatusCode(ErrorCode.SUCCESS.getCode());
-            response.setBody(JSONObject.toJSONString(streamInfo));
+        response.setBody(JSONObject.toJSONString(streamInfo));
-            sendResponse(response);
+        return response;
-        });
-
-        return null;
     }
 
     /**

src/main/java/com/genersoft/iot/vmp/service/redisMsg/service/RedisRpcPlayServiceImpl.java

@@ -212,20 +212,13 @@ public class RedisRpcPlayServiceImpl implements IRedisRpcPlayService {
     }
 
     @Override
-    public void playProxy(String serverId, int id, ErrorCallback<StreamInfo> callback) {
+    public StreamInfo playProxy(String serverId, int id) {
         RedisRpcRequest request = buildRequest("streamProxy/play", id);
-        request.setToId(serverId);
         RedisRpcResponse response = redisRpcConfig.request(request, userSetting.getPlayTimeout(), TimeUnit.SECONDS);
-        if (response == null) {
+        if (response != null && response.getStatusCode() == ErrorCode.SUCCESS.getCode()) {
-            callback.run(ErrorCode.ERROR100.getCode(), ErrorCode.ERROR100.getMsg(), null);
+            return JSON.parseObject(response.getBody().toString(), StreamInfo.class);
-        }else {
-            if (response.getStatusCode() == ErrorCode.SUCCESS.getCode()) {
-                StreamInfo streamInfo = JSON.parseObject(response.getBody().toString(), StreamInfo.class);
-                callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo);
-            }else {
-                callback.run(response.getStatusCode(), response.getBody().toString(), null);
-            }
         }
+        return null;
     }
 
     @Override

src/main/java/com/genersoft/iot/vmp/streamProxy/controller/StreamProxyController.java

@@ -7,15 +7,12 @@ import com.genersoft.iot.vmp.conf.exception.ControllerException;
 import com.genersoft.iot.vmp.conf.security.JwtUtils;
 import com.genersoft.iot.vmp.media.bean.MediaServer;
 import com.genersoft.iot.vmp.media.service.IMediaServerService;
-import com.genersoft.iot.vmp.service.bean.ErrorCallback;
-import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
 import com.genersoft.iot.vmp.streamProxy.bean.StreamProxy;
 import com.genersoft.iot.vmp.streamProxy.bean.StreamProxyParam;
 import com.genersoft.iot.vmp.streamProxy.service.IStreamProxyPlayService;
 import com.genersoft.iot.vmp.streamProxy.service.IStreamProxyService;
 import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
 import com.genersoft.iot.vmp.vmanager.bean.StreamContent;
-import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
 import com.github.pagehelper.PageInfo;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
@@ -23,10 +20,8 @@ import io.swagger.v3.oas.annotations.security.SecurityRequirement;
 import io.swagger.v3.oas.annotations.tags.Tag;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
 import org.springframework.web.bind.annotation.*;
-import org.springframework.web.context.request.async.DeferredResult;
 
 import javax.servlet.http.HttpServletRequest;
 import java.net.MalformedURLException;
@@ -94,7 +89,7 @@ public class StreamProxyController {
     })
     @PostMapping(value = "/save")
     @ResponseBody
-    public DeferredResult<WVPResult<StreamContent>> save(HttpServletRequest request, @RequestBody StreamProxyParam param){
+    public StreamContent save(@RequestBody StreamProxyParam param){
         log.info("添加代理： " + JSONObject.toJSONString(param));
         if (ObjectUtils.isEmpty(param.getMediaServerId())) {
             param.setMediaServerId("auto");
@@ -102,39 +97,18 @@ public class StreamProxyController {
         if (ObjectUtils.isEmpty(param.getType())) {
             param.setType("default");
         }
-        DeferredResult<WVPResult<StreamContent>> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue());
-        ErrorCallback<StreamInfo> callback = (code, msg, streamInfo) -> {
-            if (code == InviteErrorCode.SUCCESS.getCode()) {
-                WVPResult<StreamContent> wvpResult = WVPResult.success();
-                if (streamInfo != null) {
-                    if (userSetting.getUseSourceIpAsStreamIp()) {
-                        streamInfo=streamInfo.clone();//深拷贝
-                        String host;
-                        try {
-                            URL url=new URL(request.getRequestURL().toString());
-                            host=url.getHost();
-                        } catch (MalformedURLException e) {
-                            host=request.getLocalAddr();
-                        }
-                        streamInfo.changeStreamIp(host);
-                    }
-                    if (!ObjectUtils.isEmpty(streamInfo.getMediaServer().getTranscodeSuffix())
-                            && !"null".equalsIgnoreCase(streamInfo.getMediaServer().getTranscodeSuffix())) {
-                        streamInfo.setStream(streamInfo.getStream() + "_" + streamInfo.getMediaServer().getTranscodeSuffix());
-                    }
-                    wvpResult.setData(new StreamContent(streamInfo));
-                }else {
-                    wvpResult.setCode(code);
-                    wvpResult.setMsg(msg);
-                }
 
-                result.setResult(wvpResult);
+        StreamInfo streamInfo =  streamProxyService.save(param);
+        if (param.isEnable()) {
+            if (streamInfo == null) {
+                throw new ControllerException(ErrorCode.ERROR100.getCode(), ErrorCode.ERROR100.getMsg());
             }else {
-                result.setResult(WVPResult.fail(code, msg));
+                return new StreamContent(streamInfo);
             }
-        };
+        }else {
-        streamProxyService.save(param, callback);
+            return null;
-        return result;
+        }
+
     }
 
     @Operation(summary = "新增代理", security = @SecurityRequirement(name = JwtUtils.HEADER), parameters = {
@@ -219,46 +193,25 @@ public class StreamProxyController {
     @ResponseBody
     @Operation(summary = "启用代理", security = @SecurityRequirement(name = JwtUtils.HEADER))
     @Parameter(name = "id", description = "代理Id", required = true)
-    public DeferredResult<WVPResult<StreamContent>> start(HttpServletRequest request, int id){
+    public StreamContent start(HttpServletRequest request, int id){
         log.info("播放代理： {}", id);
-        StreamProxy streamProxy = streamProxyService.getStreamProxy(id);
+        StreamInfo streamInfo = streamProxyPlayService.start(id, null, null);
-        Assert.notNull(streamProxy, "代理信息不存在");
+        if (streamInfo == null) {
-
+            throw new ControllerException(ErrorCode.ERROR100.getCode(), ErrorCode.ERROR100.getMsg());
-        DeferredResult<WVPResult<StreamContent>> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue());
+        }else {
-
+            if (userSetting.getUseSourceIpAsStreamIp()) {
-        ErrorCallback<StreamInfo> callback = (code, msg, streamInfo) -> {
+                streamInfo=streamInfo.clone();//深拷贝
-            if (code == InviteErrorCode.SUCCESS.getCode()) {
+                String host;
-                WVPResult<StreamContent> wvpResult = WVPResult.success();
+                try {
-                if (streamInfo != null) {
+                    URL url=new URL(request.getRequestURL().toString());
-                    if (userSetting.getUseSourceIpAsStreamIp()) {
+                    host=url.getHost();
-                        streamInfo=streamInfo.clone();//深拷贝
+                } catch (MalformedURLException e) {
-                        String host;
+                    host=request.getLocalAddr();
-                        try {
-                            URL url=new URL(request.getRequestURL().toString());
-                            host=url.getHost();
-                        } catch (MalformedURLException e) {
-                            host=request.getLocalAddr();
-                        }
-                        streamInfo.changeStreamIp(host);
-                    }
-                    if (!ObjectUtils.isEmpty(streamInfo.getMediaServer().getTranscodeSuffix())
-                            && !"null".equalsIgnoreCase(streamInfo.getMediaServer().getTranscodeSuffix())) {
-                        streamInfo.setStream(streamInfo.getStream() + "_" + streamInfo.getMediaServer().getTranscodeSuffix());
-                    }
-                    wvpResult.setData(new StreamContent(streamInfo));
-                }else {
-                    wvpResult.setCode(code);
-                    wvpResult.setMsg(msg);
                 }
-
+                streamInfo.changeStreamIp(host);
-                result.setResult(wvpResult);
-            }else {
-                result.setResult(WVPResult.fail(code, msg));
             }
-        };
+            return new StreamContent(streamInfo);
-
+        }
-        streamProxyPlayService.start(id, null, callback);
-        return result;
     }
 
     @GetMapping(value = "/stop")

src/main/java/com/genersoft/iot/vmp/streamProxy/dao/StreamProxyMapper.java

@@ -92,5 +92,5 @@ public interface StreamProxyMapper {
             " SET pulling=#{pulling}, media_server_id = #{mediaServerId}, " +
             " stream_key = #{streamKey} " +
             " WHERE id=#{id}")
-    void updateStream(StreamProxy streamProxy);
+    void addStream(StreamProxy streamProxy);
 }

src/main/java/com/genersoft/iot/vmp/streamProxy/service/IStreamProxyPlayService.java

@@ -4,13 +4,13 @@ import com.genersoft.iot.vmp.common.StreamInfo;
 import com.genersoft.iot.vmp.service.bean.ErrorCallback;
 import com.genersoft.iot.vmp.streamProxy.bean.StreamProxy;
 
-import javax.validation.constraints.NotNull;
-
 public interface IStreamProxyPlayService {
 
-    void start(int id, Boolean record, ErrorCallback<StreamInfo> callback);
+    StreamInfo start(int id, Boolean record, ErrorCallback<StreamInfo> callback);
 
-    void startProxy(@NotNull StreamProxy streamProxy, ErrorCallback<StreamInfo> callback);
+    void start(int id, ErrorCallback<StreamInfo> callback);
+
+    StreamInfo startProxy(StreamProxy streamProxy);
 
     void stop(int id);
 

src/main/java/com/genersoft/iot/vmp/streamProxy/service/IStreamProxyService.java

@@ -2,7 +2,6 @@ package com.genersoft.iot.vmp.streamProxy.service;
 
 import com.genersoft.iot.vmp.common.StreamInfo;
 import com.genersoft.iot.vmp.media.bean.MediaServer;
-import com.genersoft.iot.vmp.service.bean.ErrorCallback;
 import com.genersoft.iot.vmp.streamProxy.bean.StreamProxy;
 import com.genersoft.iot.vmp.streamProxy.bean.StreamProxyParam;
 import com.genersoft.iot.vmp.vmanager.bean.ResourceBaseInfo;
@@ -16,7 +15,7 @@ public interface IStreamProxyService {
      * 保存视频代理
      * @param param
      */
-    void save(StreamProxyParam param, ErrorCallback<StreamInfo> callback);
+    StreamInfo save(StreamProxyParam param);
 
     /**
      * 分页查询
@@ -39,7 +38,7 @@ public interface IStreamProxyService {
      * @param stream
      * @return
      */
-    void startByAppAndStream(String app, String stream, ErrorCallback<StreamInfo> callback);
+    boolean startByAppAndStream(String app, String stream);
 
     /**
      * 停用用视频代理

src/main/java/com/genersoft/iot/vmp/streamProxy/service/impl/StreamProxyPlayServiceImpl.java

@@ -4,10 +4,12 @@ import com.genersoft.iot.vmp.common.StreamInfo;
 import com.genersoft.iot.vmp.conf.DynamicTask;
 import com.genersoft.iot.vmp.conf.UserSetting;
 import com.genersoft.iot.vmp.conf.exception.ControllerException;
+import com.genersoft.iot.vmp.media.bean.MediaInfo;
 import com.genersoft.iot.vmp.media.bean.MediaServer;
 import com.genersoft.iot.vmp.media.event.hook.Hook;
 import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
 import com.genersoft.iot.vmp.media.event.hook.HookType;
+import com.genersoft.iot.vmp.media.event.media.MediaArrivalEvent;
 import com.genersoft.iot.vmp.media.service.IMediaServerService;
 import com.genersoft.iot.vmp.service.bean.ErrorCallback;
 import com.genersoft.iot.vmp.service.bean.InviteErrorCode;
@@ -18,12 +20,16 @@ import com.genersoft.iot.vmp.streamProxy.service.IStreamProxyPlayService;
 import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.event.EventListener;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
 
-import javax.validation.constraints.NotNull;
+import javax.sip.message.Response;
 import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * 视频代理业务
@@ -50,42 +56,107 @@ public class StreamProxyPlayServiceImpl implements IStreamProxyPlayService {
     @Autowired
     private IRedisRpcPlayService redisRpcPlayService;
 
+    private ConcurrentHashMap<Integer, ErrorCallback<StreamInfo>> callbackMap = new ConcurrentHashMap<>();
+
+    private ConcurrentHashMap<Integer, StreamInfo> streamInfoMap = new ConcurrentHashMap<>();
+
+    /**
+     * 流到来的处理
+     */
+    @Async("taskExecutor")
+    @Transactional
+    @EventListener
+    public void onApplicationEvent(MediaArrivalEvent event) {
+        if ("rtsp".equals(event.getSchema())) {
+            StreamProxy streamProxy = streamProxyMapper.selectOneByAppAndStream(event.getApp(), event.getStream());
+            if (streamProxy != null) {
+                ErrorCallback<StreamInfo> callback = callbackMap.remove(streamProxy.getId());
+                StreamInfo streamInfo = streamInfoMap.remove(streamProxy.getId());
+                if (callback != null && streamInfo != null) {
+                    callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo);
+                }
+            }
+        }
+    }
+
     @Override
-    public void start(int id, Boolean record, ErrorCallback<StreamInfo> callback) {
+    public void start(int id, ErrorCallback<StreamInfo> callback) {
+        StreamProxy streamProxy = streamProxyMapper.select(id);
+        if (streamProxy == null) {
+            throw new ControllerException(ErrorCode.ERROR404.getCode(), "代理信息未找到");
+        }
+        StreamInfo streamInfo = startProxy(streamProxy);
+        if (streamInfo == null) {
+            callback.run(Response.BUSY_HERE, "busy here", null);
+            return;
+        }
+        callbackMap.put(id, callback);
+        streamInfoMap.put(id, streamInfo);
+
+        MediaServer mediaServer = mediaServerService.getOne(streamProxy.getMediaServerId());
+        if (mediaServer != null) {
+            MediaInfo mediaInfo = mediaServerService.getMediaInfo(mediaServer, streamProxy.getApp(), streamProxy.getStream());
+            if (mediaInfo != null) {
+                callbackMap.remove(id);
+                streamInfoMap.remove(id);
+                callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo);
+            }
+        }
+    }
+
+    @Override
+    public StreamInfo start(int id, Boolean record, ErrorCallback<StreamInfo> callback) {
         log.info("[拉流代理]， 开始拉流，ID：{}", id);
         StreamProxy streamProxy = streamProxyMapper.select(id);
         if (streamProxy == null) {
             throw new ControllerException(ErrorCode.ERROR404.getCode(), "代理信息未找到");
         }
-        log.info("[拉流代理] 类型： {}， app：{}, stream: {}, 流地址： {}", streamProxy.getType(), streamProxy.getApp(), streamProxy.getStream(), streamProxy.getSrcUrl());
         if (record != null) {
             streamProxy.setEnableMp4(record);
         }
+        if (streamProxy.getMediaServerId() != null) {
+            StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStreamWithCheck(streamProxy.getApp(), streamProxy.getStream(), streamProxy.getMediaServerId(), false);
+            if (streamInfo != null) {
+                callbackMap.remove(id);
+                streamInfoMap.remove(id);
+                callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo);
+                return streamInfo;
+            }
+        }
 
-        startProxy(streamProxy, callback);
+        StreamInfo streamInfo = startProxy(streamProxy);
+        if (callback != null) {
+            // 设置流超时的定时任务
+            String timeOutTaskKey = UUID.randomUUID().toString();
+            Hook rtpHook = Hook.getInstance(HookType.on_media_arrival, streamProxy.getApp(), streamProxy.getStream(), streamInfo.getMediaServer().getId());
+            dynamicTask.startDelay(timeOutTaskKey, () -> {
+                log.info("[拉流代理]， 收流超时，ID：{}", id);
+                // 收流超时
+                subscribe.removeSubscribe(rtpHook);
+                callback.run(InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode(), InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getMsg(), streamInfo);
+            }, userSetting.getPlayTimeout());
+
+            // 开启流到来的监听
+            subscribe.addSubscribe(rtpHook, (hookData) -> {
+                dynamicTask.stop(timeOutTaskKey);
+                // hook响应
+                callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo);
+                subscribe.removeSubscribe(rtpHook);
+            });
+        }
+        return streamInfo;
     }
 
     @Override
-    public void startProxy(@NotNull StreamProxy streamProxy, ErrorCallback<StreamInfo> callback){
+    public StreamInfo startProxy(StreamProxy streamProxy){
         if (!streamProxy.isEnable()) {
-            callback.run(ErrorCode.ERROR100.getCode(), "代理未启用", null);
+            return null;
-            return;
         }
         if (streamProxy.getServerId() == null) {
             streamProxy.setServerId(userSetting.getServerId());
         }
         if (!userSetting.getServerId().equals(streamProxy.getServerId())) {
-            log.info("[拉流代理] 由其他服务{}管理", streamProxy.getServerId());
+            return redisRpcPlayService.playProxy(streamProxy.getServerId(), streamProxy.getId());
-            redisRpcPlayService.playProxy(streamProxy.getServerId(), streamProxy.getId(), callback);
-            return;
-        }
-
-        if (streamProxy.getMediaServerId() != null) {
-            StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStreamWithCheck(streamProxy.getApp(), streamProxy.getStream(), streamProxy.getMediaServerId(), null, false);
-            if (streamInfo != null) {
-                callback.run(ErrorCode.SUCCESS.getCode(), ErrorCode.SUCCESS.getMsg(), streamInfo);
-                return;
-            }
         }
 
         MediaServer mediaServer;
@@ -98,32 +169,12 @@ public class StreamProxyPlayServiceImpl implements IStreamProxyPlayService {
         if (mediaServer == null) {
             throw new ControllerException(ErrorCode.ERROR100.getCode(), mediaServerId == null?"未找到可用的媒体节点":"未找到节点" + mediaServerId);
         }
-
+        StreamInfo streamInfo = mediaServerService.startProxy(mediaServer, streamProxy);
-        // 设置流超时的定时任务
-        String timeOutTaskKey = UUID.randomUUID().toString();
-        Hook rtpHook = Hook.getInstance(HookType.on_media_arrival, streamProxy.getApp(), streamProxy.getStream(), mediaServer.getId());
-        dynamicTask.startDelay(timeOutTaskKey, () -> {
-            log.info("[拉流代理] 收流超时，app：{}，stream: {}", streamProxy.getApp(), streamProxy.getStream());
-            // 收流超时
-            subscribe.removeSubscribe(rtpHook);
-            callback.run(InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getCode(), InviteErrorCode.ERROR_FOR_STREAM_TIMEOUT.getMsg(), null);
-        }, userSetting.getPlayTimeout());
-
-        // 开启流到来的监听
-        subscribe.addSubscribe(rtpHook, (hookData) -> {
-            log.info("[拉流代理] 收流成功，app：{}，stream: {}", hookData.getApp(), hookData.getStream());
-            dynamicTask.stop(timeOutTaskKey);
-            StreamInfo streamInfo = mediaServerService.getStreamInfoByAppAndStream(mediaServer, hookData.getApp(), hookData.getStream(), hookData.getMediaInfo(), null);
-            // hook响应
-            callback.run(InviteErrorCode.SUCCESS.getCode(), InviteErrorCode.SUCCESS.getMsg(), streamInfo);
-            subscribe.removeSubscribe(rtpHook);
-        });
-
-        mediaServerService.startProxy(mediaServer, streamProxy);
         if (mediaServerId == null || !mediaServerId.equals(mediaServer.getId())) {
             streamProxy.setMediaServerId(mediaServer.getId());
-            streamProxyMapper.updateStream(streamProxy);
+            streamProxyMapper.addStream(streamProxy);
         }
+        return streamInfo;
     }
 
     @Override

src/main/java/com/genersoft/iot/vmp/streamProxy/service/impl/StreamProxyServiceImpl.java

@@ -15,7 +15,6 @@ 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.hook.OriginType;
-import com.genersoft.iot.vmp.service.bean.ErrorCallback;
 import com.genersoft.iot.vmp.storager.IRedisCatchStorage;
 import com.genersoft.iot.vmp.streamProxy.bean.StreamProxy;
 import com.genersoft.iot.vmp.streamProxy.bean.StreamProxyParam;
@@ -110,9 +109,7 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
         // 拉流代理
         StreamProxy streamProxyByAppAndStream = getStreamProxyByAppAndStream(event.getApp(), event.getStream());
         if (streamProxyByAppAndStream != null && streamProxyByAppAndStream.isEnableDisableNoneReader()) {
-            startByAppAndStream(event.getApp(), event.getStream(), ((code, msg, data) -> {
+            startByAppAndStream(event.getApp(), event.getStream());
-                log.info("[拉流代理] 自动点播成功， app： {}， stream: {}", event.getApp(), event.getStream());
-            }));
         }
     }
 
@@ -139,7 +136,7 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
 
     @Override
     @Transactional
-    public void save(StreamProxyParam param, ErrorCallback<StreamInfo> callback) {
+    public StreamInfo save(StreamProxyParam param) {
         // 兼容旧接口
         StreamProxy streamProxyInDb = getStreamProxyByAppAndStream(param.getApp(), param.getStream());
         if (streamProxyInDb != null && streamProxyInDb.getPulling() != null && streamProxyInDb.getPulling()) {
@@ -162,7 +159,9 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
         }
 
         if (param.isEnable()) {
-            playService.startProxy(streamProxy, callback);
+            return playService.startProxy(streamProxy);
+        } else {
+            return null;
         }
     }
 
@@ -248,12 +247,13 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
 
 
     @Override
-    public void startByAppAndStream(String app, String stream, ErrorCallback<StreamInfo> callback) {
+    public boolean startByAppAndStream(String app, String stream) {
         StreamProxy streamProxy = streamProxyMapper.selectOneByAppAndStream(app, stream);
         if (streamProxy == null) {
             throw new ControllerException(ErrorCode.ERROR404.getCode(), "代理信息未找到");
         }
-        playService.startProxy(streamProxy, callback);
+        StreamInfo streamInfo = playService.startProxy(streamProxy);
+        return streamInfo != null;
     }
 
     @Override
@@ -406,7 +406,7 @@ public class StreamProxyServiceImpl implements IStreamProxyService {
         streamProxy.setPulling(status);
         streamProxy.setMediaServerId(mediaServerId);
         streamProxy.setUpdateTime(DateUtil.getNow());
-        streamProxyMapper.updateStream(streamProxy);
+        streamProxyMapper.addStream(streamProxy);
 
         streamProxy.setGbStatus(status ? "ON" : "OFF");
         if (streamProxy.getGbId() > 0) {

web/src/views/streamPush/buildPushStreamUrl.vue

@@ -1,163 +0,0 @@
-<template>
-  <div id="addUser" v-loading="isLoging">
-    <el-dialog
-      v-el-drag-dialog
-      title="构建推流地址"
-      width="40%"
-      top="2rem"
-      :close-on-click-modal="false"
-      :visible.sync="showDialog"
-      :destroy-on-close="true"
-      @close="close()"
-    >
-      <div id="shared" style="margin-right: 20px;">
-        <el-form ref="buildFrom" status-icon label-width="80px">
-          <el-form-item label="应用名" prop="app">
-            <el-input v-model="app" autocomplete="off" />
-          </el-form-item>
-          <el-form-item label="流ID" prop="stream">
-            <el-input v-model="stream" autocomplete="off" />
-          </el-form-item>
-          <el-form-item label="媒体节点" prop="mediaServerId">
-            <el-select v-model="mediaServer" placeholder="请选择" style="width: 100%">
-              <el-option
-                v-for="item in mediaServerList"
-                :key="item.id"
-                :label="item.id"
-                :value="item"
-              />
-            </el-select>
-          </el-form-item>
-          <el-form-item label="地址" prop="url">
-            <div style="width: 100%" v-if="rtc" title="点击拷贝">
-              <el-tag size="medium" @click="copyUrl(rtc)">
-                <i class="el-icon-document-copy"/>
-                {{ rtc }}
-              </el-tag>
-            </div>
-            <div style="width: 100%" v-if="rtsp" title="点击拷贝">
-              <el-tag size="medium" @click="copyUrl(rtsp)">
-                <i class="el-icon-document-copy"/>
-                {{ rtsp }}
-              </el-tag>
-            </div>
-            <div style="width: 100%" v-if="rtmp" title="点击拷贝">
-              <el-tag size="medium" @click="copyUrl(rtmp)">
-                <i class="el-icon-document-copy"/>
-                {{ rtmp }}
-              </el-tag>
-            </div>
-            <div style="width: 100%" v-if="rtcs" title="点击拷贝">
-              <el-tag size="medium" @click="copyUrl(rtcs)">
-                <i class="el-icon-document-copy" />
-                {{ rtcs }}
-              </el-tag>
-            </div>
-          </el-form-item>
-          <el-form-item>
-            <div style="float: right;">
-              <el-button type="primary" @click="onSubmit">确认</el-button>
-            </div>
-          </el-form-item>
-        </el-form>
-      </div>
-    </el-dialog>
-  </div>
-</template>
-
-<script>
-
-import elDragDialog from '@/directive/el-drag-dialog'
-import crypto from "crypto";
-
-
-export default {
-  name: 'BuildPushStreamUrl',
-  directives: { elDragDialog },
-  props: {},
-  data() {
-    return {
-      showDialog: false,
-      app: null,
-      stream: null,
-      mediaServer: null,
-      mediaServerList: [],
-      pushKey: null,
-      endCallback: null
-    }
-  },
-  computed: {
-    sign(){
-      if (!this.pushKey) {
-        return ''
-      }
-      return crypto.createHash('md5').update(this.pushKey, 'utf8').digest('hex')
-    },
-    rtsp(){
-      if (!this.mediaServer || !this.stream || !this.app) {
-        return ''
-      }
-      crypto.createHash('md5').update(this.pushKey, 'utf8').digest('hex')
-      return `rtsp://${this.mediaServer.streamIp}:${this.mediaServer.rtspPort}/${this.app}/${this.stream}?sign=${this.sign}`
-    },
-    rtmp(){
-      if (!this.mediaServer || !this.stream || !this.app) {
-        return ''
-      }
-      return `rtmp://${this.mediaServer.streamIp}:${this.mediaServer.rtmpPort}/${this.app}/${this.stream}?sign=${this.sign}`
-    },
-    rtc(){
-      if (!this.mediaServer || !this.stream || !this.app) {
-        return ''
-      }
-      return `http://${this.mediaServer.streamIp}:${this.mediaServer.httpPort}/index/api/webrtc?app=${this.app}&stream=${this.stream}&sign=${this.sign}`
-    },
-    rtcs(){
-      if (!this.mediaServer || !this.stream || !this.app) {
-        return ''
-      }
-      return `https://${this.mediaServer.streamIp}:${this.mediaServer.httpSSlPort}/index/api/webrtc?app=${this.app}&stream=${this.stream}&sign=${this.sign}`
-    }
-  },
-  created() {
-    this.initData()
-  },
-  methods: {
-    openDialog: function(callback) {
-      this.endCallback = callback
-      this.showDialog = true
-    },
-    onSubmit: function() {
-
-    },
-    close: function() {
-      this.showDialog = false
-      this.app = null
-      this.stream = null
-      this.mediaServer = null
-      this.endCallback = null
-      this.mediaServerList = []
-    },
-    initData: function() {
-      this.loading = true
-      this.$store.dispatch('server/getMediaServerList').then(data => {
-        this.mediaServerList = data
-      })
-      this.$store.dispatch('user/getUserInfo').then(data => {
-        this.pushKey = data.pushKey
-      })
-    },
-    copyUrl: function(dropdownItem) {
-      console.log(dropdownItem)
-      this.$copyText(dropdownItem).then((e) => {
-        this.$message.success({
-          showClose: true,
-          message: '成功拷贝到粘贴板'
-        })
-      }, (e) => {
-
-      })
-    },
-  }
-}
-</script>

web/src/views/streamPush/index.vue

@@ -45,27 +45,24 @@
         <el-form-item>
           <el-button icon="el-icon-plus" style="margin-right: 1rem;" type="primary" @click="addStream">添加
           </el-button>
-          <el-button-group>
+          <el-button icon="el-icon-upload2" style="margin-right: 1rem;" @click="importChannel">
-            <el-button icon="el-icon-upload2" @click="importChannel">
+            通道导入
-              通道导入
+          </el-button>
-            </el-button>
+          <el-button icon="el-icon-download" style="margin-right: 1rem;">
-            <el-button icon="el-icon-download">
+            <a
-              <a
+              style="text-align: center; text-decoration: none"
-                style="text-align: center; text-decoration: none"
+              href="/static/file/推流通道导入.zip"
-                href="/static/file/推流通道导入.zip"
+              download="推流通道导入.zip"
-                download="推流通道导入.zip"
+            >下载模板</a>
-              >下载模板</a>
+          </el-button>
-            </el-button>
-          </el-button-group>
           <el-button
             icon="el-icon-delete"
-            style="margin-left: 1rem;"
+            style="margin-right: 1rem;"
             :disabled="multipleSelection.length === 0"
             type="danger"
             @click="batchDel"
           >移除
           </el-button>
-          <el-button icon="el-icon-chicken" @click="buildPushStream">生成推流地址</el-button>
         </el-form-item>
         <el-form-item style="float: right;">
           <el-button icon="el-icon-refresh-right" circle @click="refresh()" />
@@ -138,7 +135,6 @@
     <addStreamTOGB ref="addStreamTOGB" />
     <importChannel ref="importChannel" />
     <stream-push-edit v-if="streamPush" :stream-push="streamPush" :close-edit="closeEdit" style="height: calc(100vh - 90px);" />
-    <buildPushStreamUrl ref="buildPushStreamUrl" />
   </div>
 </template>
 
@@ -147,7 +143,6 @@ import devicePlayer from '../dialog/devicePlayer.vue'
 import addStreamTOGB from '../dialog/pushStreamEdit.vue'
 import importChannel from '../dialog/importChannel.vue'
 import StreamPushEdit from './edit.vue'
-import buildPushStreamUrl from './buildPushStreamUrl.vue'
 
 export default {
   name: 'PushList',
@@ -155,8 +150,7 @@ export default {
     StreamPushEdit,
     devicePlayer,
     addStreamTOGB,
-    importChannel,
+    importChannel
-    buildPushStreamUrl
   },
   data() {
     return {
@@ -293,9 +287,6 @@ export default {
     },
     refresh: function() {
       this.initData()
-    },
-    buildPushStream: function() {
-      this.$refs.buildPushStreamUrl.openDialog()
     }
   }
 }