单人rtc异常断线处理添加

42b326a2 · Future · e5fc0ced · 42b326a2 · 42b326a2 · 42b326a2
Commit 42b326a2 authored Jun 03, 2022 by Future
10 changed files
--- a/core/src/main/java/com/wecloud/im/entity/ImRtcRecord.java
+++ b/core/src/main/java/com/wecloud/im/entity/ImRtcRecord.java
@@ -58,6 +58,9 @@ public class ImRtcRecord extends BaseEntity {
    @ApiModelProperty("音视频结束时间")
    private Date endTime;
+    @ApiModelProperty("通话过程中离线次数统计 用于区分异常断线使用")
+    private Integer offlineTimes;
    @ApiModelProperty("创建时间")
    private Date createTime;

--- a/core/src/main/java/com/wecloud/im/sdk/enums/SingleRtcOperateTypeEnum.java
+++ b/core/src/main/java/com/wecloud/im/sdk/enums/SingleRtcOperateTypeEnum.java
+package com.wecloud.im.sdk.enums;
+import io.geekidea.springbootplus.framework.common.enums.BaseEnum;
+/**
+ * @Author Future
+ * @Date 2022/6/3 13:12
+ * @Description 单人音视频操作类型枚举
+ */
+public enum SingleRtcOperateTypeEnum implements BaseEnum {
+    /**
+     * 1 - 同意进入频道
+     */
+    JOIN(1, "join"),
+    /**
+     * 2 - 拒接进入频道
+     */
+    REJECT(2, "reject"),
+    /**
+     * 2 - 离开频道
+     */
+    LEAVE(3, "leave"),
+    /**
+     * 4 - 异常原因断开
+     */
+    ABNORMAL_DISCONNECT(4, "abnormalDisconnect");
+    SingleRtcOperateTypeEnum(int code, String desc) {
+        this.code = code;
+        this.desc = desc;
+    }
+    private final Integer code;
+    private final String desc;
+    @Override
+    public Integer getCode() {
+        return this.code;
+    }
+    @Override
+    public String getDesc() {
+        return this.desc;
+    }
+}
--- a/core/src/main/java/com/wecloud/im/service/ImRtcRecordService.java
+++ b/core/src/main/java/com/wecloud/im/service/ImRtcRecordService.java
@@ -24,7 +24,8 @@ public interface ImRtcRecordService extends BaseService<ImRtcRecord> {
    /**
     * 更新音视频记录
     * @param channelId
-     * @param type 1-同意进入频道  2-拒接进入频道  3-主动挂断(离开频道)
+     * @param type 1-同意进入频道  2-拒接进入频道  3-主动挂断(离开频道) 4-异常原因断开
+     * @see com.wecloud.im.sdk.enums.SingleRtcOperateTypeEnum
     */
    void updateRtcRecord(Long channelId, Integer type);

--- a/core/src/main/java/com/wecloud/im/service/impl/ImRtcRecordServiceImpl.java
+++ b/core/src/main/java/com/wecloud/im/service/impl/ImRtcRecordServiceImpl.java
@@ -10,6 +10,7 @@ import com.wecloud.im.mapper.ImRtcRecordMapper;
 import com.wecloud.im.param.rtc.CreateRtcChannelParam;
 import com.wecloud.im.param.rtc.RtcRecordParam;
 import com.wecloud.im.sdk.enums.RtcStateEnum;
+import com.wecloud.im.sdk.enums.SingleRtcOperateTypeEnum;
 import com.wecloud.im.service.ImClientService;
 import com.wecloud.im.service.ImRtcRecordService;
 import com.wecloud.utils.SnowflakeUtil;
@@ -66,17 +67,21 @@ public class ImRtcRecordServiceImpl extends BaseServiceImpl<ImRtcRecordMapper, I
            if (rtcRecord == null) {
                return;
            }
-            if (type == 1) {
+            if (SingleRtcOperateTypeEnum.JOIN.getCode().equals(type)) {
                // 同意进入频道
                rtcRecord.setState(RtcStateEnum.ING.getCode());
                rtcRecord.setStartTime(new Date());
-            } else if (type == 2) {
+            } else if (SingleRtcOperateTypeEnum.REJECT.getCode().equals(type)) {
                // 拒接进入频道
                rtcRecord.setState(RtcStateEnum.END.getCode());
-            } else {
+            } else if (SingleRtcOperateTypeEnum.LEAVE.getCode().equals(type)) {
                // 主动挂断(离开频道)
                rtcRecord.setState(RtcStateEnum.END.getCode());
                rtcRecord.setEndTime(new Date());
+            } else if (SingleRtcOperateTypeEnum.ABNORMAL_DISCONNECT.getCode().equals(type)) {
+                // 异常原因（客户端已不在线）
+                rtcRecord.setState(RtcStateEnum.END.getCode());
+                rtcRecord.setEndTime(new Date());
            }
            this.updateById(rtcRecord);
        } catch (Exception e) {

--- a/core/src/main/java/com/wecloud/im/ws/cache/UserStateListener.java
+++ b/core/src/main/java/com/wecloud/im/ws/cache/UserStateListener.java
 package com.wecloud.im.ws.cache;
-import com.wecloud.im.entity.ImClient;
 import com.wecloud.im.ws.utils.SpringBeanUtils;
 import org.springframework.boot.context.event.ApplicationStartedEvent;
 import org.springframework.context.event.EventListener;

--- a/core/src/main/java/com/wecloud/rtc/service/RtcService.java
+++ b/core/src/main/java/com/wecloud/rtc/service/RtcService.java
@@ -43,4 +43,9 @@ public interface RtcService {
     */
    Boolean candidateForward(CandidateForwardParam candidateForwardParam);
+    /**
+     * 异常断线判断处理
+     */
+    void abnormalDisconnect();
 }
--- a/core/src/main/java/com/wecloud/rtc/service/impl/RtcServiceImpl.java
+++ b/core/src/main/java/com/wecloud/rtc/service/impl/RtcServiceImpl.java
 package com.wecloud.rtc.service.impl;
-import io.geekidea.springbootplus.framework.common.api.ApiCode;
+import com.alibaba.fastjson.JSON;
-import io.geekidea.springbootplus.framework.common.api.ApiResult;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
-import io.geekidea.springbootplus.framework.common.exception.BusinessException;
-import lombok.extern.slf4j.Slf4j;
-import java.util.List;
-import org.apache.commons.collections4.CollectionUtils;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
 import com.wecloud.im.entity.ImApplication;
 import com.wecloud.im.entity.ImClient;
+import com.wecloud.im.entity.ImRtcRecord;
 import com.wecloud.im.param.rtc.CandidateForwardParam;
 import com.wecloud.im.param.rtc.CreateRtcChannelParam;
 import com.wecloud.im.param.rtc.CreateRtcChannelResult;
@@ -20,6 +12,7 @@ import com.wecloud.im.param.rtc.JoinRtcChannelParam;
 import com.wecloud.im.param.rtc.LeaveRtcChannelParam;
 import com.wecloud.im.param.rtc.RejectRtcChannelParam;
 import com.wecloud.im.param.rtc.SdpForwardParam;
+import com.wecloud.im.sdk.enums.SingleRtcOperateTypeEnum;
 import com.wecloud.im.service.ImApplicationService;
 import com.wecloud.im.service.ImClientBlacklistService;
 import com.wecloud.im.service.ImClientService;
@@ -36,11 +29,22 @@ import com.wecloud.rtc.service.MangerRtcCacheService;
 import com.wecloud.rtc.service.RtcService;
 import com.wecloud.rtc.service.WsRtcWrite;
 import com.wecloud.utils.SnowflakeUtil;
+import io.geekidea.springbootplus.framework.common.api.ApiCode;
+import io.geekidea.springbootplus.framework.common.api.ApiResult;
+import io.geekidea.springbootplus.framework.common.exception.BusinessException;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.CollectionUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import java.util.List;
 @Slf4j
 @Service
 public class RtcServiceImpl extends UserStateListener implements RtcService {
+    private static final Integer MAX_OFFLINE_TIMES = 3;
    @Autowired
    private ImApplicationService imApplicationService;
@@ -119,7 +123,7 @@ public class RtcServiceImpl extends UserStateListener implements RtcService {
        wsRtcWrite.rtcCall(rtcCallResponse, toClient.getId());
-        // 创建通话记录  todo 优化为mq推送
+        // 创建通话记录
        imRtcRecordService.createRtcRecord(createRtcChannelParam, createRtcChannelResult.getChannelId(), currentClient);
        // TODO 待开发 下发安卓和ios系统推送
@@ -153,8 +157,8 @@ public class RtcServiceImpl extends UserStateListener implements RtcService {
            wsRtcWrite.clientJoin(rtcSdpForwardResponse, Long.valueOf(toClientId));
        }
-        // 更新通话记录  todo 优化为mq推送
+        // 更新通话记录
-        imRtcRecordService.updateRtcRecord(joinRtcChannelParam.getChannelId(), 1);
+        imRtcRecordService.updateRtcRecord(joinRtcChannelParam.getChannelId(), SingleRtcOperateTypeEnum.JOIN.getCode());
        return true;
    }
@@ -187,8 +191,8 @@ public class RtcServiceImpl extends UserStateListener implements RtcService {
            mangerRtcCacheService.delChannelInfo(rejectRtcChannelParam.getChannelId());
        }
-        // 更新通话记录  todo 优化为mq推送
+        // 更新通话记录
-        imRtcRecordService.updateRtcRecord(rejectRtcChannelParam.getChannelId(), 2);
+        imRtcRecordService.updateRtcRecord(rejectRtcChannelParam.getChannelId(), SingleRtcOperateTypeEnum.REJECT.getCode());
        return true;
    }
@@ -198,8 +202,8 @@ public class RtcServiceImpl extends UserStateListener implements RtcService {
        ImClient currentClient = imClientService.getCurrentClient();
        this.leave(leaveRtcChannelParam, currentClient);
-        // 更新通话记录  todo 优化为mq推送
+        // 更新通话记录
-        imRtcRecordService.updateRtcRecord(leaveRtcChannelParam.getChannelId(), 3);
+        imRtcRecordService.updateRtcRecord(leaveRtcChannelParam.getChannelId(), SingleRtcOperateTypeEnum.LEAVE.getCode());
        return true;
    }
@@ -313,6 +317,37 @@ public class RtcServiceImpl extends UserStateListener implements RtcService {
        return true;
    }
+    @Override
+    public void abnormalDisconnect() {
+        List<ImRtcRecord> records = imRtcRecordService.list(new QueryWrapper<ImRtcRecord>().lambda()
+                        .ne(ImRtcRecord::getState, 3));
+        if (CollectionUtils.isEmpty(records)) {
+            return;
+        }
+        for (ImRtcRecord record : records) {
+            try {
+                if (record.getOfflineTimes() >= MAX_OFFLINE_TIMES) {
+                    // 最大离线次数到阈值，做离线处理
+                    imRtcRecordService.updateRtcRecord(record.getChannelId(), SingleRtcOperateTypeEnum.ABNORMAL_DISCONNECT.getCode());
+                } else {
+                    // 判断频道内两个人是否离线 只要其中一人离线 则认定为离线  将该通话离线次数 +1
+                    ImClient fromClient = imClientService.getCacheImClient(record.getFkAppid(), record.getFromClientId());
+                    ImClient toClient = imClientService.getCacheImClient(record.getFkAppid(), record.getToClientId());
+                    boolean isFromClientOnline = userStateCacheManager.isOnline(fromClient.getId());
+                    boolean isToClientOnline = userStateCacheManager.isOnline(toClient.getId());
+                    if (isFromClientOnline && isToClientOnline) {
+                        // 双方均在线 跳过
+                        continue;
+                    }
+                    record.setOfflineTimes(record.getOfflineTimes() + 1);
+                    imRtcRecordService.updateById(record);
+                }
+            } catch (Exception e) {
+                log.info("单人音视频 {} ，处理异常 ", JSON.toJSONString(record), e);
+            }
+        }
+    }
    /**
     * 判断是否被拉黑

--- a/docs/db/feature-cluster增量.sql
+++ b/docs/db/feature-cluster增量.sql
 -- 在feature-cluster 2021年12月22日之后，需要执行的的sql增量脚本
@@ -185,6 +185,14 @@ CREATE TABLE `im_rtc_record`
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='单人音视频聊天记录表';
+ALTER TABLE im_rtc_record
+    ADD COLUMN `offline_times` tinyint(1) unsigned DEFAULT '0' COMMENT '通话过程中离线次数统计';
+ALTER TABLE `im_rtc_record`
+    ADD INDEX `idx_state_create_time`(`state`, `create_time`);
 ALTER TABLE im_conversation
    ADD COLUMN `is_encrypt` tinyint(1) unsigned DEFAULT '0' COMMENT '是否加密聊天： 1-是 0-否';

--- a/scheduled/src/main/java/io/geekidea/springbootplus/scheduled/MultiMeetScheduled.java
+++ b/scheduled/src/main/java/io/geekidea/springbootplus/scheduled/MultiMeetScheduled.java
@@ -26,7 +26,6 @@ public class MultiMeetScheduled {
    public void callingTimeout() {
        log.info("呼叫超时处理开始...");
        multiMeetService.callingTimeout();
-        log.info("呼叫超时处理结束...");
    }
    /**
@@ -37,7 +36,6 @@ public class MultiMeetScheduled {
    public void disconnect() {
        log.info("异常断线处理开始...");
        multiMeetService.disconnect();
-        log.info("异常断线处理结束...");
    }
 }
--- a/scheduled/src/main/java/io/geekidea/springbootplus/scheduled/SingleRtcScheduled.java
+++ b/scheduled/src/main/java/io/geekidea/springbootplus/scheduled/SingleRtcScheduled.java
+package io.geekidea.springbootplus.scheduled;
+import com.wecloud.rtc.service.RtcService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import javax.annotation.Resource;
+/**
+ * @Author Future
+ * @Date 2022/6/3 11:43
+ * @Description 单人rtc定时器调度
+ */
+@Slf4j
+@Component
+public class SingleRtcScheduled {
+    @Resource
+    private RtcService rtcService;
+    /**
+     * 呼叫异常停止处理 查出正在通话的记录，检测是否在线，如未在线超过5次，按断线处理
+     */
+    @Scheduled(cron = "*/13 * * * * ?")
+    public void abnormalDisconnect() {
+        log.info("单人音视频异常断线处理开始...");
+        rtcService.abnormalDisconnect();
+    }
+}