package com.fs.websocket.service;

import com.alibaba.fastjson.JSONObject;
import com.fs.common.exception.base.BaseException;
import com.fs.common.utils.StringUtils;
import com.fs.websocket.bean.SendMsgVO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.*;
import java.util.stream.Collectors;

@ServerEndpoint("/app/webSocket/{userId}")
@Component
@Slf4j
public class WebSocketServer {

    //concurrent包的线程安全，用来存放每个客户端对应的WebSocketServer的会话对象
    private static final ConcurrentHashMap<Long, Session> sessionPools = new ConcurrentHashMap<>();


    //分发消息
    public void sendMessageToAll(String message) throws IOException {
//        Collection<Session> sessions = sessionPools.values();
//        if(!CollectionUtils.isEmpty(sessions)){
            sessionPools.forEach((userId, session) -> {
                if (session.isOpen()) {
                    try {
                        // 异步发送，设置超时
                        Future<Void> future = session.getAsyncRemote().sendText(message);
                        System.out.println("分发消息，数据内容：" + message);
                        try {
                            future.get(10, TimeUnit.SECONDS);
                        } catch (TimeoutException e) {
                            // 超时关闭连接
                            session.close();
                            log.error("超时关闭连接，并移除用户：{}", userId, e);
                            sessionPools.remove(userId);
                        }
                    } catch (Exception e) {
                        log.error("分发消息失败，并移除用户: {}", userId, e);
                        sessionPools.remove(userId);
                    }
                }
            });
        log.info("分发消息结束，人数，{}",  sessionPools.size());
//        }
    }

    //指定用户发送消息
    public static void sendMessage(Session session, String message) throws IOException {
        if(session != null){
            synchronized (session) {
                log.info("发送心跳数据：{}", message);
                session.getBasicRemote().sendText(message);
            }
        }
    }

    //建立连接成功调用
    @OnOpen
    public void onOpen(Session session,  @PathParam(value = "userId") Long userId) {
        Map<String, String> params = getParams(session);
        if(userId == null || !params.containsKey("userId")){
            throw new BaseException("用户信息错误");
        }
        sessionPools.put(userId, session);
        System.out.println("已成功加入webSocket连接！当前人数为，"+ sessionPools.size());
        log.info("{}已成功加入webSocket连接！当前人数为，{}", userId, sessionPools.size());
    }


    //收到客户端信息
    @OnMessage
    public void onMessage(String message) throws IOException {
        SendMsgVO msg = JSONObject.parseObject(message, SendMsgVO.class);
        Session session;
        try {
            switch (msg.getCmd()) {
                case "heartbeat":
                    session = sessionPools.get(msg.getUserId());
                    sendMessage(session, JSONObject.toJSONString(msg));
                    break;
                case "sendMsg":
                    // 分发消息
                    sendMessageToAll(JSONObject.toJSONString(msg));
                    break;
            }
        } catch (Exception e) {
            log.error("webSocket消息发送错误，{}", e.getMessage());
        }

    }

    //关闭连接时调用
    @OnClose
    public void onClose(Session session) {
        Map<String, String> params = getParams(session);
        long userId = Long.parseLong(params.get("userId"));
        sessionPools.remove(userId);
        log.info("{}已断开webSocket连接！当前人数为，{}", userId, sessionPools.size());
    }

    //错误时调用
    @OnError
    public void onError(Session session, Throwable throwable) {
        Map<String, String> params = getParams(session);
        long userId = Long.parseLong(params.get("userId"));
        log.error("webSocket连接错误，{}，移除用户，{}", throwable.getMessage(), userId);
        throwable.printStackTrace();
    }

    private Map<String, String> getParams(Session session){
        return session.getRequestParameterMap().entrySet().stream()
                .filter(e -> e.getValue() != null && !e.getValue().isEmpty() && StringUtils.isNotEmpty(e.getValue().get(0)))
                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().get(0)));
    }
}
