推流接口

dk-modules/sample/src/main/java/org/dromara/sample/manage/controller/MegaphoneController.java

@@ -1,6 +1,9 @@
 package org.dromara.sample.manage.controller;
 
 import cn.hutool.core.convert.Convert;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baomidou.mybatisplus.extension.conditions.query.QueryChainWrapper;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.seata.common.util.StringUtils;
@@ -9,6 +12,7 @@ import io.swagger.v3.oas.annotations.tags.Tag;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import net.bytebuddy.implementation.bytecode.Throw;
+import org.apache.commons.collections.CollectionUtils;
 import org.apache.dubbo.config.annotation.DubboReference;
 import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpGet;
@@ -28,11 +32,16 @@ import org.dromara.common.sdk.mqtt.property.PropertySetPublish;
 import org.dromara.common.sdk.mqtt.services.ServicesPublish;
 import org.dromara.common.sdk.mqtt.services.ServicesReplyReceiver;
 import org.dromara.common.sdk.mqtt.services.TopicServicesRequest;
+import org.dromara.common.translation.annotation.Translation;
 import org.dromara.sample.common.util.Md5Utils;
+import org.dromara.sample.manage.mapper.IDeviceStreamMapper;
 import org.dromara.sample.manage.model.dto.*;
+import org.dromara.sample.manage.model.entity.DeviceStreamEntity;
+import org.dromara.sample.manage.service.IDeviceStreamService;
 import org.dromara.sample.media.mapper.IFileMapper;
 import org.dromara.system.api.RemoteConfigService;
 import org.json.JSONObject;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.web.bind.annotation.*;
 
 import java.net.URI;
