diff --git a/ruoyi-admin/src/main/java/org/dromara/web/controller/AuthController.java b/ruoyi-admin/src/main/java/org/dromara/web/controller/AuthController.java index fc524645a..f5fdfac2b 100644 --- a/ruoyi-admin/src/main/java/org/dromara/web/controller/AuthController.java +++ b/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(); - dto.setUserIds(List.of(userId)); - dto.setMessage(DateUtils.getTodayHour(new Date()) + "好，欢迎登录 RuoYi-Vue-Plus 后台管理系统"); - SseMessageUtils.publishMessage(dto); + PushHelper.publishMessage( + List.of(userId), + PushPayload.of( + PushTypeEnum.MESSAGE, + PushSourceEnum.BACKEND, + DateUtils.getTodayHour(new Date()) + "好，欢迎登录 RuoYi-Vue-Plus 后台管理系统", + null + ) + ); }, 5, TimeUnit.SECONDS); return R.ok(loginVo); } diff --git a/ruoyi-admin/src/main/java/org/dromara/web/controller/CaptchaController.java b/ruoyi-admin/src/main/java/org/dromara/web/controller/CaptchaController.java index 8b7e8796c..472217a6f 100644 --- a/ruoyi-admin/src/main/java/org/dromara/web/controller/CaptchaController.java +++ b/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 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 img Base64 图片数据 + * @param uuid 验证码标识 + * @param img Base64 图片数据 */ public record CaptchaVo(Boolean captchaEnabled, String uuid, String img) { } diff --git a/ruoyi-admin/src/main/resources/application-dev.yml b/ruoyi-admin/src/main/resources/application-dev.yml index 1b52fab96..6cc34d83d 100644 --- a/ruoyi-admin/src/main/resources/application-dev.yml +++ b/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 diff --git a/ruoyi-admin/src/main/resources/application.yml b/ruoyi-admin/src/main/resources/application.yml index b58c97a42..8f437e925 100644 --- a/ruoyi-admin/src/main/resources/application.yml +++ b/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 - ---- # websocket -websocket: - # 如果关闭 需要和前端开关一起关闭 - enabled: false - # 路径 - path: /resource/websocket - # 设置访问源地址 + # sse / websocket + transport: sse + # 统一访问路径 + path: /resource/message + # websocket 允许的跨域来源 allowedOrigins: '*' --- # warm-flow工作流配置 diff --git a/ruoyi-common/pom.xml b/ruoyi-common/pom.xml index 8d779e122..f6a30c47f 100644 --- a/ruoyi-common/pom.xml +++ b/ruoyi-common/pom.xml @@ -35,8 +35,7 @@ ruoyi-common-sensitive ruoyi-common-json ruoyi-common-encrypt - ruoyi-common-websocket - ruoyi-common-sse + ruoyi-common-push ruoyi-common-mqtt diff --git a/ruoyi-common/ruoyi-common-bom/pom.xml b/ruoyi-common/ruoyi-common-bom/pom.xml index 4f025fdc4..9eb04dd77 100644 --- a/ruoyi-common/ruoyi-common-bom/pom.xml +++ b/ruoyi-common/ruoyi-common-bom/pom.xml @@ -144,17 +144,10 @@ ${revision} - + org.dromara - ruoyi-common-websocket - ${revision} - - - - - org.dromara - ruoyi-common-sse + ruoyi-common-push ${revision} diff --git a/ruoyi-common/ruoyi-common-core/src/main/java/org/dromara/common/core/domain/dto/PushPayload.java b/ruoyi-common/ruoyi-common-core/src/main/java/org/dromara/common/core/domain/dto/PushPayload.java new file mode 100644 index 000000000..882f2fb3c --- /dev/null +++ b/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; + } +} diff --git a/ruoyi-common/ruoyi-common-core/src/main/java/org/dromara/common/core/enums/PushSourceEnum.java b/ruoyi-common/ruoyi-common-core/src/main/java/org/dromara/common/core/enums/PushSourceEnum.java new file mode 100644 index 000000000..1386570d0 --- /dev/null +++ b/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; +} diff --git a/ruoyi-common/ruoyi-common-core/src/main/java/org/dromara/common/core/enums/PushTypeEnum.java b/ruoyi-common/ruoyi-common-core/src/main/java/org/dromara/common/core/enums/PushTypeEnum.java new file mode 100644 index 000000000..721c7f95d --- /dev/null +++ b/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; +} diff --git a/ruoyi-common/ruoyi-common-mybatis/src/main/java/com/github/yulichang/wrapper/ext/Ext.java b/ruoyi-common/ruoyi-common-mybatis/src/main/java/com/github/yulichang/wrapper/ext/Ext.java index cb8ae0f8b..2b4e3af00 100644 --- a/ruoyi-common/ruoyi-common-mybatis/src/main/java/com/github/yulichang/wrapper/ext/Ext.java +++ b/ruoyi-common/ruoyi-common-mybatis/src/main/java/com/github/yulichang/wrapper/ext/Ext.java @@ -135,42 +135,6 @@ public interface Ext> extends IExt Children betweenIfExists(SFunction 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 Children betweenIfExists(SFunction column, Object[] values) { - Object val1 = ArrayUtils.get(values, 0); - Object val2 = ArrayUtils.get(values, 1); - return betweenIfExists(column, val1, val2); - } - /** * 当传入的集合不为空时才进行in查询 * diff --git a/ruoyi-common/ruoyi-common-oss/src/main/java/org/dromara/common/oss/client/AbstractOssClientImpl.java b/ruoyi-common/ruoyi-common-oss/src/main/java/org/dromara/common/oss/client/AbstractOssClientImpl.java index f3dd981f1..b7c5f7a64 100644 --- a/ruoyi-common/ruoyi-common-oss/src/main/java/org/dromara/common/oss/client/AbstractOssClientImpl.java +++ b/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, body); + return bucketUpload(bucket, key, file, Options.builder()); } @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) { - long size = contentLength; - try (channel; InputStream in = Channels.newInputStream(channel)) { - if (channel instanceof FileChannel fileChannel) { - size = fileChannel.size(); + public PutObjectResult bucketUpload(String bucket, String key, RandomAccessFile file) { + return bucketUpload(bucket, key, file, Options.builder()); + } + + @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, body); + return bucketUpload(bucket, key, in, contentLength, Options.builder()); } @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) { - Long contentLength = body.contentLength().orElse(null); + @NullMarked + 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 metadata = options.getMetadata(); + Collection transferListeners = options.getTransferListeners(); HandleAsyncResult 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; - return PutObjectResult.form("%s/%s".formatted(bucketUrl, key), key, response.eTag(), size == null ? 0 : size); + if (size == null) { + 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 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 bucketDownload(String bucket, String key, BiFunction downloadTransformer) { + try { + ResponseInputStream 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 download(String key, BiFunction downloadTransformer) { + return bucketDownload(defaultBucket(), key, downloadTransformer); + } + @Override public GetObjectResult download(String key, Path path) { return bucketDownload(defaultBucket(), key, path); diff --git a/ruoyi-common/ruoyi-common-oss/src/main/java/org/dromara/common/oss/client/OssClient.java b/ruoyi-common/ruoyi-common-oss/src/main/java/org/dromara/common/oss/client/OssClient.java index bb64ca4e0..4281cfa3b 100644 --- a/ruoyi-common/ruoyi-common-oss/src/main/java/org/dromara/common/oss/client/OssClient.java +++ b/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 - * + *

