同步官方futuer/boot4最新代码

ruoyi-admin/src/main/java/org/dromara/web/controller/AuthController.java

@@ -12,9 +12,12 @@ import me.zhyd.oauth.request.AuthRequest;
 import me.zhyd.oauth.utils.AuthStateUtils;
 import org.dromara.common.core.constant.SystemConstants;
 import org.dromara.common.core.domain.R;
+import org.dromara.common.core.domain.dto.PushPayload;
 import org.dromara.common.core.domain.model.LoginBody;
 import org.dromara.common.core.domain.model.RegisterBody;
 import org.dromara.common.core.domain.model.SocialLoginBody;
+import org.dromara.common.core.enums.PushSourceEnum;
+import org.dromara.common.core.enums.PushTypeEnum;
 import org.dromara.common.core.utils.DateUtils;
 import org.dromara.common.core.utils.MessageUtils;
 import org.dromara.common.core.utils.StringUtils;
@@ -23,12 +26,11 @@ import org.dromara.common.encrypt.annotation.ApiEncrypt;
 import org.dromara.common.json.utils.JsonUtils;
 import org.dromara.common.redis.annotation.RateLimiter;
 import org.dromara.common.redis.enums.LimitType;
+import org.dromara.common.push.helper.PushHelper;
 import org.dromara.common.satoken.utils.LoginHelper;
 import org.dromara.common.social.config.properties.SocialLoginConfigProperties;
 import org.dromara.common.social.config.properties.SocialProperties;
 import org.dromara.common.social.utils.SocialUtils;
-import org.dromara.common.sse.dto.SseMessageDTO;
-import org.dromara.common.sse.utils.SseMessageUtils;
 import org.dromara.system.domain.vo.SysClientVo;
 import org.dromara.system.service.ISysClientService;
 import org.dromara.system.service.ISysConfigService;
@@ -93,10 +95,15 @@ public class AuthController {
 
         Long userId = LoginHelper.getUserId();
         scheduledExecutorService.schedule(() -> {
-            SseMessageDTO dto = new SseMessageDTO();
+            PushHelper.publishMessage(
-            dto.setUserIds(List.of(userId));
+                List.of(userId),
-            dto.setMessage(DateUtils.getTodayHour(new Date()) + "好，欢迎登录 RuoYi-Vue-Plus 后台管理系统");
+                PushPayload.of(
-            SseMessageUtils.publishMessage(dto);
+                    PushTypeEnum.MESSAGE,
+                    PushSourceEnum.BACKEND,
+                    DateUtils.getTodayHour(new Date()) + "好，欢迎登录 RuoYi-Vue-Plus 后台管理系统",
+                    null
+                )
+            );
         }, 5, TimeUnit.SECONDS);
         return R.ok(loginVo);
     }

ruoyi-admin/src/main/java/org/dromara/web/controller/CaptchaController.java