@@ -69,6 +78,9 @@ public class MegaphoneController {
     @DubboReference
     private RemoteConfigService remoteConfigService;
 
+    @Resource
+    IDeviceStreamMapper deviceStreamMapper;
+
     /**
      * 喊话器-开始播放文档（对应遥控器的录音广播模式）
      * Get the topology list of all online devices in one workspace.
@@ -247,6 +259,7 @@ public class MegaphoneController {
      */
     @PostMapping("/{workspace_id}/getStreamIp")
     @Operation(summary = "获取媒体流ip。", description = "获取媒体流ip。")
+    @Transactional
     public HttpResultResponse getStreamIp(@PathVariable("workspace_id") String workspaceId,
                                           @RequestBody Map<String,Object>objectMap
     ) {
@@ -254,9 +267,42 @@ public class MegaphoneController {
         //发送mqtt
         String s = remoteConfigService.selectStreamIp();
         DisobeyDTO param = Convert.convert(DisobeyDTO.class, objectMap);
+
+        if (param.getOpen() == 1){
+            if (StringUtils.isNotEmpty(param.getModel())){
+                String[] split = param.getModel().split(",");
+                for (String type : split){
+                    List<DeviceStreamEntity> deviceStreamEntities = deviceStreamMapper.selectList(new QueryWrapper<DeviceStreamEntity>().eq("rtmp_url", param.getRtmpUrl())
+                        .eq("stream_type", type));
+                    if (CollectionUtils.isEmpty(deviceStreamEntities)){
+                        DeviceStreamEntity deviceStreamEntity = new DeviceStreamEntity();
+                        deviceStreamEntity.setStreamType(type);
+                        deviceStreamEntity.setCreateTime(new Date());
+                        deviceStreamEntity.setUpdateTime(new Date());
+                        deviceStreamEntity.setRtmpUrl(param.getRtmpUrl());
+                        deviceStreamMapper.insert(deviceStreamEntity);
+                        List<String> list = new ArrayList<>();
+                        list.add(type);
+                        DisobeyDTO disobeyDTO = new DisobeyDTO();
+                        disobeyDTO.setUrl(s);
+                        disobeyDTO.setOpen(param.getOpen());
+                        disobeyDTO.setType(list);
+                        disobeyDTO.setRtmpUrl(param.getRtmpUrl());
+                        gatewayPublish.publish(top,new CommonTopicRequest<>()
+                            .setData(Objects.requireNonNull(disobeyDTO)),1);
+                    }
+                }
+            }
+        }
+        if (param.getOpen() == 2){
             if (StringUtils.isNotEmpty(param.getModel())){
                 String[] split = param.getModel().split(",");
                 for (String type : split){
+                    List<DeviceStreamEntity> deviceStreamEntities = deviceStreamMapper.selectList(new QueryWrapper<DeviceStreamEntity>().eq("rtmp_url", param.getRtmpUrl())
+                        .eq("stream_type", type));
+                    if (CollectionUtils.isNotEmpty(deviceStreamEntities)){
+                        deviceStreamMapper.delete(new QueryWrapper<DeviceStreamEntity>().eq("rtmp_url", param.getRtmpUrl())
+                            .eq("stream_type", type));
                         List<String> list = new ArrayList<>();
                         list.add(type);
                         DisobeyDTO disobeyDTO = new DisobeyDTO();
@@ -268,6 +314,10 @@ public class MegaphoneController {
                             .setData(Objects.requireNonNull(disobeyDTO)),1);
                     }
                 }
+            }else {
+                deviceStreamMapper.delete(new QueryWrapper<DeviceStreamEntity>().eq("rtmp_url", param.getRtmpUrl()));
+            }
+        }
         return HttpResultResponse.success();
     }
 
@@ -296,6 +346,15 @@ public class MegaphoneController {
         return HttpResultResponse.success(megaphoneDTO);
     }
 
+    @GetMapping("/getStreamType")
+    @Operation(summary = "获取开启的算法类型。", description = "获取开启的算法类型。")
+    public HttpResultResponse<List<DeviceStreamEntity>> getStreamType(@RequestParam String rtmpUrl
+    ) {
+        List<DeviceStreamEntity> deviceStreamEntities = deviceStreamMapper.selectList(new LambdaQueryWrapper<DeviceStreamEntity>().eq(DeviceStreamEntity::getRtmpUrl, rtmpUrl));
+        return HttpResultResponse.success(deviceStreamEntities);
+    }
+
+
 
     /**
      * 获取媒体流ip

dk-modules/sample/src/main/java/org/dromara/sample/manage/mapper/IDeviceStreamMapper.java

@@ -0,0 +1,15 @@
+package org.dromara.sample.manage.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import org.dromara.sample.manage.model.entity.DeviceStreamEntity;
+import org.dromara.sample.manage.model.entity.PlayTextEntity;
+
+/**
+ *
+ * @author sean.zhou
+ * @date 2021/11/10
+ * @version 0.1
+ */
+public interface IDeviceStreamMapper extends BaseMapper<DeviceStreamEntity> {
+
+}

dk-modules/sample/src/main/java/org/dromara/sample/manage/model/entity/DeviceStreamEntity.java

@@ -0,0 +1,43 @@
+package org.dromara.sample.manage.model.entity;
+
+import com.baomidou.mybatisplus.annotation.*;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.io.Serializable;
+import java.util.Date;
+
+/**
+ * The entity class of the device
+ *
+ * @author sean.zhou
+ * @version 0.1
+ * @date 2021/11/10
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@TableName(value = "manage_device_stream")
+public class DeviceStreamEntity implements Serializable {
+
+    @TableId(type = IdType.AUTO)
+    private Integer id;
+
+    @TableField(value = "rtmp_url")
+    private String rtmpUrl;
+
+    @TableField(value = "stream_type")
+    private String streamType;
+
+    @TableField(fill = FieldFill.INSERT)
+    private Date createTime;
+
+    @TableField(fill = FieldFill.INSERT_UPDATE)
+    private Date updateTime;
+
+
+
+}

dk-modules/sample/src/main/java/org/dromara/sample/manage/service/IDeviceStreamService.java

@@ -0,0 +1,17 @@
+package org.dromara.sample.manage.service;
+
+import org.dromara.common.sdk.common.HttpResultResponse;
+import org.dromara.common.sdk.common.PaginationData;
+import org.dromara.sample.manage.model.entity.PlayTextEntity;
+
+import java.util.Map;
+
+/**
+ * @author sean.zhou
+ * @date 2021/11/10
+ * @version 0.1
+ */
+public interface IDeviceStreamService {
+
+
+}

dk-modules/sample/src/main/java/org/dromara/sample/manage/service/impl/DeviceStreamServiceImpl.java

@@ -0,0 +1,44 @@
+package org.dromara.sample.manage.service.impl;
+
+import cn.hutool.core.convert.Convert;
+import cn.hutool.crypto.SecureUtil;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.dubbo.config.annotation.DubboReference;
+import org.dromara.common.sdk.common.HttpResultResponse;
+import org.dromara.common.sdk.common.Pagination;
+import org.dromara.common.sdk.common.PaginationData;
+import org.dromara.common.sdk.mqtt.CommonTopicRequest;
+import org.dromara.common.sdk.mqtt.MqttGatewayPublish;
+import org.dromara.sample.feign.RemoteSystemFeign;
+import org.dromara.sample.manage.mapper.IPlayTextMapper;
+import org.dromara.sample.manage.model.dto.StreamTypeDTO;
+import org.dromara.sample.manage.model.entity.PlayTextEntity;
+import org.dromara.sample.manage.service.IDeviceStreamService;
+import org.dromara.sample.manage.service.IPlayTextService;
+import org.dromara.system.api.RemoteConfigService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.UUID;
+
+/**
+ *
+ * @author sean.zhou
+ * @version 0.1
+ * @date 2021/11/10
+ */
+@Service
+@Slf4j
+@Transactional
+public class DeviceStreamServiceImpl implements IDeviceStreamService {
+
+}