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

package com.ym.service.Impl;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.google.common.collect.Lists;
import com.ym.http.HttpHelper;
import com.ym.mec.biz.dal.dto.TencentData;
import com.ym.mec.biz.dal.entity.ImLiveBroadcastRoom;
import com.ym.mec.biz.dal.entity.ImLiveRoomVideo;
import com.ym.mec.biz.service.ImLiveBroadcastRoomService;
import com.ym.mec.biz.service.ImLiveRoomVideoService;
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.pojo.IMApiResultInfo;
import com.ym.pojo.IMUserOnlineInfo;
import com.ym.pojo.RecordConfig;
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;
import org.springframework.stereotype.Service;

import java.net.HttpURLConnection;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * @author hgw
 * Created by 2022-02-21
 */
@Slf4j
@Service
public class LiveRoomServiceImpl implements LiveRoomService {

    @Autowired
    private IMHelper imHelper;
    @Autowired
    private HttpHelper httpHelper;
    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    private ImLiveRoomVideoService imLiveRoomVideoService;
    @Autowired
    private ImLiveBroadcastRoomService imLiveBroadcastRoomService;
    @Autowired
    private StoragePluginContext storagePluginContext;

    /**
     * 创建房间-聊天室
     *
     * @param roomId   房间Uid
     * @param roomName 房间名称
     */
    public IMApiResultInfo createLiveRoom(String roomId, String roomName) {
        IMApiResultInfo resultInfo;
        try {
            resultInfo = imHelper.createChatRoom(roomId, roomName);
        } catch (Exception e) {
            log.error("create chatRoom error >>>", e.getCause());
            throw new BizException("创建聊天室失败!");
        }
        if (!resultInfo.isSuccess()) {
            log.error("create chatRoom error: {}", resultInfo.getErrorMessage());
            throw new BizException("创建聊天室失败!");
        }
        log.info("create chatRoom success: {}", roomId);
        return resultInfo;
    }

    /**
     * 销毁房间-聊天室
     *
     * @param roomId 房间Uid
     */
    public IMApiResultInfo destroyLiveRoom(String roomId) {
        //删除服务器房间
        List<String> deleteRoomIds = new ArrayList<String>() {{
            add(roomId);
        }};
        IMApiResultInfo resultInfo;
        try {
            resultInfo = imHelper.deleteChrm(deleteRoomIds);
        } catch (Exception e) {
            throw new BizException("关闭聊天室失败!");
        }
        if (!resultInfo.isSuccess()) {
            log.error("destroy chatRoom error: {}", resultInfo.getErrorMessage());
            throw new BizException("关闭聊天室失败!");
        }
        return resultInfo;
    }

    /**
     * 发送消息
     *
     * @param message
     */
    public IMApiResultInfo publishRoomMessage(ImRoomMessage message) {
        log.info("publishRoomMessage message : {}", JSONObject.toJSONString(message));
        IMApiResultInfo resultInfo;
        try {
            resultInfo = imHelper.publishRoomMessage(message.getFromUserId(), message.getToChatroomId(), message);
        } catch (Exception e) {
            throw new BizException("消息发送失败" + e.getMessage());
        }
        if (!resultInfo.isSuccess()) {
            log.error("publishRoomMessage chatRoom error: {}", resultInfo.getErrorMessage());
            throw new BizException("消息发送失败!");
        }
        return resultInfo;
    }

    @Override
    public void startRecord(String roomId, String videoResolution) throws Exception {
        log.error("开始录制直播：roomId : {} ", roomId);
        JSONObject paramJson = new JSONObject();
        paramJson.put("sessionId", getRoomSessionId(roomId));
        RecordConfig recordConfig = new RecordConfig();
        Optional.ofNullable(videoResolution)
                .ifPresent(recordConfig::setVideoResolution);
        paramJson.put("config", recordConfig);
        String body = paramJson.toJSONString();
        HttpURLConnection conn = httpHelper.createIMRtcPostHttpConnection("/rtc/record/start.json", "application/json", roomId);
        httpHelper.setBodyParameter(body, conn);
        String returnResult = httpHelper.returnResult(conn, body);
        JSONObject resultObject = JSONObject.parseObject(returnResult);
        String code = resultObject.getString("code");
        if (!"200".equals(code)) {
            log.error("直播视频录制失败：resultInfo : {} ", returnResult);
        }
        String recordId = resultObject.getString("recordId");
        ImLiveRoomVideo video = imLiveRoomVideoService.getOne(new QueryWrapper<ImLiveRoomVideo>().eq("room_uid_", roomId).eq("record_id_", recordId).eq("type", 0));
        if (Objects.nonNull(video)) {
            return;
        }
        imLiveRoomVideoService.save(initImLiveRoomVideo(roomId, recordId, new Date()));
    }

