Merge branch 'master_dev_gbt_0319' into 'master_dev'

fix: 视频点播问题修复

See merge request yunteng/thingskit!371

+--修改流媒体表名称--
+alter table tk_media_server rename to tk_video_zlmediakit;
+--先删除status字段再加新类型--
+alter table tk_video_zlmediakit drop status;
+alter table tk_video_zlmediakit add status BOOLEAN;
+COMMENT ON COLUMN "public"."tk_video_zlmediakit"."status" IS '流媒体服务状态：false离线 true在线';
\ No newline at end of file

@@ -42,6 +42,7 @@ import org.thingsboard.server.common.data.page.PageLink;
 import org.thingsboard.server.common.data.yunteng.constant.FastIotConstants;
 import org.thingsboard.server.common.data.yunteng.dto.DeviceDTO;
 import org.thingsboard.server.common.data.yunteng.dto.TkDeviceStateLogDTO;
+import org.thingsboard.server.common.data.yunteng.enums.StatusEnum;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsgDataType;
 import org.thingsboard.server.common.msg.TbMsgMetaData;
@@ -54,6 +55,7 @@ import org.thingsboard.server.dao.tenant.TenantService;
 import org.thingsboard.server.dao.timeseries.TimeseriesService;
 import org.thingsboard.server.dao.yunteng.service.TkDeviceService;
 import org.thingsboard.server.dao.yunteng.service.TkDeviceStateLogService;
+import org.thingsboard.server.dao.yunteng.service.media.TkVideoChannelService;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.service.partition.AbstractPartitionBasedService;
@@ -94,6 +96,7 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
     private final TbClusterService clusterService;
     private final PartitionService partitionService;
     private final TkDeviceService tkDeviceService;
+    private final TkVideoChannelService channelService;
     private final TkDeviceStateLogService tkDeviceStateLogService;
 
     private TelemetrySubscriptionService tsSubService;
@@ -121,7 +124,7 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
     public DefaultDeviceStateService(TenantService tenantService, DeviceService deviceService,
                                      AttributesService attributesService, TimeseriesService tsService,
                                      TbClusterService clusterService, PartitionService partitionService,
-                                     TkDeviceService tkDeviceService,TkDeviceStateLogService tkDeviceStateLogService) {
+                                     TkDeviceService tkDeviceService, TkVideoChannelService channelService, TkDeviceStateLogService tkDeviceStateLogService) {
         this.tenantService = tenantService;
         this.deviceService = deviceService;
         this.attributesService = attributesService;
@@ -129,6 +132,7 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
         this.clusterService = clusterService;
         this.partitionService = partitionService;
         this.tkDeviceService = tkDeviceService;
+        this.channelService = channelService;
         this.tkDeviceStateLogService = tkDeviceStateLogService;
     }
 
@@ -420,6 +424,7 @@ public class DefaultDeviceStateService extends AbstractPartitionBasedService<Dev
                     save(deviceId, INACTIVITY_ALARM_TIME, ts);
                     pushRuleEngineMessage(stateData, INACTIVITY_EVENT);
                     //thingskit update tkDevice state
