diff --git a/pigx-common/pigx-common-bom/pom.xml b/pigx-common/pigx-common-bom/pom.xml index f69a32bad..e95d4243d 100644 --- a/pigx-common/pigx-common-bom/pom.xml +++ b/pigx-common/pigx-common-bom/pom.xml @@ -120,6 +120,11 @@ pigx-common-xss ${pigx.version} + + com.pig4cloud + pigx-common-websocket + ${pigx.version} + com.pig4cloud pigx-common-excel diff --git a/pigx-common/pigx-common-websocket/pom.xml b/pigx-common/pigx-common-websocket/pom.xml new file mode 100755 index 000000000..a651c211a --- /dev/null +++ b/pigx-common/pigx-common-websocket/pom.xml @@ -0,0 +1,39 @@ + + + 4.0.0 + + com.pig4cloud + pigx-common + 4.1.0 + + + pigx-common-websocket + jar + + pigx websocket + + + + org.springframework.boot + spring-boot-starter-websocket + + + + org.springframework.data + spring-data-redis + + + + cn.hutool + hutool-json + ${hutool.version} + + + + com.pig4cloud + pigx-common-security + + + diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/LocalMessageDistributorConfiguration.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/LocalMessageDistributorConfiguration.java new file mode 100644 index 000000000..baeedc499 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/LocalMessageDistributorConfiguration.java @@ -0,0 +1,30 @@ +package com.pig4cloud.pigx.common.websocket.config; + +import com.pig4cloud.pigx.common.websocket.distribute.LocalMessageDistributor; +import com.pig4cloud.pigx.common.websocket.distribute.MessageDistributor; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * 本地的消息分发器配置 + * + * @author hccake + */ +@ConditionalOnProperty(prefix = WebSocketProperties.PREFIX, name = "message-distributor", + havingValue = MessageDistributorTypeConstants.LOCAL) +@Configuration(proxyBeanMethods = false) +public class LocalMessageDistributorConfiguration { + + /** + * 本地基于内存的消息分发，不支持集群 + * @return LocalMessageDistributor + */ + @Bean + @ConditionalOnMissingBean(MessageDistributor.class) + public LocalMessageDistributor messageDistributor() { + return new LocalMessageDistributor(); + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/MessageDistributorTypeConstants.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/MessageDistributorTypeConstants.java new file mode 100644 index 000000000..9e53ec911 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/MessageDistributorTypeConstants.java @@ -0,0 +1,26 @@ +package com.pig4cloud.pigx.common.websocket.config; + +/** + * @author hccake + */ +public final class MessageDistributorTypeConstants { + + private MessageDistributorTypeConstants() { + } + + /** + * 本地 + */ + public static final String LOCAL = "local"; + + /** + * 基于 Redis PUB/SUB + */ + public static final String REDIS = "redis"; + + /** + * 自定义 + */ + public static final String CUSTOM = "custom"; + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/RedisMessageDistributorConfiguration.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/RedisMessageDistributorConfiguration.java new file mode 100644 index 000000000..dc5f98efd --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/RedisMessageDistributorConfiguration.java @@ -0,0 +1,70 @@ +package com.pig4cloud.pigx.common.websocket.config; + +import com.pig4cloud.pigx.common.websocket.distribute.MessageDistributor; +import com.pig4cloud.pigx.common.websocket.distribute.RedisMessageDistributor; +import com.pig4cloud.pigx.common.websocket.distribute.RedisWebsocketMessageListener; +import lombok.RequiredArgsConstructor; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.listener.PatternTopic; +import org.springframework.data.redis.listener.RedisMessageListenerContainer; + +import javax.annotation.PostConstruct; + +/** + * 基于 Redis Pub/Sub 的消息分发器 + * + * @author hccake + */ +@ConditionalOnClass(StringRedisTemplate.class) +@ConditionalOnProperty(prefix = WebSocketProperties.PREFIX, name = "message-distributor", + havingValue = MessageDistributorTypeConstants.REDIS, matchIfMissing = true) +@Configuration(proxyBeanMethods = false) +public class RedisMessageDistributorConfiguration { + + @Bean + @ConditionalOnMissingBean(MessageDistributor.class) + public RedisMessageDistributor messageDistributor(StringRedisTemplate stringRedisTemplate) { + return new RedisMessageDistributor(stringRedisTemplate); + } + + @Bean + @ConditionalOnBean(RedisMessageDistributor.class) + @ConditionalOnMissingBean + public RedisWebsocketMessageListener redisWebsocketMessageListener(StringRedisTemplate stringRedisTemplate) { + return new RedisWebsocketMessageListener(stringRedisTemplate); + } + + @Bean + @ConditionalOnBean(RedisMessageDistributor.class) + @ConditionalOnMissingBean + public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory) { + RedisMessageListenerContainer container = new RedisMessageListenerContainer(); + container.setConnectionFactory(connectionFactory); + return container; + } + + @Configuration(proxyBeanMethods = false) + @ConditionalOnMissingBean(MessageDistributor.class) + @RequiredArgsConstructor + static class RedisMessageListenerRegisterConfiguration { + + private final RedisMessageListenerContainer redisMessageListenerContainer; + + private final RedisWebsocketMessageListener redisWebsocketMessageListener; + + @PostConstruct + public void addMessageListener() { + redisMessageListenerContainer.addMessageListener(redisWebsocketMessageListener, + new PatternTopic(RedisWebsocketMessageListener.CHANNEL)); + } + + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketAutoConfiguration.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketAutoConfiguration.java new file mode 100644 index 000000000..7f819b32a --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketAutoConfiguration.java @@ -0,0 +1,50 @@ +package com.pig4cloud.pigx.common.websocket.config; + +import com.pig4cloud.pigx.common.websocket.handler.JsonMessageHandler; +import com.pig4cloud.pigx.common.websocket.holder.JsonMessageHandlerHolder; +import lombok.RequiredArgsConstructor; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Import; +import org.springframework.web.socket.WebSocketHandler; +import org.springframework.web.socket.config.annotation.EnableWebSocket; +import org.springframework.web.socket.config.annotation.WebSocketConfigurer; +import org.springframework.web.socket.server.HandshakeInterceptor; + +import javax.annotation.PostConstruct; +import java.util.List; + +/** + * websocket自动配置 + * + * @author Yakir + */ +@Import(WebSocketHandlerConfig.class) +@EnableWebSocket +@RequiredArgsConstructor +public class WebSocketAutoConfiguration { + + private final WebSocketProperties webSocketProperties; + + private final List jsonMessageHandlerList; + + @Bean + @ConditionalOnMissingBean + public WebSocketConfigurer webSocketConfigurer(List handshakeInterceptor, + WebSocketHandler webSocketHandler) { + return registry -> registry.addHandler(webSocketHandler, webSocketProperties.getPath()) + .setAllowedOrigins(webSocketProperties.getAllowOrigins()) + .addInterceptors(handshakeInterceptor.toArray(new HandshakeInterceptor[0])); + } + + /** + * 初始化时将所有的jsonMessageHandler注册到JsonMessageHandlerHolder中 + */ + @PostConstruct + public void initJsonMessageHandlerHolder() { + for (JsonMessageHandler jsonMessageHandler : jsonMessageHandlerList) { + JsonMessageHandlerHolder.addHandler(jsonMessageHandler); + } + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketHandlerConfig.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketHandlerConfig.java new file mode 100644 index 000000000..98f4c755d --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketHandlerConfig.java @@ -0,0 +1,71 @@ +package com.pig4cloud.pigx.common.websocket.config; + +import com.pig4cloud.pigx.common.websocket.custom.PigxSessionKeyGenerator; +import com.pig4cloud.pigx.common.websocket.custom.UserAttributeHandshakeInterceptor; +import com.pig4cloud.pigx.common.websocket.handler.CustomWebSocketHandler; +import com.pig4cloud.pigx.common.websocket.handler.PingJsonMessageHandler; +import com.pig4cloud.pigx.common.websocket.handler.PlanTextMessageHandler; +import com.pig4cloud.pigx.common.websocket.holder.MapSessionWebSocketHandlerDecorator; +import com.pig4cloud.pigx.common.websocket.holder.SessionKeyGenerator; +import lombok.RequiredArgsConstructor; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.web.socket.WebSocketHandler; +import org.springframework.web.socket.handler.TextWebSocketHandler; +import org.springframework.web.socket.server.HandshakeInterceptor; + +/** + * @author Hccake 2021/1/5 + * @version 1.0 + */ +@RequiredArgsConstructor +@EnableConfigurationProperties(WebSocketProperties.class) +public class WebSocketHandlerConfig { + + private final WebSocketProperties webSocketProperties; + + @Bean + @ConditionalOnMissingBean(SessionKeyGenerator.class) + public SessionKeyGenerator sessionKeyGenerator() { + return new PigxSessionKeyGenerator(); + } + + @Bean + public HandshakeInterceptor handshakeInterceptor() { + return new UserAttributeHandshakeInterceptor(); + } + + @Bean + @ConditionalOnMissingBean({ TextWebSocketHandler.class, PlanTextMessageHandler.class }) + public WebSocketHandler webSocketHandler1(@Autowired(required = false) SessionKeyGenerator sessionKeyGenerator) { + CustomWebSocketHandler customWebSocketHandler = new CustomWebSocketHandler(); + if (webSocketProperties.isMapSession()) { + return new MapSessionWebSocketHandlerDecorator(customWebSocketHandler, sessionKeyGenerator); + } + return customWebSocketHandler; + } + + @Bean + @ConditionalOnBean(PlanTextMessageHandler.class) + @ConditionalOnMissingBean(TextWebSocketHandler.class) + public WebSocketHandler webSocketHandler2(@Autowired(required = false) SessionKeyGenerator sessionKeyGenerator, + PlanTextMessageHandler planTextMessageHandler) { + CustomWebSocketHandler customWebSocketHandler = new CustomWebSocketHandler(planTextMessageHandler); + if (webSocketProperties.isMapSession()) { + return new MapSessionWebSocketHandlerDecorator(customWebSocketHandler, sessionKeyGenerator); + } + return customWebSocketHandler; + } + + @Bean + @ConditionalOnProperty(prefix = WebSocketProperties.PREFIX, name = "heartbeat", havingValue = "true", + matchIfMissing = true) + public PingJsonMessageHandler pingJsonMessageHandler() { + return new PingJsonMessageHandler(); + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketMessageSender.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketMessageSender.java new file mode 100644 index 000000000..b8335718b --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketMessageSender.java @@ -0,0 +1,61 @@ +package com.pig4cloud.pigx.common.websocket.config; + +import cn.hutool.json.JSONUtil; +import com.pig4cloud.pigx.common.websocket.holder.WebSocketSessionHolder; +import com.pig4cloud.pigx.common.websocket.message.JsonWebSocketMessage; +import lombok.extern.slf4j.Slf4j; +import org.springframework.web.socket.TextMessage; +import org.springframework.web.socket.WebSocketSession; + +import java.io.IOException; +import java.util.Collection; + +/** + * @author Hccake 2021/1/4 + * @version 1.0 + */ +@Slf4j +public class WebSocketMessageSender { + + public static void broadcast(String message) { + Collection sessions = WebSocketSessionHolder.getSessions(); + for (WebSocketSession session : sessions) { + send(session, message); + } + } + + public static boolean send(Object sessionKey, String message) { + WebSocketSession session = WebSocketSessionHolder.getSession(sessionKey); + if (session == null) { + log.trace("[send] 当前 sessionKey：{} 对应 session 不在本服务中", sessionKey); + return false; + } + else { + return send(session, message); + } + } + + public static void send(WebSocketSession session, JsonWebSocketMessage message) { + send(session, JSONUtil.toJsonStr(message)); + } + + public static boolean send(WebSocketSession session, String message) { + if (session == null) { + log.error("[send] session 为 null"); + return false; + } + if (!session.isOpen()) { + log.error("[send] session 已经关闭"); + return false; + } + try { + session.sendMessage(new TextMessage(message)); + } + catch (IOException e) { + log.error("[send] session({}) 发送消息({}) 异常", session, message, e); + return false; + } + return true; + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketProperties.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketProperties.java new file mode 100644 index 000000000..8bdc84cee --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketProperties.java @@ -0,0 +1,49 @@ +package com.pig4cloud.pigx.common.websocket.config; + +import lombok.Data; +import org.springframework.boot.context.properties.ConfigurationProperties; + +/** + * websocket props + * + * @author Yakir + */ +@Data +@ConfigurationProperties(WebSocketProperties.PREFIX) +public class WebSocketProperties { + + public static final String PREFIX = "pigx.websocket"; + + /** + * 路径: 无参: /ws 有参: PathVariable: 单参: /ws/{test} 多参: /ws/{test1}/{test2} query: + * /ws?uid=1&name=test + * + */ + private String path = "/ws/info"; + + /** + * 允许访问源 + */ + private String allowOrigins = "*"; + + /** + * 是否支持部分消息 + */ + private boolean supportPartialMessages = false; + + /** + * 心跳处理 + */ + private boolean heartbeat = true; + + /** + * 是否开启对session的映射记录 + */ + private boolean mapSession = true; + + /** + * 消息分发器：local | redis，默认 local, 如果自定义的话，可以配置为其他任意值 + */ + private String messageDistributor = MessageDistributorTypeConstants.LOCAL; + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/custom/PigxSessionKeyGenerator.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/custom/PigxSessionKeyGenerator.java new file mode 100644 index 000000000..84c39dce9 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/custom/PigxSessionKeyGenerator.java @@ -0,0 +1,33 @@ +package com.pig4cloud.pigx.common.websocket.custom; + +import com.pig4cloud.pigx.common.security.service.PigxUser; +import com.pig4cloud.pigx.common.websocket.holder.SessionKeyGenerator; +import org.springframework.context.annotation.Configuration; +import org.springframework.web.socket.WebSocketSession; + +/** + * @author lengleng + * @date 2021/10/4 websocket session 标识生成规则 + */ +@Configuration +public class PigxSessionKeyGenerator implements SessionKeyGenerator { + + /** + * 获取当前session的唯一标识 + * @param webSocketSession 当前session + * @return session唯一标识 + */ + @Override + public Object sessionKey(WebSocketSession webSocketSession) { + + Object obj = webSocketSession.getAttributes().get("USER_KEY_ATTR_NAME"); + + if (obj instanceof PigxUser) { + PigxUser user = (PigxUser) obj; + return String.format("%s%s", user.getTenantId(), user.getUsername()); + } + + return null; + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/custom/UserAttributeHandshakeInterceptor.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/custom/UserAttributeHandshakeInterceptor.java new file mode 100644 index 000000000..19c416dfb --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/custom/UserAttributeHandshakeInterceptor.java @@ -0,0 +1,52 @@ +package com.pig4cloud.pigx.common.websocket.custom; + +import com.pig4cloud.pigx.common.security.service.PigxUser; +import com.pig4cloud.pigx.common.security.util.SecurityUtils; +import org.springframework.http.server.ServerHttpRequest; +import org.springframework.http.server.ServerHttpResponse; +import org.springframework.web.socket.WebSocketHandler; +import org.springframework.web.socket.server.HandshakeInterceptor; + +import java.util.Map; + +/** + * @author lengleng + * @date 2021/10/4 + */ +public class UserAttributeHandshakeInterceptor implements HandshakeInterceptor { + + /** + * Invoked before the handshake is processed. + * @param request the current request + * @param response the current response + * @param wsHandler the target WebSocket handler + * @param attributes the attributes from the HTTP handshake to associate with the + * WebSocket session; the provided attributes are copied, the original map is not + * used. + * @return whether to proceed with the handshake ({@code true}) or abort + * ({@code false}) + */ + @Override + public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, + Map attributes) throws Exception { + // 由于 WebSocket 握手是由 http 升级的，携带 token 已经被 Security 拦截验证了，所以可以直接获取到用户 + PigxUser user = SecurityUtils.getUser(); + attributes.put("USER_KEY_ATTR_NAME", user); + return true; + } + + /** + * Invoked after the handshake is done. The response status and headers indicate the + * results of the handshake, i.e. whether it was successful or not. + * @param request the current request + * @param response the current response + * @param wsHandler the target WebSocket handler + * @param exception an exception raised during the handshake, or {@code null} if none + */ + @Override + public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, + Exception exception) { + + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/LocalMessageDistributor.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/LocalMessageDistributor.java new file mode 100644 index 000000000..c5abdfd16 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/LocalMessageDistributor.java @@ -0,0 +1,20 @@ +package com.pig4cloud.pigx.common.websocket.distribute; + +/** + * 本地消息分发，直接进行发送 + * + * @author Hccake 2021/1/12 + * @version 1.0 + */ +public class LocalMessageDistributor implements MessageDistributor, MessageSender { + + /** + * 消息分发 + * @param messageDO 发送的消息 + */ + @Override + public void distribute(MessageDO messageDO) { + doSend(messageDO); + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/MessageDO.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/MessageDO.java new file mode 100644 index 000000000..5f9601bd9 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/MessageDO.java @@ -0,0 +1,39 @@ +package com.pig4cloud.pigx.common.websocket.distribute; + +import lombok.Data; +import lombok.experimental.Accessors; + +import java.util.List; + +/** + * @author Hccake 2021/1/12 + * @version 1.0 + */ +@Data +@Accessors(chain = true) +public class MessageDO { + + /** + * 是否广播 + */ + private Boolean needBroadcast; + + /** + * sessionKeys + */ + private List sessionKeys; + + /** + * 需要发送的消息文本 + */ + private String messageText; + + /** + * 构建需要广播的message + * @author lingting 2021-03-25 17:28 + */ + public static MessageDO broadcastMessage(String text) { + return new MessageDO().setMessageText(text).setNeedBroadcast(true); + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/MessageDistributor.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/MessageDistributor.java new file mode 100644 index 000000000..0c7ccc693 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/MessageDistributor.java @@ -0,0 +1,17 @@ +package com.pig4cloud.pigx.common.websocket.distribute; + +/** + * 消息分发器 + * + * @author Hccake 2021/1/12 + * @version 1.0 + */ +public interface MessageDistributor { + + /** + * 消息分发 + * @param messageDO 发送的消息 + */ + void distribute(MessageDO messageDO); + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/MessageSender.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/MessageSender.java new file mode 100644 index 000000000..f963f0352 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/MessageSender.java @@ -0,0 +1,34 @@ +package com.pig4cloud.pigx.common.websocket.distribute; + +import cn.hutool.core.collection.CollectionUtil; +import com.pig4cloud.pigx.common.websocket.config.WebSocketMessageSender; + +import java.util.List; + +/** + * @author Hccake 2021/1/12 + * @version 1.0 + */ +public interface MessageSender { + + /** + * 发送消息 + * @param messageDO 发送的消息 + */ + default void doSend(MessageDO messageDO) { + Boolean needBroadcast = messageDO.getNeedBroadcast(); + String messageText = messageDO.getMessageText(); + List sessionKeys = messageDO.getSessionKeys(); + if (needBroadcast != null && needBroadcast) { + // 广播信息 + WebSocketMessageSender.broadcast(messageText); + } + else if (CollectionUtil.isNotEmpty(sessionKeys)) { + // 指定用户发送 + for (Object sessionKey : sessionKeys) { + WebSocketMessageSender.send(sessionKey, messageText); + } + } + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/RedisMessageDistributor.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/RedisMessageDistributor.java new file mode 100644 index 000000000..380d9acab --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/RedisMessageDistributor.java @@ -0,0 +1,28 @@ +package com.pig4cloud.pigx.common.websocket.distribute; + +import cn.hutool.json.JSONUtil; +import lombok.RequiredArgsConstructor; +import org.springframework.data.redis.core.StringRedisTemplate; + +/** + * 消息分发器 + * + * @author Hccake 2021/1/12 + * @version 1.0 + */ +@RequiredArgsConstructor +public class RedisMessageDistributor implements MessageDistributor { + + private final StringRedisTemplate stringRedisTemplate; + + /** + * 消息分发 + * @param messageDO 发送的消息 + */ + @Override + public void distribute(MessageDO messageDO) { + String str = JSONUtil.toJsonStr(messageDO); + stringRedisTemplate.convertAndSend(RedisWebsocketMessageListener.CHANNEL, str); + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/RedisWebsocketMessageListener.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/RedisWebsocketMessageListener.java new file mode 100644 index 000000000..227907075 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/RedisWebsocketMessageListener.java @@ -0,0 +1,38 @@ +package com.pig4cloud.pigx.common.websocket.distribute; + +import cn.hutool.json.JSONUtil; +import lombok.RequiredArgsConstructor; +import org.springframework.data.redis.connection.Message; +import org.springframework.data.redis.connection.MessageListener; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.serializer.RedisSerializer; + +/** + * redis订阅 websocket 发送消息，接收到消息时进行推送 + * + * @author Hccake 2021/1/12 + * @version 1.0 + */ +@RequiredArgsConstructor +public class RedisWebsocketMessageListener implements MessageListener, MessageSender { + + public static final String CHANNEL = "websocket-send"; + + private final StringRedisTemplate stringRedisTemplate; + + @Override + public void onMessage(Message message, byte[] bytes) { + byte[] channelBytes = message.getChannel(); + RedisSerializer stringSerializer = stringRedisTemplate.getStringSerializer(); + String channel = stringSerializer.deserialize(channelBytes); + + // 这里没有使用通配符，所以一定是true + if (CHANNEL.equals(channel)) { + byte[] bodyBytes = message.getBody(); + String body = stringSerializer.deserialize(bodyBytes); + MessageDO messageDO = JSONUtil.toBean(body, MessageDO.class); + doSend(messageDO); + } + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/CustomWebSocketHandler.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/CustomWebSocketHandler.java new file mode 100644 index 000000000..6748a22b9 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/CustomWebSocketHandler.java @@ -0,0 +1,73 @@ +package com.pig4cloud.pigx.common.websocket.handler; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.core.json.JsonReadFeature; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.pig4cloud.pigx.common.websocket.holder.JsonMessageHandlerHolder; +import com.pig4cloud.pigx.common.websocket.message.AbstractJsonWebSocketMessage; +import com.pig4cloud.pigx.common.websocket.message.JsonWebSocketMessage; +import lombok.extern.slf4j.Slf4j; +import org.springframework.web.socket.TextMessage; +import org.springframework.web.socket.WebSocketSession; +import org.springframework.web.socket.handler.TextWebSocketHandler; + +/** + * @author Hccake 2020/12/31 + * @version 1.0 + */ +@Slf4j +public class CustomWebSocketHandler extends TextWebSocketHandler { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + static { + // 有特殊需要转义字符, 不报错 + MAPPER.enable(JsonReadFeature.ALLOW_UNESCAPED_CONTROL_CHARS.mappedFeature()); + } + + private PlanTextMessageHandler planTextMessageHandler; + + public CustomWebSocketHandler() { + } + + public CustomWebSocketHandler(PlanTextMessageHandler planTextMessageHandler) { + this.planTextMessageHandler = planTextMessageHandler; + } + + @Override + public void handleTextMessage(WebSocketSession session, TextMessage message) throws JsonProcessingException { + // 空消息不处理 + if (message.getPayloadLength() == 0) { + return; + } + + // 消息类型必有一属性type，先解析，获取该属性 + String payload = message.getPayload(); + JsonNode jsonNode = MAPPER.readTree(payload); + JsonNode typeNode = jsonNode.get(AbstractJsonWebSocketMessage.TYPE_FIELD); + + if (typeNode == null) { + if (planTextMessageHandler != null) { + planTextMessageHandler.handle(session, payload); + } + else { + log.error("[handleTextMessage] 普通文本消息（{}）没有对应的消息处理器", payload); + } + } + else { + String messageType = typeNode.asText(); + // 获得对应的消息处理器 + JsonMessageHandler jsonMessageHandler = JsonMessageHandlerHolder.getHandler(messageType); + if (jsonMessageHandler == null) { + log.error("[handleTextMessage] 消息类型（{}）不存在对应的消息处理器", messageType); + return; + } + // 消息处理 + Class messageClass = jsonMessageHandler.getMessageClass(); + JsonWebSocketMessage websocketMessageJson = MAPPER.treeToValue(jsonNode, messageClass); + jsonMessageHandler.handle(session, websocketMessageJson); + } + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/JsonMessageHandler.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/JsonMessageHandler.java new file mode 100644 index 000000000..87b3e4945 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/JsonMessageHandler.java @@ -0,0 +1,31 @@ +package com.pig4cloud.pigx.common.websocket.handler; + +import com.pig4cloud.pigx.common.websocket.message.JsonWebSocketMessage; +import org.springframework.web.socket.WebSocketSession; + +/** + * @author Hccake 2021/1/4 + * @version 1.0 + */ +public interface JsonMessageHandler { + + /** + * JsonWebSocketMessage 类型消息处理 + * @param session 当前接收 session + * @param message 当前接收到的 message + */ + void handle(WebSocketSession session, T message); + + /** + * 当前处理器处理的消息类型 + * @return messageType + */ + String type(); + + /** + * 当前处理器对应的消息Class + * @return Class + */ + Class getMessageClass(); + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/PingJsonMessageHandler.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/PingJsonMessageHandler.java new file mode 100644 index 000000000..aea06792b --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/PingJsonMessageHandler.java @@ -0,0 +1,34 @@ +package com.pig4cloud.pigx.common.websocket.handler; + +import com.pig4cloud.pigx.common.websocket.config.WebSocketMessageSender; +import com.pig4cloud.pigx.common.websocket.message.JsonWebSocketMessage; +import com.pig4cloud.pigx.common.websocket.message.PingJsonWebSocketMessage; +import com.pig4cloud.pigx.common.websocket.message.PongJsonWebSocketMessage; +import com.pig4cloud.pigx.common.websocket.message.WebSocketMessageTypeEnum; +import org.springframework.web.socket.WebSocketSession; + +/** + * 心跳处理，接收到客户端的ping时，立刻回复一个pong + * + * @author Hccake 2021/1/4 + * @version 1.0 + */ +public class PingJsonMessageHandler implements JsonMessageHandler { + + @Override + public void handle(WebSocketSession session, PingJsonWebSocketMessage message) { + JsonWebSocketMessage pongJsonWebSocketMessage = new PongJsonWebSocketMessage(); + WebSocketMessageSender.send(session, pongJsonWebSocketMessage); + } + + @Override + public String type() { + return WebSocketMessageTypeEnum.PING.getValue(); + } + + @Override + public Class getMessageClass() { + return PingJsonWebSocketMessage.class; + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/PlanTextMessageHandler.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/PlanTextMessageHandler.java new file mode 100644 index 000000000..12267d196 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/handler/PlanTextMessageHandler.java @@ -0,0 +1,21 @@ +package com.pig4cloud.pigx.common.websocket.handler; + +import org.springframework.web.socket.WebSocketSession; + +/** + * 普通文本类型（非指定json类型）的消息处理器 即消息不满足于我们定义的Json类型消息时，所使用的处理器 + * + * @see JsonWebSocketMessage + * @author Hccake 2021/1/5 + * @version 1.0 + */ +public interface PlanTextMessageHandler { + + /** + * 普通文本消息处理 + * @param session 当前接收消息的session + * @param message 文本消息 + */ + void handle(WebSocketSession session, String message); + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/JsonMessageHandlerHolder.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/JsonMessageHandlerHolder.java new file mode 100644 index 000000000..d19ea85bf --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/JsonMessageHandlerHolder.java @@ -0,0 +1,27 @@ +package com.pig4cloud.pigx.common.websocket.holder; + +import com.pig4cloud.pigx.common.websocket.handler.JsonMessageHandler; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +/** + * @author Hccake 2021/1/4 + * @version 1.0 + */ +public final class JsonMessageHandlerHolder { + + private JsonMessageHandlerHolder() { + } + + private static final Map MESSAGE_HANDLER_MAP = new ConcurrentHashMap<>(); + + public static JsonMessageHandler getHandler(String type) { + return MESSAGE_HANDLER_MAP.get(type); + } + + public static void addHandler(JsonMessageHandler jsonMessageHandler) { + MESSAGE_HANDLER_MAP.put(jsonMessageHandler.type(), jsonMessageHandler); + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/MapSessionWebSocketHandlerDecorator.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/MapSessionWebSocketHandlerDecorator.java new file mode 100644 index 000000000..74c3ac43d --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/MapSessionWebSocketHandlerDecorator.java @@ -0,0 +1,46 @@ +package com.pig4cloud.pigx.common.websocket.holder; + +import org.springframework.web.socket.CloseStatus; +import org.springframework.web.socket.WebSocketHandler; +import org.springframework.web.socket.WebSocketSession; +import org.springframework.web.socket.handler.WebSocketHandlerDecorator; + +/** + * WebSocketHandler 装饰器，该装饰器主要用于在开启和关闭连接时，进行session的映射存储与释放 + * + * @author Hccake 2020/12/31 + * @version 1.0 + */ +public class MapSessionWebSocketHandlerDecorator extends WebSocketHandlerDecorator { + + private final SessionKeyGenerator sessionKeyGenerator; + + public MapSessionWebSocketHandlerDecorator(WebSocketHandler delegate, SessionKeyGenerator sessionKeyGenerator) { + super(delegate); + this.sessionKeyGenerator = sessionKeyGenerator; + } + + /** + * websocket 连接时执行的动作 + * @param session websocket session 对象 + * @throws Exception 异常对象 + */ + @Override + public void afterConnectionEstablished(final WebSocketSession session) throws Exception { + Object sessionKey = sessionKeyGenerator.sessionKey(session); + WebSocketSessionHolder.addSession(sessionKey, session); + } + + /** + * websocket 关闭连接时执行的动作 + * @param session websocket session 对象 + * @param closeStatus 关闭状态对象 + * @throws Exception 异常对象 + */ + @Override + public void afterConnectionClosed(final WebSocketSession session, CloseStatus closeStatus) throws Exception { + Object sessionKey = sessionKeyGenerator.sessionKey(session); + WebSocketSessionHolder.removeSession(sessionKey); + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/SessionKeyGenerator.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/SessionKeyGenerator.java new file mode 100644 index 000000000..bccb5dd3c --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/SessionKeyGenerator.java @@ -0,0 +1,20 @@ +package com.pig4cloud.pigx.common.websocket.holder; + +import org.springframework.web.socket.WebSocketSession; + +/** + * WebSocketSession 唯一标识生成器 + * + * @author Hccake 2021/1/5 + * @version 1.0 + */ +public interface SessionKeyGenerator { + + /** + * 获取当前session的唯一标识 + * @param webSocketSession 当前session + * @return session唯一标识 + */ + Object sessionKey(WebSocketSession webSocketSession); + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/WebSocketSessionHolder.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/WebSocketSessionHolder.java new file mode 100644 index 000000000..e06cc3c17 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/holder/WebSocketSessionHolder.java @@ -0,0 +1,65 @@ +package com.pig4cloud.pigx.common.websocket.holder; + +import org.springframework.web.socket.WebSocketSession; + +import java.util.Collection; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +/** + * WebSocketSession 持有者 主要用于保存当前所有在线的会话信息 + * + * @author Hccake 2021/1/4 + * @version 1.0 + */ +public final class WebSocketSessionHolder { + + private WebSocketSessionHolder() { + } + + private static final Map USER_SESSION_MAP = new ConcurrentHashMap<>(); + + /** + * 添加一个 session + * @param sessionKey session 唯一标识 + * @param session 待添加的 WebSocketSession + */ + public static void addSession(Object sessionKey, WebSocketSession session) { + USER_SESSION_MAP.put(sessionKey, session); + } + + /** + * 删除一个 session + * @param sessionKey session唯一标识 + */ + public static void removeSession(Object sessionKey) { + USER_SESSION_MAP.remove(sessionKey); + } + + /** + * 获取指定标识的 session + * @param sessionKey session唯一标识 + * @return WebSocketSession 该标识对应的 session + */ + public static WebSocketSession getSession(Object sessionKey) { + return USER_SESSION_MAP.get(sessionKey); + } + + /** + * 获取当前所有在线的 session + * @return Collection session集合 + */ + public static Collection getSessions() { + return USER_SESSION_MAP.values(); + } + + /** + * 获取所有在线的用户标识 + * @return Set session唯一标识集合 + */ + public static Set getSessionKeys() { + return USER_SESSION_MAP.keySet(); + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/message/AbstractJsonWebSocketMessage.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/message/AbstractJsonWebSocketMessage.java new file mode 100644 index 000000000..25b340859 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/message/AbstractJsonWebSocketMessage.java @@ -0,0 +1,22 @@ +package com.pig4cloud.pigx.common.websocket.message; + +/** + * @author Hccake 2021/1/4 + * @version 1.0 + */ +public abstract class AbstractJsonWebSocketMessage implements JsonWebSocketMessage { + + public static final String TYPE_FIELD = "type"; + + private final String type; + + protected AbstractJsonWebSocketMessage(String type) { + this.type = type; + } + + @Override + public String getType() { + return type; + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/message/JsonWebSocketMessage.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/message/JsonWebSocketMessage.java new file mode 100644 index 000000000..4cfdc951f --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/message/JsonWebSocketMessage.java @@ -0,0 +1,15 @@ +package com.pig4cloud.pigx.common.websocket.message; + +/** + * @author Hccake 2021/1/4 + * @version 1.0 + */ +public interface JsonWebSocketMessage { + + /** + * 消息类型，主要用于匹配对应的消息处理器 + * @return 当前消息类型 + */ + String getType(); + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/message/PingJsonWebSocketMessage.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/message/PingJsonWebSocketMessage.java new file mode 100644 index 000000000..1faf12689 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/message/PingJsonWebSocketMessage.java @@ -0,0 +1,13 @@ +package com.pig4cloud.pigx.common.websocket.message; + +/** + * @author Hccake 2021/1/4 + * @version 1.0 + */ +public class PingJsonWebSocketMessage extends AbstractJsonWebSocketMessage { + + public PingJsonWebSocketMessage() { + super(WebSocketMessageTypeEnum.PING.getValue()); + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/message/PongJsonWebSocketMessage.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/message/PongJsonWebSocketMessage.java new file mode 100644 index 000000000..02f5c3432 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/message/PongJsonWebSocketMessage.java @@ -0,0 +1,13 @@ +package com.pig4cloud.pigx.common.websocket.message; + +/** + * @author Hccake 2021/1/4 + * @version 1.0 + */ +public class PongJsonWebSocketMessage extends AbstractJsonWebSocketMessage { + + public PongJsonWebSocketMessage() { + super(WebSocketMessageTypeEnum.PONG.getValue()); + } + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/message/WebSocketMessageTypeEnum.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/message/WebSocketMessageTypeEnum.java new file mode 100644 index 000000000..7e4604afc --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/message/WebSocketMessageTypeEnum.java @@ -0,0 +1,18 @@ +package com.pig4cloud.pigx.common.websocket.message; + +import lombok.Getter; +import lombok.RequiredArgsConstructor; + +/** + * @author Hccake 2021/1/4 + * @version 1.0 + */ +@Getter +@RequiredArgsConstructor +public enum WebSocketMessageTypeEnum { + + PING("ping"), PONG("pong"); + + private final String value; + +} diff --git a/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/package-info.java b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/package-info.java new file mode 100644 index 000000000..e8d61df69 --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/package-info.java @@ -0,0 +1,6 @@ +/** + * 此包内代码 来源至 BallCat + * 关于BallCat: 组织旨在为项目快速开发提供一系列的基础能力，方便使用者根据项目需求快速进行功能拓展。 + * https://github.com/ballcat-projects/ballcat + */ +package com.pig4cloud.pigx.common.websocket; diff --git a/pigx-common/pigx-common-websocket/src/main/resources/META-INF/spring.factories b/pigx-common/pigx-common-websocket/src/main/resources/META-INF/spring.factories new file mode 100644 index 000000000..ddaf06a0c --- /dev/null +++ b/pigx-common/pigx-common-websocket/src/main/resources/META-INF/spring.factories @@ -0,0 +1,4 @@ +org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ + com.pig4cloud.pigx.common.websocket.config.WebSocketAutoConfiguration,\ + com.pig4cloud.pigx.common.websocket.config.RedisMessageDistributorConfiguration,\ + com.pig4cloud.pigx.common.websocket.config.LocalMessageDistributorConfiguration diff --git a/pigx-common/pom.xml b/pigx-common/pom.xml index df9180090..fda6ef0ca 100644 --- a/pigx-common/pom.xml +++ b/pigx-common/pom.xml @@ -50,6 +50,7 @@ pigx-common-swagger pigx-common-test pigx-common-transaction + pigx-common-websocket pigx-common-xss diff --git a/pigx-visual/pigx-oa-platform/pom.xml b/pigx-visual/pigx-oa-platform/pom.xml index 929c285d2..e43a941eb 100644 --- a/pigx-visual/pigx-oa-platform/pom.xml +++ b/pigx-visual/pigx-oa-platform/pom.xml @@ -76,10 +76,10 @@ com.alibaba druid-spring-boot-starter - + - org.springframework.boot - spring-boot-starter-websocket + com.pig4cloud + pigx-common-websocket diff --git a/pigx-visual/pigx-oa-platform/src/main/java/com/pig4cloud/pigx/act/config/WebSocketConfig.java b/pigx-visual/pigx-oa-platform/src/main/java/com/pig4cloud/pigx/act/config/WebSocketConfig.java deleted file mode 100644 index 2f669918c..000000000 --- a/pigx-visual/pigx-oa-platform/src/main/java/com/pig4cloud/pigx/act/config/WebSocketConfig.java +++ /dev/null @@ -1,85 +0,0 @@ -package com.pig4cloud.pigx.act.config; - -import cn.hutool.core.util.ObjectUtil; -import cn.hutool.core.util.StrUtil; -import com.pig4cloud.pigx.common.core.constant.CommonConstants; -import com.pig4cloud.pigx.common.data.tenant.TenantContextHolder; -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; -import org.springframework.context.annotation.Configuration; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.simp.config.ChannelRegistration; -import org.springframework.messaging.simp.stomp.StompCommand; -import org.springframework.messaging.simp.stomp.StompHeaderAccessor; -import org.springframework.messaging.support.ChannelInterceptor; -import org.springframework.messaging.support.MessageHeaderAccessor; -import org.springframework.security.core.context.SecurityContextHolder; -import org.springframework.security.oauth2.provider.OAuth2Authentication; -import org.springframework.security.oauth2.provider.token.ResourceServerTokenServices; -import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker; -import org.springframework.web.socket.config.annotation.StompEndpointRegistry; -import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer; - -/** - * @author new - *

- * WebSocket配置类 - */ -@Slf4j -@Configuration -@RequiredArgsConstructor -@EnableWebSocketMessageBroker -public class WebSocketConfig implements WebSocketMessageBrokerConfigurer { - - private final ResourceServerTokenServices tokenService; - - @Override - public void registerStompEndpoints(StompEndpointRegistry registry) { - registry.addEndpoint("/ws").setAllowedOriginPatterns("*").withSockJS(); - } - - @Override - public void configureClientInboundChannel(ChannelRegistration registration) { - registration.interceptors(new ChannelInterceptor() { - @Override - public Message preSend(Message message, MessageChannel channel) { - StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class); - - // 判断是否首次连接请求 - if (StompCommand.CONNECT.equals(accessor.getCommand())) { - - // 处理租户 - String tenant = accessor.getFirstNativeHeader(CommonConstants.TENANT_ID); - if (StrUtil.isNotBlank(tenant)) { - TenantContextHolder.setTenantId(Integer.parseInt(tenant)); - - } - else { - TenantContextHolder.setTenantId(CommonConstants.TENANT_ID_1); - } - - // 验证令牌信息 - String tokens = accessor.getFirstNativeHeader("Authorization"); - log.info("webSocket token is {}", tokens); - if (StrUtil.isBlank(tokens)) { - return null; - } - - OAuth2Authentication auth2Authentication = tokenService.loadAuthentication(tokens.split(" ")[1]); - if (ObjectUtil.isNotNull(auth2Authentication)) { - SecurityContextHolder.getContext().setAuthentication(auth2Authentication); - accessor.setUser(auth2Authentication::getName); - return message; - } - else { - return null; - } - } - // 不是首次连接，已经成功登陆 - return message; - } - }); - } - -} diff --git a/pigx-visual/pigx-oa-platform/src/main/java/com/pig4cloud/pigx/act/listener/LeaveProcessTaskListener.java b/pigx-visual/pigx-oa-platform/src/main/java/com/pig4cloud/pigx/act/listener/LeaveProcessTaskListener.java index 1729c30b1..e54a115d9 100644 --- a/pigx-visual/pigx-oa-platform/src/main/java/com/pig4cloud/pigx/act/listener/LeaveProcessTaskListener.java +++ b/pigx-visual/pigx-oa-platform/src/main/java/com/pig4cloud/pigx/act/listener/LeaveProcessTaskListener.java @@ -17,10 +17,6 @@ package com.pig4cloud.pigx.act.listener; -import java.util.ArrayList; -import java.util.List; -import java.util.stream.Collectors; - import cn.hutool.core.collection.CollUtil; import com.pig4cloud.pigx.admin.api.entity.SysUser; import com.pig4cloud.pigx.admin.api.feign.RemoteUserService; @@ -28,10 +24,15 @@ import com.pig4cloud.pigx.common.core.util.R; import com.pig4cloud.pigx.common.core.util.SpringContextHolder; import com.pig4cloud.pigx.common.data.tenant.TenantContextHolder; import com.pig4cloud.pigx.common.security.util.SecurityUtils; +import com.pig4cloud.pigx.common.websocket.distribute.MessageDO; +import com.pig4cloud.pigx.common.websocket.distribute.RedisMessageDistributor; import lombok.extern.slf4j.Slf4j; import org.activiti.engine.delegate.DelegateTask; import org.activiti.engine.delegate.TaskListener; -import org.springframework.messaging.simp.SimpMessagingTemplate; + +import java.util.ArrayList; +import java.util.List; +import java.util.stream.Collectors; /** * @author lengleng @@ -46,7 +47,7 @@ public class LeaveProcessTaskListener implements TaskListener { */ @Override public void notify(DelegateTask delegateTask) { - SimpMessagingTemplate simpMessagingTemplate = SpringContextHolder.getBean(SimpMessagingTemplate.class); + RedisMessageDistributor messageDistributor = SpringContextHolder.getBean(RedisMessageDistributor.class); RemoteUserService userService = SpringContextHolder.getBean(RemoteUserService.class); R> result = userService.ancestorUsers(SecurityUtils.getUser().getUsername()); @@ -64,13 +65,13 @@ public class LeaveProcessTaskListener implements TaskListener { remindUserList.addAll(userList); } - remindUserList.forEach(user -> { - - // 订阅通道 /task/租户ID/用户名称/remind - String target = String.format("/task/%s/%s/remind", TenantContextHolder.getTenantId(), - SecurityUtils.getUser().getUsername()); - simpMessagingTemplate.convertAndSend(target, delegateTask.getName()); - }); + // websocket 发送消息 + List sessionKey = remindUserList.stream().map(s -> TenantContextHolder.getTenantId() + s) + .collect(Collectors.toList()); + MessageDO messageDO = new MessageDO(); + messageDO.setSessionKeys(sessionKey); + messageDO.setMessageText(String.format("协同办公 %s 的任务需要您处理", delegateTask.getName())); + messageDistributor.distribute(messageDO); } }