Merge branch 'dev_20230222_live' of http://git.dayaedu.com/yonge/mec into dev_20230222_live

mec-biz/src/main/java/com/ym/mec/biz/dal/dto/TencentData.java

@@ -7,6 +7,7 @@ import com.alibaba.fastjson.annotation.JSONField;
 import com.ym.mec.biz.dal.enums.ETencentGroupType;
 import com.ym.mec.biz.dal.enums.ETencentImCallbackCommand;
 import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
@@ -23,8 +24,6 @@ import java.util.List;
  * @author liujunchi
  * @date 2023-03-06
  */
-@NoArgsConstructor
-@Data
 public class TencentData {
 
     // 群成员离开之后回调对象
@@ -196,42 +195,86 @@ public class TencentData {
     @ApiModel("推断流事件回调通知")
     public static class CallbackStreamStateEvent implements Serializable {
 
+
+        @ApiModelProperty("推流域名")
         @JSONField(name = "app")
         private String app;
+
+        @ApiModelProperty("用户 APPID")
         @JSONField(name = "appid")
         private Integer appid;
+
+        @ApiModelProperty("推流路径")
         @JSONField(name = "appname")
         private String appname;
+
+        @ApiModelProperty("同直播流名称")
         @JSONField(name = "channel_id")
         private String channelId;
+
+        @ApiModelProperty("断流事件通知推流时长，单位毫秒")
+        @JSONField(name = "push_duration")
+        private String pushDuration;
+
+        @ApiModelProperty("推断流错误码")
         @JSONField(name = "errcode")
         private Integer errcode;
+
+        @ApiModelProperty("推断流错误描述")
         @JSONField(name = "errmsg")
         private String errmsg;
+
+        @ApiModelProperty("事件消息产生的 UNIX 时间戳")
         @JSONField(name = "event_time")
         private Integer eventTime;
+
+        @ApiModelProperty("事件类型: 推流：1；断流：0")
         @JSONField(name = "event_type")
         private Integer eventType;
+
+        @ApiModelProperty("判断是否为国内外推流。1-6为国内，7-200为国外")
         @JSONField(name = "set_id")
         private Integer setId;
+
+        @ApiModelProperty("直播接入点的 IP")
         @JSONField(name = "node")
         private String node;
+
+        @ApiModelProperty("消息序列号，标识一次推流活动，一次推流活动会产生相同序列号的推流和断流消息")
         @JSONField(name = "sequence")
         private String sequence;
+
+        @ApiModelProperty("直播流名称")
         @JSONField(name = "stream_id")
         private String streamId;
+
+        @ApiModelProperty("用户推流 URL 所带参数")
         @JSONField(name = "stream_param")
         private String streamParam;
+
+        @ApiModelProperty("用户推流 IP")
         @JSONField(name = "user_ip")
         private String userIp;
+
+        @ApiModelProperty("视频宽度，最开始推流回调的时候若视频头部信息缺失，可能为0")
         @JSONField(name = "width")
         private Integer width;
+
+        @ApiModelProperty("视频高度，最开始推流回调的时候若视频头部信息缺失，可能为0")
         @JSONField(name = "height")
         private Integer height;
+
+        @ApiModelProperty("事件通知安全签名 sign = MD5（key + t）")
         @JSONField(name = "sign")
         private String sign;
+
+        @ApiModelProperty("过期时间，事件通知签名过期 UNIX 时间戳")
         @JSONField(name = "t")
         private Integer t;
+
+        public static CallbackStreamStateEvent from(String jsonString) {
+            return JSON.parseObject(jsonString, CallbackStreamStateEvent.class);
+        }
     }
 
     @Data
@@ -241,53 +284,102 @@ public class TencentData {
     @ApiModel("流录制事件回调通知")
     public static class CallbackSteamRecordEvent implements Serializable {
 
+        @ApiModelProperty("事件类型：直播录制： 100")
         @JSONField(name = "event_type")
         private Integer eventType;
+
+        @ApiModelProperty("用户 APPID")
         @JSONField(name = "appid")
         private Integer appid;
+
+        @ApiModelProperty("推流域名")
         @JSONField(name = "app")
         private String app;
+
+        @ApiModelProperty("json对象字符串，用户自定义数据")
         @JSONField(name = "callback_ext")
         private String callbackExt;
+
+        @ApiModelProperty("推流路径")
         @JSONField(name = "appname")
         private String appname;
+
+        @ApiModelProperty("直播流名称")
         @JSONField(name = "stream_id")
         private String streamId;
+
+        @ApiModelProperty("同直播流名称")
         @JSONField(name = "channel_id")
         private String channelId;
+
+        @ApiModelProperty("点播 file ID，在 云点播平台 可以唯一定位一个点播视频文件")
         @JSONField(name = "file_id")
         private String fileId;
+
+        @ApiModelProperty("点播文件ID")
         @JSONField(name = "record_file_id")
         private String recordFileId;
+
+        @ApiModelProperty("文件格式： FLV，HLS，MP4，AAC")
         @JSONField(name = "file_format")
         private String fileFormat;
+
+        @ApiModelProperty("录制任务 ID")
         @JSONField(name = "task_id")
         private String taskId;
+
+        @ApiModelProperty("录制任务开始写文件的时间；不能以该值作为录制内容的开始时间，录制内容的开始时间 = end_time – duration")
         @JSONField(name = "start_time")
         private Integer startTime;
+
+        @ApiModelProperty("录制任务结束写文件的时间")
         @JSONField(name = "end_time")
         private Integer endTime;
+
+        @ApiModelProperty("录制任务开始写文件的时间，微秒部分")
         @JSONField(name = "start_time_usec")
         private Integer startTimeUsec;
+
+        @ApiModelProperty("录制任务结束写文件的时间，微秒部分")
         @JSONField(name = "end_time_usec")
         private Integer endTimeUsec;
+
+        @ApiModelProperty("录制文件时长，单位秒")
         @JSONField(name = "duration")
         private Integer duration;
+
+        @ApiModelProperty("录制文件大小，单位字节")
         @JSONField(name = "file_size")
         private Integer fileSize;
+
+        @ApiModelProperty("用户推流 URL 所带参数（自定义）")
         @JSONField(name = "stream_param")
         private String streamParam;
+
+        @ApiModelProperty("录制文件下载 URL")
         @JSONField(name = "video_url")
         private String videoUrl;
+
+        @ApiModelProperty("录制开始拉流收到的首帧 pts ")
         @JSONField(name = "media_start_time")
         private Integer mediaStartTime;
+
+        @ApiModelProperty("录制从转码拉流录制对应的码率（单位 kbps）")
         @JSONField(name = "record_bps")
         private Integer recordBps;
+
+        @ApiModelProperty("事件通知安全签名 sign = MD5（key + t）")
         @JSONField(name = "sign")
         private String sign;
+
+        @ApiModelProperty("过期时间，事件通知签名过期 UNIX 时间戳")
         @JSONField(name = "t")
         private Integer t;
 
+
+        public static CallbackSteamRecordEvent from(String json) {
+            return JSON.parseObject(json, CallbackSteamRecordEvent.class);
+        }
     }
 
     @Data
@@ -296,32 +388,63 @@ public class TencentData {
     @ApiModel("流异常事件回调通知")
     public static class CallbackStreamExceptionEvent implements Serializable {
 
-
+        @ApiModelProperty("用户 APPID")
         @JSONField(name = "appid")
         private Integer appid;
+
+        @ApiModelProperty("推流事件回调时间（单位ms）")
         @JSONField(name = "data_time")
         private Long dataTime;
+
+        @ApiModelProperty("推流域名")
         @JSONField(name = "domain")
         private String domain;
+
+        @ApiModelProperty("事件类型：推流异常：321")
         @JSONField(name = "event_type")
         private Integer eventType;
+
         @JSONField(name = "interface")
         private String interfaceX;
         @JSONField(name = "path")
         private String path;
+
+        @ApiModelProperty("有异常事件时，上报间隔（单位ms）")
         @JSONField(name = "report_interval")
         private Integer reportInterval;
+
         @JSONField(name = "sequence")
         private String sequence;
+
+        @ApiModelProperty("直播流名称")
         @JSONField(name = "stream_id")
         private String streamId;
+
+        @ApiModelProperty("用户推流 URL 所带参数")
         @JSONField(name = "stream_param")
         private String streamParam;
+
         @JSONField(name = "timeout")
         private Integer timeout;
+
+        @ApiModelProperty("详细异常事件事件组")
         @JSONField(name = "abnormal_event")
         private String abnormalEvent;
+
+        public static CallbackStreamExceptionEvent from(String json) {
+            return JSON.parseObject(json, CallbackStreamExceptionEvent.class);
+        }
     }
 
+    @Data
+    @Builder
+    @NoArgsConstructor
+    @AllArgsConstructor
+    @ApiModel("流事件回调结果")
+    public static class StreamEventCallbackResult implements Serializable {
+
+        @ApiModelProperty("事件响应状态")
+        private Integer code;
+    }
 
 }

mec-biz/src/main/java/com/ym/mec/biz/service/impl/ImLiveBroadcastRoomServiceImpl.java

@@ -2,6 +2,7 @@ package com.ym.mec.biz.service.impl;
 
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.baomidou.mybatisplus.core.metadata.IPage;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
@@ -32,6 +33,7 @@ import com.ym.mec.biz.dal.entity.ImLiveBroadcastRoom;
 import com.ym.mec.biz.dal.entity.ImLiveBroadcastRoomData;
 import com.ym.mec.biz.dal.entity.ImLiveBroadcastRoomMember;
 import com.ym.mec.biz.dal.entity.ImLiveRoomBlack;
+import com.ym.mec.biz.dal.entity.ImLiveRoomVideo;
 import com.ym.mec.biz.dal.enums.MessageTypeEnum;
 import com.ym.mec.biz.dal.page.LiveRoomGoodsOrderQueryInfo;
 import com.ym.mec.biz.dal.vo.*;
@@ -1312,7 +1314,7 @@ public class ImLiveBroadcastRoomServiceImpl extends ServiceImpl<ImLiveBroadcastR
                 RTCRoom.RecordResp recordResp = pluginService.rtcRoomRecordStart(RTCRequest.RecordStart.builder()
                         .startTime(dateTime.plusMillis(5).getMillis() / 1000)
                         .endTime(dateTime.plusDays(1).getMillis() / 1000)
-                        .steamName(MessageFormat.format("{0}_{1}", imLiveBroadcastRoomVo.getRoomUid(), String.valueOf(imLiveBroadcastRoomVo.getSpeakerId())))
+                        .streamName(MessageFormat.format("{0}_{1}", imLiveBroadcastRoomVo.getRoomUid(), String.valueOf(imLiveBroadcastRoomVo.getSpeakerId())))
                         .sessionId(rtcRoom.getSessionId())
                         .config(RTCRequest.RecordConfig.builder()
                                 .videoResolution(videoResolution)

mec-im/src/main/java/com/ym/config/ResourceServerConfig.java

@@ -16,6 +16,7 @@ public class ResourceServerConfig extends ResourceServerConfigurerAdapter {
                         "/group/dismiss", "/room/statusImMsg", "/history/get", "/user/statusImUser", "/liveRoom/recordSync",
                         "/liveRoom/publishRoomMsg", "/liveRoom/destroy", "/liveRoom/create", "/liveRoom/startRecord",
                         "/liveRoom/stopRecord", "/liveRoom/userExistInRoom","/liveRoom/checkOnline",
+                        "/user/tencentStreamEventCallback", "/user/tencentStreamRecordCallback", "/user/tencentStreamExceptionCallback",
                         "/liveRoom/addUserUnableSpeak","/liveRoom/removeUserUnableSpeak","/liveRoom/syncChatRoomStatus","/liveRoom/tencentImCallback")
                 .permitAll().anyRequest().authenticated().and().csrf().disable();
     }

mec-im/src/main/java/com/ym/controller/UserController.java

@@ -8,6 +8,7 @@ import com.ym.mec.biz.dal.dto.TencentImCallbackResult;
 import com.ym.mec.biz.dal.enums.ETencentImCallbackCommand;
 import com.ym.mec.biz.service.ImLiveBroadcastRoomService;
 import com.ym.mec.common.entity.ImUserState;
+import com.ym.service.LiveRoomService;
 import com.ym.service.UserService;
 import io.rong.models.user.UserModel;
 import io.swagger.annotations.ApiOperation;
@@ -27,6 +28,8 @@ public class UserController {
     private UserService userService;
     @Autowired
     private ImLiveBroadcastRoomService imLiveBroadcastRoomService;
+    @Autowired
+    private LiveRoomService liveRoomService;
 
     @RequestMapping(value = "/register", method = RequestMethod.POST)
     public Object register(@RequestBody UserModel userModel) throws Exception {
@@ -93,4 +96,51 @@ public class UserController {
 
         return new TencentImCallbackResult();
     }
+
+
+    @ApiOperation("腾讯云直播-推流 回调接口")
+    @PostMapping(value = "/tencentStreamEventCallback")
+    public TencentData.StreamEventCallbackResult tencentStreamEventCallback(@RequestBody String body) {
+
+        log.info("tencentStreamEventCallback body：{}", body);
+
+        TencentData.CallbackStreamStateEvent event = TencentData.CallbackStreamStateEvent.from(body);
+
+        // 断流事件通知
+        if (event.getEventType() == 0) {
+            // 自动关闭录制
+            liveRoomService.stopTencentLiveVideoRecord(event.getStreamId());
+        }
+
+        // 推流事件通知
+        if (event.getEventType() == 1) {
+            // 自动开启录制
+            liveRoomService.strartTencentLiveVideoRecord(event.getStreamId());
+        }
+
+        return TencentData.StreamEventCallbackResult.builder().code(0).build();
+    }
+
+    @ApiOperation("腾讯云直播-录制 回调接口")
+    @PostMapping(value = "/tencentStreamRecordCallback")
+    public TencentData.StreamEventCallbackResult tencentStreamRecordCallback(@RequestBody String body) {
+
+        log.info("tencentStreamRecordCallback body：{}", body);
+
+        TencentData.CallbackSteamRecordEvent event = TencentData.CallbackSteamRecordEvent.from(body);
+
+        log.info("taskId={}, url={}", event.getTaskId(), event.getVideoUrl());
+
+        return TencentData.StreamEventCallbackResult.builder().code(0).build();
+    }
+
+    @ApiOperation("腾讯云直播-推流异常 回调接口")
+    @PostMapping(value = "/tencentStreamExceptionCallback")
+    public TencentData.StreamEventCallbackResult tencentStreamExceptionCallback(@RequestBody String body) {
+
+        log.info("tencentStreamExceptionCallback body：{}", body);
+
+        return TencentData.StreamEventCallbackResult.builder().code(0).build();
+    }
+
 }

mec-im/src/main/java/com/ym/service/Impl/LiveRoomServiceImpl.java

@@ -1,7 +1,12 @@
 package com.ym.service.Impl;
 
+import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.microsvc.toolkit.middleware.live.LivePluginContext;
+import com.microsvc.toolkit.middleware.live.impl.TencentCloudLivePlugin;
+import com.microsvc.toolkit.middleware.live.message.RTCRequest;
+import com.microsvc.toolkit.middleware.live.message.RTCRoom;
 import com.ym.http.HttpHelper;
 import com.ym.mec.biz.dal.entity.ImLiveRoomVideo;
 import com.ym.mec.biz.service.ImLiveRoomVideoService;
@@ -9,7 +14,6 @@ import com.ym.mec.common.entity.ImRoomMessage;
 import com.ym.mec.common.exception.BizException;
 import com.ym.mec.im.IMHelper;
 import com.ym.mec.thirdparty.storage.StoragePluginContext;
-import com.ym.mec.thirdparty.storage.provider.KS3StoragePlugin;
 import com.ym.pojo.IMApiResultInfo;
 import com.ym.pojo.IMUserOnlineInfo;
 import com.ym.pojo.RecordConfig;
@@ -17,6 +21,7 @@ import com.ym.pojo.RecordNotify;
 import com.ym.service.LiveRoomService;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang.StringUtils;
+import org.joda.time.DateTime;
 import org.redisson.api.RBucket;
 import org.redisson.api.RedissonClient;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -44,6 +49,8 @@ public class LiveRoomServiceImpl implements LiveRoomService {
     private ImLiveRoomVideoService imLiveRoomVideoService;
     @Autowired
     private StoragePluginContext storagePluginContext;
+    @Autowired
+    private LivePluginContext livePluginContext;
 
     /**
      * 创建房间-聊天室
@@ -319,4 +326,81 @@ public class LiveRoomServiceImpl implements LiveRoomService {
         return sessionId;
     }
 
+
+    /**
+     * 创建tencent云直播录制记录
+     *
+     * @param streamId 推流Id
+     */
+    @Override
+    public void strartTencentLiveVideoRecord(String streamId) {
+
+        // 直播间ROOM_UID
+        String roomId = streamId.split("_")[0];
+
+        // 创建直播录制
+        RTCRequest.RecordStart recordStart = RTCRequest.RecordStart.builder()
+                .streamName(streamId)
+                .extra("")
+                .endTime(DateTime.now().plusHours(23).getMillis())
+                .build();
+        // 创建录制任务失败，重试3次后，发送IM消息通知主播老师
+        int maxRetry = 0;
+        // 录制任务ID
+        String taskId = "";
+        do {
+            try {
+
+                // 创建录制任务
+                RTCRoom.RecordResp resp = livePluginContext.getPluginService(TencentCloudLivePlugin.PLUGIN_NAME)
+                        .rtcRoomRecordStart(recordStart);
+
+                taskId = resp.getRecordId();
+                if (StringUtils.isNotBlank(taskId)) {
+                    maxRetry = 3;
+                }
+                log.info("createTencentLiveRoomVideoRecord resp={}", JSON.toJSONString(resp));
+            } catch (Exception e) {
+                log.error("创建直播录制失败", e);
+            }
+        } while (maxRetry++ < 3);
+
+        // 生成录制记录
+        ImLiveRoomVideo video = imLiveRoomVideoService.getOne(new QueryWrapper<ImLiveRoomVideo>()
+                .eq("room_uid_", roomId)
+                .eq("record_id_", taskId)
+                .eq("type", 0));
+        if (Objects.nonNull(video)) {
+            return;
+        }
+        imLiveRoomVideoService.save(initImLiveRoomVideo(roomId, taskId, new Date()));
+    }
+
+    /**
+     * 关闭tencent云直播录制记录
+     *
+     * @param streamId 推流Id
+     */
+    @Override
+    public void stopTencentLiveVideoRecord(String streamId) {
+
+        // 关闭录制任务失败，重试3次后，发送IM消息通知主播老师
+        int maxRetry = 0;
+        do {
+            try {
+
+                // 关闭直播录制
+                RTCRoom.RecordResp ret = livePluginContext.getPluginService(TencentCloudLivePlugin.PLUGIN_NAME)
+                        .rtcRoomRecordStop(streamId);
+
+                if (StringUtils.isNotBlank(ret.getRequestId())) {
+                    // 重试最大次数
+                    maxRetry = 3;
+                }
+            } catch (Exception e) {
+                log.error("关闭直播录制失败", e);
+            }
+        } while (maxRetry++ < 3);
+
+    }
 }

mec-im/src/main/java/com/ym/service/LiveRoomService.java

@@ -62,4 +62,16 @@ public interface LiveRoomService {
      * @param userId  用户id
      */
     boolean removeUserUnableSpeak(String roomUid, String userId);
+
+    /**
+     * 开始tencent云直播录制记录
+     * @param streamId 推流Id
+     */
+    void strartTencentLiveVideoRecord(String streamId);
+
+    /**
+     * 关闭tencent云直播录制记录
+     * @param streamId 推流Id
+     */
+    void stopTencentLiveVideoRecord(String streamId);
 }