+                    channelService.updateVideoChannelState(deviceId.getId().toString(), StatusEnum.OFFLINE);
                     if(stateData.getState().getLastActivityTime()>0){
                         String tenantId = stateData.getTenantId().getId().toString();
                         String tbDeviceId = deviceId.getId().toString();

@@ -24,10 +24,8 @@ import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.protobuf.ByteString;
-import java.util.List;
-import java.util.Optional;
-import java.util.Set;
-import java.util.UUID;
+
+import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.locks.Lock;
@@ -886,10 +884,18 @@ public class DefaultTransportApiService implements TransportApiService {
             dataDecodingEncodingService.decode(requestMsg.getContext().toByteArray());
         allChannel.ifPresent(
             d -> {
-              channelService.clearDeviceChannel(d.get(0).getCameraCode());
+              String cameraCode = d.get(0).getCameraCode();
+              Map<String, String> playingChannels = channelService.findPlayingChannel(cameraCode);
+              channelService.clearDeviceChannel(cameraCode);
               List<TkVideoChannelEntity> chanel =
                   d.stream()
-                      .map(item -> item.getEntity(TkVideoChannelEntity.class))
+                      .map(item -> {
+                        String channelId = item.getChannelId();
+                        if(playingChannels.containsKey(channelId)){
+                          item.setStreamId(playingChannels.get(channelId));
+                        }
+                        return item.getEntity(TkVideoChannelEntity.class);
+                      })
                       .collect(Collectors.toList());
               channelService.insertBatch(chanel, chanel.size());
             });

@@ -82,7 +82,7 @@ public class TkVideoControlServiceImpl implements TkVideoControlService {
     // 找到可以使用的流媒体
     MediaServerDTO mediaServer =
         tkMediaServerService.getMediaServerForPlay(tenantId, sipDeviceDTO.getMediaServerId());
-    if (null == mediaServer) {
+    if (null == mediaServer || !mediaServer.isStatus()) {
       throw new TkDataValidationException(
           ErrorMessage.NOT_FOUND_MEDIA_SERVER_FOR_PLAY.getMessage());
     }

@@ -28,6 +28,7 @@ import org.thingsboard.server.dao.tenant.TenantService;
 import org.thingsboard.server.dao.timeseries.TimeseriesService;
 import org.thingsboard.server.dao.yunteng.service.TkDeviceService;
 import org.thingsboard.server.dao.yunteng.service.TkDeviceStateLogService;
+import org.thingsboard.server.dao.yunteng.service.media.TkVideoChannelService;
 import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.cluster.TbClusterService;
 
@@ -59,14 +60,15 @@ public class DefaultDeviceStateServiceTest {
     TkDeviceService tkDeviceService;
     @Mock
     TkDeviceStateLogService tkDeviceStateLogService;
-
+    @Mock
+    TkVideoChannelService channelService;
     DeviceId deviceId = DeviceId.fromString("00797a3b-7aeb-4b5b-b57a-c2a810d0f112");
 
     DefaultDeviceStateService service;
 
     @Before
     public void setUp() {
-        service = spy(new DefaultDeviceStateService(tenantService, deviceService, attributesService, tsService, clusterService, partitionService,tkDeviceService,tkDeviceStateLogService));
+        service = spy(new DefaultDeviceStateService(tenantService, deviceService, attributesService, tsService, clusterService, partitionService,tkDeviceService, channelService, tkDeviceStateLogService));
     }
 
     @Test
@@ -86,4 +88,4 @@ public class DefaultDeviceStateServiceTest {
         Mockito.verify(service, times(1)).fetchDeviceStateData(deviceId);
     }
 
-}
\ No newline at end of file
+}

@@ -138,7 +138,7 @@ public final class ModelConstants {
     public static final String TK_DEVICE_ACCESS_INFORMATION = "tk_device_access_information";
 
     /** ZLMediaKit 流媒体表 */
-    public static final String TK_MEDIA_SERVER_NAME = "tk_media_server";
+    public static final String TK_MEDIA_SERVER_NAME = "tk_video_zlmediakit";
 
     /** 视频通道表 */
     public static final String TK_VIDEO_CHANNEL_NAME = "tk_video_channel";

@@ -136,6 +136,7 @@ public enum ErrorMessage {
 
   DEVICE_NOT_ONLINE(400111,"设备不在线，无法执行相关操作！"),
   GBT_VIDEO_REPETITION(400112,"该设备通道号视频已存在！"),
+  SSRC_NO_ENABLED(400113,"未获取到可用的SSRC资源"),
   HAVE_NO_PERMISSION(500002,"没有修改权限"),
   NOT_ALLOED_ISOLATED_IN_MONOLITH(500003,"【monolith】模式下，不能选择【isolated】类型的租户配置");
 

@@ -23,7 +23,7 @@ public class TkMediaServerEntity extends TenantBaseEntity {
   private Integer rtspSslPort;
   private String secret;
   private boolean rtpEnable;
-  private Integer status;
+  private boolean status;
   private String rtpPortRange;
   private String sendRtpPortRange;
   private int recordAssistPort;

@@ -8,6 +8,7 @@ import org.thingsboard.server.common.data.yunteng.config.media.UserSetting;
 import org.thingsboard.server.common.data.yunteng.constant.FastIotConstants;
 import org.thingsboard.server.common.data.yunteng.core.cache.CacheUtils;
 import org.thingsboard.server.common.data.yunteng.core.exception.TkDataValidationException;
+import org.thingsboard.server.common.data.yunteng.core.message.ErrorMessage;
 
 /** ssrc使用 */
 @Component
@@ -88,7 +89,7 @@ public class SSRCFactory {
     String redisKey = SSRC_INFO_KEY + userSetting.getServerId() + "_" + mediaServerId;
     Optional<Set<String>> all = cacheUtils.get(cacheName, redisKey);
     if (all.isEmpty()) {
-      throw new TkDataValidationException("ssrc已经用完");
+      throw new TkDataValidationException(ErrorMessage.SSRC_NO_ENABLED.getMessage());
     } else {
       // 在集合中移除并返回第一个成员。
       Set<String> list = all.get();

@@ -3,9 +3,12 @@ package org.thingsboard.server.dao.yunteng.impl.media;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.metadata.IPage;
 import java.time.LocalDateTime;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.stream.Collectors;
+
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
@@ -74,6 +77,23 @@ public class TkVideoChannelServiceImpl
   }
 
   @Override
+  public Map<String, String> findPlayingChannel(String cameraCode) {
+    List<TkVideoChannelEntity> channelList =
+            baseMapper.selectList(
+                    new LambdaQueryWrapper<TkVideoChannelEntity>()
+                            .isNotNull(TkVideoChannelEntity::getStreamId)
+                            .eq(TkVideoChannelEntity::getCameraCode, cameraCode));
+    Map<String, String> result = new HashMap<>();
+    if(channelList == null){
+      return result;
+    }
+    channelList.forEach(item ->{
+      result.put(item.getChannelId(),item.getStreamId());
+    });
+    return result;
+  }
+
+  @Override
   public VideoChanelDTO updateChannelInfo(VideoChanelDTO videoChanelDTO) {
     if (StringUtils.isEmpty(videoChanelDTO.getId())) {
       throw new TkDataValidationException(ErrorMessage.INVALID_PARAMETER.getMessage());
@@ -103,6 +123,14 @@ public class TkVideoChannelServiceImpl
         .orElse(false);
   }
 
+  @Override
+  public void updateVideoChannelState(String tbDeviceId, StatusEnum online) {
+    TkVideoChannelEntity entity = new TkVideoChannelEntity();
+    entity.setStatus(online);
+    entity.setStreamId(null);
+    baseMapper.update(entity,new LambdaQueryWrapper<TkVideoChannelEntity>()
+            .eq(TkVideoChannelEntity::getDeviceId, tbDeviceId));
+  }
 
   @Override
   public List<VideoChanelDTO> getListByDeviceId(String deviceId,String tenantId) {

@@ -3,6 +3,7 @@ package org.thingsboard.server.dao.yunteng.service.media;
 import java.util.List;
 import java.util.Map;
 import org.thingsboard.server.common.data.yunteng.dto.sip.VideoChanelDTO;
+import org.thingsboard.server.common.data.yunteng.enums.StatusEnum;
 import org.thingsboard.server.common.data.yunteng.utils.tools.TkPageData;
 import org.thingsboard.server.dao.yunteng.entities.TkVideoChannelEntity;
 import org.thingsboard.server.dao.yunteng.service.BaseService;
@@ -27,7 +28,12 @@ public interface TkVideoChannelService extends BaseService<TkVideoChannelEntity>
    * @param cameraCode
    * @return
    */
-  int clearDeviceChannel(String cameraCode);
+  int clearDeviceChannel(String cameraCode);  /**
+   * 查询点播中的摄像头通道
+   * @param cameraCode
+   * @return
+   */
+  Map<String,String> findPlayingChannel(String cameraCode);
 
   /**
    * 更新视频通道信息。
@@ -46,7 +52,14 @@ public interface TkVideoChannelService extends BaseService<TkVideoChannelEntity>
    * @return true 更新成功 false更新失败
    */
   boolean updateVideoChannelStreamId(
-      String streamId, String deviceCode, String channelId);
+          String streamId, String deviceCode, String channelId);
+  /**
+   * 更新视频通道状态
+   *
+   * @param tbDeviceId 设备ID
+   * @return true 更新成功 false更新失败
+   */
+  void updateVideoChannelState(String tbDeviceId, StatusEnum online);
 
   /**
    * 根据设备id获取通道号集合