    private ImLiveRoomVideo initImLiveRoomVideo(String roomId, String recordId, Date now) {
        ImLiveRoomVideo video = new ImLiveRoomVideo();
        video.setRoomUid(roomId);
        video.setRecordId(recordId);
        video.setStartTime(now);
        video.setType(0);
        video.setCreatedTime(now);
        return video;
    }

    @Override
    public void stopRecord(String roomId) throws Exception {
        JSONObject paramJson = new JSONObject();
        paramJson.put("sessionId", getRoomSessionId(roomId));
        String body = paramJson.toJSONString();
        HttpURLConnection conn = httpHelper.createIMRtcPostHttpConnection("/rtc/record/stop.json", "application/json", roomId);
        httpHelper.setBodyParameter(body, conn);
        redissonClient.getBucket("sessionId:" + roomId).delete();
        log.info("结束录制直播 roomId :{},{}", roomId, httpHelper.returnResult(conn, body));
    }

    @Override
    public void recordSync(RecordNotify recordNotify) {
        if (recordNotify.getCode().equals(200)) {
            if (Objects.nonNull(recordNotify.getType()) && recordNotify.getType() == 4) {
                //云端录制文件地址
                String fileUrl = storagePluginContext.getPublicUrl(recordNotify.getOutput().getFileUrl(),"live-rewind");
                String roomId = recordNotify.getRoomId();
                //写入数据库
                try {
                    Date now = new Date();
                    //获取最后一次录制视频
                    ImLiveRoomVideo video = imLiveRoomVideoService.getDao().getLastRecord(roomId, recordNotify.getRecordId());
                    if (Objects.isNull(video)) {
                        log.error("获取录制视频失败：roomId : {} ,recordId:{}", roomId, recordNotify.getRecordId());
                        return;
                    }
                    if (StringUtils.isNotEmpty(video.getUrl())) {
                        //保存切片
                        ImLiveRoomVideo imLiveRoomVideo = initImLiveRoomVideo(roomId, recordNotify.getRecordId(), now);
                        imLiveRoomVideo.setStartTime(video.getEndTime());
                        imLiveRoomVideo.setEndTime(now);
                        imLiveRoomVideo.setUrl(fileUrl);
                        imLiveRoomVideo.setType(2);
                        imLiveRoomVideoService.save(imLiveRoomVideo);
                        return;
                    } else {
                        video.setEndTime(now);
                        video.setType(2);
                        video.setUrl(fileUrl);
                        video.setCreatedTime(now);
                        imLiveRoomVideoService.updateById(video);
                    }
                } catch (Exception e) {
                    log.error("recordSync >>>>  error : {}", e.getMessage());
                }
            }
        }
    }

    /**
     * 查询用户是否在聊天室
     *
     * @param chatroomId 要查询的聊天室 ID（必传）
     * @param userId     要查询的用户 ID（必传）
     * @return true 在聊天室，false 不在聊天室
     * <p>触发融云退出聊天室机制将用户踢出
     * <p>聊天室中用户在离线 30 秒后有新消息产生时或离线后聊天室中产生 30 条消息时会被自动退出聊天室
     * <p>此状态需要聊天室中有新消息时才会进行同步
     */
    public boolean userExistInRoom(String chatroomId, String userId) {
        log.info("userExistInRoom chatroomId : {}  userId : {}", chatroomId, userId);
        IMApiResultInfo resultInfo;
        try {
            resultInfo = imHelper.isInChartRoom(chatroomId, userId);
        } catch (Exception e) {
            throw new BizException("查询失败" + e.getMessage());
        }
        if (!resultInfo.isSuccess()) {
            log.error("userExistInRoom  chatroomId : {}  userId : {}", chatroomId, userId);
            throw new BizException("查询失败!");
        }
        return resultInfo.isSuccess() && resultInfo.getIsInChrm();
    }

    /**
     * 查询用户是否在线
     * 注意：断网时，用户状态返回可能不准确
     *
     * @param userId 要查询的用户 ID（必传）
     * @return true 在线，false 不在线
     */
    public boolean checkOnline(String userId) {
        log.info("checkOnline userId : {}", userId);
        IMUserOnlineInfo resultInfo;
        try {
            resultInfo = imHelper.checkOnline(userId);
        } catch (Exception e) {
            return false;
        }
        if (!resultInfo.isSuccess()) {
            log.error("checkOnline userId : {}", userId);
            return false;
        }
        log.info("checkOnline userId : {}  resultInfo code:{} status: {}", userId, resultInfo.getCode(), resultInfo.getStatus());
        return Objects.equals("1", resultInfo.getStatus());
    }

