广播模式修改

themis-exam/src/main/java/com/qmth/themis/exam/listener/service/impl/MqOeLogicServiceImpl.java

@@ -32,6 +32,7 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * @Description: mq 考生端执行逻辑 impl
@@ -94,6 +95,7 @@ public class MqOeLogicServiceImpl implements MqOeLogicService {
         Gson gson = new Gson();
         ConcurrentHashMap<String, WebSocketOeServer> webSocketMap = WebSocketOeServer.getWebSocketMap();
         String tag = mqDto.getTag();
+        AtomicBoolean match = new AtomicBoolean(false);
         if (Objects.equals(MqTagEnum.OE_MONITOR_FINISH.name(), tag)) {//强制离线交卷
             Set examRecordId = JacksonUtil.readJson(String.valueOf(mqDto.getBody()), Set.class);
             examRecordId.forEach(s -> {
@@ -105,6 +107,7 @@ public class MqOeLogicServiceImpl implements MqOeLogicService {
                     ExamStudentCacheBean examStudentCacheBean = teExamStudentService.getExamStudentCacheBean(examStudentId);
                     WebSocketOeServer webSocketOeServer = webSocketMap.get(clientWebsocketId);
                     if (Objects.nonNull(webSocketOeServer.getRecordId()) && webSocketOeServer.getRecordId().longValue() == recordId.longValue()) {
+                        match.set(true);
                         Map map = new HashMap<>();
                         map.put("form", mqDto.getObjName());
                         map.put(SystemConstant.RECORD_ID, recordId);
@@ -116,26 +119,7 @@ public class MqOeLogicServiceImpl implements MqOeLogicService {
                     }
                 }
             });
-        }
-//        else if (Objects.equals(MqTagEnum.OE_HARD_FINISH.name(), tag)) {//手动交卷
-//            Set examRecordId = JacksonUtil.readJson(String.valueOf(mqDto.getBody()), Set.class);
-//            examRecordId.forEach(s -> {
-//                Long recordId = Long.parseLong(String.valueOf(s));
-//                if (Objects.nonNull(webSocketMap.get(recordId))) {
-//                    Long examStudentId = ExamRecordCacheUtil.getExamStudentId(recordId);
-//                    ExamStudentCacheBean examStudentCacheBean = teExamStudentService.getExamStudentCacheBean(examStudentId);
-//                    WebSocketOeServer webSocketOeServer = webSocketMap.get(recordId);
-//                    Map map = new HashMap<>();
-//                    map.put(SystemConstant.RECORD_ID, recordId);
-//                    map.put(SystemConstant.MESSAGE, FinishTypeEnum.valueOf(String.valueOf(mqDto.getProperties().get("type"))).getCode());
-//                    WebsocketDto websocketDto = new WebsocketDto(WebsocketTypeEnum.HAND_STOP_EXAM.name(), map);
-//                    webSocketOeServer.sendMessage(websocketDto);
-//                    TEExamStudentLog teExamStudentLog = new TEExamStudentLog(mqDto.getType().name(), mqDto.getType().getCode(), mqDto.getType().getCode(), examStudentCacheBean.getStudentId(), examStudentId, recordId);
-//                    teExamStudentLogService.save(teExamStudentLog);
-//                }
-//            });
-//        }
-        else if (Objects.equals(MqTagEnum.OE_WARNING_FINISH.name(), tag)) {//预警交卷
+        } else if (Objects.equals(MqTagEnum.OE_WARNING_FINISH.name(), tag)) {//预警交卷
             Set examRecordId = JacksonUtil.readJson(String.valueOf(mqDto.getBody()), Set.class);
             examRecordId.forEach(s -> {
                 Long recordId = Long.parseLong(String.valueOf(s));
@@ -145,6 +129,7 @@ public class MqOeLogicServiceImpl implements MqOeLogicService {
                     ExamStudentCacheBean examStudentCacheBean = teExamStudentService.getExamStudentCacheBean(examStudentId);
                     WebSocketOeServer webSocketOeServer = webSocketMap.get(clientWebsocketId);
                     if (Objects.nonNull(webSocketOeServer.getRecordId()) && webSocketOeServer.getRecordId().longValue() == recordId.longValue()) {
+                        match.set(true);
                         Map map = new HashMap<>();
                         map.put(SystemConstant.RECORD_ID, recordId);
                         map.put(SystemConstant.BREACH_STATUS, FinishTypeEnum.valueOf(String.valueOf(mqDto.getProperties().get("type"))).getCode());
@@ -164,6 +149,7 @@ public class MqOeLogicServiceImpl implements MqOeLogicService {
                 Long examActivityId = ExamRecordCacheUtil.getExamActivityId(recordId);
                 WebSocketOeServer webSocketOeServer = webSocketMap.get(clientWebsocketId);
                 if (Objects.nonNull(webSocketOeServer.getRecordId()) && webSocketOeServer.getRecordId().longValue() == recordId.longValue()) {
+                    match.set(true);
                     Map<String, Object> prop = mqDto.getProperties();
                     Map map = new HashMap<>();
                     map.put(SystemConstant.RECORD_ID, recordId);
@@ -188,6 +174,7 @@ public class MqOeLogicServiceImpl implements MqOeLogicService {
                 ExamStudentCacheBean examStudentCacheBean = teExamStudentService.getExamStudentCacheBean(examStudentId);
                 WebSocketOeServer webSocketOeServer = webSocketMap.get(clientWebsocketId);
                 if (Objects.nonNull(webSocketOeServer.getRecordId()) && webSocketOeServer.getRecordId().longValue() == recordId.longValue()) {
+                    match.set(true);
                     Map map = new HashMap<>();
                     map.put(SystemConstant.RECORD_ID, recordId);
                     WebsocketDto websocketDto = new WebsocketDto(WebsocketTypeEnum.INVIGILATE_LIVENESS_VERIFY.name(), map);
@@ -207,6 +194,7 @@ public class MqOeLogicServiceImpl implements MqOeLogicService {
             if (Objects.nonNull(webSocketMap.get(clientWebsocketId))) {
                 WebSocketOeServer webSocketOeServer = webSocketMap.get(clientWebsocketId);
                 if (Objects.nonNull(webSocketOeServer.getRecordId()) && webSocketOeServer.getRecordId().longValue() == recordId.longValue()) {
+                    match.set(true);
                     WebsocketDto websocketDto = null;
                     switch (tag.toUpperCase()) {
                         case "OE_WEBSOCKET_MOBILE_ANSWER_READY":
@@ -234,11 +222,13 @@ public class MqOeLogicServiceImpl implements MqOeLogicService {
                 }
             }
         }
-        mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);
-        TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
-        tmRocketMessage.setBody(JacksonUtil.parseJson(tmRocketMessage.getBody()));
-        tmRocketMessageService.saveOrUpdate(tmRocketMessage);
-        redisUtil.delete(key, mqDto.getId());
+        if (match.get()) {
+            mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);
+            TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
+            tmRocketMessage.setBody(JacksonUtil.parseJson(tmRocketMessage.getBody()));
+            tmRocketMessageService.saveOrUpdate(tmRocketMessage);
+            redisUtil.delete(key, mqDto.getId());
+        }
     }
 
     /**
@@ -256,6 +246,7 @@ public class MqOeLogicServiceImpl implements MqOeLogicService {
     public void execMqOeMobileLogic(MqDto mqDto, String key) {
         Gson gson = new Gson();
         String tag = mqDto.getTag();
+        AtomicBoolean match = new AtomicBoolean(false);
         ConcurrentHashMap<String, WebSocketMobileServer> webSocketMap = WebSocketMobileServer.getWebSocketMap();
         if (Objects.equals(MqTagEnum.EXAM_STOP.name(), tag)//考试退出
                 || Objects.equals(MqTagEnum.EXAM_START.name(), tag)) {//考试开始
@@ -264,6 +255,7 @@ public class MqOeLogicServiceImpl implements MqOeLogicService {
             if (Objects.nonNull(clientWebsocketId) && Objects.nonNull(webSocketMap.get(clientWebsocketId + "-" + MonitorVideoSourceEnum.MOBILE_FIRST.name()))) {
                 WebSocketMobileServer webSocketMobileServer = webSocketMap.get(clientWebsocketId + "-" + MonitorVideoSourceEnum.MOBILE_FIRST.name());
                 if (Objects.nonNull(webSocketMobileServer.getRecordId()) && webSocketMobileServer.getRecordId().longValue() == recordId.longValue()) {
+                    match.set(true);
                     WebsocketDto websocketDto = null;
                     switch (tag.toUpperCase()) {
                         case "EXAM_STOP":
@@ -281,6 +273,7 @@ public class MqOeLogicServiceImpl implements MqOeLogicService {
             if (Objects.nonNull(clientWebsocketId) && Objects.nonNull(webSocketMap.get(clientWebsocketId + "-" + MonitorVideoSourceEnum.MOBILE_SECOND.name()))) {
                 WebSocketMobileServer webSocketMobileServer = webSocketMap.get(clientWebsocketId + "-" + MonitorVideoSourceEnum.MOBILE_SECOND.name());
                 if (Objects.nonNull(webSocketMobileServer.getRecordId()) && webSocketMobileServer.getRecordId().longValue() == recordId.longValue()) {
+                    match.set(true);
                     WebsocketDto websocketDto = null;
                     switch (tag.toUpperCase()) {
                         case "EXAM_STOP":
@@ -296,10 +289,12 @@ public class MqOeLogicServiceImpl implements MqOeLogicService {
                 }
             }
         }
-        mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);
-        TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
-        tmRocketMessage.setBody(JacksonUtil.parseJson(tmRocketMessage.getBody()));
-        tmRocketMessageService.saveOrUpdate(tmRocketMessage);
-        redisUtil.delete(key, mqDto.getId());
+        if (match.get()) {
+            mqDto.setAck(SystemConstant.STANDARD_ACK_TYPE);
+            TMRocketMessage tmRocketMessage = gson.fromJson(gson.toJson(mqDto), TMRocketMessage.class);
+            tmRocketMessage.setBody(JacksonUtil.parseJson(tmRocketMessage.getBody()));
+            tmRocketMessageService.saveOrUpdate(tmRocketMessage);
+            redisUtil.delete(key, mqDto.getId());
+        }
     }
 }