@@ -61,7 +61,7 @@ public class CaptchaController {
     @RateLimiter(key = "#phonenumber", time = 60, count = 1)
     @GetMapping("/resource/sms/code")
     public R<Void> smsCode(@NotBlank(message = "{user.phonenumber.not.blank}") String phoneNumber) {
-        if (RegexValidator.isMobile(phoneNumber)) {
+        if (!RegexValidator.isMobile(phoneNumber)) {
             return R.fail("请输入正确的手机号！");
         }
         String key = GlobalConstants.CAPTCHA_CODE_KEY + phoneNumber;
@@ -91,7 +91,7 @@ public class CaptchaController {
         if (!mailProperties.getEnabled()) {
             return R.fail("当前系统没有开启邮箱功能！");
         }
-        if (RegexValidator.isEmail(email)) {
+        if (!RegexValidator.isEmail(email)) {
             return R.fail("请输入正确的邮箱地址！");
         }
         SpringUtils.getAopProxy(this).emailCodeImpl(email);
@@ -168,8 +168,8 @@ public class CaptchaController {
      * 图片验证码响应对象。
      *
      * @param captchaEnabled 是否启用验证码
-     * @param uuid 验证码标识
+     * @param uuid           验证码标识
-     * @param img Base64 图片数据
+     * @param img            Base64 图片数据
      */
     public record CaptchaVo(Boolean captchaEnabled, String uuid, String img) {
     }

ruoyi-admin/src/main/resources/application-dev.yml

@@ -1,7 +1,7 @@
 --- # 监控中心配置
 spring.boot.admin.client:
   # 增加客户端开关
-  enabled: true
+  enabled: false
   url: http://localhost:9090/admin
   instance:
     service-host-type: IP
@@ -13,7 +13,7 @@ spring.boot.admin.client:
 
 --- # snail-job 配置
 snail-job:
-  enabled: true
+  enabled: false
   # 需要在 SnailJob 后台组管理创建对应名称的组,然后创建任务的时候选择对应的组,才能正确分派任务
   group: "ruoyi_group"
   # SnailJob 接入验证令牌 详见 script/sql/ry_job.sql `sj_group_config` 表
@@ -102,7 +102,7 @@ spring.data:
     # 数据库索引
     database: 0
     # redis 密码必须配置
-    password: ruoyi123
+#    password: ruoyi123
     # 连接超时时间
     timeout: 10s
     # 是否开启ssl

ruoyi-admin/src/main/resources/application.yml

@@ -213,18 +213,14 @@ management:
     logfile:
       external-file: ./logs/sys-console.log
 
---- # 默认/推荐使用sse推送
+--- # 统一消息推送配置
-sse:
+message:
   enabled: true
-  path: /resource/sse
+  # sse / websocket
-
+  transport: sse
---- # websocket
+  # 统一访问路径
-websocket:
+  path: /resource/message
-  # 如果关闭 需要和前端开关一起关闭
+  # websocket 允许的跨域来源
-  enabled: false
-  # 路径
-  path: /resource/websocket
-  # 设置访问源地址
   allowedOrigins: '*'
 
 --- # warm-flow工作流配置

ruoyi-common/pom.xml

@@ -35,8 +35,7 @@
         <module>ruoyi-common-sensitive</module>
         <module>ruoyi-common-json</module>
         <module>ruoyi-common-encrypt</module>
-        <module>ruoyi-common-websocket</module>
+        <module>ruoyi-common-push</module>
-        <module>ruoyi-common-sse</module>
         <module>ruoyi-common-mqtt</module>
     </modules>
 

ruoyi-common/ruoyi-common-bom/pom.xml

@@ -144,17 +144,10 @@
                 <version>${revision}</version>
             </dependency>
 
-            <!-- WebSocket模块 -->
+            <!-- 消息推送模块 -->
             <dependency>
                 <groupId>org.dromara</groupId>
-                <artifactId>ruoyi-common-websocket</artifactId>
+                <artifactId>ruoyi-common-push</artifactId>
-                <version>${revision}</version>
-            </dependency>
-
-            <!-- SSE模块 -->
-            <dependency>
-                <groupId>org.dromara</groupId>
-                <artifactId>ruoyi-common-sse</artifactId>
                 <version>${revision}</version>
             </dependency>
 

ruoyi-common/ruoyi-common-core/src/main/java/org/dromara/common/core/domain/dto/PushPayload.java

@@ -0,0 +1,76 @@
+package org.dromara.common.core.domain.dto;
+
+import lombok.Data;
+import org.dromara.common.core.enums.PushSourceEnum;
+import org.dromara.common.core.enums.PushTypeEnum;
+import org.dromara.common.core.utils.StringUtils;
+
+import java.io.Serial;
+import java.io.Serializable;
+
+/**
+ * 推送给前端的统一消息体
+ *
+ * @author Lion Li
+ */
+@Data
+public class PushPayload implements Serializable {
+
+    @Serial
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * 消息类型
+     */
+    private String type;
+
+    /**
+     * 消息来源
+     */
+    private String source;
+
+    /**
+     * 文本消息
+     */
+    private String message;
+
+    /**
+     * 扩展数据
+     */
+    private Object data;
+
+    /**
+     * 前端跳转路径
+     */
+    private String path;
+
+    /**
+     * 时间戳
+     */
+    private Long timestamp;
+
+    public static PushPayload of(String type, String source, String message, Object data) {
+        PushPayload payload = new PushPayload();
+        payload.setType(StringUtils.defaultIfBlank(type, PushTypeEnum.MESSAGE.getType()));
+        payload.setSource(StringUtils.defaultIfBlank(source, PushSourceEnum.BACKEND.getSource()));
+        payload.setMessage(message);
+        payload.setData(data);
+        payload.setTimestamp(System.currentTimeMillis());
+        return payload;
+    }
+
+    public static PushPayload of(PushTypeEnum type, PushSourceEnum source, String message, Object data) {
+        return of(
+            type == null ? null : type.getType(),
+            source == null ? null : source.getSource(),
+            message,
+            data
+        );
+    }
+
+    public static PushPayload of(PushTypeEnum type, PushSourceEnum source, String message, Object data, String path) {
+        PushPayload payload = of(type, source, message, data);
+        payload.setPath(path);
+        return payload;
+    }
+}

ruoyi-common/ruoyi-common-core/src/main/java/org/dromara/common/core/enums/PushSourceEnum.java

@@ -0,0 +1,41 @@
+package org.dromara.common.core.enums;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+/**
+ * 推送消息来源枚举
+ *
+ * @author Lion Li
+ */
+@Getter
+@AllArgsConstructor
+public enum PushSourceEnum {
+
+    /**
+     * 后端系统消息
+     */
+    BACKEND("backend"),
+
+    /**
+     * 通知公告
+     */
+    NOTICE("notice"),
+
+    /**
+     * 工作流
+     */
+    WORKFLOW("workflow"),
+
+    /**
+     * 大模型
+     */
+    LLM("llm"),
+
+    /**
+     * 客户端消息
+     */
+    CLIENT("client");
+
+    private final String source;
+}

ruoyi-common/ruoyi-common-core/src/main/java/org/dromara/common/core/enums/PushTypeEnum.java

@@ -0,0 +1,36 @@
+package org.dromara.common.core.enums;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+/**
+ * 推送消息类型枚举
+ *
+ * @author Lion Li
+ */
+@Getter
+@AllArgsConstructor
+public enum PushTypeEnum {
+
+    /**
+     * 通用消息
+     */
+    MESSAGE("message"),
+
+    /**
+     * 通知公告
+     */
+    NOTICE("notice"),
+
+    /**
+     * 大模型消息
+     */
+    LLM("llm"),
+
+    /**
+     * 自定义消息
+     */
+    CUSTOM("custom");
+
+    private final String type;
+}

ruoyi-common/ruoyi-common-mybatis/src/main/java/com/github/yulichang/wrapper/ext/Ext.java

@@ -135,42 +135,6 @@ public interface Ext<Children extends MPJLambdaWrapper<?>> extends IExt<Children
         return getChildren();
     }
 
-
-    /**
-     * 如果值存在才进行的时间范围查询
-     *
-     * @param timeFunc 时间查询字段
-     * @param start    开始时间
-     * @param end      结束时间
-     */
-    default <T> Children betweenIfExists(SFunction<T, ?> timeFunc, Object start, Object end) {
-        if (ObjectUtil.isNotNull(start) && ObjectUtil.isNotNull(end)) {
-            getChildren().between(timeFunc, start, end);
-            return getChildren();
-        }
-        if (ObjectUtil.isNotNull(start)) {
-            getChildren().ge(timeFunc, start);
-            return getChildren();
-        }
-        if (ObjectUtil.isNotNull(end)) {
-            getChildren().le(timeFunc, end);
-            return getChildren();
-        }
-        return getChildren();
-    }
-
-    /**
-     * 时间范围查询
-     *
-     * @param column 时间查询字段
-     * @param values 时间范围
-     */
-    default <T> Children betweenIfExists(SFunction<T, ?> column, Object[] values) {
-        Object val1 = ArrayUtils.get(values, 0);
-        Object val2 = ArrayUtils.get(values, 1);
-        return betweenIfExists(column, val1, val2);
-    }
-
     /**
      * 当传入的集合不为空时才进行in查询
      *

ruoyi-common/ruoyi-common-oss/src/main/java/org/dromara/common/oss/client/AbstractOssClientImpl.java

@@ -2,14 +2,16 @@ package org.dromara.common.oss.client;
 
 import cn.hutool.core.lang.Assert;
 import cn.hutool.core.util.IdUtil;
-import lombok.extern.slf4j.Slf4j;
 import org.dromara.common.core.utils.StringUtils;
 import org.dromara.common.oss.config.OssClientConfig;
 import org.dromara.common.oss.exception.S3StorageException;
 import org.dromara.common.oss.io.OutputStreamDownloadSubscriber;
 import org.dromara.common.oss.model.GetObjectResult;
 import org.dromara.common.oss.model.HandleAsyncResult;
+import org.dromara.common.oss.model.Options;
 import org.dromara.common.oss.model.PutObjectResult;
+import org.jspecify.annotations.NullMarked;
+import software.amazon.awssdk.core.ResponseInputStream;
 import software.amazon.awssdk.core.async.AsyncRequestBody;
 import software.amazon.awssdk.core.async.AsyncResponseTransformer;
 import software.amazon.awssdk.core.async.ResponsePublisher;
@@ -23,13 +25,13 @@ import software.amazon.awssdk.transfer.s3.progress.TransferListener;
 
 import java.io.*;
 import java.nio.channels.Channels;
-import java.nio.channels.FileChannel;
 import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.SeekableByteChannel;
 import java.nio.channels.WritableByteChannel;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.time.Duration;
-import java.time.LocalDateTime;
+import java.time.ZoneOffset;
 import java.util.Collection;
 import java.util.Map;
 import java.util.Objects;
@@ -45,7 +47,6 @@ import java.util.function.Function;
  *
  * @author 秋辞未寒
  */
-@Slf4j
 public abstract class AbstractOssClientImpl implements OssClient {
 
     private final AtomicBoolean initialized = new AtomicBoolean(false);
@@ -203,20 +204,32 @@ public abstract class AbstractOssClientImpl implements OssClient {
     }
 
     @Override
-    public PutObjectResult bucketUpload(String bucket, String key, Path path) {
+    public PutObjectResult bucketUpload(String bucket, String key, Path path, Options options) {
         AsyncRequestBody body = AsyncRequestBody.fromFile(path);
-        return bucketUpload(bucket, key, body);
+        return bucketUpload(bucket, key, body, options);
+    }
+
+    @Override
+    public PutObjectResult bucketUpload(String bucket, String key, Path path) {
+        return bucketUpload(bucket, key, path, Options.builder());
+    }
+
+    @Override
+    public PutObjectResult bucketUpload(String bucket, String key, File file, Options options) {
+        AsyncRequestBody body = AsyncRequestBody.fromFile(file);
+        return bucketUpload(bucket, key, body, options);
     }
 
     @Override
     public PutObjectResult bucketUpload(String bucket, String key, File file) {
-        AsyncRequestBody body = AsyncRequestBody.fromFile(file);
+        return bucketUpload(bucket, key, file, Options.builder());
-        return bucketUpload(bucket, key, body);
     }
 
     @Override
-    public PutObjectResult bucketUpload(String bucket, String key, RandomAccessFile file) {
+    public PutObjectResult bucketUpload(String bucket, String key, RandomAccessFile file, Options options) {
         try {
+            // 以文件的大小为准
+            options.setLength(file.length());
             return bucketUpload(bucket, key, file.getChannel(), -1L);
         } catch (Exception e) {
             if (e instanceof S3StorageException ex) {
@@ -227,13 +240,21 @@ public abstract class AbstractOssClientImpl implements OssClient {
     }
 
     @Override
-    public PutObjectResult bucketUpload(String bucket, String key, ReadableByteChannel channel, long contentLength) {
+    public PutObjectResult bucketUpload(String bucket, String key, RandomAccessFile file) {
-        long size = contentLength;
+        return bucketUpload(bucket, key, file, Options.builder());
-        try (channel; InputStream in = Channels.newInputStream(channel)) {
+    }
-            if (channel instanceof FileChannel fileChannel) {
+
-                size = fileChannel.size();
+    @Override
+    public PutObjectResult bucketUpload(String bucket, String key, ReadableByteChannel channel, long contentLength, Options options) {
+        // 让调用者自行处理通道的关闭
+        InputStream in = Channels.newInputStream(channel);
+        try {
+            // 如果可以实时获取文件大小，则优先是有实时获取的
+            long size = contentLength;
+            if (channel instanceof SeekableByteChannel byteChannel) {
+                size = byteChannel.size();
             }
-            return bucketUpload(bucket, key, in, size);
+            return bucketUpload(bucket, key, in, size, options);
         } catch (Exception e) {
             if (e instanceof S3StorageException ex) {
                 throw ex;
@@ -242,16 +263,27 @@ public abstract class AbstractOssClientImpl implements OssClient {
         }
     }
 
+    @Override
+    public PutObjectResult bucketUpload(String bucket, String key, ReadableByteChannel channel, long contentLength) {
+        return bucketUpload(bucket, key, channel, contentLength, Options.builder());
+    }
+
+    @Override
+    public PutObjectResult bucketUpload(String bucket, String key, InputStream in, long contentLength, Options options) {
+        options.setLength(contentLength);
+        AsyncRequestBody body = AsyncRequestBody.fromInputStream(in, contentLength, asyncExecutor);
+        return bucketUpload(bucket, key, body, options);
+    }
+
     @Override
     public PutObjectResult bucketUpload(String bucket, String key, InputStream in, long contentLength) {
-        AsyncRequestBody body = AsyncRequestBody.fromInputStream(in, contentLength, asyncExecutor);
+        return bucketUpload(bucket, key, in, contentLength, Options.builder());
-        return bucketUpload(bucket, key, body);
     }
 
     @Override
-    public PutObjectResult bucketUpload(String bucket, String key, byte[] data) {
+    public PutObjectResult bucketUpload(String bucket, String key, byte[] data, Options options) {
         try (ByteArrayInputStream in = new ByteArrayInputStream(data)) {
-            return bucketUpload(bucket, key, in, data.length);
+            return bucketUpload(bucket, key, in, data.length, options);
         } catch (Exception e) {
             if (e instanceof S3StorageException ex) {
                 throw ex;
@@ -260,15 +292,28 @@ public abstract class AbstractOssClientImpl implements OssClient {
         }
     }
 
+    @Override
+    public PutObjectResult bucketUpload(String bucket, String key, byte[] data) {
+        return bucketUpload(bucket, key, data, Options.builder());
+    }
 
-    private PutObjectResult bucketUpload(String bucket, String key, AsyncRequestBody body) {
+    @NullMarked
-        Long contentLength = body.contentLength().orElse(null);
+    private PutObjectResult bucketUpload(String bucket, String key, AsyncRequestBody body, Options options) {
+        // 优先使用body中的内容大小，如果不存在，再获取可选项中的
+        Long contentLength = body.contentLength().orElse(options.getLength());
+        // 优先使用body中的内容类型，如果不存在，再获取可选项中的
+        String contentType = StringUtils.isBlank(options.getContentType()) ? body.contentType() : options.getContentType();
+        String md5Digest = options.getMd5Digest();
+        Map<String, String> metadata = options.getMetadata();
+        Collection<TransferListener> transferListeners = options.getTransferListeners();
         HandleAsyncResult<PutObjectResponse> result = doCustomUpload(body, builder -> {
             builder.bucket(bucket)
                 .key(key)
+                .contentMD5(md5Digest)
+                .contentType(contentType)
                 .contentLength(contentLength)
-            ;
+                .metadata(metadata);
-        });
+        }, transferListeners);
         if (result.isFailure()) {
             throw S3StorageException.form(result.error());
         }
@@ -277,11 +322,13 @@ public abstract class AbstractOssClientImpl implements OssClient {
             throw S3StorageException.form("response is empty.");
         }
         PutObjectResponse response = opt.get();
-        String bucketUrl = config.getBucketUrl(bucket);
         // 不知道什么原因导致 response.size() 返回了一个 null size ，此处做一个适配...
         Long size = response.size();
-        size = size == null ? contentLength : size;
+        if (size == null) {
-        return PutObjectResult.form("%s/%s".formatted(bucketUrl, key), key, response.eTag(), size == null ? 0 : size);
+            size = contentLength == null ? 0 : contentLength;
+        }
+        String bucketUrl = config.getBucketUrl(bucket);
+        return PutObjectResult.form("%s/%s".formatted(bucketUrl, key), key, response.eTag(), size);
     }
 
     @Override
@@ -309,7 +356,7 @@ public abstract class AbstractOssClientImpl implements OssClient {
         try {
             ResponsePublisher<GetObjectResponse> publisher = doCustomDownload(builder -> builder.bucket(bucket).key(key), AsyncResponseTransformer.toPublisher(), null);
             GetObjectResult getObjectResult = buildGetObjectResult(key, publisher.response());
-            publisher.subscribe(downloadSubscriber);
+            publisher.subscribe(downloadSubscriber).join();
             return getObjectResult;
         } catch (Exception e) {
             if (e instanceof S3StorageException ex) {
@@ -319,6 +366,21 @@ public abstract class AbstractOssClientImpl implements OssClient {
         }
     }
 
+    @Override
+    public <T> T bucketDownload(String bucket, String key, BiFunction<GetObjectResult, InputStream, T> downloadTransformer) {
+        try {
+            ResponseInputStream<GetObjectResponse> responseInputStream = doCustomDownload(builder -> builder.bucket(bucket).key(key), AsyncResponseTransformer.toBlockingInputStream(), null);
+            GetObjectResponse response = responseInputStream.response();
+            GetObjectResult getObjectResult = buildGetObjectResult(key, response);
+            return downloadTransformer.apply(getObjectResult, responseInputStream);
+        } catch (Exception e) {
+            if (e instanceof S3StorageException ex) {
+                throw ex;
+            }
+            throw S3StorageException.form(e);
+        }
+    }
+
     @Override
     public GetObjectResult bucketDownload(String bucket, String key, Path path) {
         try (OutputStream out = Files.newOutputStream(path)) {
@@ -362,7 +424,7 @@ public abstract class AbstractOssClientImpl implements OssClient {
         return GetObjectResult.form(
             key,
             response.eTag(),
-            LocalDateTime.from(response.lastModified()),
+            response.lastModified().atOffset(ZoneOffset.UTC).toLocalDateTime(),
             response.contentLength(),
             response.contentType(),
             response.contentDisposition(),
@@ -411,31 +473,61 @@ public abstract class AbstractOssClientImpl implements OssClient {
         }
     }
 
+    @Override
+    public PutObjectResult upload(String key, Path path, Options options) {
+        return bucketUpload(defaultBucket(), key, path, options);
+    }
+
     @Override
     public PutObjectResult upload(String key, Path path) {
         return bucketUpload(defaultBucket(), key, path);
     }
 
+    @Override
+    public PutObjectResult upload(String key, File file, Options options) {
+        return bucketUpload(defaultBucket(), key, file, options);
+    }
+
     @Override
     public PutObjectResult upload(String key, File file) {
         return bucketUpload(defaultBucket(), key, file);
     }
 
+    @Override
+    public PutObjectResult upload(String key, RandomAccessFile file, Options options) {
+        return bucketUpload(defaultBucket(), key, file, options);
+    }
+
     @Override
     public PutObjectResult upload(String key, RandomAccessFile file) {
         return bucketUpload(defaultBucket(), key, file);
     }
 
+    @Override
+    public PutObjectResult upload(String key, ReadableByteChannel channel, long contentLength, Options options) {
+        return bucketUpload(defaultBucket(), key, channel, contentLength, options);
+    }
+
     @Override
     public PutObjectResult upload(String key, ReadableByteChannel channel, long contentLength) {
         return bucketUpload(defaultBucket(), key, channel, contentLength);
     }
 
+    @Override
+    public PutObjectResult upload(String key, InputStream in, long contentLength, Options options) {
+        return bucketUpload(defaultBucket(), key, in, contentLength, options);
+    }
+
     @Override
     public PutObjectResult upload(String key, InputStream in, long contentLength) {
         return bucketUpload(defaultBucket(), key, in, contentLength);
     }
 
+    @Override
+    public PutObjectResult upload(String key, byte[] data, Options options) {
+        return bucketUpload(defaultBucket(), key, data, options);
+    }
+
     @Override
     public PutObjectResult upload(String key, byte[] data) {
         return bucketUpload(defaultBucket(), key, data);
@@ -446,6 +538,11 @@ public abstract class AbstractOssClientImpl implements OssClient {
         return bucketDownload(defaultBucket(), key, downloadSubscriber);
     }
 
+    @Override
+    public <T> T download(String key, BiFunction<GetObjectResult, InputStream, T> downloadTransformer) {
+        return bucketDownload(defaultBucket(), key, downloadTransformer);
+    }
+
     @Override
     public GetObjectResult download(String key, Path path) {
         return bucketDownload(defaultBucket(), key, path);

ruoyi-common/ruoyi-common-oss/src/main/java/org/dromara/common/oss/client/OssClient.java

@@ -5,6 +5,7 @@ import org.dromara.common.oss.config.OssClientConfig;
 import org.dromara.common.oss.io.OutputStreamDownloadSubscriber;
 import org.dromara.common.oss.model.GetObjectResult;
 import org.dromara.common.oss.model.HandleAsyncResult;
+import org.dromara.common.oss.model.Options;
 import org.dromara.common.oss.model.PutObjectResult;
 import software.amazon.awssdk.core.async.AsyncRequestBody;
 import software.amazon.awssdk.core.async.AsyncResponseTransformer;
@@ -43,12 +44,12 @@ public interface OssClient extends AutoCloseable {
 
     /**
      * S3 存储客户端ID
-     *
+     * <p>
      * 用于标识客户端，初始化后不允许更改
      *
      * @return S3 存储客户端ID
      */
-    default String clientId(){
+    default String clientId() {
         return IdUtil.fastSimpleUUID();
     }
 
@@ -136,6 +137,17 @@ public interface OssClient extends AutoCloseable {
      */
     HandleAsyncResult<PutObjectResponse> doCustomUpload(AsyncRequestBody body, Consumer<PutObjectRequest.Builder> putObjectRequestBuilderConsumer);
 
+    /**
+     * 将本地路径对应的文件上传到指定存储桶。
+     *
+     * @param bucket 存储桶名称
+     * @param key    对象键
+     * @param path   文件路径
+     * @param options 可选项
+     * @return 上传结果
+     */
+    PutObjectResult bucketUpload(String bucket, String key, Path path, Options options);
+
     /**
      * 将本地路径对应的文件上传到指定存储桶。
      *
@@ -146,6 +158,17 @@ public interface OssClient extends AutoCloseable {
      */
     PutObjectResult bucketUpload(String bucket, String key, Path path);
 
+    /**
+     * 将文件上传到指定存储桶。
+     *
+     * @param bucket 存储桶名称
+     * @param key    对象键
+     * @param file   文件对象
+     * @param options 可选项
+     * @return 上传结果
+     */
+    PutObjectResult bucketUpload(String bucket, String key, File file, Options options);
+
     /**
      * 将文件上传到指定存储桶。
      *
@@ -156,6 +179,17 @@ public interface OssClient extends AutoCloseable {
      */
     PutObjectResult bucketUpload(String bucket, String key, File file);
 
+    /**
+     * 将随机访问文件上传到指定存储桶。
+     *
+     * @param bucket 存储桶名称
+     * @param key    对象键
+     * @param file   随机访问文件
+     * @param options 可选项
+     * @return 上传结果
+     */
+    PutObjectResult bucketUpload(String bucket, String key, RandomAccessFile file, Options options);
+
     /**
      * 将随机访问文件上传到指定存储桶。
      *
@@ -166,6 +200,18 @@ public interface OssClient extends AutoCloseable {
      */
     PutObjectResult bucketUpload(String bucket, String key, RandomAccessFile file);
 
+    /**
+     * 将可读通道中的数据上传到指定存储桶。
+     *
+     * @param bucket        存储桶名称
+     * @param key           对象键
+     * @param channel       数据通道
+     * @param contentLength 内容长度
+     * @param options 可选项
+     * @return 上传结果
+     */
+    PutObjectResult bucketUpload(String bucket, String key, ReadableByteChannel channel, long contentLength, Options options);
+
     /**
      * 将可读通道中的数据上传到指定存储桶。
      *
@@ -177,6 +223,18 @@ public interface OssClient extends AutoCloseable {
      */
     PutObjectResult bucketUpload(String bucket, String key, ReadableByteChannel channel, long contentLength);
 
+    /**
+     * 将可读通道中的数据上传到指定存储桶。
+     *
+     * @param bucket        存储桶名称
+     * @param key           对象键
+     * @param in            输入流
+     * @param contentLength 内容长度
+     * @param options 可选项
+     * @return 上传结果
+     */
+    PutObjectResult bucketUpload(String bucket, String key, InputStream in, long contentLength, Options options);
+
     /**
      * 将输入流中的数据上传到指定存储桶。
      *
@@ -188,6 +246,17 @@ public interface OssClient extends AutoCloseable {
      */
     PutObjectResult bucketUpload(String bucket, String key, InputStream in, long contentLength);
 
+    /**
+     * 将字节数组上传到指定存储桶。
+     *
+     * @param bucket 存储桶名称
+     * @param key    对象键
+     * @param data   字节数组
+     * @param options 可选项
+     * @return 上传结果
+     */
+    PutObjectResult bucketUpload(String bucket, String key, byte[] data, Options options);
+
     /**
      * 将字节数组上传到指定存储桶。
      *
@@ -219,6 +288,16 @@ public interface OssClient extends AutoCloseable {
      */
     GetObjectResult bucketDownload(String bucket, String key, OutputStreamDownloadSubscriber downloadSubscriber);
 
+    /**
+     * 将指定存储桶中的对象下载到转换器中，由使用者决定返回值。
+     *
+     * @param bucket              存储桶名称
+     * @param key                 对象键
+     * @param downloadTransformer 下载转换器
+     * @return 下载结果
+     */
+    <T> T bucketDownload(String bucket, String key, BiFunction<GetObjectResult, InputStream, T> downloadTransformer);
+
     /**
      * 将指定存储桶中的对象下载到本地路径。
      *
@@ -299,6 +378,16 @@ public interface OssClient extends AutoCloseable {
      */
     String bucketPresignPutUrl(String bucket, String key, Duration expiredTime, Map<String, String> metadata);
 
+    /**
+     * 将本地路径对应的文件上传到默认存储桶。
+     *
+     * @param key  对象键
+     * @param path 文件路径
+     * @param options 可选项
+     * @return 上传结果
+     */
+    PutObjectResult upload(String key, Path path, Options options);
+
     /**
      * 将本地路径对应的文件上传到默认存储桶。
      *
@@ -308,6 +397,16 @@ public interface OssClient extends AutoCloseable {
      */
     PutObjectResult upload(String key, Path path);
 
+    /**
+     * 将文件上传到默认存储桶。
+     *
+     * @param key  对象键
+     * @param file 文件对象
+     * @param options 可选项
+     * @return 上传结果
+     */
+    PutObjectResult upload(String key, File file, Options options);
+
     /**
      * 将文件上传到默认存储桶。
      *
@@ -317,6 +416,16 @@ public interface OssClient extends AutoCloseable {
      */
     PutObjectResult upload(String key, File file);
 
+    /**
+     * 将随机访问文件上传到默认存储桶。
+     *
+     * @param key  对象键
+     * @param file 随机访问文件
+     * @param options 可选项
+     * @return 上传结果
+     */
+    PutObjectResult upload(String key, RandomAccessFile file, Options options);
+
     /**
      * 将随机访问文件上传到默认存储桶。
      *
@@ -326,6 +435,17 @@ public interface OssClient extends AutoCloseable {
      */
     PutObjectResult upload(String key, RandomAccessFile file);
 
+    /**
+     * 将可读通道中的数据上传到默认存储桶。
+     *
+     * @param key    对象键
+     * @param channel            数据通道
+     * @param contentLength 内容长度
+     * @param options 可选项
+     * @return 上传结果
+     */
+    PutObjectResult upload(String key, ReadableByteChannel channel, long contentLength, Options options);
+
     /**
      * 将可读通道中的数据上传到默认存储桶。
      *
@@ -336,6 +456,17 @@ public interface OssClient extends AutoCloseable {
      */
     PutObjectResult upload(String key, ReadableByteChannel channel, long contentLength);
 
+    /**
+     * 将输入流中的数据上传到默认存储桶。
+     *
+     * @param key    对象键
+     * @param in            输入流
+     * @param contentLength 内容长度
+     * @param options 可选项
+     * @return 上传结果
+     */
+    PutObjectResult upload(String key, InputStream in, long contentLength, Options options);
+
     /**
      * 将输入流中的数据上传到默认存储桶。
      *
@@ -346,6 +477,16 @@ public interface OssClient extends AutoCloseable {
      */
     PutObjectResult upload(String key, InputStream in, long contentLength);
 
+    /**
+     * 将字节数组上传到默认存储桶。
+     *
+     * @param key    对象键
+     * @param data   字节数组
+     * @param options 可选项
+     * @return 上传结果
+     */
+    PutObjectResult upload(String key, byte[] data, Options options);
+
     /**
      * 将字节数组上传到默认存储桶。
      *
@@ -364,6 +505,15 @@ public interface OssClient extends AutoCloseable {
      */
     GetObjectResult download(String key, OutputStreamDownloadSubscriber downloadSubscriber);
 
+    /**
+     * 将指定存储桶中的对象下载到转换器中，由使用者决定返回值。
+     *
+     * @param key                 对象键
+     * @param downloadTransformer 下载转换器
+     * @return 下载结果
+     */
+    <T> T download(String key, BiFunction<GetObjectResult, InputStream, T> downloadTransformer);
+
     /**
      * 将默认存储桶中的对象下载到本地路径。
      *

ruoyi-common/ruoyi-common-oss/src/main/java/org/dromara/common/oss/factory/OssFactory.java

@@ -66,7 +66,7 @@ public class OssFactory {
             CLIENT_CACHE.put(configKey, client);
             return client;
         } finally {
-            LOCK.lock();
+            LOCK.unlock();
         }
     }
 

ruoyi-common/ruoyi-common-oss/src/main/java/org/dromara/common/oss/io/OutputStreamDownloadSubscriber.java

@@ -3,7 +3,6 @@ package org.dromara.common.oss.io;
 import org.dromara.common.oss.exception.S3StorageException;
 
 import java.io.FileOutputStream;
-import java.io.IOException;
 import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.nio.channels.Channels;
@@ -19,11 +18,15 @@ public class OutputStreamDownloadSubscriber implements Consumer<ByteBuffer>, Aut
 
     private final WritableByteChannel channel;
 
-    private OutputStreamDownloadSubscriber(WritableByteChannel channel) {
+    private final boolean allowAutoClose;
+
+    private OutputStreamDownloadSubscriber(WritableByteChannel channel, boolean allowAutoClose) {
         this.channel = channel;
+        this.allowAutoClose = allowAutoClose;
     }
 
-    private OutputStreamDownloadSubscriber(OutputStream out) {
+    private OutputStreamDownloadSubscriber(OutputStream out, boolean allowAutoClose) {
+        this.allowAutoClose = allowAutoClose;
         // 创建可写入的字节通道
         if (out instanceof FileOutputStream outputStream) {
             // 如果是文件输入流，直接获取文件输出流的 Channel
@@ -35,18 +38,18 @@ public class OutputStreamDownloadSubscriber implements Consumer<ByteBuffer>, Aut
 
     @Override
     public void accept(ByteBuffer byteBuffer) {
-        try (channel) {
+        try {
             while (byteBuffer.hasRemaining()) {
                 channel.write(byteBuffer);
             }
-        } catch (IOException e) {
+        } catch (Exception e) {
             throw S3StorageException.form(e);
         }
     }
 
     @Override
     public void close() throws Exception {
-        if (channel.isOpen()) {
+        if (channel.isOpen() && allowAutoClose) {
             channel.close();
         }
     }
@@ -58,7 +61,18 @@ public class OutputStreamDownloadSubscriber implements Consumer<ByteBuffer>, Aut
      * @return 输出流下载订阅器
      */
     public static OutputStreamDownloadSubscriber create(OutputStream out) {
-        return new OutputStreamDownloadSubscriber(out);
+        return create(out, false);
+    }
+
+    /**
+     * 创建一个输出流下载订阅器
+     *
+     * @param out            输出流
+     * @param allowAutoClose 是否允许自动关闭流
+     * @return 输出流下载订阅器
+     */
+    public static OutputStreamDownloadSubscriber create(OutputStream out, boolean allowAutoClose) {
+        return new OutputStreamDownloadSubscriber(out, allowAutoClose);
     }
 
     /**
@@ -68,7 +82,18 @@ public class OutputStreamDownloadSubscriber implements Consumer<ByteBuffer>, Aut
      * @return 输出流下载订阅器
      */
     public static OutputStreamDownloadSubscriber create(WritableByteChannel channel) {
-        return new OutputStreamDownloadSubscriber(channel);
+        return create(channel, false);
+    }
+
+    /**
+     * 创建一个输出流下载订阅器
+     *
+     * @param channel        可写字节通道
+     * @param allowAutoClose 是否允许自动关闭流
+     * @return 输出流下载订阅器
+     */
+    public static OutputStreamDownloadSubscriber create(WritableByteChannel channel, boolean allowAutoClose) {
+        return new OutputStreamDownloadSubscriber(channel, allowAutoClose);
     }
 
 }

ruoyi-common/ruoyi-common-push/pom.xml

@@ -9,10 +9,10 @@
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
-    <artifactId>ruoyi-common-websocket</artifactId>
+    <artifactId>ruoyi-common-push</artifactId>
 
     <description>
-        ruoyi-common-websocket 模块
+        ruoyi-common-push 模块
     </description>
 
     <dependencies>

ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/config/MessageAutoConfiguration.java

@@ -0,0 +1,17 @@
+package org.dromara.common.push.config;
+
+import org.dromara.common.push.properties.MessageProperties;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+
+/**
+ * 统一消息推送公共自动装配。
+ *
+ * @author Lion Li
+ */
+@AutoConfiguration
+@ConditionalOnProperty(prefix = "message", name = "enabled", havingValue = "true", matchIfMissing = true)
+@EnableConfigurationProperties(MessageProperties.class)
+public class MessageAutoConfiguration {
+}

ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/config/MessageSseConfiguration.java

@@ -0,0 +1,33 @@
+package org.dromara.common.push.config;
+
+import org.dromara.common.push.controller.SseController;
+import org.dromara.common.push.core.SseEmitterSessionManager;
+import org.dromara.common.push.listener.MessageTopicListener;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+
+/**
+ * SSE 消息推送自动装配。
+ *
+ * @author Lion Li
+ */
+@AutoConfiguration(after = MessageAutoConfiguration.class)
+@ConditionalOnProperty(prefix = "message", name = "transport", havingValue = "sse", matchIfMissing = true)
+public class MessageSseConfiguration {
+
+    @Bean
+    public SseEmitterSessionManager sseEmitterManager() {
+        return new SseEmitterSessionManager();
+    }
+
+    @Bean
+    public MessageTopicListener messageTopicListener(SseEmitterSessionManager manager) {
+        return new MessageTopicListener(manager);
+    }
+
+    @Bean
+    public SseController sseController(SseEmitterSessionManager manager) {
+        return new SseController(manager);
+    }
+}

ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/config/MessageWebSocketConfiguration.java

@@ -0,0 +1,55 @@
+package org.dromara.common.push.config;
+
+import org.dromara.common.push.listener.MessageTopicListener;
+import org.dromara.common.push.core.WebSocketSessionManager;
+import org.dromara.common.push.handler.PlusWebSocketHandler;
+import org.dromara.common.push.interceptor.PlusWebSocketInterceptor;
+import org.dromara.common.push.properties.MessageProperties;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+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;
+
+/**
+ * WebSocket 消息推送自动装配。
+ *
+ * @author Lion Li
+ */
+@EnableWebSocket
+@AutoConfiguration(after = MessageAutoConfiguration.class)
+@ConditionalOnProperty(prefix = "message", name = "transport", havingValue = "websocket")
+public class MessageWebSocketConfiguration {
+
+    @Bean
+    public WebSocketConfigurer webSocketConfigurer(HandshakeInterceptor handshakeInterceptor,
+                                                   WebSocketHandler webSocketHandler,
+                                                   MessageProperties messageProperties) {
+        return registry -> registry
+            .addHandler(webSocketHandler, messageProperties.getPath())
+            .addInterceptors(handshakeInterceptor)
+            .setAllowedOrigins(messageProperties.getAllowedOrigins());
+    }
+
+    @Bean
+    public WebSocketSessionManager webSocketSessionManager() {
+        return new WebSocketSessionManager();
+    }
+
+    @Bean
+    public HandshakeInterceptor handshakeInterceptor() {
+        return new PlusWebSocketInterceptor();
+    }
+
+    @Bean
+    public WebSocketHandler webSocketHandler(WebSocketSessionManager webSocketSessionManager) {
+        return new PlusWebSocketHandler(webSocketSessionManager);
+    }
+
+    @Bean
+    public MessageTopicListener messageTopicListener(WebSocketSessionManager webSocketSessionManager) {
+        return new MessageTopicListener(webSocketSessionManager);
+    }
+}

ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/constant/MessageConstants.java

@@ -0,0 +1,19 @@
+package org.dromara.common.push.constant;
+
+/**
+ * 模块通用消息常量定义。
+ *
+ * @author Lion Li
+ */
+public interface MessageConstants {
+
+    String LOGIN_USER_KEY = "loginUser";
+
+    String LOGIN_TOKEN_KEY = "token";
+
+    String MESSAGE_TOPIC = "global:message";
+
+    String PING = "ping";
+
+    String PONG = "pong";
+}

ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/controller/SseController.java

@@ -1,13 +1,12 @@
-package org.dromara.common.sse.controller;
+package org.dromara.common.push.controller;
 
 import cn.dev33.satoken.annotation.SaIgnore;
 import cn.dev33.satoken.stp.StpUtil;
 import lombok.RequiredArgsConstructor;
 import org.dromara.common.core.domain.R;
+import org.dromara.common.push.core.SseEmitterSessionManager;
 import org.dromara.common.satoken.utils.LoginHelper;
-import org.dromara.common.sse.core.SseEmitterManager;
 import org.springframework.beans.factory.DisposableBean;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.http.MediaType;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RestController;
@@ -19,25 +18,24 @@ import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
  * @author Lion Li
  */
 @RestController
-@ConditionalOnProperty(value = "sse.enabled", havingValue = "true")
 @RequiredArgsConstructor
 public class SseController implements DisposableBean {
 
-    private final SseEmitterManager sseEmitterManager;
+    private final SseEmitterSessionManager sessionManager;
 
     /**
      * 建立当前登录用户的 SSE 连接。
      *
      * @return SSE 发射器
      */
-    @GetMapping(value = "${sse.path}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
+    @GetMapping(value = "${message.path:/resource/message}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
     public SseEmitter connect() {
         if (!StpUtil.isLogin()) {
             return null;
         }
         String tokenValue = StpUtil.getTokenValue();
         Long userId = LoginHelper.getUserId();
-        return sseEmitterManager.connect(userId, tokenValue);
+        return sessionManager.connect(userId, tokenValue);
     }
 
     /**
@@ -46,11 +44,11 @@ public class SseController implements DisposableBean {
      * @return 操作结果
      */
     @SaIgnore
-    @GetMapping(value = "${sse.path}/close")
+    @GetMapping(value = "${message.path:/resource/message}/close")
     public R<Void> close() {
         String tokenValue = StpUtil.getTokenValue();
         Long userId = LoginHelper.getUserId();
-        sseEmitterManager.disconnect(userId, tokenValue);
+        sessionManager.disconnect(userId, tokenValue);
         return R.ok();
     }
 
@@ -61,12 +59,12 @@ public class SseController implements DisposableBean {
 //     * @param userId 目标用户的 ID
 //     * @param msg    要发送的消息内容
 //     */
-//    @GetMapping(value = "${sse.path}/send")
+//    @GetMapping(value = "${message.path:/resource/message}/send")
 //    public R<Void> send(Long userId, String msg) {
-//        SseMessageDTO dto = new SseMessageDTO();
+//        PushDTO dto = new PushDTO();
 //        dto.setUserIds(List.of(userId));
-//        dto.setMessage(msg);
+//        dto.setPayload(PushPayload.of("message", "backend", msg, null));
-//        sseEmitterManager.publishMessage(dto);
+//        sessionManager.publishMessage(dto);
 //        return R.ok();
 //    }
 //
@@ -75,9 +73,9 @@ public class SseController implements DisposableBean {
 //     *
 //     * @param msg 要发送的消息内容
 //     */
-//    @GetMapping(value = "${sse.path}/sendAll")
+//    @GetMapping(value = "${message.path:/resource/message}/sendAll")
 //    public R<Void> send(String msg) {
-//        sseEmitterManager.publishAll(msg);
+//        sessionManager.publishAll(msg);
 //        return R.ok();
 //    }
 

ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/PushSessionManager.java

@@ -0,0 +1,24 @@
+package org.dromara.common.push.core;
+
+import org.dromara.common.core.domain.dto.PushPayload;
+import org.dromara.common.push.dto.PushDTO;
+
+import java.util.function.Consumer;
+
+/**
+ * 统一推送会话管理器。
+ *
+ * @author Lion Li
+ */
+public interface PushSessionManager {
+
+    void subscribeMessage(Consumer<PushDTO> consumer);
+
+    void sendMessage(Long userId, PushPayload payload);
+
+    void sendMessage(PushPayload payload);
+
+    void publishMessage(PushDTO pushDTO);
+
+    void publishAll(PushPayload payload);
+}

ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/SseEmitterSessionManager.java

@@ -1,11 +1,14 @@
-package org.dromara.common.sse.core;
+package org.dromara.common.push.core;
 
 import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.map.MapUtil;
 import lombok.extern.slf4j.Slf4j;
+import org.dromara.common.core.domain.dto.PushPayload;
+import org.dromara.common.push.constant.MessageConstants;
+import org.dromara.common.push.dto.PushDTO;
 import org.dromara.common.core.utils.SpringUtils;
+import org.dromara.common.json.utils.JsonUtils;
 import org.dromara.common.redis.utils.RedisUtils;
-import org.dromara.common.sse.dto.SseMessageDTO;
 import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 
 import java.io.IOException;
@@ -23,16 +26,11 @@ import java.util.function.Consumer;
  * @author Lion Li
  */
 @Slf4j
-public class SseEmitterManager {
+public class SseEmitterSessionManager implements PushSessionManager {
-
-    /**
-     * 订阅的频道
-     */
-    private final static String SSE_TOPIC = "global:sse";
 
     private final static Map<Long, Map<String, SseEmitter>> USER_TOKEN_EMITTERS = new ConcurrentHashMap<>();
 
-    public SseEmitterManager() {
+    public SseEmitterSessionManager() {
         // 定时执行 SSE 心跳检测
         SpringUtils.getBean(ScheduledExecutorService.class)
             .scheduleWithFixedDelay(this::sseMonitor, 60L, 60L, TimeUnit.SECONDS);
@@ -158,8 +156,9 @@ public class SseEmitterManager {
      *
      * @param consumer 处理SSE消息的消费者函数
      */
-    public void subscribeMessage(Consumer<SseMessageDTO> consumer) {
+    @Override
-        RedisUtils.subscribe(SSE_TOPIC, SseMessageDTO.class, consumer);
+    public void subscribeMessage(Consumer<PushDTO> consumer) {
+        RedisUtils.subscribe(MessageConstants.MESSAGE_TOPIC, PushDTO.class, consumer);
     }
 
     /**
@@ -188,6 +187,30 @@ public class SseEmitterManager {
         }
     }
 
+    /**
+     * 向指定用户的全部本地 SSE 会话发送统一 JSON 消息。
+     *
+     * @param userId  要发送消息的用户id
+     * @param payload 要发送的消息体
+     */
+    @Override
+    public void sendMessage(Long userId, PushPayload payload) {
+        sendMessage(userId, JsonUtils.toJsonString(payload));
+    }
+
+    /**
+     * 向指定用户的全部本地 SSE 会话发送统一 JSON 消息。
+     *
+     * @param userId        要发送消息的用户id
+     * @param pushDTO 要发送的消息内容
+     */
+    public void sendMessage(Long userId, PushDTO pushDTO) {
+        if (pushDTO == null) {
+            return;
+        }
+        sendMessage(userId, pushDTO.getPayload());
+    }
+
     /**
      * 向当前节点所有 SSE 会话发送消息。
      *
@@ -199,19 +222,29 @@ public class SseEmitterManager {
         }
     }
 
+    /**
+     * 向当前节点所有 SSE 会话发送统一 JSON 消息。
+     *
+     * @param payload 要发送的消息体
+     */
+    @Override
+    public void sendMessage(PushPayload payload) {
+        sendMessage(JsonUtils.toJsonString(payload));
+    }
+
     /**
      * 发布 SSE 订阅消息。
      *
-     * @param sseMessageDTO 要发布的SSE消息对象
+     * @param pushDTO 要发布的SSE消息对象
      */
-    public void publishMessage(SseMessageDTO sseMessageDTO) {
+    @Override
-        SseMessageDTO broadcastMessage = new SseMessageDTO();
+    public void publishMessage(PushDTO pushDTO) {
-        broadcastMessage.setUserIds(sseMessageDTO.getUserIds());
+        RedisUtils.publish(MessageConstants.MESSAGE_TOPIC, pushDTO, consumer -> log.info(
-        broadcastMessage.setMessage(sseMessageDTO.getMessage());
+            "发送主题订阅消息topic:{} userIds:{} message:{}",
-        RedisUtils.publish(SSE_TOPIC, broadcastMessage, consumer -> {
+            MessageConstants.MESSAGE_TOPIC,
-            log.info("SSE发送主题订阅消息topic:{} session keys:{} message:{}",
+            pushDTO.getUserIds(),
-                SSE_TOPIC, sseMessageDTO.getUserIds(), sseMessageDTO.getMessage());
+            pushDTO.getPayload() == null ? null : pushDTO.getPayload().getMessage()
-        });
+        ));
     }
 
     /**
@@ -220,10 +253,21 @@ public class SseEmitterManager {
      * @param message 要发布的消息内容
      */
     public void publishAll(String message) {
-        SseMessageDTO broadcastMessage = new SseMessageDTO();
+        publishAll(PushPayload.of("message", "backend", message, null));
-        broadcastMessage.setMessage(message);
+    }
-        RedisUtils.publish(SSE_TOPIC, broadcastMessage, consumer -> {
+
-            log.info("SSE发送主题订阅消息topic:{} message:{}", SSE_TOPIC, message);
+    /**
+     * 发布 SSE 广播 JSON 消息。
+     *
+     * @param payload 要发布的消息体
+     */
+    @Override
+    public void publishAll(PushPayload payload) {
+        PushDTO dto = new PushDTO();
+        dto.setPayload(payload);
+        RedisUtils.publish(MessageConstants.MESSAGE_TOPIC, dto, consumer -> {
+            log.info("发送主题订阅消息topic:{} type:{} source:{} message:{}",
+                MessageConstants.MESSAGE_TOPIC, payload.getType(), payload.getSource(), payload.getMessage());
         });
     }
 }

ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/WebSocketSessionManager.java

@@ -0,0 +1,168 @@
+package org.dromara.common.push.core;
+
+import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.map.MapUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.dromara.common.core.domain.dto.PushPayload;
+import org.dromara.common.core.utils.SpringUtils;
+import org.dromara.common.json.utils.JsonUtils;
+import org.dromara.common.push.dto.PushDTO;
+import org.dromara.common.redis.utils.RedisUtils;
+import org.springframework.web.socket.CloseStatus;
+import org.springframework.web.socket.PongMessage;
+import org.springframework.web.socket.TextMessage;
+import org.springframework.web.socket.WebSocketMessage;
+import org.springframework.web.socket.WebSocketSession;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+
+import static org.dromara.common.push.constant.MessageConstants.MESSAGE_TOPIC;
+
+/**
+ * WebSocket 会话管理器。
+ *
+ * @author Lion Li
+ */
+@Slf4j
+public class WebSocketSessionManager implements PushSessionManager {
+
+    private static final Map<Long, Map<String, WebSocketSession>> USER_TOKEN_SESSIONS = new ConcurrentHashMap<>();
+
+    public WebSocketSessionManager() {
+        SpringUtils.getBean(ScheduledExecutorService.class)
+            .scheduleWithFixedDelay(this::sessionMonitor, 60L, 60L, TimeUnit.SECONDS);
+    }
+
+    public void connect(Long userId, String token, WebSocketSession session) {
+        Map<String, WebSocketSession> sessions = USER_TOKEN_SESSIONS.computeIfAbsent(userId, key -> new ConcurrentHashMap<>());
+        WebSocketSession oldSession = sessions.remove(token);
+        closeSession(oldSession, CloseStatus.NORMAL);
+        sessions.put(token, session);
+    }
+
+    public void disconnect(Long userId, String token) {
+        if (userId == null || token == null) {
+            return;
+        }
+        Map<String, WebSocketSession> sessions = USER_TOKEN_SESSIONS.get(userId);
+        if (MapUtil.isEmpty(sessions)) {
+            USER_TOKEN_SESSIONS.remove(userId);
+            return;
+        }
+        closeSession(sessions.remove(token), CloseStatus.NORMAL);
+        if (sessions.isEmpty()) {
+            USER_TOKEN_SESSIONS.remove(userId);
+        }
+    }
+
+    public void sessionMonitor() {
+        List<Long> toRemoveUsers = new ArrayList<>();
+        USER_TOKEN_SESSIONS.forEach((userId, sessionMap) -> {
+            if (CollUtil.isEmpty(sessionMap)) {
+                toRemoveUsers.add(userId);
+                return;
+            }
+            sessionMap.entrySet().removeIf(entry -> {
+                WebSocketSession session = entry.getValue();
+                if (session == null || !session.isOpen()) {
+                    closeSession(session, CloseStatus.NORMAL);
+                    return true;
+                }
+                return false;
+            });
+            if (sessionMap.isEmpty()) {
+                toRemoveUsers.add(userId);
+            }
+        });
+        toRemoveUsers.forEach(USER_TOKEN_SESSIONS::remove);
+    }
+
+    @Override
+    public void subscribeMessage(Consumer<PushDTO> consumer) {
+        RedisUtils.subscribe(MESSAGE_TOPIC, PushDTO.class, consumer);
+    }
+
+    @Override
+    public void sendMessage(Long userId, PushPayload payload) {
+        if (payload == null) {
+            return;
+        }
+        Map<String, WebSocketSession> sessions = USER_TOKEN_SESSIONS.get(userId);
+        if (MapUtil.isEmpty(sessions)) {
+            USER_TOKEN_SESSIONS.remove(userId);
+            return;
+        }
+        sessions.entrySet().removeIf(entry -> {
+            WebSocketSession session = entry.getValue();
+            if (session == null || !session.isOpen()) {
+                closeSession(session, CloseStatus.NORMAL);
+                return true;
+            }
+            return !sendMessage(session, new TextMessage(JsonUtils.toJsonString(payload)));
+        });
+        if (sessions.isEmpty()) {
+            USER_TOKEN_SESSIONS.remove(userId);
+        }
+    }
+
+    @Override
+    public void sendMessage(PushPayload payload) {
+        USER_TOKEN_SESSIONS.keySet().forEach(userId -> sendMessage(userId, payload));
+    }
+
+    @Override
+    public void publishMessage(PushDTO pushDTO) {
+        RedisUtils.publish(MESSAGE_TOPIC, pushDTO, consumer -> log.info(
+            "WebSocket发送主题订阅消息topic:{} userIds:{} message:{}",
+            MESSAGE_TOPIC,
+            pushDTO.getUserIds(),
+            pushDTO.getPayload() == null ? null : pushDTO.getPayload().getMessage()
+        ));
+    }
+
+    @Override
+    public void publishAll(PushPayload payload) {
+        PushDTO dto = new PushDTO();
+        dto.setPayload(payload);
+        publishMessage(dto);
+    }
+
+    public void sendPongMessage(WebSocketSession session) {
+        sendMessage(session, new PongMessage());
+    }
+
+    public void sendMessage(WebSocketSession session, String message) {
+        sendMessage(session, new TextMessage(message));
+    }
+
+    private boolean sendMessage(WebSocketSession session, WebSocketMessage<?> message) {
+        if (session == null || !session.isOpen()) {
+            log.warn("[send] session会话已经关闭");
+            return false;
+        }
+        try {
+            session.sendMessage(message);
+            return true;
+        } catch (IOException e) {
+            log.error("[send] session({}) 发送消息({}) 异常", session, message, e);
+            return false;
+        }
+    }
+
+    private void closeSession(WebSocketSession session, CloseStatus status) {
+        if (session == null) {
+            return;
+        }
+        try {
+            session.close(status);
+        } catch (Exception ignored) {
+        }
+    }
+}

ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/dto/PushDTO.java

@@ -0,0 +1,30 @@
+package org.dromara.common.push.dto;
+
+import lombok.Data;
+import org.dromara.common.core.domain.dto.PushPayload;
+
+import java.io.Serial;
+import java.io.Serializable;
+import java.util.List;
+
+/**
+ * 统一推送 DTO。
+ *
+ * @author Lion Li
+ */
+@Data
+public class PushDTO implements Serializable {
+
+    @Serial
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * 目标用户 ID 列表，为空表示广播。
+     */
+    private List<Long> userIds;
+
+    /**
+     * 推送消息体。
+     */
+    private PushPayload payload;
+}

ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/enums/MessageTransportEnum.java

@@ -0,0 +1,32 @@
+package org.dromara.common.push.enums;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+import java.util.Arrays;
+
+/**
+ * 消息推送传输方式。
+ *
+ * @author Lion Li
+ */
+@Getter
+@AllArgsConstructor
+public enum MessageTransportEnum {
+
+    SSE("sse"),
+    WEBSOCKET("websocket");
+
+    private final String code;
+
+    public boolean matches(String transport) {
+        return code.equalsIgnoreCase(transport);
+    }
+
+    public static MessageTransportEnum of(String transport) {
+        return Arrays.stream(values())
+            .filter(item -> item.matches(transport))
+            .findFirst()
+            .orElse(SSE);
+    }
+}

ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/handler/PlusWebSocketHandler.java

@@ -0,0 +1,104 @@
+package org.dromara.common.push.handler;
+
+import cn.hutool.core.util.ObjectUtil;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.dromara.common.core.domain.dto.PushPayload;
+import org.dromara.common.core.domain.model.LoginUser;
+import org.dromara.common.core.enums.PushSourceEnum;
+import org.dromara.common.core.enums.PushTypeEnum;
+import org.dromara.common.push.constant.MessageConstants;
+import org.dromara.common.push.core.WebSocketSessionManager;
+import org.dromara.common.push.dto.PushDTO;
+import org.springframework.web.socket.BinaryMessage;
+import org.springframework.web.socket.CloseStatus;
+import org.springframework.web.socket.PongMessage;
+import org.springframework.web.socket.TextMessage;
+import org.springframework.web.socket.WebSocketSession;
+import org.springframework.web.socket.handler.AbstractWebSocketHandler;
+import org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * WebSocket Handler。
+ *
+ * @author Lion Li
+ */
+@RequiredArgsConstructor
+@Slf4j
+public class PlusWebSocketHandler extends AbstractWebSocketHandler {
+
+    private final WebSocketSessionManager webSocketSessionManager;
+
+    @Override
+    public void afterConnectionEstablished(WebSocketSession session) throws IOException {
+        LoginUser loginUser = (LoginUser) session.getAttributes().get(MessageConstants.LOGIN_USER_KEY);
+        String token = (String) session.getAttributes().get(MessageConstants.LOGIN_TOKEN_KEY);
+        if (ObjectUtil.hasNull(loginUser, token)) {
+            session.close(CloseStatus.BAD_DATA);
+            log.info("[connect] invalid token received. sessionId: {}", session.getId());
+            return;
+        }
+        webSocketSessionManager.connect(
+            loginUser.getUserId(),
+            token,
+            new ConcurrentWebSocketSessionDecorator(session, 10 * 1000, 64_000)
+        );
+        log.info("[connect] sessionId: {}, userId:{}, token:{}", session.getId(), loginUser.getUserId(), token);
+    }
+
+    @Override
+    protected void handleTextMessage(WebSocketSession session, TextMessage message) {
+        LoginUser loginUser = (LoginUser) session.getAttributes().get(MessageConstants.LOGIN_USER_KEY);
+        if (ObjectUtil.isNull(loginUser)) {
+            return;
+        }
+        if (MessageConstants.PING.equalsIgnoreCase(message.getPayload())) {
+            webSocketSessionManager.sendMessage(session, MessageConstants.PONG);
+            return;
+        }
+        PushDTO dto = new PushDTO();
+        dto.setUserIds(List.of(loginUser.getUserId()));
+        dto.setPayload(PushPayload.of(
+            PushTypeEnum.CUSTOM,
+            PushSourceEnum.CLIENT,
+            message.getPayload(),
+            null
+        ));
+        webSocketSessionManager.publishMessage(dto);
+    }
+
+    @Override
+    protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) throws Exception {
+        super.handleBinaryMessage(session, message);
+    }
+
+    @Override
+    protected void handlePongMessage(WebSocketSession session, PongMessage message) {
+        webSocketSessionManager.sendPongMessage(session);
+    }
+
+    @Override
+    public void handleTransportError(WebSocketSession session, Throwable exception) {
+        log.error("[transport error] sessionId: {}, exception:{}", session.getId(), exception.getMessage());
+    }
+
+    @Override
+    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
+        LoginUser loginUser = (LoginUser) session.getAttributes().get(MessageConstants.LOGIN_USER_KEY);
+        String token = (String) session.getAttributes().get(MessageConstants.LOGIN_TOKEN_KEY);
+        if (ObjectUtil.hasNull(loginUser, token)) {
+            log.info("[disconnect] invalid token received. sessionId: {}", session.getId());
+            return;
+        }
+        webSocketSessionManager.disconnect(loginUser.getUserId(), token);
+        log.info("[disconnect] sessionId: {}, userId:{}, token:{}", session.getId(), loginUser.getUserId(), token);
+    }
+
+    @Override
+    public boolean supportsPartialMessages() {
+        return false;
+    }
+}

ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/helper/PushHelper.java

@@ -0,0 +1,80 @@
+package org.dromara.common.push.helper;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+import org.dromara.common.core.domain.dto.PushPayload;
+import org.dromara.common.core.enums.PushSourceEnum;
+import org.dromara.common.core.enums.PushTypeEnum;
+import org.dromara.common.core.utils.SpringUtils;
+import org.dromara.common.push.core.PushSessionManager;
+import org.dromara.common.push.dto.PushDTO;
+
+import java.util.List;
+
+/**
+ * 统一消息推送工具。
+ *
+ * @author Lion Li
+ */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class PushHelper {
+
+    public static void sendMessage(Long userId, String message) {
+        sendMessage(userId, buildMessage(message));
+    }
+
+    public static void sendMessage(String message) {
+        sendMessage(buildMessage(message));
+    }
+
+    public static void sendMessage(Long userId, PushPayload payload) {
+        if (!isEnabled()) {
+            return;
+        }
+        getSessionManager().sendMessage(userId, payload);
+    }
+
+    public static void sendMessage(PushPayload payload) {
+        if (!isEnabled()) {
+            return;
+        }
+        getSessionManager().sendMessage(payload);
+    }
+
+    public static void publishMessage(List<Long> userIds, PushPayload payload) {
+        PushDTO dto = new PushDTO();
+        dto.setUserIds(userIds);
+        dto.setPayload(payload);
+        publishMessage(dto);
+    }
+
+    public static void publishMessage(PushDTO dto) {
+        if (!isEnabled() || dto == null || dto.getPayload() == null) {
+            return;
+        }
+        getSessionManager().publishMessage(dto);
+    }
+
+    public static void publishAll(String message) {
+        publishAll(buildMessage(message));
+    }
+
+    public static void publishAll(PushPayload payload) {
+        if (!isEnabled()) {
+            return;
+        }
+        getSessionManager().publishAll(payload);
+    }
+
+    public static boolean isEnabled() {
+        return Boolean.TRUE.equals(SpringUtils.getProperty("message.enabled", Boolean.class, Boolean.TRUE));
+    }
+
+    private static PushSessionManager getSessionManager() {
+        return SpringUtils.getBean(PushSessionManager.class);
+    }
+
+    private static PushPayload buildMessage(String message) {
+        return PushPayload.of(PushTypeEnum.MESSAGE, PushSourceEnum.BACKEND, message, null);
+    }
+}

ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/interceptor/PlusWebSocketInterceptor.java

@@ -1,4 +1,4 @@
-package org.dromara.common.websocket.interceptor;
+package org.dromara.common.push.interceptor;
 
 import cn.dev33.satoken.exception.NotLoginException;
 import cn.dev33.satoken.stp.StpUtil;
@@ -6,6 +6,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.dromara.common.core.domain.model.LoginUser;
 import org.dromara.common.core.utils.ServletUtils;
 import org.dromara.common.core.utils.StringUtils;
+import org.dromara.common.push.constant.MessageConstants;
 import org.dromara.common.satoken.utils.LoginHelper;
 import org.springframework.http.server.ServerHttpRequest;
 import org.springframework.http.server.ServerHttpResponse;
@@ -13,45 +14,35 @@ import org.springframework.web.socket.WebSocketHandler;
 import org.springframework.web.socket.server.HandshakeInterceptor;
 
 import java.util.Map;
-
-import static org.dromara.common.websocket.constant.WebSocketConstants.LOGIN_USER_KEY;
-
 /**
- * WebSocket握手请求的拦截器
+ * WebSocket 握手拦截器。
  *
- * @author zendwang
+ * @author Lion Li
  */
 @Slf4j
 public class PlusWebSocketInterceptor implements HandshakeInterceptor {
 
-    /**
-     * WebSocket握手之前执行的前置处理方法
-     *
-     * @param request    WebSocket握手请求
-     * @param response   WebSocket握手响应
-     * @param wsHandler  WebSocket处理程序
-     * @param attributes 与WebSocket会话关联的属性
-     * @return 如果允许握手继续进行，则返回true；否则返回false
-     */
     @Override
-    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Map<String, Object> attributes) {
+    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler,
+                                   Map<String, Object> attributes) {
         try {
-            // 检查是否登录 是否有token
             LoginUser loginUser = LoginHelper.getLoginUser();
+            String tokenValue = StpUtil.getTokenValue();
+            if (loginUser == null || StringUtils.isBlank(tokenValue)) {
+                return false;
+            }
 
-            // 解决 ws 不走 mvc 拦截器问题(cloud 版本不受影响)
-            // 检查 header 与 param 里的 clientid 与 token 里的是否一致
             String headerCid = ServletUtils.getRequest().getHeader(LoginHelper.CLIENT_KEY);
             String paramCid = ServletUtils.getParameter(LoginHelper.CLIENT_KEY);
             String clientId = StpUtil.getExtra(LoginHelper.CLIENT_KEY).toString();
             if (!StringUtils.equalsAny(clientId, headerCid, paramCid)) {
-                // token 无效
                 throw NotLoginException.newInstance(StpUtil.getLoginType(),
                     "-100", "客户端ID与Token不匹配",
                     StpUtil.getTokenValue());
             }
 
-            attributes.put(LOGIN_USER_KEY, loginUser);
+            attributes.put(MessageConstants.LOGIN_USER_KEY, loginUser);
+            attributes.put(MessageConstants.LOGIN_TOKEN_KEY, tokenValue);
             return true;
         } catch (NotLoginException e) {
             log.error("WebSocket 认证失败'{}',无法访问系统资源", e.getMessage());
@@ -59,17 +50,8 @@ public class PlusWebSocketInterceptor implements HandshakeInterceptor {
         }
     }
 
-    /**
-     * WebSocket握手成功后执行的后置处理方法
-     *
-     * @param request   WebSocket握手请求
-     * @param response  WebSocket握手响应
-     * @param wsHandler WebSocket处理程序
-     * @param exception 握手过程中可能出现的异常
-     */
     @Override
-    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Exception exception) {
+    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler,
-        // 在这个方法中可以执行一些握手成功后的后续处理逻辑，比如记录日志或者其他操作
+                               Exception exception) {
     }
-
 }

ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/listener/MessageTopicListener.java

@@ -0,0 +1,44 @@
+package org.dromara.common.push.listener;
+
+import cn.hutool.core.collection.CollUtil;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.dromara.common.push.core.PushSessionManager;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.core.Ordered;
+
+/**
+ * 统一消息主题订阅监听器。
+ *
+ * @author Lion Li
+ */
+@Slf4j
+@RequiredArgsConstructor
+public class MessageTopicListener implements ApplicationRunner, Ordered {
+
+    private final PushSessionManager pushSessionManager;
+
+    @Override
+    public void run(ApplicationArguments args) {
+        pushSessionManager.subscribeMessage(message -> {
+            log.info("消息主题订阅收到消息userIds={} message={}",
+                message.getUserIds(),
+                message.getPayload() == null ? null : message.getPayload().getMessage());
+            if (message.getPayload() == null) {
+                return;
+            }
+            if (CollUtil.isNotEmpty(message.getUserIds())) {
+                message.getUserIds().forEach(userId -> pushSessionManager.sendMessage(userId, message.getPayload()));
+            } else {
+                pushSessionManager.sendMessage(message.getPayload());
+            }
+        });
+        log.info("初始化消息主题订阅监听器成功");
+    }
+
+    @Override
+    public int getOrder() {
+        return -1;
+    }
+}

ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/properties/MessageProperties.java

@@ -0,0 +1,35 @@
+package org.dromara.common.push.properties;
+
+import lombok.Data;
+import org.dromara.common.push.enums.MessageTransportEnum;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * 统一消息推送配置。
+ *
+ * @author Lion Li
+ */
+@Data
+@ConfigurationProperties("message")
+public class MessageProperties {
+
+    /**
+     * 是否启用消息推送。
+     */
+    private Boolean enabled = true;
+
+    /**
+     * 传输方式：sse / websocket。
+     */
+    private String transport = MessageTransportEnum.SSE.getCode();
+
+    /**
+     * 统一访问路径。
+     */
+    private String path = "/resource/message";
+
+    /**
+     * WebSocket 允许的跨域来源。
+     */
+    private String allowedOrigins = "*";
+}

ruoyi-common/ruoyi-common-push/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -0,0 +1,3 @@
+org.dromara.common.push.config.MessageAutoConfiguration
+org.dromara.common.push.config.MessageSseConfiguration
+org.dromara.common.push.config.MessageWebSocketConfiguration

ruoyi-common/ruoyi-common-security/src/main/java/org/dromara/common/security/config/SecurityConfig.java

@@ -39,8 +39,8 @@ import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
 public class SecurityConfig implements WebMvcConfigurer {
 
     private final SecurityProperties securityProperties;
-    @Value("${sse.path}")
+    @Value("${message.path:/resource/message}")
-    private String ssePath;
+    private String messagePath;
 
     /**
      * 注册 Sa-Token 路由拦截器并配置鉴权规则。
@@ -85,7 +85,7 @@ public class SecurityConfig implements WebMvcConfigurer {
             })).addPathPatterns("/**")
             // 排除不需要拦截的路径
             .excludePathPatterns(securityProperties.getExcludes())
-            .excludePathPatterns(ssePath);
+            .excludePathPatterns(messagePath);
     }
 
     /**

ruoyi-common/ruoyi-common-sse/pom.xml

@@ -1,36 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <parent>
-        <groupId>org.dromara</groupId>
-        <artifactId>ruoyi-common</artifactId>
-        <version>${revision}</version>
-    </parent>
-    <modelVersion>4.0.0</modelVersion>
-
-    <artifactId>ruoyi-common-sse</artifactId>
-
-    <description>
-        ruoyi-common-sse 模块
-    </description>
-
-    <dependencies>
-        <dependency>
-            <groupId>org.dromara</groupId>
-            <artifactId>ruoyi-common-core</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>org.dromara</groupId>
-            <artifactId>ruoyi-common-redis</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>org.dromara</groupId>
-            <artifactId>ruoyi-common-satoken</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>org.dromara</groupId>
-            <artifactId>ruoyi-common-json</artifactId>
-        </dependency>
-    </dependencies>
-</project>

ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/config/SseAutoConfiguration.java

@@ -1,36 +0,0 @@
-package org.dromara.common.sse.config;
-
-import org.dromara.common.sse.controller.SseController;
-import org.dromara.common.sse.core.SseEmitterManager;
-import org.dromara.common.sse.listener.SseTopicListener;
-import org.springframework.boot.autoconfigure.AutoConfiguration;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.boot.context.properties.EnableConfigurationProperties;
-import org.springframework.context.annotation.Bean;
-
-/**
- * SSE 自动装配
- *
- * @author Lion Li
- */
-@AutoConfiguration
-@ConditionalOnProperty(value = "sse.enabled", havingValue = "true")
-@EnableConfigurationProperties(SseProperties.class)
-public class SseAutoConfiguration {
-
-    @Bean
-    public SseEmitterManager sseEmitterManager() {
-        return new SseEmitterManager();
-    }
-
-    @Bean
-    public SseTopicListener sseTopicListener() {
-        return new SseTopicListener();
-    }
-
-    @Bean
-    public SseController sseController(SseEmitterManager sseEmitterManager) {
-        return new SseController(sseEmitterManager);
-    }
-
-}

ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/config/SseProperties.java

@@ -1,21 +0,0 @@
-package org.dromara.common.sse.config;
-
-import lombok.Data;
-import org.springframework.boot.context.properties.ConfigurationProperties;
-
-/**
- * SSE 配置项
- *
- * @author Lion Li
- */
-@Data
-@ConfigurationProperties("sse")
-public class SseProperties {
-
-    private Boolean enabled;
-
-    /**
-     * 路径
-     */
-    private String path;
-}

ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/dto/SseMessageDTO.java

@@ -1,29 +0,0 @@
-package org.dromara.common.sse.dto;
-
-import lombok.Data;
-
-import java.io.Serial;
-import java.io.Serializable;
-import java.util.List;
-
-/**
- * 消息的dto
- *
- * @author zendwang
- */
-@Data
-public class SseMessageDTO implements Serializable {
-
-    @Serial
-    private static final long serialVersionUID = 1L;
-
-    /**
-     * 需要推送到的session key 列表
-     */
-    private List<Long> userIds;
-
-    /**
-     * 需要发送的消息
-     */
-    private String message;
-}

ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/listener/SseTopicListener.java

@@ -1,48 +0,0 @@
-package org.dromara.common.sse.listener;
-
-import cn.hutool.core.collection.CollUtil;
-import lombok.extern.slf4j.Slf4j;
-import org.dromara.common.sse.core.SseEmitterManager;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.ApplicationArguments;
-import org.springframework.boot.ApplicationRunner;
-import org.springframework.core.Ordered;
-
-/**
- * SSE 主题订阅监听器
- *
- * @author Lion Li
- */
-@Slf4j
-public class SseTopicListener implements ApplicationRunner, Ordered {
-
-    @Autowired
-    private SseEmitterManager sseEmitterManager;
-
-    /**
-     * 在Spring Boot应用程序启动时初始化SSE主题订阅监听器
-     *
-     * @param args 应用程序参数
-     * @throws Exception 初始化过程中可能抛出的异常
-     */
-    @Override
-    public void run(ApplicationArguments args) throws Exception {
-        sseEmitterManager.subscribeMessage((message) -> {
-            log.info("SSE主题订阅收到消息session keys={} message={}", message.getUserIds(), message.getMessage());
-            // 如果key不为空就按照key发消息 如果为空就群发
-            if (CollUtil.isNotEmpty(message.getUserIds())) {
-                message.getUserIds().forEach(key -> {
-                    sseEmitterManager.sendMessage(key, message.getMessage());
-                });
-            } else {
-                sseEmitterManager.sendMessage(message.getMessage());
-            }
-        });
-        log.info("初始化SSE主题订阅监听器成功");
-    }
-
-    @Override
-    public int getOrder() {
-        return -1;
-    }
-}

ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/utils/SseMessageUtils.java

@@ -1,86 +0,0 @@
-package org.dromara.common.sse.utils;
-
-import lombok.AccessLevel;
-import lombok.NoArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import org.dromara.common.core.utils.SpringUtils;
-import org.dromara.common.sse.core.SseEmitterManager;
-import org.dromara.common.sse.dto.SseMessageDTO;
-
-/**
- * SSE工具类
- *
- * @author Lion Li
- */
-@Slf4j
-@NoArgsConstructor(access = AccessLevel.PRIVATE)
-public class SseMessageUtils {
-
-    private final static Boolean SSE_ENABLE = SpringUtils.getProperty("sse.enabled", Boolean.class, true);
-    private static SseEmitterManager MANAGER;
-
-    static {
-        if (isEnable() && MANAGER == null) {
-            MANAGER = SpringUtils.getBean(SseEmitterManager.class);
-        }
-    }
-
-    /**
-     * 向指定用户的 SSE 会话发送消息。
-     *
-     * @param userId  要发送消息的用户id
-     * @param message 要发送的消息内容
-     */
-    public static void sendMessage(Long userId, String message) {
-        if (!isEnable()) {
-            return;
-        }
-        MANAGER.sendMessage(userId, message);
-    }
-
-    /**
-     * 向当前节点上的所有 SSE 会话发送消息。
-     *
-     * @param message 要发送的消息内容
-     */
-    public static void sendMessage(String message) {
-        if (!isEnable()) {
-            return;
-        }
-        MANAGER.sendMessage(message);
-    }
-
-    /**
-     * 发布 SSE 订阅消息。
-     *
-     * @param sseMessageDTO 要发布的SSE消息对象
-     */
-    public static void publishMessage(SseMessageDTO sseMessageDTO) {
-        if (!isEnable()) {
-            return;
-        }
-        MANAGER.publishMessage(sseMessageDTO);
-    }
-
-    /**
-     * 向所有用户发布 SSE 广播消息。
-     *
-     * @param message 要发布的消息内容
-     */
-    public static void publishAll(String message) {
-        if (!isEnable()) {
-            return;
-        }
-        MANAGER.publishAll(message);
-    }
-
-    /**
-     * 判断 SSE 功能是否启用。
-     *
-     * @return 是否启用
-     */
-    public static Boolean isEnable() {
-        return SSE_ENABLE;
-    }
-
-}

ruoyi-common/ruoyi-common-sse/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -1 +0,0 @@
-org.dromara.common.sse.config.SseAutoConfiguration

ruoyi-common/ruoyi-common-web/src/main/java/org/dromara/common/web/handler/GlobalExceptionHandler.java

@@ -11,6 +11,7 @@ import org.dromara.common.core.domain.R;
 import org.dromara.common.core.exception.ServiceException;
 import org.dromara.common.core.exception.SseException;
 import org.dromara.common.core.exception.base.BaseException;
+import org.dromara.common.core.utils.SpringUtils;
 import org.dromara.common.core.utils.StreamUtils;
 import org.dromara.common.json.utils.JsonUtils;
 import org.springframework.boot.json.JsonParseException;
@@ -129,7 +130,8 @@ public class GlobalExceptionHandler {
     @ExceptionHandler(IOException.class)
     public void handleIoException(IOException e, HttpServletRequest request) {
         String requestURI = request.getRequestURI();
-        if (requestURI.contains("sse")) {
+        String path = SpringUtils.getProperty("message.path");
+        if (requestURI.contains(path)) {
             // sse 经常性连接中断 例如关闭浏览器 直接屏蔽
             return;
         }

ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/config/WebSocketConfig.java

@@ -1,63 +0,0 @@
-package org.dromara.common.websocket.config;
-
-import cn.hutool.core.util.StrUtil;
-import org.dromara.common.websocket.config.properties.WebSocketProperties;
-import org.dromara.common.websocket.handler.PlusWebSocketHandler;
-import org.dromara.common.websocket.interceptor.PlusWebSocketInterceptor;
-import org.dromara.common.websocket.listener.WebSocketTopicListener;
-import org.springframework.boot.autoconfigure.AutoConfiguration;
-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.config.annotation.EnableWebSocket;
-import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
-import org.springframework.web.socket.server.HandshakeInterceptor;
-
-/**
- * WebSocket 配置
- *
- * @author zendwang
- */
-@AutoConfiguration
-@ConditionalOnProperty(value = "websocket.enabled", havingValue = "true")
-@EnableConfigurationProperties(WebSocketProperties.class)
-@EnableWebSocket
-public class WebSocketConfig {
-
-    @Bean
-    public WebSocketConfigurer webSocketConfigurer(HandshakeInterceptor handshakeInterceptor,
-                                                   WebSocketHandler webSocketHandler, WebSocketProperties webSocketProperties) {
-        // 如果WebSocket的路径为空，则设置默认路径为 "/websocket"
-        if (StrUtil.isBlank(webSocketProperties.getPath())) {
-            webSocketProperties.setPath("/websocket");
-        }
-
-        // 如果允许跨域访问的地址为空，则设置为 "*"，表示允许所有来源的跨域请求
-        if (StrUtil.isBlank(webSocketProperties.getAllowedOrigins())) {
-            webSocketProperties.setAllowedOrigins("*");
-        }
-
-        // 返回一个WebSocketConfigurer对象，用于配置WebSocket
-        return registry -> registry
-            // 添加WebSocket处理程序和拦截器到指定路径，设置允许的跨域来源
-            .addHandler(webSocketHandler, webSocketProperties.getPath())
-            .addInterceptors(handshakeInterceptor)
-            .setAllowedOrigins(webSocketProperties.getAllowedOrigins());
-    }
-
-    @Bean
-    public HandshakeInterceptor handshakeInterceptor() {
-        return new PlusWebSocketInterceptor();
-    }
-
-    @Bean
-    public WebSocketHandler webSocketHandler() {
-        return new PlusWebSocketHandler();
-    }
-
-    @Bean
-    public WebSocketTopicListener topicListener() {
-        return new WebSocketTopicListener();
-    }
-}

ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/config/properties/WebSocketProperties.java

@@ -1,26 +0,0 @@
-package org.dromara.common.websocket.config.properties;
-
-import lombok.Data;
-import org.springframework.boot.context.properties.ConfigurationProperties;
-
-/**
- * WebSocket 配置项
- *
- * @author zendwang
- */
-@ConfigurationProperties("websocket")
-@Data
-public class WebSocketProperties {
-
-    private Boolean enabled;
-
-    /**
-     * 路径
-     */
-    private String path;
-
-    /**
-     *  设置访问源地址
-     */
-    private String allowedOrigins;
-}

ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/constant/WebSocketConstants.java

@@ -1,29 +0,0 @@
-package org.dromara.common.websocket.constant;
-
-/**
- * websocket的常量配置
- *
- * @author zendwang
- */
-public interface WebSocketConstants {
-
-    /**
-     * websocketSession中的参数的key
-     */
-    String LOGIN_USER_KEY = "loginUser";
-
-    /**
-     * 订阅的频道
-     */
-    String WEB_SOCKET_TOPIC = "global:websocket";
-
-    /**
-     * 前端心跳检查的命令
-     */
-    String PING = "ping";
-
-    /**
-     * 服务端心跳恢复的字符串
-     */
-    String PONG = "pong";
-}

ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/dto/WebSocketMessageDTO.java

@@ -1,29 +0,0 @@
-package org.dromara.common.websocket.dto;
-
-import lombok.Data;
-
-import java.io.Serial;
-import java.io.Serializable;
-import java.util.List;
-
-/**
- * 消息的dto
- *
- * @author zendwang
- */
-@Data
-public class WebSocketMessageDTO implements Serializable {
-
-    @Serial
-    private static final long serialVersionUID = 1L;
-
-    /**
-     * 需要推送到的session key 列表
-     */
-    private List<Long> sessionKeys;
-
-    /**
-     * 需要发送的消息
-     */
-    private String message;
-}

ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/handler/PlusWebSocketHandler.java

@@ -1,123 +0,0 @@
-package org.dromara.common.websocket.handler;
-
-import cn.hutool.core.util.ObjectUtil;
-import lombok.extern.slf4j.Slf4j;
-import org.dromara.common.core.domain.model.LoginUser;
-import org.dromara.common.websocket.dto.WebSocketMessageDTO;
-import org.dromara.common.websocket.holder.WebSocketSessionHolder;
-import org.dromara.common.websocket.utils.WebSocketUtils;
-import org.springframework.web.socket.*;
-import org.springframework.web.socket.handler.AbstractWebSocketHandler;
-import org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator;
-
-import java.io.IOException;
-import java.util.List;
-
-import static org.dromara.common.websocket.constant.WebSocketConstants.LOGIN_USER_KEY;
-
-/**
- * WebSocketHandler 实现类
- *
- * @author zendwang
- */
-@Slf4j
-public class PlusWebSocketHandler extends AbstractWebSocketHandler {
-
-    /**
-     * 连接成功后
-     */
-    @Override
-    public void afterConnectionEstablished(WebSocketSession session) throws IOException {
-        LoginUser loginUser = (LoginUser) session.getAttributes().get(LOGIN_USER_KEY);
-        if (ObjectUtil.isNull(loginUser)) {
-            session.close(CloseStatus.BAD_DATA);
-            log.info("[connect] invalid token received. sessionId: {}", session.getId());
-            return;
-        }
-        WebSocketSessionHolder.addSession(loginUser.getUserId(), new ConcurrentWebSocketSessionDecorator(session, 10 * 1000, 64000));
-        log.info("[connect] sessionId: {},userId:{},userType:{}", session.getId(), loginUser.getUserId(), loginUser.getUserType());
-    }
-
-    /**
-     * 处理接收到的文本消息
-     *
-     * @param session WebSocket会话
-     * @param message 接收到的文本消息
-     * @throws Exception 处理消息过程中可能抛出的异常
-     */
-    @Override
-    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
-        // 从WebSocket会话中获取登录用户信息
-        LoginUser loginUser = (LoginUser) session.getAttributes().get(LOGIN_USER_KEY);
-
-        // 创建WebSocket消息DTO对象
-        WebSocketMessageDTO messageDTO = new WebSocketMessageDTO();
-        messageDTO.setSessionKeys(List.of(loginUser.getUserId()));
-        messageDTO.setMessage(message.getPayload());
-        WebSocketUtils.publishMessage(messageDTO);
-    }
-
-    /**
-     * 处理接收到的二进制消息
-     *
-     * @param session WebSocket会话
-     * @param message 接收到的二进制消息
-     * @throws Exception 处理消息过程中可能抛出的异常
-     */
-    @Override
-    protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) throws Exception {
-        super.handleBinaryMessage(session, message);
-    }
-
-    /**
-     * 处理接收到的Pong消息（心跳监测）
-     *
-     * @param session WebSocket会话
-     * @param message 接收到的Pong消息
-     * @throws Exception 处理消息过程中可能抛出的异常
-     */
-    @Override
-    protected void handlePongMessage(WebSocketSession session, PongMessage message) throws Exception {
-        WebSocketUtils.sendPongMessage(session);
-    }
-
-    /**
-     * 处理WebSocket传输错误
-     *
-     * @param session   WebSocket会话
-     * @param exception 发生的异常
-     * @throws Exception 处理过程中可能抛出的异常
-     */
-    @Override
-    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
-        log.error("[transport error] sessionId: {} , exception:{}", session.getId(), exception.getMessage());
-    }
-
-    /**
-     * 在WebSocket连接关闭后执行清理操作
-     *
-     * @param session WebSocket会话
-     * @param status  关闭状态信息
-     */
-    @Override
-    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
-        LoginUser loginUser = (LoginUser) session.getAttributes().get(LOGIN_USER_KEY);
-        if (ObjectUtil.isNull(loginUser)) {
-            log.info("[disconnect] invalid token received. sessionId: {}", session.getId());
-            return;
-        }
-        WebSocketSessionHolder.removeSession(loginUser.getUserId());
-        log.info("[disconnect] sessionId: {},userId:{},userType:{}", session.getId(), loginUser.getUserId(), loginUser.getUserType());
-    }
-
-    /**
-     * 指示处理程序是否支持接收部分消息
-     *
-     * @return 如果支持接收部分消息，则返回true；否则返回false
-     */
-    @Override
-    public boolean supportsPartialMessages() {
-        return false;
-    }
-
-}

ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/holder/WebSocketSessionHolder.java

@@ -1,74 +0,0 @@
-package org.dromara.common.websocket.holder;
-
-import lombok.AccessLevel;
-import lombok.NoArgsConstructor;
-import org.springframework.web.socket.CloseStatus;
-import org.springframework.web.socket.WebSocketSession;
-
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-
-/**
- * WebSocketSession 用于保存当前所有在线的会话信息
- *
- * @author zendwang
- */
-@NoArgsConstructor(access = AccessLevel.PRIVATE)
-public class WebSocketSessionHolder {
-
-    private static final Map<Long, WebSocketSession> USER_SESSION_MAP = new ConcurrentHashMap<>();
-
-    /**
-     * 将WebSocket会话添加到用户会话Map中
-     *
-     * @param sessionKey 会话键，用于检索会话
-     * @param session    要添加的WebSocket会话
-     */
-    public static void addSession(Long sessionKey, WebSocketSession session) {
-        removeSession(sessionKey);
-        USER_SESSION_MAP.put(sessionKey, session);
-    }
-
-    /**
-     * 从用户会话Map中移除指定会话键对应的WebSocket会话
-     *
-     * @param sessionKey 要移除的会话键
-     */
-    public static void removeSession(Long sessionKey) {
-        WebSocketSession session = USER_SESSION_MAP.remove(sessionKey);
-        try {
-            session.close(CloseStatus.BAD_DATA);
-        } catch (Exception ignored) {
-        }
-    }
-
-    /**
-     * 根据会话键从用户会话Map中获取WebSocket会话
-     *
-     * @param sessionKey 要获取的会话键
-     * @return 与给定会话键对应的WebSocket会话，如果不存在则返回null
-     */
-    public static WebSocketSession getSessions(Long sessionKey) {
-        return USER_SESSION_MAP.get(sessionKey);
-    }
-
-    /**
-     * 获取存储在用户会话Map中所有WebSocket会话的会话键集合
-     *
-     * @return 所有WebSocket会话的会话键集合
-     */
-    public static Set<Long> getSessionsAll() {
-        return USER_SESSION_MAP.keySet();
-    }
-
-    /**
-     * 检查给定的会话键是否存在于用户会话Map中
-     *
-     * @param sessionKey 要检查的会话键
-     * @return 如果存在对应的会话键，则返回true；否则返回false
-     */
-    public static Boolean existSession(Long sessionKey) {
-        return USER_SESSION_MAP.containsKey(sessionKey);
-    }
-}

ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/listener/WebSocketTopicListener.java

@@ -1,50 +0,0 @@
-package org.dromara.common.websocket.listener;
-
-import cn.hutool.core.collection.CollUtil;
-import lombok.extern.slf4j.Slf4j;
-import org.dromara.common.websocket.holder.WebSocketSessionHolder;
-import org.dromara.common.websocket.utils.WebSocketUtils;
-import org.springframework.boot.ApplicationArguments;
-import org.springframework.boot.ApplicationRunner;
-import org.springframework.core.Ordered;
-
-/**
- * WebSocket 主题订阅监听器
- *
- * @author zendwang
- */
-@Slf4j
-public class WebSocketTopicListener implements ApplicationRunner, Ordered {
-
-    /**
-     * 在Spring Boot应用程序启动时初始化WebSocket主题订阅监听器
-     *
-     * @param args 应用程序参数
-     * @throws Exception 初始化过程中可能抛出的异常
-     */
-    @Override
-    public void run(ApplicationArguments args) throws Exception {
-        // 订阅WebSocket消息
-        WebSocketUtils.subscribeMessage((message) -> {
-            log.info("WebSocket主题订阅收到消息session keys={} message={}", message.getSessionKeys(), message.getMessage());
-            // 如果key不为空就按照key发消息 如果为空就群发
-            if (CollUtil.isNotEmpty(message.getSessionKeys())) {
-                message.getSessionKeys().forEach(key -> {
-                    if (WebSocketSessionHolder.existSession(key)) {
-                        WebSocketUtils.sendMessage(key, message.getMessage());
-                    }
-                });
-            } else {
-                WebSocketSessionHolder.getSessionsAll().forEach(key -> {
-                    WebSocketUtils.sendMessage(key, message.getMessage());
-                });
-            }
-        });
-        log.info("初始化WebSocket主题订阅监听器成功");
-    }
-
-    @Override
-    public int getOrder() {
-        return -1;
-    }
-}

ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/utils/WebSocketUtils.java

@@ -1,127 +0,0 @@
-package org.dromara.common.websocket.utils;
-
-import cn.hutool.core.collection.CollUtil;
-import lombok.AccessLevel;
-import lombok.NoArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import org.dromara.common.redis.utils.RedisUtils;
-import org.dromara.common.websocket.dto.WebSocketMessageDTO;
-import org.dromara.common.websocket.holder.WebSocketSessionHolder;
-import org.springframework.web.socket.PongMessage;
-import org.springframework.web.socket.TextMessage;
-import org.springframework.web.socket.WebSocketMessage;
-import org.springframework.web.socket.WebSocketSession;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.function.Consumer;
-
-import static org.dromara.common.websocket.constant.WebSocketConstants.WEB_SOCKET_TOPIC;
-
-/**
- * 工具类
- *
- * @author zendwang
- */
-@Slf4j
-@NoArgsConstructor(access = AccessLevel.PRIVATE)
-public class WebSocketUtils {
-
-    /**
-     * 向指定会话标识发送文本消息。
-     *
-     * @param sessionKey 要发送消息的用户id
-     * @param message    要发送的消息内容
-     */
-    public static void sendMessage(Long sessionKey, String message) {
-        WebSocketSession session = WebSocketSessionHolder.getSessions(sessionKey);
-        sendMessage(session, message);
-    }
-
-    /**
-     * 订阅 WebSocket 广播主题消息。
-     *
-     * @param consumer 处理WebSocket消息的消费者函数
-     */
-    public static void subscribeMessage(Consumer<WebSocketMessageDTO> consumer) {
-        RedisUtils.subscribe(WEB_SOCKET_TOPIC, WebSocketMessageDTO.class, consumer);
-    }
-
-    /**
-     * 按会话标识发布 WebSocket 消息。
-     *
-     * @param webSocketMessage 要发布的WebSocket消息对象
-     */
-    public static void publishMessage(WebSocketMessageDTO webSocketMessage) {
-        List<Long> unsentSessionKeys = new ArrayList<>();
-        // 当前服务内session,直接发送消息
-        for (Long sessionKey : webSocketMessage.getSessionKeys()) {
-            if (WebSocketSessionHolder.existSession(sessionKey)) {
-                WebSocketUtils.sendMessage(sessionKey, webSocketMessage.getMessage());
-                continue;
-            }
-            unsentSessionKeys.add(sessionKey);
-        }
-        // 不在当前服务内session,发布订阅消息
-        if (CollUtil.isNotEmpty(unsentSessionKeys)) {
-            WebSocketMessageDTO broadcastMessage = new WebSocketMessageDTO();
-            broadcastMessage.setSessionKeys(unsentSessionKeys);
-            broadcastMessage.setMessage(webSocketMessage.getMessage());
-            RedisUtils.publish(WEB_SOCKET_TOPIC, broadcastMessage, consumer -> {
-                log.info(" WebSocket发送主题订阅消息topic:{} session keys:{} message:{}",
-                    WEB_SOCKET_TOPIC, unsentSessionKeys, webSocketMessage.getMessage());
-            });
-        }
-    }
-
-    /**
-     * 向所有 WebSocket 会话发布广播消息。
-     *
-     * @param message 要发布的消息内容
-     */
-    public static void publishAll(String message) {
-        WebSocketMessageDTO broadcastMessage = new WebSocketMessageDTO();
-        broadcastMessage.setMessage(message);
-        RedisUtils.publish(WEB_SOCKET_TOPIC, broadcastMessage, consumer -> {
-            log.info("WebSocket发送主题订阅消息topic:{} message:{}", WEB_SOCKET_TOPIC, message);
-        });
-    }
-
-    /**
-     * 向指定会话发送 Pong 心跳消息。
-     *
-     * @param session 要发送Pong消息的WebSocket会话
-     */
-    public static void sendPongMessage(WebSocketSession session) {
-        sendMessage(session, new PongMessage());
-    }
-
-    /**
-     * 向指定 WebSocket 会话发送文本消息。
-     *
-     * @param session WebSocket会话
-     * @param message 要发送的文本消息内容
-     */
-    public static void sendMessage(WebSocketSession session, String message) {
-        sendMessage(session, new TextMessage(message));
-    }
-
-    /**
-     * 向指定 WebSocket 会话发送原始消息对象。
-     *
-     * @param session WebSocket会话
-     * @param message 要发送的WebSocket消息对象
-     */
-    private static void sendMessage(WebSocketSession session, WebSocketMessage<?> message) {
-        if (session == null || !session.isOpen()) {
-            log.warn("[send] session会话已经关闭");
-        } else {
-            try {
-                session.sendMessage(message);
-            } catch (IOException e) {
-                log.error("[send] session({}) 发送消息({}) 异常", session, message, e);
-            }
-        }
-    }
-}

ruoyi-common/ruoyi-common-websocket/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -1 +0,0 @@
-org.dromara.common.websocket.config.WebSocketConfig

ruoyi-modules/ruoyi-demo/pom.xml

@@ -85,7 +85,7 @@
 
         <dependency>
             <groupId>org.dromara</groupId>
-            <artifactId>ruoyi-common-websocket</artifactId>
+            <artifactId>ruoyi-common-push</artifactId>
         </dependency>
 
         <dependency>

ruoyi-modules/ruoyi-demo/src/main/java/org/dromara/demo/controller/WebSocketController.java

@@ -1,14 +1,18 @@
 package org.dromara.demo.controller;
 
+import org.dromara.common.core.domain.R;
+import org.dromara.common.core.domain.dto.PushPayload;
+import org.dromara.common.core.enums.PushSourceEnum;
+import org.dromara.common.core.enums.PushTypeEnum;
+import org.dromara.common.push.helper.PushHelper;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.dromara.common.core.domain.R;
-import org.dromara.common.websocket.dto.WebSocketMessageDTO;
-import org.dromara.common.websocket.utils.WebSocketUtils;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 
+import java.util.List;
+
 /**
  * WebSocket 演示案例
  *
@@ -23,11 +27,22 @@ public class WebSocketController {
     /**
      * 发布消息
      *
-     * @param dto 发送内容
+     * @param userId 目标用户
+     * @param message 发送内容
      */
     @GetMapping("/send")
-    public R<Void> send(WebSocketMessageDTO dto) throws InterruptedException {
+    public R<Void> send(Long userId, String message) {
-        WebSocketUtils.publishMessage(dto);
+        PushPayload payload = PushPayload.of(
+            PushTypeEnum.MESSAGE,
+            PushSourceEnum.BACKEND,
+            message,
+            null
+        );
+        if (userId == null) {
+            PushHelper.publishAll(payload);
+        } else {
+            PushHelper.publishMessage(List.of(userId), payload);
+        }
         return R.ok("操作成功");
     }
 }

ruoyi-modules/ruoyi-generator/src/main/java/org/dromara/generator/constant/GenConstants.java

@@ -73,30 +73,30 @@ public interface GenConstants {
      * BO对象 不需要添加字段
      */
     String[] COLUMNNAME_NOT_ADD = {"create_dept", "create_by", "create_time", "del_flag", "update_by",
-        "update_time", "version", "tenant_id"};
+        "update_time", "version"};
 
     /**
      * BO对象 不需要编辑字段
      */
     String[] COLUMNNAME_NOT_EDIT = {"create_dept", "create_by", "create_time", "del_flag", "update_by",
-        "update_time", "version", "tenant_id"};
+        "update_time", "version"};
 
     /**
      * VO对象 不需要返回字段
      */
     String[] COLUMNNAME_NOT_LIST = {"create_dept", "create_by", "create_time", "del_flag", "update_by",
-        "update_time", "version", "tenant_id"};
+        "update_time", "version"};
 
     /**
      * BO对象 不需要查询字段
      */
     String[] COLUMNNAME_NOT_QUERY = {"id", "create_dept", "create_by", "create_time", "del_flag", "update_by",
-        "update_time", "remark", "version", "tenant_id"};
+        "update_time", "remark", "version"};
 
     /**
      * Entity基类字段
      */
-    String[] BASE_ENTITY = {"createDept", "createBy", "createTime", "updateBy", "updateTime", "tenantId"};
+    String[] BASE_ENTITY = {"createDept", "createBy", "createTime", "updateBy", "updateTime"};
 
     /**
      * 文本框

ruoyi-modules/ruoyi-system/pom.xml

@@ -82,12 +82,7 @@
 
         <dependency>
             <groupId>org.dromara</groupId>
-            <artifactId>ruoyi-common-websocket</artifactId>
+            <artifactId>ruoyi-common-push</artifactId>
-        </dependency>
-
-        <dependency>
-            <groupId>org.dromara</groupId>
-            <artifactId>ruoyi-common-sse</artifactId>
         </dependency>
 
     </dependencies>

ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/controller/system/SysNoticeController.java

@@ -4,12 +4,15 @@ import cn.dev33.satoken.annotation.SaCheckPermission;
 import lombok.RequiredArgsConstructor;
 import org.dromara.common.core.domain.PageResult;
 import org.dromara.common.core.domain.R;
+import org.dromara.common.core.domain.dto.PushPayload;
+import org.dromara.common.core.enums.PushSourceEnum;
+import org.dromara.common.core.enums.PushTypeEnum;
 import org.dromara.common.core.service.DictService;
 import org.dromara.common.log.annotation.Log;
 import org.dromara.common.log.enums.BusinessType;
 import org.dromara.common.mybatis.core.page.PageQuery;
+import org.dromara.common.push.helper.PushHelper;
 import org.dromara.common.redis.annotation.RepeatSubmit;
-import org.dromara.common.sse.utils.SseMessageUtils;
 import org.dromara.common.web.core.BaseController;
 import org.dromara.system.domain.bo.SysNoticeBo;
 import org.dromara.system.domain.vo.SysNoticeVo;
@@ -17,6 +20,9 @@ import org.dromara.system.service.ISysNoticeService;
 import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.*;
 
+import java.util.HashMap;
+import java.util.Map;
+
 /**
  * 公告 信息操作处理
  *
@@ -72,7 +78,18 @@ public class SysNoticeController extends BaseController {
             return R.fail();
         }
         String type = dictService.getDictLabel("sys_notice_type", notice.getNoticeType());
-        SseMessageUtils.publishAll("[" + type + "] " + notice.getNoticeTitle());
+        Map<String, Object> data = new HashMap<>(4);
+        data.put("noticeType", notice.getNoticeType());
+        data.put("noticeTypeLabel", type);
+        data.put("noticeTitle", notice.getNoticeTitle());
+        data.put("noticeId", notice.getNoticeId());
+        PushHelper.publishAll(PushPayload.of(
+            PushTypeEnum.NOTICE,
+            PushSourceEnum.NOTICE,
+            "[" + type + "] " + notice.getNoticeTitle(),
+            data,
+            "/system/notice?noticeId=" + notice.getNoticeId()
+        ));
         return R.ok();
     }
 

ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/controller/system/SysOssController.java

@@ -2,7 +2,6 @@ package org.dromara.system.controller.system;
 
 
 import cn.dev33.satoken.annotation.SaCheckPermission;
-import jakarta.servlet.http.HttpServletResponse;
 import jakarta.validation.constraints.NotEmpty;
 import lombok.RequiredArgsConstructor;
 import org.dromara.common.core.domain.PageResult;
@@ -16,6 +15,7 @@ import org.dromara.system.domain.bo.SysOssBo;
 import org.dromara.system.domain.vo.SysOssVo;
 import org.dromara.system.service.ISysOssService;
 import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
 import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.*;
 import org.springframework.web.multipart.MultipartFile;
@@ -83,13 +83,12 @@ public class SysOssController extends BaseController {
      * 下载OSS对象
      *
      * @param ossId OSS对象ID
-     * @param response HTTP 响应
      * @throws IOException IO 异常
      */
     @SaCheckPermission("system:oss:download")
     @GetMapping("/download/{ossId}")
-    public void download(@PathVariable Long ossId, HttpServletResponse response) throws IOException {
+    public ResponseEntity<byte[]> download(@PathVariable Long ossId) throws IOException {
-        ossService.download(ossId, response);
+        return ossService.download(ossId);
     }
 
     /**

ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/domain/bo/SysUserProfileBo.java

@@ -51,6 +51,6 @@ public class SysUserProfileBo implements Serializable {
     /**
      * 用户性别（0男 1女 2未知）
      */
-    private String sex;
+    private String gender;
 
 }

ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/domain/vo/ProfileUserVo.java

@@ -58,7 +58,7 @@ public class ProfileUserVo implements Serializable {
     /**
      * 用户性别（0男 1女 2未知）
      */
-    private String sex;
+    private String gender;
 
     /**
      * 头像地址

ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/mapper/SysDeptMapper.java

@@ -75,7 +75,6 @@ public interface SysDeptMapper extends BaseMapperPlus<SysDept, SysDeptVo> {
         return this.selectList(new MPJLambdaWrapper<SysDept>()
             .select(SysDept::getDeptId)
             .findInSet(SysDept::getAncestors, parentId));
-
     }
 
 

ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/service/ISysOssService.java

@@ -1,14 +1,13 @@
 package org.dromara.system.service;
 
-import jakarta.servlet.http.HttpServletResponse;
 import org.dromara.common.core.domain.PageResult;
 import org.dromara.common.mybatis.core.page.PageQuery;
 import org.dromara.system.domain.bo.SysOssBo;
 import org.dromara.system.domain.vo.SysOssVo;
+import org.springframework.http.ResponseEntity;
 import org.springframework.web.multipart.MultipartFile;
 
 import java.io.File;
-import java.io.IOException;
 import java.util.Collection;
 import java.util.List;
 
@@ -64,9 +63,8 @@ public interface ISysOssService {
      * 文件下载方法，支持一次性下载完整文件
      *
      * @param ossId    OSS对象ID
-     * @param response HttpServletResponse对象，用于设置响应头和向客户端发送文件内容
      */
-    void download(Long ossId, HttpServletResponse response) throws IOException;
+    ResponseEntity<byte[]> download(Long ossId);
 
     /**
      * 删除OSS对象存储

ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/service/impl/SysDeptServiceImpl.java

@@ -114,20 +114,8 @@ public class SysDeptServiceImpl implements ISysDeptService, DeptService {
                 List<Long> deptIds = baseMapper.selectDeptAndChildById(bo.getBelongDeptId());
                 x.in(SysDept::getDeptId, deptIds);
             });
-
-//            //部门树搜索
-//            wrapper.in(SysUser::getDeptId, SysDept.class,
-//                in -> in
-//                    .select(SysDept::getDeptId)
-//                    .and(q -> q
-//                        .eq(SysDept::getDeptId, bo.getBelongDeptId())
-//                        .or()
-//                        .findInSetIfExists(SysDept::getAncestors, bo.getDeptId())
-//                    )
-//            );
         }
 
-
         return wrapper;
     }
 

ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/service/impl/SysNoticeServiceImpl.java

@@ -97,7 +97,9 @@ public class SysNoticeServiceImpl implements ISysNoticeService {
     @Override
     public int insertNotice(SysNoticeBo bo) {
         SysNotice notice = MapstructUtils.convert(bo, SysNotice.class);
-        return baseMapper.insert(notice);
+        int rows = baseMapper.insert(notice);
+        bo.setNoticeId(notice.getNoticeId());
+        return rows;
     }
 
     /**

ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/service/impl/SysOssServiceImpl.java

@@ -2,11 +2,15 @@ package org.dromara.system.service.impl;
 
 import cn.hutool.core.bean.BeanUtil;
 import cn.hutool.core.convert.Convert;
+import cn.hutool.core.io.IoUtil;
 import cn.hutool.core.util.ObjectUtil;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.github.yulichang.wrapper.MPJLambdaWrapper;
 import jakarta.servlet.http.HttpServletResponse;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.dromara.common.core.constant.CacheNames;
 import org.dromara.common.core.domain.PageResult;
 import org.dromara.common.core.domain.dto.OssDTO;
@@ -22,7 +26,7 @@ import org.dromara.common.mybatis.core.page.PageQuery;
 import org.dromara.common.oss.client.OssClient;
 import org.dromara.common.oss.enums.AccessPolicy;
 import org.dromara.common.oss.factory.OssFactory;
-import org.dromara.common.oss.model.GetObjectResult;
+import org.dromara.common.oss.model.Options;
 import org.dromara.common.oss.model.PutObjectResult;
 import org.dromara.common.oss.util.S3ObjectUtil;
 import org.dromara.system.domain.SysOss;
@@ -33,12 +37,15 @@ import org.dromara.system.mapper.SysOssMapper;
 import org.dromara.system.service.ISysOssService;
 import org.jetbrains.annotations.NotNull;
 import org.springframework.cache.annotation.Cacheable;
+import org.springframework.http.HttpHeaders;
 import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Service;
 import org.springframework.web.multipart.MultipartFile;
 
 import java.io.File;
 import java.io.IOException;
+import java.io.InputStream;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -50,6 +57,7 @@ import java.util.Map;
  *
  * @author Lion Li
  */
+@Slf4j
 @RequiredArgsConstructor
 @Service
 public class SysOssServiceImpl implements ISysOssService, OssService {
@@ -180,20 +188,34 @@ public class SysOssServiceImpl implements ISysOssService, OssService {
     /**
      * 文件下载方法，支持一次性下载完整文件
      *
-     * @param ossId    OSS对象ID
+     * @param ossId OSS对象ID
-     * @param response HttpServletResponse对象，用于设置响应头和向客户端发送文件内容
      */
     @Override
-    public void download(Long ossId, HttpServletResponse response) throws IOException {
+    public ResponseEntity<byte[]> download(Long ossId) {
         SysOssVo sysOss = SpringUtils.getAopProxy(this).getById(ossId);
         if (ObjectUtil.isNull(sysOss)) {
             throw new ServiceException("文件数据不存在!");
         }
-        FileUtils.setAttachmentResponseHeader(response, sysOss.getOriginalName());
+        String percentEncodedFileName = FileUtils.percentEncode(sysOss.getOriginalName());
-        response.setContentType(MediaType.APPLICATION_OCTET_STREAM_VALUE + "; charset=UTF-8");
+        return OssFactory.instance(sysOss.getService())
-        OssClient instance = OssFactory.instance(sysOss.getService());
+            .download(sysOss.getFileName(), (result, inputStream) -> {
-        GetObjectResult result = instance.download(sysOss.getFileName(), response.getOutputStream());
+                // 尝试解析媒体类型，如果解析失败，则使用 application/octet-stream
-        response.setContentLengthLong(result.size());
+                MediaType mediaType;
+                try {
+                    mediaType = MediaType.parseMediaType(result.contentType());
+                } catch (Exception e) {
+                    mediaType = MediaType.APPLICATION_OCTET_STREAM;
+                }
+                // 构建响应实体
+                return ResponseEntity.ok()
+                    .header(HttpHeaders.ACCESS_CONTROL_EXPOSE_HEADERS, "Content-Disposition,download-filename")
+                    .header(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=%s;filename*=utf-8''%s".formatted(percentEncodedFileName, percentEncodedFileName))
+                    .header("download-filename", percentEncodedFileName)
+                    .contentType(mediaType)
+                    .contentLength(result.size())
+                    .body(IoUtil.readBytes(inputStream));
+            });
+
     }
 
     /**
@@ -213,7 +235,9 @@ public class SysOssServiceImpl implements ISysOssService, OssService {
         OssClient instance = OssFactory.instance();
         try {
             String pathKey = S3ObjectUtil.buildPathKey(originalfileName);
-            PutObjectResult result = instance.upload(pathKey, file.getInputStream(), file.getSize());
+            InputStream inputStream = file.getInputStream();
+            PutObjectResult result = instance.upload(pathKey, inputStream, file.getSize(), Options.builder().setContentType(file.getContentType()));
+            IoUtil.close(inputStream);
             SysOssExt ext1 = new SysOssExt();
             ext1.setFileSize(file.getSize());
             ext1.setContentType(file.getContentType());
@@ -239,7 +263,7 @@ public class SysOssServiceImpl implements ISysOssService, OssService {
         String suffix = StringUtils.substring(originalfileName, originalfileName.lastIndexOf("."), originalfileName.length());
         OssClient instance = OssFactory.instance();
         String pathKey = S3ObjectUtil.buildPathKey(originalfileName);
-        PutObjectResult result = instance.upload(pathKey, file);
+        PutObjectResult result = instance.upload(pathKey, file, Options.builder().setContentType(FileUtils.getMimeType(file.toPath())));
         SysOssExt ext1 = new SysOssExt();
         ext1.setFileSize(result.size());
         // 保存文件信息

ruoyi-modules/ruoyi-workflow/pom.xml

@@ -20,7 +20,7 @@
 
         <dependency>
             <groupId>org.dromara</groupId>
-            <artifactId>ruoyi-common-sse</artifactId>
+            <artifactId>ruoyi-common-push</artifactId>
         </dependency>
 
         <dependency>

ruoyi-modules/ruoyi-workflow/src/main/java/org/dromara/workflow/service/impl/FlwCommonServiceImpl.java

@@ -4,14 +4,16 @@ import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.util.ObjectUtil;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.dromara.common.core.domain.dto.PushPayload;
 import org.dromara.common.core.domain.dto.UserDTO;
+import org.dromara.common.core.enums.PushSourceEnum;
+import org.dromara.common.core.enums.PushTypeEnum;
 import org.dromara.common.core.exception.ServiceException;
 import org.dromara.common.core.utils.SpringUtils;
 import org.dromara.common.core.utils.StreamUtils;
 import org.dromara.common.core.utils.StringUtils;
 import org.dromara.common.mail.utils.MailUtils;
-import org.dromara.common.sse.dto.SseMessageDTO;
+import org.dromara.common.push.helper.PushHelper;
-import org.dromara.common.sse.utils.SseMessageUtils;
 import org.dromara.warm.flow.core.FlowEngine;
 import org.dromara.warm.flow.core.entity.Node;
 import org.dromara.warm.flow.orm.entity.FlowTask;
@@ -92,10 +94,12 @@ public class FlwCommonServiceImpl implements IFlwCommonService {
             try {
                 switch (messageTypeEnum) {
                     case SYSTEM_MESSAGE -> {
-                        SseMessageDTO dto = new SseMessageDTO();
+                        PushHelper.publishMessage(userIds, PushPayload.of(
-                        dto.setUserIds(userIds);
+                            PushTypeEnum.MESSAGE,
-                        dto.setMessage(message);
+                            PushSourceEnum.WORKFLOW,
-                        SseMessageUtils.publishMessage(dto);
+                            message,
+                            null
+                        ));
                     }
                     case EMAIL_MESSAGE -> MailUtils.sendText(emails, subject, message);
                     case SMS_MESSAGE -> {