    /**
     * 添加禁言成员-默认禁言120分钟
     *
     * @param roomUid 房间uid
     * @param userId  用户id
     */
    public boolean addUserUnableSpeak(String roomUid, String userId) {
        log.info("addUserUnableToSpeak chatroomId : {}  userId : {}", roomUid, userId);
        IMApiResultInfo resultInfo;
        try {
            resultInfo = imHelper.addUserUnableSpeak(roomUid, userId, "120");
        } catch (Exception e) {
            log.error("addUserUnableToSpeak chatroomId error: {}  userId : {}", roomUid, userId);
            return false;
        }
        if (!resultInfo.isSuccess()) {
            log.error("addUserUnableToSpeak chatroomId : {}  userId : {}", roomUid, userId);
            return false;
        }
        return true;
    }

    /**
     * 移除禁言成员
     *
     * @param roomUid 房间uid
     * @param userId  用户id
     */
    public boolean removeUserUnableSpeak(String roomUid, String userId) {
        log.info("removeUserUnableSpeak chatroomId : {}  userId : {}", roomUid, userId);
        IMApiResultInfo resultInfo;
        try {
            resultInfo = imHelper.removeUserUnableSpeak(roomUid, userId);
        } catch (Exception e) {
            log.error("removeUserUnableSpeak chatroomId error: {}  userId : {}", roomUid, userId);
            return false;
        }
        if (!resultInfo.isSuccess()) {
            log.error("removeUserUnableSpeak chatroomId : {}  userId : {}", roomUid, userId);
            return false;
        }
        return true;
    }

    public String getRoomSessionId(String roomId) {
        RBucket<String> bucket = redissonClient.getBucket("sessionId:" + roomId);
        if (bucket.isExists()) {
            return bucket.get();
        }
        HttpURLConnection conn;
        JSONObject resultObject;

        try {
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("roomId", roomId);
            conn = httpHelper.createIMRtcPostHttpConnection("/rtc/room/query.json", "application/json", null);
            httpHelper.setBodyParameter(jsonObject.toJSONString(), conn);
            String returnResult = httpHelper.returnResult(conn, jsonObject.toJSONString());
            resultObject = JSONObject.parseObject(returnResult);
        } catch (Exception e) {
            throw new BizException("获取sessionId失败", e.getCause());
        }

        String sessionId;
        if ("200".equals(resultObject.getString("code"))) {
            sessionId = resultObject.getString("sessionId");
            bucket.set(sessionId, 1, TimeUnit.HOURS);
            log.info("getRoomSessionId roomId : {}  sessionId : {}", roomId, sessionId);
        } else {
            log.error("获取sessionId失败 roomId : {} returnResult:{}", roomId, resultObject);
            throw new BizException("获取sessionId失败");
        }
        return sessionId;
    }

    /**
     * 生成直播录制信息
     *
     * @param event TencentData.CallbackSteamRecordEvent
     */
    @Override
    public void createLiveRoomVideoRecord(TencentData.CallbackSteamRecordEvent event) {

        // 直播间ROOM_UID
        String roomId = event.getStreamId().split("_")[0];

        // 录制开始时间
        long startTime =  (event.getEndTime() - event.getDuration()) * 1000L;
        //保存切片
        ImLiveRoomVideo imLiveRoomVideo = initImLiveRoomVideo(roomId, event.getTaskId(), DateTime.now().toDate());
        imLiveRoomVideo.setStartTime(new DateTime(startTime).toDate());
        imLiveRoomVideo.setEndTime(new DateTime(event.getEndTime() * 1000L).toDate());
        // 强制替换http为https
        imLiveRoomVideo.setUrl(event.getVideoUrl().replace("http://", "https://"));
        imLiveRoomVideo.setType(2);

        // 回放记录已存在，直接忽略
        ImLiveRoomVideo video = imLiveRoomVideoService.lambdaQuery()
                .eq(ImLiveRoomVideo::getRoomUid, imLiveRoomVideo.getRoomUid())
                .eq(ImLiveRoomVideo::getRecordId, imLiveRoomVideo.getRecordId())
                .last("LIMIT 1")
                .one();
        if (Objects.isNull(video)) {

            ImLiveBroadcastRoom room = imLiveBroadcastRoomService.getByRoomUid(imLiveRoomVideo.getRoomUid());
            if (Objects.nonNull(room) && "LIVE".equals(room.getGroupType())) {

                if (StringUtils.isBlank(room.getVideoRecord()) || Arrays.stream(room.getVideoRecord().split(","))
                        .anyMatch(x -> x.equals(imLiveRoomVideo.getRecordId()))) {
                    imLiveRoomVideoService.save(imLiveRoomVideo);
                }
            } else {
                imLiveRoomVideoService.save(imLiveRoomVideo);
            }

        }

    }
}