raex_mmo/src/main/java/com/izouma/meta/websocket/LivePublicScreenChatWebsocket.java

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<String, Session> clients = new ConcurrentHashMap();

    private static final Map<String, MetaVisitor> 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<PublicScreenChat> 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<Long> 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<String, Session> clients, String message, String userId) {
        PublicScreenChat publicScreenChat = JSON.parseObject(message, PublicScreenChat.class);
        Set<String> 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));
    }
}