临时修改exam模块两个websocket连接鉴权，去掉time失效判断和token比对，方便压测使用

themis-business/src/main/java/com/qmth/themis/business/enums/MonitorVideoSourceEnum.java

@@ -25,4 +25,13 @@ public enum MonitorVideoSourceEnum {
     public String getCode() {
         return code;
     }
+
+    public static MonitorVideoSourceEnum findByName(String name) {
+        for (MonitorVideoSourceEnum value : MonitorVideoSourceEnum.values()) {
+            if (value.toString().equalsIgnoreCase(name)) {
+                return value;
+            }
+        }
+        return null;
+    }
 }

themis-common/src/main/java/com/qmth/themis/common/enums/ExceptionResultEnum.java

@@ -11,47 +11,47 @@ public enum ExceptionResultEnum {
     /**
      * 考生端错误
      */
-    NOT_FOUND_EXAM_RECORD(500,500020,"未找到考试记录"),
+    NOT_FOUND_EXAM_RECORD(500, 500020, "未找到考试记录"),
 
-    NOT_FOUND_EXAM_STUDENT(500,500021,"未找到考生"),
+    NOT_FOUND_EXAM_STUDENT(500, 500021, "未找到考生"),
 
-    EXAM_ALREADY_FINISHED(500,500022,"该考试已结束"),
+    EXAM_ALREADY_FINISHED(500, 500022, "该考试已结束"),
 
-    EXAM_ID_NOT_EQUALY(500,500023,"考试记录的学生Id和当前登录用户不一致"),
+    EXAM_ID_NOT_EQUALY(500, 500023, "考试记录的学生Id和当前登录用户不一致"),
 
-    QR_CODE_EXPIRE(500,500024,"二维码已过期"),
+    QR_CODE_EXPIRE(500, 500024, "二维码已过期"),
 
-    ORG_ID_OR_EXAM_ID_NOT_CHOOSE(500,500025,"机构id或考试id必须有一个"),
+    ORG_ID_OR_EXAM_ID_NOT_CHOOSE(500, 500025, "机构id或考试id必须有一个"),
 
-    STUDENT_NOT_ALLOW_LOGIN(500,500026,"该学生正在考试，不能登录"),
+    STUDENT_NOT_ALLOW_LOGIN(500, 500026, "该学生正在考试，不能登录"),
 
-    EXAM_STATUS_NOT_NULL(500,500027,"考试状态不能为空"),
+    EXAM_STATUS_NOT_NULL(500, 500027, "考试状态不能为空"),
 
-    EXAM_STATUS_NOT_ALLOW_IP(500,500028,"考生IP不被允许"),
+    EXAM_STATUS_NOT_ALLOW_IP(500, 500028, "考生IP不被允许"),
 
-    EXAM_LEFT_COUNT_IS_ZERO(500,500029,"没有剩余考试次数"),
+    EXAM_LEFT_COUNT_IS_ZERO(500, 500029, "没有剩余考试次数"),
 
-    NOT_FOUND_EXAM_ACTIVITY(500,500030,"未找到考试场次"),
+    NOT_FOUND_EXAM_ACTIVITY(500, 500030, "未找到考试场次"),
 
-    NOT_ALLOW_MAX_START_TIME(500,500031,"没有到允许开考的时间"),
+    NOT_ALLOW_MAX_START_TIME(500, 500031, "没有到允许开考的时间"),
 
-    MAX_START_TIME_FINISHED(500,500032,"允许开考的时间已结束"),
+    MAX_START_TIME_FINISHED(500, 500032, "允许开考的时间已结束"),
 
-    NOT_FOUND_EXAM_COURSE(500,500033,"未找到考试科目"),
+    NOT_FOUND_EXAM_COURSE(500, 500033, "未找到考试科目"),
 
-    EXAM_COURSE_OR_EXAM_PAPER_NOT_BIND(500,500034,"考试科目未绑定试卷"),
+    EXAM_COURSE_OR_EXAM_PAPER_NOT_BIND(500, 500034, "考试科目未绑定试卷"),
 
-    NOT_FOUND_EXAM_PAPER(500,500035,"未找到试卷"),
+    NOT_FOUND_EXAM_PAPER(500, 500035, "未找到试卷"),
 
-    CLIENT_NET_OFFLINE(500,500036,"客户端网络离线"),
+    CLIENT_NET_OFFLINE(500, 500036, "客户端网络离线"),
 
-    CLIENT_CAMERA_OFFLINE(500,500037,"客户端摄像头离线"),
+    CLIENT_CAMERA_OFFLINE(500, 500037, "客户端摄像头离线"),
 
-    CLIENT_SCREEN_OFFLINE(500,500038,"客户端屏幕离线"),
+    CLIENT_SCREEN_OFFLINE(500, 500038, "客户端屏幕离线"),
 
-    MOBILE_FIRST_OFFLINE(500,500039,"移动端第一机位离线"),
+    MOBILE_FIRST_OFFLINE(500, 500039, "移动端第一机位离线"),
 
-    MOBILE_SECOND_OFFLINE(500,500040,"移动端第二机位离线"),
+    MOBILE_SECOND_OFFLINE(500, 500040, "移动端第二机位离线"),
 
     /**
      * 系统预置
@@ -192,6 +192,10 @@ public enum ExceptionResultEnum {
 
     AUTHORIZATION_INVALID(401, 401008, "authorization无效"),
 
+    SOURCE_INVALID(401, 401009, "source无效"),
+
+    RECORD_ID_INVALID(401, 401009, "recordId无效"),
+
     /**
      * 404
      */
@@ -233,17 +237,19 @@ public enum ExceptionResultEnum {
 
     REQUEST_AWAIT(500, 500016, "请求等待:请稍后重试"),
 
-    ATTACHMENT_ERROR(500, 500017,"上传文件失败"),
+    ATTACHMENT_ERROR(500, 500017, "上传文件失败"),
 
     FINISH_TYPE_ERROR(500, 500018, "考试结束类型错误"),
 
     EXAM_STUDENT_ENABLE(500, 500019, "考生已停用");
 
     private int statusCode;
+
     private int code;
+
     private String message;
 
-    ExceptionResultEnum(int code,int statusCode, String message) {
+    ExceptionResultEnum(int code, int statusCode, String message) {
         this.code = code;
         this.statusCode = statusCode;
         this.message = message;
@@ -260,4 +266,4 @@ public enum ExceptionResultEnum {
     public String getMessage() {
         return message;
     }
-    }
+}

themis-exam/src/main/java/com/qmth/themis/exam/websocket/WebSocketMobileServer.java

@@ -22,6 +22,7 @@ import com.qmth.themis.common.signature.SignatureType;
 import com.qmth.themis.exam.listener.service.MqOeLogicService;
 import com.qmth.themis.exam.websocketTemplete.WebSocketMobileMessageTemplete;
 import com.qmth.themis.mq.templete.Concurrently;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 import org.apache.rocketmq.common.message.MessageExt;
@@ -50,22 +51,53 @@ import java.util.concurrent.ConcurrentHashMap;
 @ServerEndpoint("/ws/mobile")
 @Component
 public class WebSocketMobileServer implements Concurrently {
+
     private final static Logger log = LoggerFactory.getLogger(WebSocketMobileServer.class);
+
     private volatile static ConcurrentHashMap<String, WebSocketMobileServer> webSocketMap = new ConcurrentHashMap<>();
+
     /**
      * 与某个客户端的连接会话，需要通过它来给客户端发送数据
      */
     private Session session = null;
+
     private String sessionId = null, ip = null;
+
     private Long recordId = null;
+
     private String platform = null, deviceId = null, Authorization = null;
+
     private Long time = null;
+
     private RedisUtil redisUtil;
+
     private Long updateTime = null;
+
     private Map<String, Object> tranMap = null;
+
     private MonitorVideoSourceEnum source = null;
+
     private String url = "/ws/mobile";
 
+    private static String getStringParameter(Map<String, List<String>> parameters, String key) {
+        List<String> list = parameters.get(key);
+        if (list == null || list.isEmpty()) {
+            return null;
+        }
+        return list.get(0);
+    }
+
+    private static Long getLongParameter(Map<String, List<String>> parameters, String key) {
+        String value = getStringParameter(parameters, key);
+        if (value != null) {
+            try {
+                return Long.parseLong(value);
+            } catch (Exception e) {
+            }
+        }
+        return null;
+    }
+
     /**
      * 连接建立成功调用的方法
      */
@@ -75,66 +107,90 @@ public class WebSocketMobileServer implements Concurrently {
         if (Objects.isNull(mapParameter)) {
             throw new BusinessException(ExceptionResultEnum.PARAMS_ILLEGALITY);
         }
-        log.info("mapParameter:{}", JacksonUtil.parseJson(mapParameter));
-        log.info("uri:{}", session.getRequestURI());
-        if (Objects.isNull(mapParameter.get("platform")) || mapParameter.get("platform").size() == 0) {
+        log.info("Mobile websocket connect incoming, parameters:{}", JacksonUtil.parseJson(mapParameter));
+        this.platform = getStringParameter(mapParameter, "platform");
+        this.deviceId = getStringParameter(mapParameter, "deviceId");
+        this.Authorization = getStringParameter(mapParameter, "Authorization");
+        this.time = getLongParameter(mapParameter, "time");
+        this.recordId = getLongParameter(mapParameter, "recordId");
+        this.source = MonitorVideoSourceEnum.findByName(getStringParameter(mapParameter, "source"));
+        if (this.time == null) {
+            throw new BusinessException(ExceptionResultEnum.TIME_INVALID);
+        }
+        if (this.recordId == null) {
+            throw new BusinessException(ExceptionResultEnum.RECORD_ID_INVALID);
+        }
+        if (this.source == null) {
+            throw new BusinessException(ExceptionResultEnum.SOURCE_INVALID);
+        }
+        if (StringUtils.isBlank(this.platform)) {
             throw new BusinessException(ExceptionResultEnum.PLATFORM_INVALID);
         }
-        this.platform = String.valueOf(mapParameter.get("platform").get(0));
-        if (Objects.isNull(mapParameter.get("deviceId")) || mapParameter.get("deviceId").size() == 0) {
+        if (StringUtils.isBlank(this.deviceId)) {
             throw new BusinessException(ExceptionResultEnum.DEVICE_ID_INVALID);
         }
-        this.deviceId = String.valueOf(mapParameter.get("deviceId").get(0));
-        this.Authorization = String.valueOf(mapParameter.get("Authorization").get(0));
-        this.time = Long.parseLong(String.valueOf(mapParameter.get("time").get(0)));
-        this.recordId = Long.parseLong(String.valueOf(mapParameter.get("recordId").get(0)));
-        this.source = MonitorVideoSourceEnum.valueOf(mapParameter.get("source").get(0));
-//        final SignatureInfo info = SignatureInfo
-//                .parse(Authorization);
-        if (!SystemConstant.expire(this.time.longValue())) {
-            final SignatureInfo info = SignatureInfo
-                    .parse(SystemConstant.GET, url, this.time, this.Authorization);
-            if (Objects.nonNull(info) && info.getType() == SignatureType.TOKEN) {
-                String sessionId = info.getInvoker();
-                redisUtil = SpringContextHolder.getBean(RedisUtil.class);
-                TBSession tbSession = (TBSession) redisUtil.getUserSession(sessionId);
-                if (Objects.isNull(tbSession)) {
-                    throw new BusinessException(ExceptionResultEnum.LOGIN_NO);
-                } else {
-                    if (info.validate(tbSession.getAccessToken()) && info.getTimestamp() < tbSession.getExpireTime()
-                            && platform.equalsIgnoreCase(tbSession.getPlatform()) && Objects.equals(deviceId, tbSession.getDeviceId())) {
-                        this.session = session;
-                        session.setMaxIdleTimeout(SystemConstant.WEBSOCKET_MAX_TIME_OUT);
-                        this.sessionId = tbSession.getId();
-                        if (webSocketMap.containsKey(this.recordId + "-" + this.source.name())) {
-                            webSocketMap.remove(this.recordId + "-" + this.source.name());
-                            webSocketMap.put(this.recordId + "-" + this.source.name(), this);
-                        } else {
-                            webSocketMap.put(this.recordId + "-" + this.source.name(), this);
-//                        addOnlineCount();
-                        }
-//                    log.info("用户连接:{},当前在线人数为:{}", this.sessionId, getOnlineCount());
-                        log.info("用户连接:{}", this.sessionId);
-                        InetSocketAddress addr = (InetSocketAddress) WebsocketUtil.getFieldInstance(this.session.getAsyncRemote(), "base#socketWrapper#socket#sc#remoteAddress");
-                        this.ip = addr.toString().replace("/", "").split(":")[0];
-//                    this.sendMessage("ip[" + this.ip + "]连接成功");
-                        log.info("ip[:{}]连接成功", this.ip);
-                        tranMap = new HashMap<>();
-                        tranMap.put("recordId", this.recordId);
-                        tranMap.put("deviceId", this.deviceId);
-                        tranMap.put("ip", this.ip);
-                        this.updateTime = System.currentTimeMillis();
-                        tranMap.put("updateTime", this.updateTime);
-                    } else {
-                        throw new BusinessException(ExceptionResultEnum.AUTHORIZATION_ERROR);
-                    }
-                }
-            } else {
-                throw new BusinessException(ExceptionResultEnum.AUTHORIZATION_ERROR);
-            }
-        } else {
+        //校验时间戳是否过期
+        if (SystemConstant.expire(time)) {
+            log.warn("Authorization faile: time expired, server time=" + System.currentTimeMillis());
+            //throw new BusinessException(ExceptionResultEnum.AUTHORIZATION_ERROR);
+        }
+        //校验签名信息
+        final SignatureInfo info = SignatureInfo.parse(SystemConstant.GET, url, this.time, this.Authorization);
+        if (info == null) {
+            log.warn("Authorization faile: signature decode error");
+            throw new BusinessException(ExceptionResultEnum.AUTHORIZATION_ERROR);
+        }
+        if (SignatureType.TOKEN != info.getType()) {
+            log.warn("Authorization faile: signature type is not Token");
+            throw new BusinessException(ExceptionResultEnum.AUTHORIZATION_ERROR);
+        }
+        //校验session
+        String sessionId = info.getInvoker();
+        redisUtil = SpringContextHolder.getBean(RedisUtil.class);
+        TBSession tbSession = (TBSession) redisUtil.getUserSession(sessionId);
+        if (Objects.isNull(tbSession)) {
+            log.warn("Authorization faile: session id not exists: " + sessionId);
+            throw new BusinessException(ExceptionResultEnum.LOGIN_NO);
+        }
+        if (tbSession.getExpireTime() < System.currentTimeMillis() || info.getTimestamp() > tbSession.getExpireTime()) {
+            log.warn("Authorization faile: session has expired, expire time=" + tbSession.getExpireTime());
+            throw new BusinessException(ExceptionResultEnum.LOGIN_NO);
+        }
+        if (!info.validate(tbSession.getAccessToken())) {
+            log.warn("Authorization faile: access token invalid, session token is " + tbSession.getAccessToken());
+            //throw new BusinessException(ExceptionResultEnum.AUTHORIZATION_ERROR);
+        }
+        if (!tbSession.getPlatform().equalsIgnoreCase(platform)) {
+            log.warn("Authorization faile: platform invalid, session platform is " + tbSession.getPlatform());
+            throw new BusinessException(ExceptionResultEnum.AUTHORIZATION_ERROR);
+        }
+        if (!tbSession.getDeviceId().equalsIgnoreCase(deviceId)) {
+            log.warn("Authorization faile: deviceId invalid, session deviceId is " + tbSession.getDeviceId());
             throw new BusinessException(ExceptionResultEnum.AUTHORIZATION_ERROR);
         }
+        this.session = session;
+        session.setMaxIdleTimeout(SystemConstant.WEBSOCKET_MAX_TIME_OUT);
+        this.sessionId = tbSession.getId();
+        if (webSocketMap.containsKey(this.recordId + "-" + this.source.name())) {
+            webSocketMap.remove(this.recordId + "-" + this.source.name());
+            webSocketMap.put(this.recordId + "-" + this.source.name(), this);
+        } else {
+            webSocketMap.put(this.recordId + "-" + this.source.name(), this);
+            //                        addOnlineCount();
+        }
+        //                    log.info("用户连接:{},当前在线人数为:{}", this.sessionId, getOnlineCount());
+        log.info("用户连接:{}", this.sessionId);
+        InetSocketAddress addr = (InetSocketAddress) WebsocketUtil
+                .getFieldInstance(this.session.getAsyncRemote(), "base#socketWrapper#socket#sc#remoteAddress");
+        this.ip = addr.toString().replace("/", "").split(":")[0];
+        //                    this.sendMessage("ip[" + this.ip + "]连接成功");
+        log.info("ip[:{}]连接成功", this.ip);
+        tranMap = new HashMap<>();
+        tranMap.put("recordId", this.recordId);
+        tranMap.put("deviceId", this.deviceId);
+        tranMap.put("ip", this.ip);
+        this.updateTime = System.currentTimeMillis();
+        tranMap.put("updateTime", this.updateTime);
     }
 
     /**
@@ -146,7 +202,7 @@ public class WebSocketMobileServer implements Concurrently {
         if (webSocketMap.containsKey(this.recordId + "-" + this.source.name())) {
             webSocketMap.remove(this.recordId + "-" + this.source.name());
             //从set中删除
-//            subOnlineCount();
+            //            subOnlineCount();
             //判断是否是正常退出
             ExamRecordCacheUtil.setMonitorStatus(recordId, this.source, MonitorStatusSourceEnum.STOP, true);
             ConcurrentHashMap<Long, WebSocketOeServer> webSocketMap = WebSocketOeServer.getWebSocketMap();
@@ -159,7 +215,7 @@ public class WebSocketMobileServer implements Concurrently {
                 webSocketOeServer.sendMessage(websocketDto);
             }
         }
-//        log.info("用户退出:{},当前在线人数为:{},updateTime:{}", this.sessionId, getOnlineCount(), this.updateTime);
+        //        log.info("用户退出:{},当前在线人数为:{},updateTime:{}", this.sessionId, getOnlineCount(), this.updateTime);
         log.info("用户退出:{},updateTime:{}", this.sessionId, this.updateTime);
     }
 
@@ -179,13 +235,17 @@ public class WebSocketMobileServer implements Concurrently {
                 JSONObject jsonObject = JSONObject.parseObject(message);
                 log.info("onMessage:{}", jsonObject.toJSONString());
                 if (Objects.nonNull(jsonObject)) {
-                    WebSocketMobileMessageTemplete webSocketMobileMessageTemplete = SpringContextHolder.getBean(WebSocketMobileMessageTemplete.class);
+                    WebSocketMobileMessageTemplete webSocketMobileMessageTemplete = SpringContextHolder
+                            .getBean(WebSocketMobileMessageTemplete.class);
                     Gson gson = new Gson();
                     WebsocketDto websocketDto = gson.fromJson(gson.toJson(jsonObject), WebsocketDto.class);
                     jsonObject.getJSONObject("body").put("recordId", this.recordId);
                     websocketDto.setBody(jsonObject.getJSONObject("body"));
-                    Method method = webSocketMobileMessageTemplete.getClass().getDeclaredMethod(WebsocketTypeEnum.valueOf(websocketDto.getType()).getDesc(), String.class);
-                    WebsocketDto result = (WebsocketDto) method.invoke(webSocketMobileMessageTemplete, String.valueOf(websocketDto.getBody()));
+                    Method method = webSocketMobileMessageTemplete.getClass()
+                            .getDeclaredMethod(WebsocketTypeEnum.valueOf(websocketDto.getType()).getDesc(),
+                                    String.class);
+                    WebsocketDto result = (WebsocketDto) method
+                            .invoke(webSocketMobileMessageTemplete, String.valueOf(websocketDto.getBody()));
                     this.sendMessage(result);
                 }
             } catch (Exception e) {
@@ -218,45 +278,45 @@ public class WebSocketMobileServer implements Concurrently {
         this.updateTime = System.currentTimeMillis();
     }
 
-//    /**
-//     * 获取在线人数
-//     *
-//     * @return
-//     */
-//    public synchronized int getOnlineCount() {
-//        Object o = redisUtil.get(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT);
-//        return Objects.isNull(o) ? 0 : (int) o;
-//    }
-//
-//    /**
-//     * 在线人数加一
-//     */
-//    public synchronized void addOnlineCount() {
-//        if (redisUtil.lock(SystemConstant.REDIS_LOCK_WEBSOCKET_PREFIX + this.sessionId, SystemConstant.REDIS_LOCK_WEBSOCKET_TIME_OUT)) {
-//            Object o = redisUtil.get(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT);
-//            int count = 0;
-//            if (Objects.nonNull(o)) {
-//                count = (int) o;
-//            }
-//            count++;
-//            redisUtil.set(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT, count);
-//        }
-//    }
-//
-//    /**
-//     * 在线人数减一
-//     */
-//    public synchronized void subOnlineCount() {
-//        if (redisUtil.lock(SystemConstant.REDIS_LOCK_WEBSOCKET_PREFIX + this.sessionId, SystemConstant.REDIS_LOCK_WEBSOCKET_TIME_OUT)) {
-//            Object o = redisUtil.get(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT);
-//            int count = 0;
-//            if (Objects.nonNull(o)) {
-//                count = (int) o;
-//            }
-//            count--;
-//            redisUtil.set(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT, count < 0 ? 0 : count);
-//        }
-//    }
+    //    /**
+    //     * 获取在线人数
+    //     *
+    //     * @return
+    //     */
+    //    public synchronized int getOnlineCount() {
+    //        Object o = redisUtil.get(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT);
+    //        return Objects.isNull(o) ? 0 : (int) o;
+    //    }
+    //
+    //    /**
+    //     * 在线人数加一
+    //     */
+    //    public synchronized void addOnlineCount() {
+    //        if (redisUtil.lock(SystemConstant.REDIS_LOCK_WEBSOCKET_PREFIX + this.sessionId, SystemConstant.REDIS_LOCK_WEBSOCKET_TIME_OUT)) {
+    //            Object o = redisUtil.get(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT);
+    //            int count = 0;
+    //            if (Objects.nonNull(o)) {
+    //                count = (int) o;
+    //            }
+    //            count++;
+    //            redisUtil.set(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT, count);
+    //        }
+    //    }
+    //
+    //    /**
+    //     * 在线人数减一
+    //     */
+    //    public synchronized void subOnlineCount() {
+    //        if (redisUtil.lock(SystemConstant.REDIS_LOCK_WEBSOCKET_PREFIX + this.sessionId, SystemConstant.REDIS_LOCK_WEBSOCKET_TIME_OUT)) {
+    //            Object o = redisUtil.get(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT);
+    //            int count = 0;
+    //            if (Objects.nonNull(o)) {
+    //                count = (int) o;
+    //            }
+    //            count--;
+    //            redisUtil.set(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT, count < 0 ? 0 : count);
+    //        }
+    //    }
 
     public static ConcurrentHashMap<String, WebSocketMobileServer> getWebSocketMap() {
         return webSocketMap;
@@ -267,7 +327,8 @@ public class WebSocketMobileServer implements Concurrently {
     }
 
     @Override
-    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
+    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
+            ConsumeConcurrentlyContext consumeConcurrentlyContext) {
         RedisUtil redisUtil = SpringContextHolder.getBean(RedisUtil.class);
         MqOeLogicService mqOeLogicService = SpringContextHolder.getBean(MqOeLogicService.class);
         MqDto mqDto = null;
@@ -282,7 +343,10 @@ public class WebSocketMobileServer implements Concurrently {
                 if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
                     mqOeLogicService.execMqMaxReconsumeTime(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
                 } else {
-                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && Objects.nonNull(redisUtil.get(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId())) && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
+                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE
+                            && Objects.nonNull(redisUtil.get(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId()))
+                            && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(),
+                            SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
                         try {
                             mqOeLogicService.execMqOeMobileLogic(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
                             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

themis-exam/src/main/java/com/qmth/themis/exam/websocket/WebSocketOeServer.java

@@ -23,6 +23,7 @@ import com.qmth.themis.exam.config.ExamConstant;
 import com.qmth.themis.exam.listener.service.MqOeLogicService;
 import com.qmth.themis.exam.websocketTemplete.WebSocketOeMessageTemplete;
 import com.qmth.themis.mq.templete.Concurrently;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 import org.apache.rocketmq.common.message.MessageExt;
@@ -50,21 +51,51 @@ import java.util.concurrent.ConcurrentHashMap;
 @ServerEndpoint("/ws/oe")
 @Component
 public class WebSocketOeServer implements Concurrently {
+
     private final static Logger log = LoggerFactory.getLogger(WebSocketOeServer.class);
+
     private volatile static ConcurrentHashMap<Long, WebSocketOeServer> webSocketMap = new ConcurrentHashMap<>();
+
     /**
      * 与某个客户端的连接会话，需要通过它来给客户端发送数据
      */
     private Session session = null;
+
     private String sessionId = null, ip = null;
+
     private Long recordId = null;
+
     private String platform = null, deviceId = null, Authorization = null;
+
     private Long time = null;
+
     private RedisUtil redisUtil;
+
     private Long updateTime = null;
+
     private Map<String, Object> tranMap = null;
+
     private String url = "/ws/oe";
 
+    private static String getStringParameter(Map<String, List<String>> parameters, String key) {
+        List<String> list = parameters.get(key);
+        if (list == null || list.isEmpty()) {
+            return null;
+        }
+        return list.get(0);
+    }
+
+    private static Long getLongParameter(Map<String, List<String>> parameters, String key) {
+        String value = getStringParameter(parameters, key);
+        if (value != null) {
+            try {
+                return Long.parseLong(value);
+            } catch (Exception e) {
+            }
+        }
+        return null;
+    }
+
     /**
      * 连接建立成功调用的方法
      */
@@ -74,73 +105,97 @@ public class WebSocketOeServer implements Concurrently {
         if (Objects.isNull(mapParameter)) {
             throw new BusinessException(ExceptionResultEnum.PARAMS_ILLEGALITY);
         }
-        log.info("mapParameter:{}", JacksonUtil.parseJson(mapParameter));
-        log.info("uri:{}", session.getRequestURI());
-        if (Objects.isNull(mapParameter.get("platform")) || mapParameter.get("platform").size() == 0) {
+        log.info("OE websocket connect incoming, parameters:{}", JacksonUtil.parseJson(mapParameter));
+        this.platform = getStringParameter(mapParameter, "platform");
+        this.deviceId = getStringParameter(mapParameter, "deviceId");
+        this.Authorization = getStringParameter(mapParameter, "Authorization");
+        this.time = getLongParameter(mapParameter, "time");
+        this.recordId = getLongParameter(mapParameter, "recordId");
+        if (this.time == null) {
+            throw new BusinessException(ExceptionResultEnum.TIME_INVALID);
+        }
+        if (this.recordId == null) {
+            throw new BusinessException(ExceptionResultEnum.RECORD_ID_INVALID);
+        }
+        if (StringUtils.isBlank(this.platform)) {
             throw new BusinessException(ExceptionResultEnum.PLATFORM_INVALID);
         }
-        this.platform = String.valueOf(mapParameter.get("platform").get(0));
-        if (Objects.isNull(mapParameter.get("deviceId")) || mapParameter.get("deviceId").size() == 0) {
+        if (StringUtils.isBlank(this.deviceId)) {
             throw new BusinessException(ExceptionResultEnum.DEVICE_ID_INVALID);
         }
-        this.deviceId = String.valueOf(mapParameter.get("deviceId").get(0));
-        this.Authorization = String.valueOf(mapParameter.get("Authorization").get(0));
-        this.time = Long.parseLong(String.valueOf(mapParameter.get("time").get(0)));
-        this.recordId = Long.parseLong(String.valueOf(mapParameter.get("recordId").get(0)));
-//        final SignatureInfo info = SignatureInfo
-//                .parse(Authorization);
-        if (!SystemConstant.expire(this.time.longValue())) {
-            final SignatureInfo info = SignatureInfo
-                    .parse(SystemConstant.GET, url, this.time, this.Authorization);
-            if (Objects.nonNull(info) && info.getType() == SignatureType.TOKEN) {
-                String sessionId = info.getInvoker();
-                redisUtil = SpringContextHolder.getBean(RedisUtil.class);
-                TBSession tbSession = (TBSession) redisUtil.getUserSession(sessionId);
-                if (Objects.isNull(tbSession)) {
-                    throw new BusinessException(ExceptionResultEnum.LOGIN_NO);
-                } else {
-                    if (info.validate(tbSession.getAccessToken()) && info.getTimestamp() < tbSession.getExpireTime()
-                            && platform.equalsIgnoreCase(tbSession.getPlatform()) && Objects.equals(deviceId, tbSession.getDeviceId())) {
-                        this.session = session;
-                        session.setMaxIdleTimeout(SystemConstant.WEBSOCKET_MAX_TIME_OUT);
-                        this.sessionId = tbSession.getId();
-                        if (webSocketMap.containsKey(this.recordId)) {
-                            webSocketMap.remove(this.recordId);
-                            webSocketMap.put(this.recordId, this);
-                        } else {
-                            webSocketMap.put(this.recordId, this);
-                            addOnlineCount();
-                        }
-                        log.info("用户连接:{},当前在线人数为:{}", this.sessionId, getOnlineCount());
-                        InetSocketAddress addr = (InetSocketAddress) WebsocketUtil.getFieldInstance(this.session.getAsyncRemote(), "base#socketWrapper#socket#sc#remoteAddress");
-                        this.ip = addr.toString().replace("/", "").split(":")[0];
-//                    this.sendMessage("ip[" + this.ip + "]连接成功");
-                        log.info("ip[:{}]连接成功", this.ip);
-                        ExamRecordCacheUtil.setClientWebsocketStatus(recordId, WebsocketStatusEnum.ON_LINE, false);
-                        ExamRecordCacheUtil.setClientCurrentIp(recordId, this.ip, false);
-                        ExamRecordCacheUtil.setClientWebsocketId(recordId, this.session.getId(), false);
-                        Long clientLastSyncTime = System.currentTimeMillis();
-                        ExamRecordCacheUtil.setClientLastSyncTime(recordId, clientLastSyncTime, false);
-                        String[] columns = new String[]{ExamRecordFieldEnum.client_websocket_status.name(), ExamRecordFieldEnum.client_current_ip.name(), ExamRecordFieldEnum.client_websocket_id.name(), ExamRecordFieldEnum.client_last_sync_time.name()};
-                        Object[] values = new Object[]{WebsocketStatusEnum.ON_LINE, this.ip, this.session.getId(), clientLastSyncTime};
-                        TOeExamRecordService tOeExamRecordService = SpringContextHolder.getBean(TOeExamRecordService.class);
-                        tOeExamRecordService.dataUpdatesMq(recordId, columns, values);
-                        tranMap = new HashMap<>();
-                        tranMap.put("recordId", this.recordId);
-                        tranMap.put("deviceId", this.deviceId);
-                        tranMap.put("ip", this.ip);
-                        this.updateTime = System.currentTimeMillis();
-                        tranMap.put("updateTime", this.updateTime);
-                    } else {
-                        throw new BusinessException(ExceptionResultEnum.AUTHORIZATION_ERROR);
-                    }
-                }
-            } else {
-                throw new BusinessException(ExceptionResultEnum.AUTHORIZATION_ERROR);
-            }
-        } else {
+        //校验时间戳是否过期
+        if (SystemConstant.expire(time)) {
+            log.warn("Authorization faile: time expired, server time=" + System.currentTimeMillis());
+            //throw new BusinessException(ExceptionResultEnum.AUTHORIZATION_ERROR);
+        }
+        //校验签名信息
+        final SignatureInfo info = SignatureInfo.parse(SystemConstant.GET, url, this.time, this.Authorization);
+        if (info == null) {
+            log.warn("Authorization faile: signature decode error");
+            throw new BusinessException(ExceptionResultEnum.AUTHORIZATION_ERROR);
+        }
+        if (SignatureType.TOKEN != info.getType()) {
+            log.warn("Authorization faile: signature type is not Token");
             throw new BusinessException(ExceptionResultEnum.AUTHORIZATION_ERROR);
         }
+        //校验session
+        String sessionId = info.getInvoker();
+        redisUtil = SpringContextHolder.getBean(RedisUtil.class);
+        TBSession tbSession = (TBSession) redisUtil.getUserSession(sessionId);
+        if (Objects.isNull(tbSession)) {
+            log.warn("Authorization faile: session id not exists: " + sessionId);
+            throw new BusinessException(ExceptionResultEnum.LOGIN_NO);
+        }
+        if (tbSession.getExpireTime() < System.currentTimeMillis() || info.getTimestamp() > tbSession.getExpireTime()) {
+            log.warn("Authorization faile: session has expired, expire time=" + tbSession.getExpireTime());
+            throw new BusinessException(ExceptionResultEnum.LOGIN_NO);
+        }
+        if (!info.validate(tbSession.getAccessToken())) {
+            log.warn("Authorization faile: access token invalid, session token is " + tbSession.getAccessToken());
+            //throw new BusinessException(ExceptionResultEnum.AUTHORIZATION_ERROR);
+        }
+        if (!tbSession.getPlatform().equalsIgnoreCase(platform)) {
+            log.warn("Authorization faile: platform invalid, session platform is " + tbSession.getPlatform());
+            throw new BusinessException(ExceptionResultEnum.AUTHORIZATION_ERROR);
+        }
+        if (!tbSession.getDeviceId().equalsIgnoreCase(deviceId)) {
+            log.warn("Authorization faile: deviceId invalid, session deviceId is " + tbSession.getDeviceId());
+            throw new BusinessException(ExceptionResultEnum.AUTHORIZATION_ERROR);
+        }
+        this.session = session;
+        session.setMaxIdleTimeout(SystemConstant.WEBSOCKET_MAX_TIME_OUT);
+        this.sessionId = tbSession.getId();
+        if (webSocketMap.containsKey(this.recordId)) {
+            webSocketMap.remove(this.recordId);
+            webSocketMap.put(this.recordId, this);
+        } else {
+            webSocketMap.put(this.recordId, this);
+            addOnlineCount();
+        }
+        log.info("用户连接:{},当前在线人数为:{}", this.sessionId, getOnlineCount());
+        InetSocketAddress addr = (InetSocketAddress) WebsocketUtil
+                .getFieldInstance(this.session.getAsyncRemote(), "base#socketWrapper#socket#sc#remoteAddress");
+        this.ip = addr.toString().replace("/", "").split(":")[0];
+        //                    this.sendMessage("ip[" + this.ip + "]连接成功");
+        log.info("ip[:{}]连接成功", this.ip);
+        ExamRecordCacheUtil.setClientWebsocketStatus(recordId, WebsocketStatusEnum.ON_LINE, false);
+        ExamRecordCacheUtil.setClientCurrentIp(recordId, this.ip, false);
+        ExamRecordCacheUtil.setClientWebsocketId(recordId, this.session.getId(), false);
+        Long clientLastSyncTime = System.currentTimeMillis();
+        ExamRecordCacheUtil.setClientLastSyncTime(recordId, clientLastSyncTime, false);
+        String[] columns = new String[] { ExamRecordFieldEnum.client_websocket_status.name(),
+                ExamRecordFieldEnum.client_current_ip.name(), ExamRecordFieldEnum.client_websocket_id.name(),
+                ExamRecordFieldEnum.client_last_sync_time.name() };
+        Object[] values = new Object[] { WebsocketStatusEnum.ON_LINE, this.ip, this.session.getId(),
+                clientLastSyncTime };
+        TOeExamRecordService tOeExamRecordService = SpringContextHolder.getBean(TOeExamRecordService.class);
+        tOeExamRecordService.dataUpdatesMq(recordId, columns, values);
+        tranMap = new HashMap<>();
+        tranMap.put("recordId", this.recordId);
+        tranMap.put("deviceId", this.deviceId);
+        tranMap.put("ip", this.ip);
+        this.updateTime = System.currentTimeMillis();
+        tranMap.put("updateTime", this.updateTime);
     }
 
     /**
@@ -157,21 +212,25 @@ public class WebSocketOeServer implements Concurrently {
             Date now = new Date();
             ExamRecordCacheUtil.setClientWebsocketStatus(recordId, WebsocketStatusEnum.OFF_LINE, true);
             ExamRecordStatusEnum status = ExamRecordCacheUtil.getStatus(this.recordId);
-            if (!Objects.equals(status, ExamRecordStatusEnum.FIRST_PREPARE) && !Objects.equals(status, ExamRecordStatusEnum.FINISHED) && !Objects.equals(status, ExamRecordStatusEnum.PERSISTED)) {
+            if (!Objects.equals(status, ExamRecordStatusEnum.FIRST_PREPARE) && !Objects
+                    .equals(status, ExamRecordStatusEnum.FINISHED) && !Objects
+                    .equals(status, ExamRecordStatusEnum.PERSISTED)) {
                 //大于等于超时时间，说明规定时间内都没有通信，非正常退出，因为期间会有心跳更新updateTime
                 if ((now.getTime() - this.updateTime) / 1000 >= SystemConstant.WEBSOCKET_MAX_TIME_OUT / 1000) {
                     log.info("超时退出");
                     //发送延时mq消息start
                     MqDtoService mqDtoService = SpringContextHolder.getBean(MqDtoService.class);
                     String level = "2m";
-//                    String level = "30s";
+                    //                    String level = "30s";
                     Integer time = SystemConstant.mqDelayLevel.get(level);
                     LocalDateTime dt = LocalDateTime.now();
                     dt = dt.plusMinutes(Long.parseLong(level.replace("m", "")));
-//                    dt = dt.plusSeconds(Long.parseLong(level.replace("s", "")));
+                    //                    dt = dt.plusSeconds(Long.parseLong(level.replace("s", "")));
                     tranMap.put("timeOut", time);
                     tranMap.put("mqExecTime", dt.toInstant(ZoneOffset.of("+8")).toEpochMilli());
-                    MqDto mqDto = new MqDto(MqTopicEnum.THEMIS_TOPIC.getCode(), MqTagEnum.OE_UN_NORMAL.name(), MqTagEnum.OE_UN_NORMAL, MqTagEnum.OE_UN_NORMAL, String.valueOf(this.recordId), this.tranMap, this.sessionId);
+                    MqDto mqDto = new MqDto(MqTopicEnum.THEMIS_TOPIC.getCode(), MqTagEnum.OE_UN_NORMAL.name(),
+                            MqTagEnum.OE_UN_NORMAL, MqTagEnum.OE_UN_NORMAL, String.valueOf(this.recordId), this.tranMap,
+                            this.sessionId);
                     mqDtoService.assembleSendAsyncDelayMsg(mqDto);
                     //发送延时mq消息end
                 } else {
@@ -201,13 +260,17 @@ public class WebSocketOeServer implements Concurrently {
                 JSONObject jsonObject = JSONObject.parseObject(message);
                 log.info("onMessage:{}", jsonObject.toJSONString());
                 if (Objects.nonNull(jsonObject)) {
-                    WebSocketOeMessageTemplete webSocketOeMessageTemplete = SpringContextHolder.getBean(WebSocketOeMessageTemplete.class);
+                    WebSocketOeMessageTemplete webSocketOeMessageTemplete = SpringContextHolder
+                            .getBean(WebSocketOeMessageTemplete.class);
                     Gson gson = new Gson();
                     WebsocketDto websocketDto = gson.fromJson(gson.toJson(jsonObject), WebsocketDto.class);
                     jsonObject.getJSONObject("body").put("recordId", this.recordId);
                     websocketDto.setBody(jsonObject.getJSONObject("body"));
-                    Method method = webSocketOeMessageTemplete.getClass().getDeclaredMethod(WebsocketTypeEnum.valueOf(websocketDto.getType()).getDesc(), String.class);
-                    WebsocketDto result = (WebsocketDto) method.invoke(webSocketOeMessageTemplete, String.valueOf(websocketDto.getBody()));
+                    Method method = webSocketOeMessageTemplete.getClass()
+                            .getDeclaredMethod(WebsocketTypeEnum.valueOf(websocketDto.getType()).getDesc(),
+                                    String.class);
+                    WebsocketDto result = (WebsocketDto) method
+                            .invoke(webSocketOeMessageTemplete, String.valueOf(websocketDto.getBody()));
                     this.sendMessage(result);
                 }
             } catch (Exception e) {
@@ -257,7 +320,8 @@ public class WebSocketOeServer implements Concurrently {
      * 在线人数加一
      */
     public synchronized void addOnlineCount() {
-        if (redisUtil.lock(SystemConstant.REDIS_LOCK_WEBSOCKET_PREFIX + this.sessionId, SystemConstant.REDIS_LOCK_WEBSOCKET_TIME_OUT)) {
+        if (redisUtil.lock(SystemConstant.REDIS_LOCK_WEBSOCKET_PREFIX + this.sessionId,
+                SystemConstant.REDIS_LOCK_WEBSOCKET_TIME_OUT)) {
             try {
                 Object o = redisUtil.get(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT);
                 int count = 0;
@@ -278,7 +342,8 @@ public class WebSocketOeServer implements Concurrently {
      * 在线人数减一
      */
     public synchronized void subOnlineCount() {
-        if (redisUtil.lock(SystemConstant.REDIS_LOCK_WEBSOCKET_PREFIX + this.sessionId, SystemConstant.REDIS_LOCK_WEBSOCKET_TIME_OUT)) {
+        if (redisUtil.lock(SystemConstant.REDIS_LOCK_WEBSOCKET_PREFIX + this.sessionId,
+                SystemConstant.REDIS_LOCK_WEBSOCKET_TIME_OUT)) {
             try {
                 Object o = redisUtil.get(SystemConstant.WEBSOCKET_OE_ONLINE_COUNT);
                 int count = 0;
@@ -296,8 +361,8 @@ public class WebSocketOeServer implements Concurrently {
     }
 
     @Override
-    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext
-            consumeConcurrentlyContext) {
+    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
+            ConsumeConcurrentlyContext consumeConcurrentlyContext) {
         RedisUtil redisUtil = SpringContextHolder.getBean(RedisUtil.class);
         MqOeLogicService mqOeLogicService = SpringContextHolder.getBean(MqOeLogicService.class);
         MqDto mqDto = null;
@@ -312,7 +377,10 @@ public class WebSocketOeServer implements Concurrently {
                 if (reconsumeTime >= SystemConstant.MAXRECONSUMETIMES) {
                     mqOeLogicService.execMqMaxReconsumeTime(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
                 } else {
-                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE && Objects.nonNull(redisUtil.get(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId())) && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(), SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
+                    if (Objects.nonNull(mqDto.getAck()) && mqDto.getAck().intValue() != SystemConstant.STANDARD_ACK_TYPE
+                            && Objects.nonNull(redisUtil.get(SystemConstant.MQ_TOPIC_BUFFER_LIST, mqDto.getId()))
+                            && redisUtil.lock(SystemConstant.REDIS_LOCK_MQ_PREFIX + mqDto.getId(),
+                            SystemConstant.REDIS_LOCK_MQ_TIME_OUT)) {
                         try {
                             mqOeLogicService.execMqOeLogic(mqDto, SystemConstant.MQ_TOPIC_BUFFER_LIST);
                             return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;