package com.izouma.meta.websocket; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONArray; import com.alibaba.fastjson.JSONObject; import com.izouma.meta.config.Constants; import com.izouma.meta.domain.MetaVisitor; import com.izouma.meta.domain.PublicScreenChat; import com.izouma.meta.dto.PublicScreenChatExceptionMsg; import com.izouma.meta.repo.MetaVisitorRepo; import com.izouma.meta.repo.PublicScreenChatRepo; import com.izouma.meta.service.ContentAuditService; import com.izouma.meta.utils.ApplicationContextUtil; import lombok.extern.slf4j.Slf4j; import org.apache.commons.collections.CollectionUtils; import org.apache.commons.lang.StringUtils; import org.springframework.stereotype.Service; import javax.websocket.*; import javax.websocket.server.PathParam; import javax.websocket.server.ServerEndpoint; import java.time.LocalDateTime; import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; @Service @ServerEndpoint(value = "/websocket/live/public/screen/{userId}") @Slf4j public class LivePublicScreenChatWebsocket { /** * 当前在线的客户端map */ private static final Map clients = new ConcurrentHashMap(); private static final Map user = new ConcurrentHashMap<>(); private PublicScreenChatRepo publicScreenChatRepo; private final String PREFIX = "meta-live-chat:"; private ContentAuditService contentAuditService; private WebsocketCommon websocketCommon; private MetaVisitorRepo metaVisitorRepo; private void init() { if (Objects.isNull(metaVisitorRepo)) { metaVisitorRepo = (MetaVisitorRepo) ApplicationContextUtil.getBean("metaVisitorRepo"); } if (Objects.isNull(publicScreenChatRepo)) { publicScreenChatRepo = (PublicScreenChatRepo) ApplicationContextUtil.getBean("publicScreenChatRepo"); } if (Objects.isNull(contentAuditService)) { contentAuditService = (ContentAuditService) ApplicationContextUtil.getBean("contentAuditService"); } if (Objects.isNull(websocketCommon)) { websocketCommon = (WebsocketCommon) ApplicationContextUtil.getBean("websocketCommon"); } } @OnOpen public void onOpen(@PathParam("userId") String userId, Session session) { init(); MetaVisitor metaVisitor = metaVisitorRepo.findById(Long.parseLong(userId)).orElse(null); if (Objects.isNull(metaVisitor)) { log.error("当前游客信息不存在!"); return; } if (!user.containsKey(userId)) { user.put(userId, metaVisitor); } // 判断当前玩家是否在其他地方登陆 if (clients.containsKey(PREFIX.concat(metaVisitor.getNickname()))) { String msg = String.format("已在别处登陆,sessionId为[%S],正在为您关闭本连接", session.getId()); exceptionHandle(userId, new PublicScreenChatExceptionMsg(1, msg)); try { log.info("关闭session连接"); clients.get(PREFIX.concat(userId)).close(); } catch (Exception e) { exceptionHandle(userId, new PublicScreenChatExceptionMsg(1, String.format("session close throw exception[%S]", e))); return; } } log.info("现在来连接的sessionId:" + session.getId() + "玩家id:" + userId + "玩家昵称" + metaVisitor.getNickname()); clients.put(PREFIX.concat(userId), session); String format = String.format("玩家[%S][%S]进入大厅", userId, metaVisitor.getNickname()); PublicScreenChat publicScreenChat; try { publicScreenChat = savePublicScreenChat(metaVisitor, format, true); } catch (Exception e) { String errMsg = String.format("玩家进入大厅,保存信息发生异常[%S]", e); exceptionHandle(userId, new PublicScreenChatExceptionMsg(2, errMsg)); return; } websocketCommon.sendMessageToOther(clients, JSON.toJSONString(publicScreenChat), PREFIX.concat(userId)); } @OnError public void onError(Session session, Throwable error) { error.printStackTrace(); // 异常处理 log.error(String.format("sessionId[%S]的服务端发生了错误:[%S]", session.getId(), error)); } @OnClose public void onClose(@PathParam("userId") String userId, Session session) { init(); MetaVisitor metaVisitor = user.get(userId); if (Objects.isNull(metaVisitor)) { metaVisitor = metaVisitorRepo.findById(Long.parseLong(userId)).orElse(null); } if (Objects.isNull(metaVisitor)) { String errMsg = "游客信息为空!"; exceptionHandle(userId, new PublicScreenChatExceptionMsg(4, errMsg)); return; } log.info(String.format("sessionId:[%S] userId:[%S] is closed", session.getId(), userId)); String format = String.format("玩家[%S][%S]离开大厅", userId, metaVisitor.getNickname()); PublicScreenChat publicScreenChat; try { publicScreenChat = savePublicScreenChat(metaVisitor, format, true); } catch (Exception e) { String errMsg = String.format("玩家离开大厅,保存信息发生异常[%S]", e); exceptionHandle(userId, new PublicScreenChatExceptionMsg(2, errMsg)); return; } websocketCommon.sendMessageToOther(clients, JSON.toJSONString(publicScreenChat), PREFIX.concat(userId)); clients.remove(PREFIX.concat(userId)); } @OnMessage public void onMessage(@PathParam("userId") String userId, String message, Session session) { init(); MetaVisitor metaVisitor = metaVisitorRepo.findById(Long.parseLong(userId)).orElse(null); if (Objects.isNull(metaVisitor)) { String errMsg = "游客信息为空!"; exceptionHandle(userId, new PublicScreenChatExceptionMsg(4, errMsg)); return; } if (metaVisitor.isTaboo()) { String errMsg = "当前用户已被禁言"; log.info(errMsg); exceptionHandle(userId, new PublicScreenChatExceptionMsg(5, errMsg)); return; } if (StringUtils.isBlank(message)) { String errMsg = "Illegal parameter : message can not be null"; exceptionHandle(userId, new PublicScreenChatExceptionMsg(4, errMsg)); return; } JSONObject jsonObject = JSON.parseObject(message); if (!jsonObject.containsKey("type")) { String errMsg = "Illegal parameter : type can not be null"; exceptionHandle(userId, new PublicScreenChatExceptionMsg(4, errMsg)); return; } String type = jsonObject.getString("type"); if (StringUtils.isBlank(type)) { String errMsg = "Illegal parameter : type can not be null"; exceptionHandle(userId, new PublicScreenChatExceptionMsg(4, errMsg)); return; } // 心跳检测 if (Constants.HEART_RECEIVE.equals(type)) { log.info(String.format("sessionId:[%S] userId:[%S] 连接正常", session.getId(), userId)); websocketCommon.sendMessageToSingle(clients, Constants.HEART_RETURN, PREFIX.concat(userId)); return; } // 查询需要撤回的消息 if (Constants.MESSAGE_RECALL.equals(type)) { List publicScreenChats = publicScreenChatRepo.findAllByTypeAndRecallAndDel(2, 2, false); websocketCommon.sendMessageToSingle(clients, JSONObject.toJSONString(publicScreenChats), PREFIX.concat(userId)); return; } // 消息撤回确定 if (Constants.MESSAGE_RECALL_CONFIRM.equals(type)) { if (!jsonObject.containsKey("ids")) { String errMsg = "Illegal parameter : ids can not be null"; exceptionHandle(userId, new PublicScreenChatExceptionMsg(4, errMsg)); return; } JSONArray jsonArray = jsonObject.getJSONArray("ids"); List ids = jsonArray.toJavaList(Long.class); if (CollectionUtils.isEmpty(ids)) { String errMsg = "Illegal parameter : ids can not be null"; exceptionHandle(userId, new PublicScreenChatExceptionMsg(4, errMsg)); return; } ids.forEach(id -> { publicScreenChatRepo.recallConfirm(id); }); websocketCommon.sendMessageToSingle(clients, "消息已被撤回", PREFIX.concat(userId)); return; } // 正常发送消息 if (Constants.MESSAGE_NORMAL.equals(type)) { if (!jsonObject.containsKey("value")) { String errMsg = "Illegal parameter : value can not be null"; exceptionHandle(userId, new PublicScreenChatExceptionMsg(4, errMsg)); return; } String value = jsonObject.getString("value"); if (StringUtils.isBlank(value)) { String errMsg = "Illegal parameter : value can not be null"; exceptionHandle(userId, new PublicScreenChatExceptionMsg(4, errMsg)); return; } log.info(String.format("收到来自userId[%S]的消息[%S]", userId, value)); if (!contentAuditService.auditText(value)) { savePublicScreenChat(metaVisitor, value, false); exceptionHandle(userId, new PublicScreenChatExceptionMsg(3, "消息包含非法内容")); return; } PublicScreenChat publicScreenChat; try { publicScreenChat = savePublicScreenChat(metaVisitor, value, true); } catch (Exception e) { String errMsg = String.format("玩家发送消息,保存信息发生异常[%S]", e); exceptionHandle(userId, new PublicScreenChatExceptionMsg(2, errMsg)); return; } sendMessageToAll(clients, JSON.toJSONString(publicScreenChat), PREFIX.concat(userId)); } } /** * 给所有用户发消息 * * @param clients 在线客户端 * @param message 消息 */ public void sendMessageToAll(Map clients, String message, String userId) { PublicScreenChat publicScreenChat = JSON.parseObject(message, PublicScreenChat.class); Set userIds = clients.keySet(); userIds.forEach(id -> { try { publicScreenChat.setMyself(id.equals(userId)); log.info(String.format("服务器给所有在线用户发送消息,当前在线人员为[%S]。消息:[%S]", id, JSON.toJSONString(publicScreenChat))); clients.get(id).getBasicRemote().sendText(JSON.toJSONString(publicScreenChat)); } catch (Exception e) { log.error(String.format("send message [%S] to [%S] throw exception [%S]:", JSON.toJSONString(publicScreenChat), id, e)); } }); } private PublicScreenChat savePublicScreenChat(MetaVisitor metaVisitor, String messageInfo, boolean illegal) { PublicScreenChat publicScreenChat = new PublicScreenChat(); publicScreenChat.setUserId(metaVisitor.getId().toString()); publicScreenChat.setSex(metaVisitor.getSex()); publicScreenChat.setNickname(metaVisitor.getNickname()); publicScreenChat.setMessageInfo(messageInfo); publicScreenChat.setTime(LocalDateTime.now()); publicScreenChat.setIllegal(illegal); publicScreenChat.setRecall(1); publicScreenChat.setType(2); return publicScreenChatRepo.save(publicScreenChat); } private void exceptionHandle(String userId, PublicScreenChatExceptionMsg msg) { log.error(JSON.toJSONString(msg)); // 推送消息给该玩家 websocketCommon.sendMessageToSingle(clients, JSON.toJSONString(msg), PREFIX.concat(userId)); } }