✨ Introducing new features. support sse

pigx-common/pigx-common-bom/pom.xml

@@ -139,6 +139,11 @@
                 <artifactId>pigx-common-xss</artifactId>
                 <version>${pigx.version}</version>
             </dependency>
+            <dependency>
+                <groupId>com.pig4cloud</groupId>
+                <artifactId>pigx-common-sse</artifactId>
+                <version>${pigx.version}</version>
+            </dependency>
             <dependency>
                 <groupId>com.pig4cloud</groupId>
                 <artifactId>pigx-common-websocket</artifactId>

pigx-common/pigx-common-sse/pom.xml

@@ -0,0 +1,33 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xmlns="http://maven.apache.org/POM/4.0.0"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>com.pig4cloud</groupId>
+		<artifactId>pigx-common</artifactId>
+		<version>5.3.0</version>
+	</parent>
+
+	<artifactId>pigx-common-sse</artifactId>
+	<packaging>jar</packaging>
+
+	<description>pigx sse 服务端推送</description>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.data</groupId>
+			<artifactId>spring-data-redis</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>cn.hutool</groupId>
+			<artifactId>hutool-json</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>com.pig4cloud</groupId>
+			<artifactId>pigx-common-security</artifactId>
+		</dependency>
+	</dependencies>
+</project>

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();
+	}
+
+}

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";
+
+}

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));
+		}
+
+	}
+
+}

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();
+	}
+
+}

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;
+    }
+
+}

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<SseEmitter> 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;
+	}
+
+}

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;
+
+}

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);
+	}
+
+}

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<Object> 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);
+	}
+
+}

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);
+
+}

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<Object> sessionKeys = messageDO.getSessionKeys();
+		if (needBroadcast != null && needBroadcast) {
+			// 广播信息
+			SseEmitterMessageSender.broadcast(messageText);
+		}
+		else if (CollectionUtil.isNotEmpty(sessionKeys)) {
+			// 指定用户发送
+			for (Object sessionKey : sessionKeys) {
+				SseEmitterMessageSender.send(sessionKey, messageText);
+			}
+		}
+	}
+
+}

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<Object> sessionKeyList = new ArrayList<>(messageDO.getSessionKeys());
+		messageDO.setSessionKeys(sessionKeyList);
+
+		String str = JSONUtil.toJsonStr(messageDO);
+		stringRedisTemplate.convertAndSend(RedisSseEmitterMessageListener.CHANNEL, str);
+	}
+
+}

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<String> 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);
+		}
+	}
+
+}

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<String, SseEmitter> 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> sseEmitter集合
+	 */
+	public static Collection<SseEmitter> getSseEmitters() {
+		return USER_SSE_EMITTER_MAP.values();
+	}
+
+	/**
+	 * 获取所有在线的用户标识
+	 * @return Set<Object> sseEmitter唯一标识集合
+	 */
+	public static Set<String> getSseEmitterKeys() {
+		return USER_SSE_EMITTER_MAP.keySet();
+	}
+
+}

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

pigx-common/pom.xml

@@ -52,6 +52,7 @@
 		<module>pigx-common-sentinel</module>
 		<module>pigx-common-sequence</module>
 		<module>pigx-common-swagger</module>
+		<module>pigx-common-sse</module>
 		<module>pigx-common-websocket</module>
 		<module>pigx-common-xss</module>
 	</modules>