* 用于标识客户端，初始化后不允许更改 * * @return S3 存储客户端ID */ - default String clientId(){ + default String clientId() { return IdUtil.fastSimpleUUID(); } @@ -136,6 +137,17 @@ public interface OssClient extends AutoCloseable { */ HandleAsyncResult doCustomUpload(AsyncRequestBody body, Consumer 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 bucketDownload(String bucket, String key, BiFunction downloadTransformer); + /** * 将指定存储桶中的对象下载到本地路径。 * @@ -299,6 +378,16 @@ public interface OssClient extends AutoCloseable { */ String bucketPresignPutUrl(String bucket, String key, Duration expiredTime, Map 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 download(String key, BiFunction downloadTransformer); + /** * 将默认存储桶中的对象下载到本地路径。 * diff --git a/ruoyi-common/ruoyi-common-oss/src/main/java/org/dromara/common/oss/factory/OssFactory.java b/ruoyi-common/ruoyi-common-oss/src/main/java/org/dromara/common/oss/factory/OssFactory.java index c2a85a3b0..45a1bf04a 100644 --- a/ruoyi-common/ruoyi-common-oss/src/main/java/org/dromara/common/oss/factory/OssFactory.java +++ b/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(); } } diff --git a/ruoyi-common/ruoyi-common-oss/src/main/java/org/dromara/common/oss/io/OutputStreamDownloadSubscriber.java b/ruoyi-common/ruoyi-common-oss/src/main/java/org/dromara/common/oss/io/OutputStreamDownloadSubscriber.java index 229098141..14f96a517 100644 --- a/ruoyi-common/ruoyi-common-oss/src/main/java/org/dromara/common/oss/io/OutputStreamDownloadSubscriber.java +++ b/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, 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, 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, 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, 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); } } diff --git a/ruoyi-common/ruoyi-common-websocket/pom.xml b/ruoyi-common/ruoyi-common-push/pom.xml similarity index 94% rename from ruoyi-common/ruoyi-common-websocket/pom.xml rename to ruoyi-common/ruoyi-common-push/pom.xml index 0587cd79a..4d888e05f 100644 --- a/ruoyi-common/ruoyi-common-websocket/pom.xml +++ b/ruoyi-common/ruoyi-common-push/pom.xml @@ -9,10 +9,10 @@ 4.0.0 - ruoyi-common-websocket + ruoyi-common-push - ruoyi-common-websocket 模块 + ruoyi-common-push 模块 diff --git a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/config/MessageAutoConfiguration.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/config/MessageAutoConfiguration.java new file mode 100644 index 000000000..7ea031130 --- /dev/null +++ b/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 { +} diff --git a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/config/MessageSseConfiguration.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/config/MessageSseConfiguration.java new file mode 100644 index 000000000..02b9d78ff --- /dev/null +++ b/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); + } +} diff --git a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/config/MessageWebSocketConfiguration.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/config/MessageWebSocketConfiguration.java new file mode 100644 index 000000000..b1bb6ec2e --- /dev/null +++ b/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); + } +} diff --git a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/constant/MessageConstants.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/constant/MessageConstants.java new file mode 100644 index 000000000..3f66957af --- /dev/null +++ b/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"; +} diff --git a/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/controller/SseController.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/controller/SseController.java similarity index 71% rename from ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/controller/SseController.java rename to ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/controller/SseController.java index 2967d6736..ce42857a8 100644 --- a/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/controller/SseController.java +++ b/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 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 send(Long userId, String msg) { -// SseMessageDTO dto = new SseMessageDTO(); +// PushDTO dto = new PushDTO(); // dto.setUserIds(List.of(userId)); -// dto.setMessage(msg); -// sseEmitterManager.publishMessage(dto); +// dto.setPayload(PushPayload.of("message", "backend", msg, null)); +// 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 send(String msg) { -// sseEmitterManager.publishAll(msg); +// sessionManager.publishAll(msg); // return R.ok(); // } diff --git a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/PushSessionManager.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/PushSessionManager.java new file mode 100644 index 000000000..384a83ec7 --- /dev/null +++ b/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 consumer); + + void sendMessage(Long userId, PushPayload payload); + + void sendMessage(PushPayload payload); + + void publishMessage(PushDTO pushDTO); + + void publishAll(PushPayload payload); +} diff --git a/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/core/SseEmitterManager.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/SseEmitterSessionManager.java similarity index 72% rename from ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/core/SseEmitterManager.java rename to ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/SseEmitterSessionManager.java index d8504d703..042e0759d 100644 --- a/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/core/SseEmitterManager.java +++ b/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 { - - /** - * 订阅的频道 - */ - private final static String SSE_TOPIC = "global:sse"; +public class SseEmitterSessionManager implements PushSessionManager { private final static Map> 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 consumer) { - RedisUtils.subscribe(SSE_TOPIC, SseMessageDTO.class, consumer); + @Override + public void subscribeMessage(Consumer 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) { - SseMessageDTO broadcastMessage = new SseMessageDTO(); - broadcastMessage.setUserIds(sseMessageDTO.getUserIds()); - broadcastMessage.setMessage(sseMessageDTO.getMessage()); - RedisUtils.publish(SSE_TOPIC, broadcastMessage, consumer -> { - log.info("SSE发送主题订阅消息topic:{} session keys:{} message:{}", - SSE_TOPIC, sseMessageDTO.getUserIds(), sseMessageDTO.getMessage()); - }); + @Override + public void publishMessage(PushDTO pushDTO) { + RedisUtils.publish(MessageConstants.MESSAGE_TOPIC, pushDTO, consumer -> log.info( + "发送主题订阅消息topic:{} userIds:{} message:{}", + MessageConstants.MESSAGE_TOPIC, + pushDTO.getUserIds(), + 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(); - broadcastMessage.setMessage(message); - RedisUtils.publish(SSE_TOPIC, broadcastMessage, consumer -> { - log.info("SSE发送主题订阅消息topic:{} message:{}", SSE_TOPIC, message); + publishAll(PushPayload.of("message", "backend", message, null)); + } + + /** + * 发布 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()); }); } } diff --git a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/WebSocketSessionManager.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/WebSocketSessionManager.java new file mode 100644 index 000000000..30f34df79 --- /dev/null +++ b/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> 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 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 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 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 consumer) { + RedisUtils.subscribe(MESSAGE_TOPIC, PushDTO.class, consumer); + } + + @Override + public void sendMessage(Long userId, PushPayload payload) { + if (payload == null) { + return; + } + Map 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) { + } + } +} diff --git a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/dto/PushDTO.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/dto/PushDTO.java new file mode 100644 index 000000000..8a31df0b5 --- /dev/null +++ b/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 userIds; + + /** + * 推送消息体。 + */ + private PushPayload payload; +} diff --git a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/enums/MessageTransportEnum.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/enums/MessageTransportEnum.java new file mode 100644 index 000000000..ee5660e1b --- /dev/null +++ b/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); + } +} diff --git a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/handler/PlusWebSocketHandler.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/handler/PlusWebSocketHandler.java new file mode 100644 index 000000000..f704c9245 --- /dev/null +++ b/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; + } +} diff --git a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/helper/PushHelper.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/helper/PushHelper.java new file mode 100644 index 000000000..dedf5e462 --- /dev/null +++ b/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 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); + } +} diff --git a/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/interceptor/PlusWebSocketInterceptor.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/interceptor/PlusWebSocketInterceptor.java similarity index 53% rename from ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/interceptor/PlusWebSocketInterceptor.java rename to ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/interceptor/PlusWebSocketInterceptor.java index c70e37729..a830be5f9 100644 --- a/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/interceptor/PlusWebSocketInterceptor.java +++ b/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 attributes) { + public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, + Map 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) { } - } diff --git a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/listener/MessageTopicListener.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/listener/MessageTopicListener.java new file mode 100644 index 000000000..4f1bbad7a --- /dev/null +++ b/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; + } +} diff --git a/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/properties/MessageProperties.java b/ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/properties/MessageProperties.java new file mode 100644 index 000000000..f444123ef --- /dev/null +++ b/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 = "*"; +} diff --git a/ruoyi-common/ruoyi-common-push/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/ruoyi-common/ruoyi-common-push/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports new file mode 100644 index 000000000..2e02f912d --- /dev/null +++ b/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 diff --git a/ruoyi-common/ruoyi-common-security/src/main/java/org/dromara/common/security/config/SecurityConfig.java b/ruoyi-common/ruoyi-common-security/src/main/java/org/dromara/common/security/config/SecurityConfig.java index e94f4dfc1..eee99c74c 100644 --- a/ruoyi-common/ruoyi-common-security/src/main/java/org/dromara/common/security/config/SecurityConfig.java +++ b/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}") - private String ssePath; + @Value("${message.path:/resource/message}") + private String messagePath; /** * 注册 Sa-Token 路由拦截器并配置鉴权规则。 @@ -85,7 +85,7 @@ public class SecurityConfig implements WebMvcConfigurer { })).addPathPatterns("/**") // 排除不需要拦截的路径 .excludePathPatterns(securityProperties.getExcludes()) - .excludePathPatterns(ssePath); + .excludePathPatterns(messagePath); } /** diff --git a/ruoyi-common/ruoyi-common-sse/pom.xml b/ruoyi-common/ruoyi-common-sse/pom.xml deleted file mode 100644 index ae44c988e..000000000 --- a/ruoyi-common/ruoyi-common-sse/pom.xml +++ /dev/null @@ -1,36 +0,0 @@ - - - - org.dromara - ruoyi-common - ${revision} - - 4.0.0 - - ruoyi-common-sse - - - ruoyi-common-sse 模块 - - - - - org.dromara - ruoyi-common-core - - - org.dromara - ruoyi-common-redis - - - org.dromara - ruoyi-common-satoken - - - org.dromara - ruoyi-common-json - - - diff --git a/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/config/SseAutoConfiguration.java b/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/config/SseAutoConfiguration.java deleted file mode 100644 index 0cf8054ed..000000000 --- a/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/config/SseAutoConfiguration.java +++ /dev/null @@ -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); - } - -} diff --git a/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/config/SseProperties.java b/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/config/SseProperties.java deleted file mode 100644 index ce4e1732d..000000000 --- a/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/config/SseProperties.java +++ /dev/null @@ -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; -} diff --git a/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/dto/SseMessageDTO.java b/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/dto/SseMessageDTO.java deleted file mode 100644 index 1adc5108c..000000000 --- a/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/dto/SseMessageDTO.java +++ /dev/null @@ -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 userIds; - - /** - * 需要发送的消息 - */ - private String message; -} diff --git a/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/listener/SseTopicListener.java b/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/listener/SseTopicListener.java deleted file mode 100644 index 7a4dff13e..000000000 --- a/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/listener/SseTopicListener.java +++ /dev/null @@ -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; - } -} diff --git a/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/utils/SseMessageUtils.java b/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/utils/SseMessageUtils.java deleted file mode 100644 index b145b70a7..000000000 --- a/ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/utils/SseMessageUtils.java +++ /dev/null @@ -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; - } - -} diff --git a/ruoyi-common/ruoyi-common-sse/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/ruoyi-common/ruoyi-common-sse/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports deleted file mode 100644 index b80971390..000000000 --- a/ruoyi-common/ruoyi-common-sse/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports +++ /dev/null @@ -1 +0,0 @@ -org.dromara.common.sse.config.SseAutoConfiguration diff --git a/ruoyi-common/ruoyi-common-web/src/main/java/org/dromara/common/web/handler/GlobalExceptionHandler.java b/ruoyi-common/ruoyi-common-web/src/main/java/org/dromara/common/web/handler/GlobalExceptionHandler.java index c59f0f2ac..d02089956 100644 --- a/ruoyi-common/ruoyi-common-web/src/main/java/org/dromara/common/web/handler/GlobalExceptionHandler.java +++ b/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; } diff --git a/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/config/WebSocketConfig.java b/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/config/WebSocketConfig.java deleted file mode 100644 index ef5cfc96f..000000000 --- a/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/config/WebSocketConfig.java +++ /dev/null @@ -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(); - } -} diff --git a/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/config/properties/WebSocketProperties.java b/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/config/properties/WebSocketProperties.java deleted file mode 100644 index d629fe55a..000000000 --- a/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/config/properties/WebSocketProperties.java +++ /dev/null @@ -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; -} diff --git a/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/constant/WebSocketConstants.java b/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/constant/WebSocketConstants.java deleted file mode 100644 index e243279d9..000000000 --- a/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/constant/WebSocketConstants.java +++ /dev/null @@ -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"; -} diff --git a/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/dto/WebSocketMessageDTO.java b/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/dto/WebSocketMessageDTO.java deleted file mode 100644 index 452475c40..000000000 --- a/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/dto/WebSocketMessageDTO.java +++ /dev/null @@ -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 sessionKeys; - - /** - * 需要发送的消息 - */ - private String message; -} diff --git a/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/handler/PlusWebSocketHandler.java b/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/handler/PlusWebSocketHandler.java deleted file mode 100644 index b905ae066..000000000 --- a/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/handler/PlusWebSocketHandler.java +++ /dev/null @@ -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; - } - -} diff --git a/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/holder/WebSocketSessionHolder.java b/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/holder/WebSocketSessionHolder.java deleted file mode 100644 index 9c2372b85..000000000 --- a/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/holder/WebSocketSessionHolder.java +++ /dev/null @@ -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 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 getSessionsAll() { - return USER_SESSION_MAP.keySet(); - } - - /** - * 检查给定的会话键是否存在于用户会话Map中 - * - * @param sessionKey 要检查的会话键 - * @return 如果存在对应的会话键，则返回true；否则返回false - */ - public static Boolean existSession(Long sessionKey) { - return USER_SESSION_MAP.containsKey(sessionKey); - } -} diff --git a/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/listener/WebSocketTopicListener.java b/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/listener/WebSocketTopicListener.java deleted file mode 100644 index 0ad39affe..000000000 --- a/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/listener/WebSocketTopicListener.java +++ /dev/null @@ -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; - } -} diff --git a/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/utils/WebSocketUtils.java b/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/utils/WebSocketUtils.java deleted file mode 100644 index 2d2c0c921..000000000 --- a/ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/utils/WebSocketUtils.java +++ /dev/null @@ -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 consumer) { - RedisUtils.subscribe(WEB_SOCKET_TOPIC, WebSocketMessageDTO.class, consumer); - } - - /** - * 按会话标识发布 WebSocket 消息。 - * - * @param webSocketMessage 要发布的WebSocket消息对象 - */ - public static void publishMessage(WebSocketMessageDTO webSocketMessage) { - List 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); - } - } - } -} diff --git a/ruoyi-common/ruoyi-common-websocket/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/ruoyi-common/ruoyi-common-websocket/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports deleted file mode 100644 index c3a7305a3..000000000 --- a/ruoyi-common/ruoyi-common-websocket/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports +++ /dev/null @@ -1 +0,0 @@ -org.dromara.common.websocket.config.WebSocketConfig diff --git a/ruoyi-modules/ruoyi-demo/pom.xml b/ruoyi-modules/ruoyi-demo/pom.xml index c61fb47a0..ec948ceee 100644 --- a/ruoyi-modules/ruoyi-demo/pom.xml +++ b/ruoyi-modules/ruoyi-demo/pom.xml @@ -85,7 +85,7 @@ org.dromara - ruoyi-common-websocket + ruoyi-common-push diff --git a/ruoyi-modules/ruoyi-demo/src/main/java/org/dromara/demo/controller/WebSocketController.java b/ruoyi-modules/ruoyi-demo/src/main/java/org/dromara/demo/controller/WebSocketController.java index 481a0f5a9..92c738a71 100644 --- a/ruoyi-modules/ruoyi-demo/src/main/java/org/dromara/demo/controller/WebSocketController.java +++ b/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 send(WebSocketMessageDTO dto) throws InterruptedException { - WebSocketUtils.publishMessage(dto); + public R send(Long userId, String message) { + 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("操作成功"); } } diff --git a/ruoyi-modules/ruoyi-generator/src/main/java/org/dromara/generator/constant/GenConstants.java b/ruoyi-modules/ruoyi-generator/src/main/java/org/dromara/generator/constant/GenConstants.java index d01156091..5c7ac3e81 100644 --- a/ruoyi-modules/ruoyi-generator/src/main/java/org/dromara/generator/constant/GenConstants.java +++ b/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"}; /** * 文本框 diff --git a/ruoyi-modules/ruoyi-system/pom.xml b/ruoyi-modules/ruoyi-system/pom.xml index a24883dff..eaa6ab0c4 100644 --- a/ruoyi-modules/ruoyi-system/pom.xml +++ b/ruoyi-modules/ruoyi-system/pom.xml @@ -82,12 +82,7 @@ org.dromara - ruoyi-common-websocket - - - - org.dromara - ruoyi-common-sse + ruoyi-common-push diff --git a/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/controller/system/SysNoticeController.java b/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/controller/system/SysNoticeController.java index 039e5a6cb..2b400e260 100644 --- a/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/controller/system/SysNoticeController.java +++ b/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 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(); } diff --git a/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/controller/system/SysOssController.java b/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/controller/system/SysOssController.java index ea2c28058..36f520ab0 100644 --- a/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/controller/system/SysOssController.java +++ b/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 { - ossService.download(ossId, response); + public ResponseEntity download(@PathVariable Long ossId) throws IOException { + return ossService.download(ossId); } /** diff --git a/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/domain/bo/SysUserProfileBo.java b/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/domain/bo/SysUserProfileBo.java index d3d1e28c1..2513f30b7 100644 --- a/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/domain/bo/SysUserProfileBo.java +++ b/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; } diff --git a/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/domain/vo/ProfileUserVo.java b/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/domain/vo/ProfileUserVo.java index 18a761038..afb81c5f8 100644 --- a/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/domain/vo/ProfileUserVo.java +++ b/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; /** * 头像地址 diff --git a/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/mapper/SysDeptMapper.java b/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/mapper/SysDeptMapper.java index 894172002..39a8d87a4 100644 --- a/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/mapper/SysDeptMapper.java +++ b/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/mapper/SysDeptMapper.java @@ -75,7 +75,6 @@ public interface SysDeptMapper extends BaseMapperPlus { return this.selectList(new MPJLambdaWrapper() .select(SysDept::getDeptId) .findInSet(SysDept::getAncestors, parentId)); - } diff --git a/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/service/ISysOssService.java b/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/service/ISysOssService.java index da8d5fa0b..52870e874 100644 --- a/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/service/ISysOssService.java +++ b/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 download(Long ossId); /** * 删除OSS对象存储 diff --git a/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/service/impl/SysDeptServiceImpl.java b/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/service/impl/SysDeptServiceImpl.java index 74a4af636..37507d144 100644 --- a/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/service/impl/SysDeptServiceImpl.java +++ b/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 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; } diff --git a/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/service/impl/SysNoticeServiceImpl.java b/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/service/impl/SysNoticeServiceImpl.java index 1456eb8fd..674cc86ec 100644 --- a/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/service/impl/SysNoticeServiceImpl.java +++ b/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; } /** diff --git a/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/service/impl/SysOssServiceImpl.java b/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/service/impl/SysOssServiceImpl.java index b658e443d..ce3ab78b9 100644 --- a/ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/service/impl/SysOssServiceImpl.java +++ b/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 response HttpServletResponse对象，用于设置响应头和向客户端发送文件内容 + * @param ossId OSS对象ID */ @Override - public void download(Long ossId, HttpServletResponse response) throws IOException { + public ResponseEntity download(Long ossId) { SysOssVo sysOss = SpringUtils.getAopProxy(this).getById(ossId); if (ObjectUtil.isNull(sysOss)) { throw new ServiceException("文件数据不存在!"); } - FileUtils.setAttachmentResponseHeader(response, sysOss.getOriginalName()); - response.setContentType(MediaType.APPLICATION_OCTET_STREAM_VALUE + "; charset=UTF-8"); - OssClient instance = OssFactory.instance(sysOss.getService()); - GetObjectResult result = instance.download(sysOss.getFileName(), response.getOutputStream()); - response.setContentLengthLong(result.size()); + String percentEncodedFileName = FileUtils.percentEncode(sysOss.getOriginalName()); + return OssFactory.instance(sysOss.getService()) + .download(sysOss.getFileName(), (result, inputStream) -> { + // 尝试解析媒体类型，如果解析失败，则使用 application/octet-stream + 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()); // 保存文件信息 diff --git a/ruoyi-modules/ruoyi-workflow/pom.xml b/ruoyi-modules/ruoyi-workflow/pom.xml index 9034f9688..40d6b072b 100644 --- a/ruoyi-modules/ruoyi-workflow/pom.xml +++ b/ruoyi-modules/ruoyi-workflow/pom.xml @@ -20,7 +20,7 @@ org.dromara - ruoyi-common-sse + ruoyi-common-push diff --git a/ruoyi-modules/ruoyi-workflow/src/main/java/org/dromara/workflow/service/impl/FlwCommonServiceImpl.java b/ruoyi-modules/ruoyi-workflow/src/main/java/org/dromara/workflow/service/impl/FlwCommonServiceImpl.java index 57ed2af71..d10a05717 100644 --- a/ruoyi-modules/ruoyi-workflow/src/main/java/org/dromara/workflow/service/impl/FlwCommonServiceImpl.java +++ b/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.sse.utils.SseMessageUtils; +import org.dromara.common.push.helper.PushHelper; 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(); - dto.setUserIds(userIds); - dto.setMessage(message); - SseMessageUtils.publishMessage(dto); + PushHelper.publishMessage(userIds, PushPayload.of( + PushTypeEnum.MESSAGE, + PushSourceEnum.WORKFLOW, + message, + null + )); } case EMAIL_MESSAGE -> MailUtils.sendText(emails, subject, message); case SMS_MESSAGE -> {