[RPC] 推流播放

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

@@ -28,4 +28,7 @@ public interface IRedisRpcService {
     void subscribeCatalog(int id, int cycle);
 
     void subscribeMobilePosition(int id, int cycle, int interval);
+
+    void play(Integer id, ErrorCallback<StreamInfo> callback);
+
 }

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

@@ -18,6 +18,7 @@ import com.genersoft.iot.vmp.service.ISendRtpServerService;
 import com.genersoft.iot.vmp.service.redisMsg.dto.RedisRpcController;
 import com.genersoft.iot.vmp.service.redisMsg.dto.RedisRpcMapping;
 import com.genersoft.iot.vmp.service.redisMsg.dto.RpcController;
+import com.genersoft.iot.vmp.streamPush.service.IStreamPushPlayService;
 import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -47,6 +48,9 @@ public class RedisRpcStreamPushController extends RpcController {
     @Autowired
     private RedisTemplate<Object, Object> redisTemplate;
 
+    @Autowired
+    private IStreamPushPlayService streamPushPlayService;
+
 
     private void sendResponse(RedisRpcResponse response){
         log.info("[redis-rpc] >> {}", response);
@@ -172,4 +176,26 @@ public class RedisRpcStreamPushController extends RpcController {
         return response;
     }
 
+    /**
+     * 停止监听流上线
+     */
+    @RedisRpcMapping("play")
+    public RedisRpcResponse play(RedisRpcRequest request) {
+        int id = Integer.parseInt(request.getParam().toString());
+        RedisRpcResponse response = request.getResponse();
+        if (id <= 0) {
+            response.setStatusCode(ErrorCode.ERROR400.getCode());
+            response.setBody("param error");
+            return response;
+        }
+        streamPushPlayService.start(id, (code, msg, data) -> {
+            if (code == ErrorCode.SUCCESS.getCode()) {
+                response.setStatusCode(ErrorCode.SUCCESS.getCode());
+                response.setBody(data);
+                sendResponse(response);
+            }
+        }, null, null);
+        return null;
+    }
+
 }

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

@@ -15,6 +15,7 @@ import com.genersoft.iot.vmp.media.event.hook.HookSubscribe;
 import com.genersoft.iot.vmp.media.event.hook.HookType;
 import com.genersoft.iot.vmp.media.service.IMediaServerService;
 import com.genersoft.iot.vmp.service.ISendRtpServerService;
+import com.genersoft.iot.vmp.service.bean.ErrorCallback;
 import com.genersoft.iot.vmp.service.redisMsg.IRedisRpcService;
 import com.genersoft.iot.vmp.vmanager.bean.ErrorCode;
 import com.genersoft.iot.vmp.vmanager.bean.WVPResult;
@@ -221,4 +222,10 @@ public class RedisRpcServiceImpl implements IRedisRpcService {
         RedisRpcRequest request = buildRequest("device/subscribeMobilePosition", jsonObject);
         redisRpcConfig.request(request, 10);
     }
+
+    @Override
+    public void play(Integer id, ErrorCallback<StreamInfo> callback) {
+        RedisRpcRequest request = buildRequest("streamPush/play", id);
+        redisRpcConfig.request(request, 10);
+    }
 }

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

@@ -245,7 +245,7 @@ public class StreamPushController {
     @GetMapping(value = "/start")
     @ResponseBody
     @Operation(summary = "开始播放", security = @SecurityRequirement(name = JwtUtils.HEADER))
-    public DeferredResult<WVPResult<StreamContent>> batchStop(Integer id){
+    public DeferredResult<WVPResult<StreamContent>> start(Integer id){
         Assert.notNull(id, "推流ID不可为NULL");
         DeferredResult<WVPResult<StreamContent>> result = new DeferredResult<>(userSetting.getPlayTimeout().longValue());
         result.onTimeout(()->{

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

@@ -53,6 +53,11 @@ public class StreamPushPlayServiceImpl implements IStreamPushPlayService {
         StreamPush streamPush = streamPushMapper.queryOne(id);
         Assert.notNull(streamPush, "推流信息未找到");
 
+        if (!userSetting.getServerId().equals(streamPush.getServerId())) {
+            redisRpcService.play(id, callback);
+            return;
+        }
+
         MediaServer mediaServer = mediaServerService.getOne(streamPush.getMediaServerId());
         Assert.notNull(mediaServer, "节点" + streamPush.getMediaServerId() + "未找到");
         MediaInfo mediaInfo = mediaServerService.getMediaInfo(mediaServer, streamPush.getApp(), streamPush.getStream());