增加 websocket相关代码

service/trade-service-source/trade-service-source-dao/src/main/java/com/hchbox/config/SellerUserHandlerMethodArgumentResolver.java

@@ -24,7 +24,7 @@ public class SellerUserHandlerMethodArgumentResolver implements HandlerMethodArg
     }
 
     @Override
-    public Object resolveArgument(MethodParameter parameter, ModelAndViewContainer container,
+    public Object resolveArgument(MethodParameter methodParameter, ModelAndViewContainer container,
                                   NativeWebRequest request, WebDataBinderFactory factory) {
         // session解析账号
         SellerAccount sellerAccount = (SellerAccount) request.getAttribute(AuthConstants.SELLER_ACCOUNT, DefaultConstants.ZERO_NUMBER);

service/trade-service-source/trade-service-source-taobao/src/main/java/com/hchbox/hander/RateRefundSuccessHandler.java

@@ -1,5 +1,6 @@
 package com.hchbox.hander;
 import com.hchbox.hander.message.TaoBaoRefundMessage;
+import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Service;
 
 /**
@@ -7,6 +8,7 @@ import org.springframework.stereotype.Service;
  *
  * @author t102 2015-8-26 下午3:30:06
  */
+@RequiredArgsConstructor
 @Service("rateRefundSuccessHandler")
 public class RateRefundSuccessHandler  implements TaoBaoHandler<TaoBaoRefundMessage> {
 

service/trade-service-wwsend/pom.xml

@@ -24,6 +24,12 @@
             <artifactId>trade-service-wwsend-task</artifactId>
             <version>1.0-SNAPSHOT</version>
         </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-websocket</artifactId>
+        </dependency>
+
     </dependencies>
 
     <build>

service/trade-service-wwsend/trade-service-wwsend-task/pom.xml

@@ -43,6 +43,10 @@
             <groupId>org.springframework.cloud</groupId>
             <artifactId>spring-cloud-starter-openfeign</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-websocket</artifactId>
+        </dependency>
 
     </dependencies>
 

service/trade-service-wwsend/trade-service-wwsend-task/src/main/java/com/hchbox/config/HttpAuthHandler.java

@@ -0,0 +1,67 @@
+package com.hchbox.config;
+
+import com.hchbox.exception.TaoMiCommException;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import org.springframework.web.socket.CloseStatus;
+import org.springframework.web.socket.TextMessage;
+import org.springframework.web.socket.WebSocketSession;
+import org.springframework.web.socket.handler.TextWebSocketHandler;
+
+/**
+ * @author Enzo
+ * @date : 2021/7/5
+ */
+@Slf4j
+@Component
+public class HttpAuthHandler extends TextWebSocketHandler {
+
+    /**
+     * socket 建立成功事件
+     *
+     * @param session session
+     * @throws Exception
+     */
+    @Override
+    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
+        Object token = session.getAttributes().get("nickname");
+        if (token != null) {
+            // 用户连接成功，放入在线用户缓存
+            WsSessionManager.add(token.toString(), session);
+        } else {
+            throw new TaoMiCommException("用户登录已经失效!");
+        }
+    }
+
+    /**
+     * 接收消息事件
+     *
+     * @param session
+     * @param message
+     * @throws Exception
+     */
+    @Override
+    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
+        super.handleTextMessage(session, message);
+    }
+
+    /**
+     * socket 断开连接时
+     *
+     * @param session
+     * @param status
+     * @throws Exception
+     */
+    @Override
+    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
+        Object sessionKey = session.getAttributes().get("nickname");
+        if (sessionKey != null) {
+            // 用户退出，移除缓存
+            WsSessionManager.remove(sessionKey.toString());
+        }
+    }
+
+
+
+
+}

service/trade-service-wwsend/trade-service-wwsend-task/src/main/java/com/hchbox/config/WebSocketConfig.java

@@ -0,0 +1,20 @@
+package com.hchbox.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.socket.server.standard.ServerEndpointExporter;
+
+/**
+ * @author Enzo
+ * @date : 2021/7/5
+ */
+@Configuration
+public class WebSocketConfig {
+
+    @Bean
+    public ServerEndpointExporter serverEndpointExporter () {
+        return new ServerEndpointExporter();
+    }
+
+
+}

service/trade-service-wwsend/trade-service-wwsend-task/src/main/java/com/hchbox/config/WsSessionManager.java

