From 5a8e4b72b11013f71db78f61241edbfccde0eac1 Mon Sep 17 00:00:00 2001 From: lbw Date: Tue, 12 Dec 2023 18:35:01 +0800 Subject: [PATCH] :sparkles: Introducing new features. support sse --- pigx-common/pigx-common-bom/pom.xml | 5 ++ pigx-common/pigx-common-sse/pom.xml | 33 +++++++++ .../LocalMessageDistributorConfiguration.java | 30 ++++++++ .../MessageDistributorTypeConstants.java | 26 +++++++ .../RedisMessageDistributorConfiguration.java | 69 +++++++++++++++++++ .../config/SseEmitterAutoConfiguration.java | 21 ++++++ .../common/sse/config/SseEmitterEndpoint.java | 45 ++++++++++++ .../sse/config/SseEmitterMessageSender.java | 59 ++++++++++++++++ .../sse/config/SseEmitterProperties.java | 28 ++++++++ .../distribute/LocalMessageDistributor.java | 20 ++++++ .../pigx/common/sse/distribute/MessageDO.java | 39 +++++++++++ .../sse/distribute/MessageDistributor.java | 17 +++++ .../common/sse/distribute/MessageSender.java | 34 +++++++++ .../distribute/RedisMessageDistributor.java | 35 ++++++++++ .../RedisSseEmitterMessageListener.java | 41 +++++++++++ .../common/sse/holder/SseEmitterHolder.java | 65 +++++++++++++++++ ...ot.autoconfigure.AutoConfiguration.imports | 3 + pigx-common/pom.xml | 1 + 18 files changed, 571 insertions(+) create mode 100755 pigx-common/pigx-common-sse/pom.xml create mode 100644 pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/LocalMessageDistributorConfiguration.java create mode 100644 pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/MessageDistributorTypeConstants.java create mode 100644 pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/RedisMessageDistributorConfiguration.java create mode 100644 pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/SseEmitterAutoConfiguration.java create mode 100644 pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/SseEmitterEndpoint.java create mode 100644 pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/SseEmitterMessageSender.java create mode 100644 pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/SseEmitterProperties.java create mode 100644 pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/LocalMessageDistributor.java create mode 100644 pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/MessageDO.java create mode 100644 pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/MessageDistributor.java create mode 100644 pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/MessageSender.java create mode 100644 pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/RedisMessageDistributor.java create mode 100644 pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/RedisSseEmitterMessageListener.java create mode 100644 pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/holder/SseEmitterHolder.java create mode 100644 pigx-common/pigx-common-sse/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports diff --git a/pigx-common/pigx-common-bom/pom.xml b/pigx-common/pigx-common-bom/pom.xml index 2d8d2be11..a1a3622be 100644 --- a/pigx-common/pigx-common-bom/pom.xml +++ b/pigx-common/pigx-common-bom/pom.xml @@ -139,6 +139,11 @@ pigx-common-xss ${pigx.version} + + com.pig4cloud + pigx-common-sse + ${pigx.version} + com.pig4cloud pigx-common-websocket diff --git a/pigx-common/pigx-common-sse/pom.xml b/pigx-common/pigx-common-sse/pom.xml new file mode 100755 index 000000000..3cb211e24 --- /dev/null +++ b/pigx-common/pigx-common-sse/pom.xml @@ -0,0 +1,33 @@ + + + 4.0.0 + + com.pig4cloud + pigx-common + 5.3.0 + + + pigx-common-sse + jar + + pigx sse 服务端推送 + + + + org.springframework.data + spring-data-redis + + + + cn.hutool + hutool-json + + + + com.pig4cloud + pigx-common-security + + + diff --git a/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/LocalMessageDistributorConfiguration.java b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/LocalMessageDistributorConfiguration.java new file mode 100644 index 000000000..0d30ad835 --- /dev/null +++ b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/LocalMessageDistributorConfiguration.java @@ -0,0 +1,30 @@ +package com.pig4cloud.pigx.common.sse.config; + +import com.pig4cloud.pigx.common.sse.distribute.LocalMessageDistributor; +import com.pig4cloud.pigx.common.sse.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 lengleng + */ +@ConditionalOnProperty(prefix = SseEmitterProperties.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-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/MessageDistributorTypeConstants.java b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/MessageDistributorTypeConstants.java new file mode 100644 index 000000000..6ae8cee0b --- /dev/null +++ b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/MessageDistributorTypeConstants.java @@ -0,0 +1,26 @@ +package com.pig4cloud.pigx.common.sse.config; + +/** + * @author lengleng + */ +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-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/RedisMessageDistributorConfiguration.java b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/RedisMessageDistributorConfiguration.java new file mode 100644 index 000000000..72010638d --- /dev/null +++ b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/RedisMessageDistributorConfiguration.java @@ -0,0 +1,69 @@ +package com.pig4cloud.pigx.common.sse.config; + +import com.pig4cloud.pigx.common.sse.distribute.MessageDistributor; +import com.pig4cloud.pigx.common.sse.distribute.RedisMessageDistributor; +import com.pig4cloud.pigx.common.sse.distribute.RedisSseEmitterMessageListener; +import jakarta.annotation.PostConstruct; +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; + +/** + * 基于 Redis Pub/Sub 的消息分发器 + * + * @author lengleng + */ +@ConditionalOnClass(StringRedisTemplate.class) +@ConditionalOnProperty(prefix = SseEmitterProperties.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 RedisSseEmitterMessageListener redisWebsocketMessageListener(StringRedisTemplate stringRedisTemplate) { + return new RedisSseEmitterMessageListener(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 RedisSseEmitterMessageListener redisWebsocketMessageListener; + + @PostConstruct + public void addMessageListener() { + redisMessageListenerContainer.addMessageListener(redisWebsocketMessageListener, + new PatternTopic(RedisSseEmitterMessageListener.CHANNEL)); + } + + } + +} diff --git a/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/SseEmitterAutoConfiguration.java b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/SseEmitterAutoConfiguration.java new file mode 100644 index 000000000..c2fa9cdec --- /dev/null +++ b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/SseEmitterAutoConfiguration.java @@ -0,0 +1,21 @@ +package com.pig4cloud.pigx.common.sse.config; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; + +/** + * websocket自动配置 + * + * @author Yakir + */ +@EnableConfigurationProperties(SseEmitterProperties.class) +public class SseEmitterAutoConfiguration { + + @Bean + @ConditionalOnMissingBean + public SseEmitterEndpoint sseEmitterEndpoint() { + return new SseEmitterEndpoint(); + } + +} diff --git a/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/SseEmitterEndpoint.java b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/SseEmitterEndpoint.java new file mode 100644 index 000000000..b39693c6c --- /dev/null +++ b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/SseEmitterEndpoint.java @@ -0,0 +1,45 @@ +package com.pig4cloud.pigx.common.sse.config; + +import com.pig4cloud.pigx.common.security.util.SecurityUtils; +import com.pig4cloud.pigx.common.sse.holder.SseEmitterHolder; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.util.Objects; + +/** + * @author lengleng + * @date 2023/12/11 + */ +@Slf4j +@RestController +@RequiredArgsConstructor +@RequestMapping("/sse") +public class SseEmitterEndpoint { + + @RequestMapping("/info") + public SseEmitter info() { + SseEmitter sseEmitter = new SseEmitter(0L); + + if (Objects.isNull(SecurityUtils.getUser())) { + log.debug("MSG: SseConnectError"); + return null; + } + log.info("MSG: SseConnect | EmitterHash: {} | ID: {}", sseEmitter.hashCode(), SecurityUtils.getUser().getId()); + sseEmitter.onCompletion(() -> { + log.info("MSG: SseConnectCompletion | EmitterHash: {} |ID: {}", sseEmitter.hashCode(), SecurityUtils.getUser().getId()); + }); + sseEmitter.onTimeout(() -> { + log.error("MSG: SseConnectTimeout | EmitterHash: {} |ID: {} ", sseEmitter.hashCode(), SecurityUtils.getUser().getId()); + }); + sseEmitter.onError(t -> { + log.error("MSG: SseConnectError | EmitterHash: {} |ID: {}", sseEmitter.hashCode(), SecurityUtils.getUser().getId()); + }); + SseEmitterHolder.addSseEmitter(SecurityUtils.getUser().getId(), sseEmitter); + return sseEmitter; + } + +} diff --git a/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/SseEmitterMessageSender.java b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/SseEmitterMessageSender.java new file mode 100644 index 000000000..c3a4b5922 --- /dev/null +++ b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/SseEmitterMessageSender.java @@ -0,0 +1,59 @@ +package com.pig4cloud.pigx.common.sse.config; + +import cn.hutool.core.util.ReflectUtil; +import com.pig4cloud.pigx.common.sse.holder.SseEmitterHolder; +import lombok.extern.slf4j.Slf4j; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.io.IOException; +import java.util.Collection; + +/** + * @author lengleng 2021/1/4 + * @version 1.0 + */ +@Slf4j +public class SseEmitterMessageSender { + + public static void broadcast(String message) { + Collection sseEmitters = SseEmitterHolder.getSseEmitters(); + for (SseEmitter sseEmitter : sseEmitters) { + send(sseEmitter, message); + } + } + + public static boolean send(Object sseEmitterKey, String message) { + SseEmitter sseEmitter = SseEmitterHolder.getSseEmitter(sseEmitterKey); + if (sseEmitter == null) { + log.info("[send] 当前 sseEmitterKey：{} 对应 sseEmitter 不在本服务中", sseEmitterKey); + return false; + } + else { + return send(sseEmitter, message); + } + } + + public static boolean send(SseEmitter sseEmitter, String message) { + if (sseEmitter == null) { + log.error("[send] sseEmitter 为 null"); + return false; + } + + Object sendFailed = ReflectUtil.getFieldValue(sseEmitter, "sendFailed"); + if ((boolean) sendFailed) { + log.error("[send] sseEmitter 已经关闭"); + SseEmitterHolder.removeSseEmitter(sseEmitter); + return false; + } + try { + sseEmitter.send(message); + } + catch (IOException e) { + log.error("[send] sseEmitter({}) 发送消息({}) 异常", sseEmitter, message, e); + SseEmitterHolder.removeSseEmitter(sseEmitter); + return false; + } + return true; + } + +} diff --git a/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/SseEmitterProperties.java b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/SseEmitterProperties.java new file mode 100644 index 000000000..238969416 --- /dev/null +++ b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/config/SseEmitterProperties.java @@ -0,0 +1,28 @@ +package com.pig4cloud.pigx.common.sse.config; + +import lombok.Data; +import org.springframework.boot.context.properties.ConfigurationProperties; + +/** + * websocket props + * + * @author Yakir + */ +@Data +@ConfigurationProperties(SseEmitterProperties.PREFIX) +public class SseEmitterProperties { + + public static final String PREFIX = "pigx.sse"; + + /** + * 路径: 无参: /ws 有参: PathVariable: 单参: /ws/{test} 多参: /ws/{test1}/{test2} query: + * /ws?uid=1&name=test + */ + private String path = "/sse/info"; + + /** + * 消息分发器：local | redis，默认 local, 如果自定义的话，可以配置为其他任意值 + */ + private String messageDistributor = MessageDistributorTypeConstants.LOCAL; + +} diff --git a/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/LocalMessageDistributor.java b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/LocalMessageDistributor.java new file mode 100644 index 000000000..9a7d92867 --- /dev/null +++ b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/LocalMessageDistributor.java @@ -0,0 +1,20 @@ +package com.pig4cloud.pigx.common.sse.distribute; + +/** + * 本地消息分发，直接进行发送 + * + * @author lengleng 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-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/MessageDO.java b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/MessageDO.java new file mode 100644 index 000000000..e1fd87688 --- /dev/null +++ b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/MessageDO.java @@ -0,0 +1,39 @@ +package com.pig4cloud.pigx.common.sse.distribute; + +import lombok.Data; +import lombok.experimental.Accessors; + +import java.util.List; + +/** + * @author lengleng 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-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/MessageDistributor.java b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/MessageDistributor.java new file mode 100644 index 000000000..a9f4981f5 --- /dev/null +++ b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/MessageDistributor.java @@ -0,0 +1,17 @@ +package com.pig4cloud.pigx.common.sse.distribute; + +/** + * 消息分发器 + * + * @author lengleng 2021/1/12 + * @version 1.0 + */ +public interface MessageDistributor { + + /** + * 消息分发 + * @param messageDO 发送的消息 + */ + void distribute(MessageDO messageDO); + +} diff --git a/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/MessageSender.java b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/MessageSender.java new file mode 100644 index 000000000..728f6f50e --- /dev/null +++ b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/MessageSender.java @@ -0,0 +1,34 @@ +package com.pig4cloud.pigx.common.sse.distribute; + +import cn.hutool.core.collection.CollectionUtil; +import com.pig4cloud.pigx.common.sse.config.SseEmitterMessageSender; + +import java.util.List; + +/** + * @author lengleng 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) { + // 广播信息 + SseEmitterMessageSender.broadcast(messageText); + } + else if (CollectionUtil.isNotEmpty(sessionKeys)) { + // 指定用户发送 + for (Object sessionKey : sessionKeys) { + SseEmitterMessageSender.send(sessionKey, messageText); + } + } + } + +} diff --git a/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/RedisMessageDistributor.java b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/RedisMessageDistributor.java new file mode 100644 index 000000000..4e23f3743 --- /dev/null +++ b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/RedisMessageDistributor.java @@ -0,0 +1,35 @@ +package com.pig4cloud.pigx.common.sse.distribute; + +import cn.hutool.json.JSONUtil; +import lombok.RequiredArgsConstructor; +import org.springframework.data.redis.core.StringRedisTemplate; + +import java.util.ArrayList; +import java.util.List; + +/** + * 消息分发器 + * + * @author lengleng 2021/1/12 + * @version 1.0 + */ +@RequiredArgsConstructor +public class RedisMessageDistributor implements MessageDistributor { + + private final StringRedisTemplate stringRedisTemplate; + + /** + * 消息分发 + * @param messageDO 发送的消息 + */ + @Override + public void distribute(MessageDO messageDO) { + // 包装 sessionKey 适配分布式多环境 + List sessionKeyList = new ArrayList<>(messageDO.getSessionKeys()); + messageDO.setSessionKeys(sessionKeyList); + + String str = JSONUtil.toJsonStr(messageDO); + stringRedisTemplate.convertAndSend(RedisSseEmitterMessageListener.CHANNEL, str); + } + +} diff --git a/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/RedisSseEmitterMessageListener.java b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/RedisSseEmitterMessageListener.java new file mode 100644 index 000000000..1da768434 --- /dev/null +++ b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/distribute/RedisSseEmitterMessageListener.java @@ -0,0 +1,41 @@ +package com.pig4cloud.pigx.common.sse.distribute; + +import cn.hutool.json.JSONUtil; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +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 lengleng 2021/1/12 + * @version 1.0 + */ +@Slf4j +@RequiredArgsConstructor +public class RedisSseEmitterMessageListener implements MessageListener, MessageSender { + + public static final String CHANNEL = "sse-send"; + + private final StringRedisTemplate stringRedisTemplate; + + @Override + public void onMessage(Message message, byte[] bytes) { + log.info("redis channel Listener message send {}", message); + 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-sse/src/main/java/com/pig4cloud/pigx/common/sse/holder/SseEmitterHolder.java b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/holder/SseEmitterHolder.java new file mode 100644 index 000000000..67f9c32fc --- /dev/null +++ b/pigx-common/pigx-common-sse/src/main/java/com/pig4cloud/pigx/common/sse/holder/SseEmitterHolder.java @@ -0,0 +1,65 @@ +package com.pig4cloud.pigx.common.sse.holder; + +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.util.Collection; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +/** + * SseEmitter 持有者 主要用于保存当前所有在线的会话信息 + * + * @author lengleng 2021/1/4 + * @version 1.0 + */ +public final class SseEmitterHolder { + + private SseEmitterHolder() { + } + + private static final Map USER_SSE_EMITTER_MAP = new ConcurrentHashMap<>(); + + /** + * 添加一个 sseEmitter + * @param sseEmitterKey sseEmitter 唯一标识 + * @param sseEmitter 待添加的 SseEmitter + */ + public static void addSseEmitter(Object sseEmitterKey, SseEmitter sseEmitter) { + USER_SSE_EMITTER_MAP.put(sseEmitterKey.toString(), sseEmitter); + } + + /** + * 删除一个 sseEmitter + * @param sseEmitterKey sseEmitter唯一标识 + */ + public static void removeSseEmitter(Object sseEmitterKey) { + USER_SSE_EMITTER_MAP.remove(sseEmitterKey.toString()); + } + + /** + * 获取指定标识的 sseEmitter + * @param sseEmitterKey sseEmitter唯一标识 + * @return SseEmitter 该标识对应的 sseEmitter + */ + public static SseEmitter getSseEmitter(Object sseEmitterKey) { + return USER_SSE_EMITTER_MAP.get(sseEmitterKey.toString()); + } + + /** + * 获取当前所有在线的 sseEmitter + * @return Collection sseEmitter集合 + */ + public static Collection getSseEmitters() { + return USER_SSE_EMITTER_MAP.values(); + } + + /** + * 获取所有在线的用户标识 + * @return Set sseEmitter唯一标识集合 + */ + public static Set getSseEmitterKeys() { + return USER_SSE_EMITTER_MAP.keySet(); + } + +} diff --git a/pigx-common/pigx-common-sse/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/pigx-common/pigx-common-sse/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports new file mode 100644 index 000000000..51b0b7f81 --- /dev/null +++ b/pigx-common/pigx-common-sse/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -0,0 +1,3 @@ +com.pig4cloud.pigx.common.sse.config.SseEmitterAutoConfiguration +com.pig4cloud.pigx.common.sse.config.RedisMessageDistributorConfiguration +com.pig4cloud.pigx.common.sse.config.LocalMessageDistributorConfiguration diff --git a/pigx-common/pom.xml b/pigx-common/pom.xml index ec3e296f1..809766cd0 100644 --- a/pigx-common/pom.xml +++ b/pigx-common/pom.xml @@ -52,6 +52,7 @@ pigx-common-sentinel pigx-common-sequence pigx-common-swagger + pigx-common-sse pigx-common-websocket pigx-common-xss