@@ -0,0 +1,88 @@
+package com.hchbox.config;
+
+import com.hchbox.exception.TaoMiCommException;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.web.socket.TextMessage;
+import org.springframework.web.socket.WebSocketSession;
+
+import java.io.IOException;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * @author Enzo
+ * @date : 2021/7/5
+ */
+@Slf4j
+public class WsSessionManager {
+
+    private WsSessionManager() {
+    }
+
+    /**
+     * 保存连接 session 的地方
+     */
+    private static final ConcurrentHashMap<String, WebSocketSession> SESSION_POOL = new ConcurrentHashMap<>();
+
+    /**
+     * 添加 session
+     *
+     * @param key key
+     */
+    public static void add(String key, WebSocketSession session) {
+        // 添加 session
+        SESSION_POOL.put(key, session);
+    }
+
+    /**
+     * 删除 session,会返回删除的 session
+     *
+     * @param key key
+     * @return 删除
+     */
+    public static WebSocketSession remove(String key) {
+        // 删除 session
+        return SESSION_POOL.remove(key);
+    }
+
+    /**
+     * 删除并同步关闭连接
+     *
+     * @param key key
+     */
+    public static void removeAndClose(String key) {
+        WebSocketSession session = remove(key);
+        if (session != null) {
+            try {
+                // 关闭连接
+                session.close();
+            } catch (IOException e) {
+                log.error("close link exception key as {}", key);
+                throw new TaoMiCommException("the remove session error");
+            }
+        }
+    }
+
+    /**
+     * 获得 session
+     *
+     * @param key key
+     * @return 获取session
+     */
+    public static WebSocketSession get(String key) {
+        // 获得 session
+        return SESSION_POOL.get(key);
+    }
+
+
+    public static void sendMessageToUser(String nickname, TextMessage message) {
+        WebSocketSession session = SESSION_POOL.get(nickname);
+        if (session != null){
+            try {
+                session.sendMessage(message);
+            } catch (IOException e) {
+                log.error("the send message error  nickname as {}",nickname);
+                throw new TaoMiCommException("the send message error");
+            }
+        }
+    }
+}

service/trade-service-wwsend/trade-service-wwsend-task/src/main/java/com/hchbox/socket/WebSocketServer.java

@@ -0,0 +1,130 @@
+package com.hchbox.socket;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import javax.websocket.*;
+import javax.websocket.server.ServerEndpoint;
+import java.io.IOException;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * @author Enzo
+ * @date : 2021/7/5
+ */
+@Slf4j
+@Component
+@ServerEndpoint(value = "/prosperous/")
+public class WebSocketServer {
+
+    @PostConstruct
+    public void init() {
+        log.info("websocket 加载");
+    }
+
+    private static final AtomicInteger ONLINE_COUNT = new AtomicInteger(0);
+    /**
+     * concurrent包的线程安全Set，用来存放每个客户端对应的Session对象。
+     */
+    private static final CopyOnWriteArraySet<Session> SESSIONS = new CopyOnWriteArraySet<>();
+
+
+    /**
+     * 连接建立成功调用的方法
+     */
+    @OnOpen
+    public void onOpen(Session session) {
+        SESSIONS.add(session);
+        // 在线数加1
+        int cnt = ONLINE_COUNT.incrementAndGet();
+        log.info("有连接加入，当前连接数为：{}", cnt);
+        sendMessage(session, "连接成功");
+    }
+
+    /**
+     * 连接关闭调用的方法
+     */
+    @OnClose
+    public void onClose(Session session) {
+        SESSIONS.remove(session);
+        int cnt = ONLINE_COUNT.decrementAndGet();
+        log.info("有连接关闭，当前连接数为：{}", cnt);
+    }
+
+    /**
+     * 收到客户端消息后调用的方法
+     *
+     * @param message 客户端发送过来的消息
+     */
+    @OnMessage
+    public void onMessage(String message, Session session) {
+        log.info("来自客户端的消息：{}", message);
+        sendMessage(session, "收到消息，消息内容：" + message);
+
+    }
+
+    /**
+     * 出现错误
+     *
+     * @param session
+     * @param error
+     */
+    @OnError
+    public void onError(Session session, Throwable error) {
+        log.error("发生错误：{}，Session ID： {}", error.getMessage(), session.getId());
+        error.printStackTrace();
+    }
+
+    /**
+     * 发送消息，实践表明，每次浏览器刷新，session会发生变化。
+     *
+     * @param session
+     * @param message
+     */
+    public static void sendMessage(Session session, String message) {
+        try {
+            session.getBasicRemote().sendText(String.format("%s (From Server，Session ID=%s)", message, session.getId()));
+        } catch (IOException e) {
+            log.error("发送消息出错：{}", e.getMessage());
+            e.printStackTrace();
+        }
+    }
+
+    /**
+     * 群发消息
+     *
+     * @param message
+     * @throws IOException
+     */
+    public static void BroadCastInfo(String message) throws IOException {
+        for (Session session : SESSIONS) {
+            if (session.isOpen()) {
+                sendMessage(session, message);
+            }
+        }
+    }
+
+    /**
+     * 指定Session发送消息
+     *
+     * @param sessionId
+     * @param message
+     * @throws IOException
+     */
+    public static void sendMessage(String message, String sessionId) throws IOException {
+        Session session = null;
+        for (Session s : SESSIONS) {
+            if (s.getId().equals(sessionId)) {
+                session = s;
+                break;
+            }
+        }
+        if (session != null) {
+            sendMessage(session, message);
+        } else {
+            log.warn("没有找到你指定ID的会话：{}", sessionId);
+        }
+    }
+}