seewe/seewis-hsl-cloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update 重构 common-sse 与 common-websocket 合并为 ruoyi-common-push 推送模块

Browse Source 
update 优化 通知公告页面增加查看详情功能 支持预览已经编辑好的富文本内容
update 优化 消息盒子相关消息可直接跳转到详情页展示富文本内容
update 完成消息盒子功能前后端联动(已读未读在前端浏览器存储)
update 工作流消息推送增加前端路由跳转
This commit is contained in:
疯狂的狮子Li
2026-03-27 19:42:28 +08:00
parent fb5b5aed29
commit 58afbf3ebf
62 changed files with 1672 additions and 921 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ruoyi-api/ruoyi-api-resource/src/main/java/org/dromara/resource/api/RemoteMessageService.java
+18
View File
@@ -1,5 +1,7 @@
package org.dromara.resource.api;
import org.dromara.resource.api.domain.dto.RemotePushPayLoad;
import java.util.List;
/**
@@ -17,10 +19,26 @@ public interface RemoteMessageService {
*/
void publishMessage(List<Long> sessionKey, String message);
/**
* 发布指定用户的结构化消息
*
* @param userIds 用户ID列表
* @param payload 推送体
*/
void publishMessagePayload(List<Long> userIds, RemotePushPayLoad payload);
/**
* 发布订阅的消息(群发)
*
* @param message 消息内容
*/
void publishAll(String message);
/**
* 发布广播结构化消息
*
* @param payload 推送体
*/
void publishAllPayload(RemotePushPayLoad payload);
}
ruoyi-api/ruoyi-api-resource/src/main/java/org/dromara/resource/api/RemoteMessageServiceStub.java
+20
View File
@@ -2,6 +2,7 @@ package org.dromara.resource.api;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.dromara.resource.api.domain.dto.RemotePushPayLoad;
import java.util.List;
@@ -31,6 +32,15 @@ public class RemoteMessageServiceStub implements RemoteMessageService {
}
}
@Override
public void publishMessagePayload(List<Long> userIds, RemotePushPayLoad payload) {
try {
remoteMessageService.publishMessagePayload(userIds, payload);
} catch (Exception e) {
log.warn("推送功能未开启或服务未找到");
}
}
/**
* 发布订阅的消息(群发)
*
@@ -44,4 +54,14 @@ public class RemoteMessageServiceStub implements RemoteMessageService {
log.warn("推送功能未开启或服务未找到");
}
}
@Override
public void publishAllPayload(RemotePushPayLoad payload) {
try {
remoteMessageService.publishAllPayload(payload);
} catch (Exception e) {
log.warn("推送功能未开启或服务未找到");
}
}
}
ruoyi-api/ruoyi-api-resource/src/main/java/org/dromara/resource/api/domain/dto/RemotePushPayLoad.java
+60
View File
@@ -0,0 +1,60 @@
package org.dromara.resource.api.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 RemotePushPayLoad implements Serializable {
@Serial
private static final long serialVersionUID = 1L;
private Long messageId;
private String type;
private String source;
private String message;
private Object data;
private String path;
private Long timestamp;
public static RemotePushPayLoad of(String type, String source, String message, Object data) {
RemotePushPayLoad payload = new RemotePushPayLoad();
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 RemotePushPayLoad 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 RemotePushPayLoad of(PushTypeEnum type, PushSourceEnum source, String message, Object data, String path) {
RemotePushPayLoad payload = of(type, source, message, data);
payload.setPath(path);
return payload;
}
}
ruoyi-common/pom.xml
+1 -2
View File
@@ -36,11 +36,10 @@
<module>ruoyi-common-sensitive</module>
<module>ruoyi-common-json</module>
<module>ruoyi-common-encrypt</module>
<module>ruoyi-common-websocket</module>
<module>ruoyi-common-push</module>
<module>ruoyi-common-social</module>
<module>ruoyi-common-nacos</module>
<module>ruoyi-common-bus</module>
<module>ruoyi-common-sse</module>
<module>ruoyi-common-mqtt</module>
</modules>
ruoyi-common/ruoyi-common-bom/pom.xml
+2 -9
View File
@@ -195,10 +195,10 @@
</dependency>
<!-- websocket模块 -->
<!-- 消息推送模块 -->
<dependency>
<groupId>org.dromara</groupId>
<artifactId>ruoyi-common-websocket</artifactId>
<artifactId>ruoyi-common-push</artifactId>
<version>${revision}</version>
</dependency>
@@ -223,13 +223,6 @@
<version>${revision}</version>
</dependency>
<!-- sse -->
<dependency>
<groupId>org.dromara</groupId>
<artifactId>ruoyi-common-sse</artifactId>
<version>${revision}</version>
</dependency>
<dependency>
<groupId>org.dromara</groupId>
<artifactId>ruoyi-common-mqtt</artifactId>
ruoyi-common/ruoyi-common-core/src/main/java/org/dromara/common/core/enums/PushSourceEnum.java
+41
View File
@@ -0,0 +1,41 @@
package org.dromara.common.core.enums;
import lombok.AllArgsConstructor;
import lombok.Getter;
/**
* 推送消息来源枚举
*
* @author Lion Li
*/
@Getter
@AllArgsConstructor
public enum PushSourceEnum {
/**
* 后端系统消息
*/
BACKEND("backend"),
/**
* 通知公告
*/
NOTICE("notice"),
/**
* 工作流
*/
WORKFLOW("workflow"),
/**
* 大模型
*/
LLM("llm"),
/**
* 客户端消息
*/
CLIENT("client");
private final String source;
}
ruoyi-common/ruoyi-common-core/src/main/java/org/dromara/common/core/enums/PushTypeEnum.java
+36
View File
@@ -0,0 +1,36 @@
package org.dromara.common.core.enums;
import lombok.AllArgsConstructor;
import lombok.Getter;
/**
* 推送消息类型枚举
*
* @author Lion Li
*/
@Getter
@AllArgsConstructor
public enum PushTypeEnum {
/**
* 通用消息
*/
MESSAGE("message"),
/**
* 通知公告
*/
NOTICE("notice"),
/**
* 大模型消息
*/
LLM("llm"),
/**
* 自定义消息
*/
CUSTOM("custom");
private final String type;
}
ruoyi-common/ruoyi-common-websocket/pom.xml → ruoyi-common/ruoyi-common-push/pom.xml
+6 -2
View File
@@ -9,10 +9,10 @@
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>ruoyi-common-websocket</artifactId>
<artifactId>ruoyi-common-push</artifactId>
<description>
ruoyi-common-websocket 模块
ruoyi-common-push 模块
</description>
<dependencies>
@@ -32,6 +32,10 @@
<groupId>org.dromara</groupId>
<artifactId>ruoyi-common-json</artifactId>
</dependency>
<dependency>
<groupId>org.dromara</groupId>
<artifactId>ruoyi-api-system</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/config/MessageAutoConfiguration.java
+17
View File
@@ -0,0 +1,17 @@
package org.dromara.common.push.config;
import org.dromara.common.push.properties.MessageProperties;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
/**
* 统一消息推送公共自动装配。
*
* @author Lion Li
*/
@AutoConfiguration
@ConditionalOnProperty(prefix = "message", name = "enabled", havingValue = "true", matchIfMissing = true)
@EnableConfigurationProperties(MessageProperties.class)
public class MessageAutoConfiguration {
}
ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/config/MessageSseConfiguration.java
+33
View File
@@ -0,0 +1,33 @@
package org.dromara.common.push.config;
import org.dromara.common.push.controller.SseController;
import org.dromara.common.push.core.SseEmitterSessionManager;
import org.dromara.common.push.listener.MessageTopicListener;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
/**
* SSE 消息推送自动装配。
*
* @author Lion Li
*/
@AutoConfiguration(after = MessageAutoConfiguration.class)
@ConditionalOnProperty(prefix = "message", name = "transport", havingValue = "sse", matchIfMissing = true)
public class MessageSseConfiguration {
@Bean
public SseEmitterSessionManager sseEmitterManager() {
return new SseEmitterSessionManager();
}
@Bean
public MessageTopicListener messageTopicListener(SseEmitterSessionManager manager) {
return new MessageTopicListener(manager);
}
@Bean
public SseController sseController(SseEmitterSessionManager manager) {
return new SseController(manager);
}
}
ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/config/MessageWebSocketConfiguration.java
+55
View File
@@ -0,0 +1,55 @@
package org.dromara.common.push.config;
import org.dromara.common.push.listener.MessageTopicListener;
import org.dromara.common.push.core.WebSocketSessionManager;
import org.dromara.common.push.handler.PlusWebSocketHandler;
import org.dromara.common.push.interceptor.PlusWebSocketInterceptor;
import org.dromara.common.push.properties.MessageProperties;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
import org.springframework.web.socket.server.HandshakeInterceptor;
/**
* WebSocket 消息推送自动装配。
*
* @author Lion Li
*/
@EnableWebSocket
@AutoConfiguration(after = MessageAutoConfiguration.class)
@ConditionalOnProperty(prefix = "message", name = "transport", havingValue = "websocket")
public class MessageWebSocketConfiguration {
@Bean
public WebSocketConfigurer webSocketConfigurer(HandshakeInterceptor handshakeInterceptor,
WebSocketHandler webSocketHandler,
MessageProperties messageProperties) {
return registry -> registry
.addHandler(webSocketHandler, messageProperties.getPath())
.addInterceptors(handshakeInterceptor)
.setAllowedOrigins(messageProperties.getAllowedOrigins());
}
@Bean
public WebSocketSessionManager webSocketSessionManager() {
return new WebSocketSessionManager();
}
@Bean
public HandshakeInterceptor handshakeInterceptor() {
return new PlusWebSocketInterceptor();
}
@Bean
public WebSocketHandler webSocketHandler(WebSocketSessionManager webSocketSessionManager) {
return new PlusWebSocketHandler(webSocketSessionManager);
}
@Bean
public MessageTopicListener messageTopicListener(WebSocketSessionManager webSocketSessionManager) {
return new MessageTopicListener(webSocketSessionManager);
}
}
ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/constant/MessageConstants.java
+19
View File
@@ -0,0 +1,19 @@
package org.dromara.common.push.constant;
/**
* 模块通用消息常量定义。
*
* @author Lion Li
*/
public interface MessageConstants {
String LOGIN_USER_KEY = "loginUser";
String LOGIN_TOKEN_KEY = "token";
String MESSAGE_TOPIC = "global:message";
String PING = "ping";
String PONG = "pong";
}
ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/controller/SseController.java → ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/controller/SseController.java
+24 -18
View File
@@ -1,12 +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;
@@ -18,33 +18,37 @@ 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 连接
* 建立当前登录用户的 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);
}
/**
* 关闭 SSE 连接
* 关闭当前登录用户的 SSE 连接。
*
* @return 操作结果
*/
@GetMapping(value = "${sse.path}/close")
@SaIgnore
@GetMapping(value = "${message.path:/resource/message}/close")
public R<Void> close() {
String tokenValue = StpUtil.getTokenValue();
Long userId = LoginHelper.getUserId();
sseEmitterManager.disconnect(userId, tokenValue);
sessionManager.disconnect(userId, tokenValue);
return R.ok();
}
@@ -55,12 +59,12 @@ public class SseController implements DisposableBean {
// * @param userId 目标用户的 ID
// * @param msg 要发送的消息内容
// */
// @GetMapping(value = "${sse.path}/send")
// @GetMapping(value = "${message.path:/resource/message}/send")
// public R<Void> send(Long userId, String msg) {
// SseMessageDTO dto = new SseMessageDTO();
// PushDTO dto = new PushDTO();
// dto.setUserIds(List.of(userId));
// dto.setMessage(msg);
// sseEmitterManager.publishMessage(dto);
// dto.setPayload(PushPayloadDTO.of("message", "backend", msg, null));
// sessionManager.publishMessage(dto);
// return R.ok();
// }
//
@@ -69,14 +73,16 @@ public class SseController implements DisposableBean {
// *
// * @param msg 要发送的消息内容
// */
// @GetMapping(value = "${sse.path}/sendAll")
// @GetMapping(value = "${message.path:/resource/message}/sendAll")
// public R<Void> send(String msg) {
// sseEmitterManager.publishAll(msg);
// sessionManager.publishAll(msg);
// return R.ok();
// }
/**
* 清理资源。此方法目前不执行任何操作，但避免因未实现而导致错误
* 容器销毁时释放资源占位实现。
*
* @throws Exception 销毁异常
*/
@Override
public void destroy() throws Exception {
ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/PushSessionManager.java
+24
View File
@@ -0,0 +1,24 @@
package org.dromara.common.push.core;
import org.dromara.common.push.dto.PushDTO;
import org.dromara.common.push.dto.PushPayloadDTO;
import java.util.function.Consumer;
/**
* 统一推送会话管理器。
*
* @author Lion Li
*/
public interface PushSessionManager {
void subscribeMessage(Consumer<PushDTO> consumer);
void sendMessage(Long userId, PushPayloadDTO payload);
void sendMessage(PushPayloadDTO payload);
void publishMessage(PushDTO pushDTO);
void publishAll(PushPayloadDTO payload);
}
ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/core/SseEmitterManager.java → ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/SseEmitterSessionManager.java
+105 -39
View File
@@ -1,13 +1,19 @@
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.push.constant.MessageConstants;
import org.dromara.common.push.dto.PushPayloadDTO;
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;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
@@ -20,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<Long, Map<String, SseEmitter>> USER_TOKEN_EMITTERS = new ConcurrentHashMap<>();
public SseEmitterManager() {
public SseEmitterSessionManager() {
// 定时执行 SSE 心跳检测
SpringUtils.getBean(ScheduledExecutorService.class)
.scheduleWithFixedDelay(this::sseMonitor, 60L, 60L, TimeUnit.SECONDS);
@@ -113,35 +114,55 @@ public class SseEmitterManager {
}
/**
* SSE心跳检测，关闭无效连接
* 执行 SSE 心跳检测并清理失效连接。
*/
public void sseMonitor() {
log.info("开始 SSE 心跳");
USER_TOKEN_EMITTERS.forEach((userId, map) ->
map.entrySet().removeIf(e -> {
final SseEmitter.SseEventBuilder heartbeat = SseEmitter.event().comment("heartbeat");
// 记录需要移除的用户ID
List<Long> toRemoveUsers = new ArrayList<>();
USER_TOKEN_EMITTERS.forEach((userId, emitterMap) -> {
if (CollUtil.isEmpty(emitterMap)) {
toRemoveUsers.add(userId);
return;
}
emitterMap.entrySet().removeIf(entry -> {
try {
e.getValue().send(SseEmitter.event().comment("heartbeat"));
entry.getValue().send(heartbeat);
return false;
} catch (Exception ex) {
log.warn("心跳失败，移除连接: userId={}, token={}", userId, e.getKey());
e.getValue().complete();
return true;
try {
entry.getValue().complete();
} catch (Exception ignore) {
// 忽略重复关闭异常
}
})
);
return true; // 发送失败 → 移除该连接
}
});
// 移除空连接用户
if (emitterMap.isEmpty()) {
toRemoveUsers.add(userId);
}
});
// 循环结束后统一清理空用户，避免并发修改异常
toRemoveUsers.forEach(USER_TOKEN_EMITTERS::remove);
}
/**
* 订阅SSE消息主题，并提供一个消费者函数来处理接收到的消息
* 订阅 SSE 广播主题消息。
*
* @param consumer 处理SSE消息的消费者函数
*/
public void subscribeMessage(Consumer<SseMessageDTO> consumer) {
RedisUtils.subscribe(SSE_TOPIC, SseMessageDTO.class, consumer);
@Override
public void subscribeMessage(Consumer<PushDTO> consumer) {
RedisUtils.subscribe(MessageConstants.MESSAGE_TOPIC, PushDTO.class, consumer);
}
/**
* 向指定的用户会话发送消息
* 向指定用户的全部本地 SSE 会话发送消息。
*
* @param userId 要发送消息的用户id
* @param message 要发送的消息内容
@@ -167,7 +188,31 @@ public class SseEmitterManager {
}
/**
* 本机全用户会话发送消息
* 向指定用户的全部本地 SSE 会话发送统一 JSON 消息。
*
* @param userId 要发送消息的用户id
* @param payload 要发送的消息体
*/
@Override
public void sendMessage(Long userId, PushPayloadDTO 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 会话发送消息。
*
* @param message 要发送的消息内容
*/
@@ -178,30 +223,51 @@ public class SseEmitterManager {
}
/**
* 发布SSE订阅消息
* 向当前节点所有 SSE 会话发送统一 JSON 消息。
*
* @param sseMessageDTO 要发布的SSE消息对象
* @param payload 要发送的消息体
*/
public void publishMessage(SseMessageDTO sseMessageDTO) {
SseMessageDTO broadcastMessage = new SseMessageDTO();
broadcastMessage.setMessage(sseMessageDTO.getMessage());
broadcastMessage.setUserIds(sseMessageDTO.getUserIds());
RedisUtils.publish(SSE_TOPIC, broadcastMessage, consumer -> {
log.info("SSE发送主题订阅消息topic:{} session keys:{} message:{}",
SSE_TOPIC, sseMessageDTO.getUserIds(), sseMessageDTO.getMessage());
});
@Override
public void sendMessage(PushPayloadDTO payload) {
sendMessage(JsonUtils.toJsonString(payload));
}
/**
* 向所有的用户发布订阅的消息(群发)
* 发布 SSE 订阅消息。
*
* @param pushDTO 要发布的SSE消息对象
*/
@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()
));
}
/**
* 发布 SSE 广播消息。
*
* @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(PushPayloadDTO.of("message", "backend", message, null));
}
/**
* 发布 SSE 广播 JSON 消息。
*
* @param payload 要发布的消息体
*/
@Override
public void publishAll(PushPayloadDTO payload) {
PushDTO dto = new PushDTO();
dto.setPayload(payload);
RedisUtils.publish(MessageConstants.MESSAGE_TOPIC, dto, consumer -> {
log.info("发送主题订阅消息topic:{} type:{} source:{} message:{}",
MessageConstants.MESSAGE_TOPIC, payload.getType(), payload.getSource(), payload.getMessage());
});
}
}
ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/core/WebSocketSessionManager.java
+168
View File
@@ -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.utils.SpringUtils;
import org.dromara.common.json.utils.JsonUtils;
import org.dromara.common.push.dto.PushPayloadDTO;
import org.dromara.common.push.dto.PushDTO;
import org.dromara.common.redis.utils.RedisUtils;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.PongMessage;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketMessage;
import org.springframework.web.socket.WebSocketSession;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import static org.dromara.common.push.constant.MessageConstants.MESSAGE_TOPIC;
/**
* WebSocket 会话管理器。
*
* @author Lion Li
*/
@Slf4j
public class WebSocketSessionManager implements PushSessionManager {
private static final Map<Long, Map<String, WebSocketSession>> USER_TOKEN_SESSIONS = new ConcurrentHashMap<>();
public WebSocketSessionManager() {
SpringUtils.getBean(ScheduledExecutorService.class)
.scheduleWithFixedDelay(this::sessionMonitor, 60L, 60L, TimeUnit.SECONDS);
}
public void connect(Long userId, String token, WebSocketSession session) {
Map<String, WebSocketSession> sessions = USER_TOKEN_SESSIONS.computeIfAbsent(userId, key -> new ConcurrentHashMap<>());
WebSocketSession oldSession = sessions.remove(token);
closeSession(oldSession, CloseStatus.NORMAL);
sessions.put(token, session);
}
public void disconnect(Long userId, String token) {
if (userId == null || token == null) {
return;
}
Map<String, WebSocketSession> sessions = USER_TOKEN_SESSIONS.get(userId);
if (MapUtil.isEmpty(sessions)) {
USER_TOKEN_SESSIONS.remove(userId);
return;
}
closeSession(sessions.remove(token), CloseStatus.NORMAL);
if (sessions.isEmpty()) {
USER_TOKEN_SESSIONS.remove(userId);
}
}
public void sessionMonitor() {
List<Long> toRemoveUsers = new ArrayList<>();
USER_TOKEN_SESSIONS.forEach((userId, sessionMap) -> {
if (CollUtil.isEmpty(sessionMap)) {
toRemoveUsers.add(userId);
return;
}
sessionMap.entrySet().removeIf(entry -> {
WebSocketSession session = entry.getValue();
if (session == null || !session.isOpen()) {
closeSession(session, CloseStatus.NORMAL);
return true;
}
return false;
});
if (sessionMap.isEmpty()) {
toRemoveUsers.add(userId);
}
});
toRemoveUsers.forEach(USER_TOKEN_SESSIONS::remove);
}
@Override
public void subscribeMessage(Consumer<PushDTO> consumer) {
RedisUtils.subscribe(MESSAGE_TOPIC, PushDTO.class, consumer);
}
@Override
public void sendMessage(Long userId, PushPayloadDTO payload) {
if (payload == null) {
return;
}
Map<String, WebSocketSession> sessions = USER_TOKEN_SESSIONS.get(userId);
if (MapUtil.isEmpty(sessions)) {
USER_TOKEN_SESSIONS.remove(userId);
return;
}
sessions.entrySet().removeIf(entry -> {
WebSocketSession session = entry.getValue();
if (session == null || !session.isOpen()) {
closeSession(session, CloseStatus.NORMAL);
return true;
}
return !sendMessage(session, new TextMessage(JsonUtils.toJsonString(payload)));
});
if (sessions.isEmpty()) {
USER_TOKEN_SESSIONS.remove(userId);
}
}
@Override
public void sendMessage(PushPayloadDTO 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(PushPayloadDTO payload) {
PushDTO dto = new PushDTO();
dto.setPayload(payload);
publishMessage(dto);
}
public void sendPongMessage(WebSocketSession session) {
sendMessage(session, new PongMessage());
}
public void sendMessage(WebSocketSession session, String message) {
sendMessage(session, new TextMessage(message));
}
private boolean sendMessage(WebSocketSession session, WebSocketMessage<?> message) {
if (session == null || !session.isOpen()) {
log.warn("[send] session会话已经关闭");
return false;
}
try {
session.sendMessage(message);
return true;
} catch (IOException e) {
log.error("[send] session({}) 发送消息({}) 异常", session, message, e);
return false;
}
}
private void closeSession(WebSocketSession session, CloseStatus status) {
if (session == null) {
return;
}
try {
session.close(status);
} catch (Exception ignored) {
}
}
}
ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/dto/SseMessageDTO.java → ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/dto/PushDTO.java
+7 -7
View File
@@ -1,4 +1,4 @@
package org.dromara.common.sse.dto;
package org.dromara.common.push.dto;
import lombok.Data;
@@ -7,23 +7,23 @@ import java.io.Serializable;
import java.util.List;
/**
* 消息的DTO
* 统一推送 DTO。
*
* @author zendwang
* @author Lion Li
*/
@Data
public class SseMessageDTO implements Serializable {
public class PushDTO implements Serializable {
@Serial
private static final long serialVersionUID = 1L;
/**
* 需要推送到的session key 列表
* 目标用户 ID 列表，为空表示广播。
*/
private List<Long> userIds;
/**
* 需要发送的消息
* 推送消息体。
*/
private String message;
private PushPayloadDTO payload;
}
ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/dto/PushPayloadDTO.java
+60
View File
@@ -0,0 +1,60 @@
package org.dromara.common.push.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 PushPayloadDTO implements Serializable {
@Serial
private static final long serialVersionUID = 1L;
private Long messageId;
private String type;
private String source;
private String message;
private Object data;
private String path;
private Long timestamp;
public static PushPayloadDTO of(String type, String source, String message, Object data) {
PushPayloadDTO payload = new PushPayloadDTO();
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 PushPayloadDTO 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 PushPayloadDTO of(PushTypeEnum type, PushSourceEnum source, String message, Object data, String path) {
PushPayloadDTO payload = of(type, source, message, data);
payload.setPath(path);
return payload;
}
}
ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/enums/MessageTransportEnum.java
+32
View File
@@ -0,0 +1,32 @@
package org.dromara.common.push.enums;
import lombok.AllArgsConstructor;
import lombok.Getter;
import java.util.Arrays;
/**
* 消息推送传输方式。
*
* @author Lion Li
*/
@Getter
@AllArgsConstructor
public enum MessageTransportEnum {
SSE("sse"),
WEBSOCKET("websocket");
private final String code;
public boolean matches(String transport) {
return code.equalsIgnoreCase(transport);
}
public static MessageTransportEnum of(String transport) {
return Arrays.stream(values())
.filter(item -> item.matches(transport))
.findFirst()
.orElse(SSE);
}
}
ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/handler/PlusWebSocketHandler.java
+104
View File
@@ -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.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.PushPayloadDTO;
import org.dromara.common.push.dto.PushDTO;
import org.dromara.system.api.model.LoginUser;
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(PushPayloadDTO.of(
PushTypeEnum.CUSTOM,
PushSourceEnum.CLIENT,
message.getPayload(),
null
));
webSocketSessionManager.publishMessage(dto);
}
@Override
protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) throws Exception {
super.handleBinaryMessage(session, message);
}
@Override
protected void handlePongMessage(WebSocketSession session, PongMessage message) {
webSocketSessionManager.sendPongMessage(session);
}
@Override
public void handleTransportError(WebSocketSession session, Throwable exception) {
log.error("[transport error] sessionId: {}, exception:{}", session.getId(), exception.getMessage());
}
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
LoginUser loginUser = (LoginUser) session.getAttributes().get(MessageConstants.LOGIN_USER_KEY);
String token = (String) session.getAttributes().get(MessageConstants.LOGIN_TOKEN_KEY);
if (ObjectUtil.hasNull(loginUser, token)) {
log.info("[disconnect] invalid token received. sessionId: {}", session.getId());
return;
}
webSocketSessionManager.disconnect(loginUser.getUserId(), token);
log.info("[disconnect] sessionId: {}, userId:{}, token:{}", session.getId(), loginUser.getUserId(), token);
}
@Override
public boolean supportsPartialMessages() {
return false;
}
}
ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/helper/PushHelper.java
+80
View File
@@ -0,0 +1,80 @@
package org.dromara.common.push.helper;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
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 org.dromara.common.push.dto.PushPayloadDTO;
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, PushPayloadDTO payload) {
if (!isEnabled()) {
return;
}
getSessionManager().sendMessage(userId, payload);
}
public static void sendMessage(PushPayloadDTO payload) {
if (!isEnabled()) {
return;
}
getSessionManager().sendMessage(payload);
}
public static void publishMessage(List<Long> userIds, PushPayloadDTO 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(PushPayloadDTO 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 PushPayloadDTO buildMessage(String message) {
return PushPayloadDTO.of(PushTypeEnum.MESSAGE, PushSourceEnum.BACKEND, message, null);
}
}
ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/interceptor/PlusWebSocketInterceptor.java
+57
View File
@@ -0,0 +1,57 @@
package org.dromara.common.push.interceptor;
import cn.dev33.satoken.exception.NotLoginException;
import cn.dev33.satoken.stp.StpUtil;
import lombok.extern.slf4j.Slf4j;
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.dromara.system.api.model.LoginUser;
import org.springframework.http.server.ServerHttpRequest;
import org.springframework.http.server.ServerHttpResponse;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.server.HandshakeInterceptor;
import java.util.Map;
/**
* WebSocket 握手拦截器。
*
* @author Lion Li
*/
@Slf4j
public class PlusWebSocketInterceptor implements HandshakeInterceptor {
@Override
public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler,
Map<String, Object> attributes) {
try {
LoginUser loginUser = LoginHelper.getLoginUser();
String tokenValue = StpUtil.getTokenValue();
if (loginUser == null || StringUtils.isBlank(tokenValue)) {
return false;
}
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)) {
throw NotLoginException.newInstance(StpUtil.getLoginType(),
"-100", "客户端ID与Token不匹配",
StpUtil.getTokenValue());
}
attributes.put(MessageConstants.LOGIN_USER_KEY, loginUser);
attributes.put(MessageConstants.LOGIN_TOKEN_KEY, tokenValue);
return true;
} catch (NotLoginException e) {
log.error("WebSocket 认证失败'{}',无法访问系统资源", e.getMessage());
return false;
}
}
@Override
public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler,
Exception exception) {
}
}
ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/listener/MessageTopicListener.java
+44
View File
@@ -0,0 +1,44 @@
package org.dromara.common.push.listener;
import cn.hutool.core.collection.CollUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.dromara.common.push.core.PushSessionManager;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.Ordered;
/**
* 统一消息主题订阅监听器。
*
* @author Lion Li
*/
@Slf4j
@RequiredArgsConstructor
public class MessageTopicListener implements ApplicationRunner, Ordered {
private final PushSessionManager pushSessionManager;
@Override
public void run(ApplicationArguments args) {
pushSessionManager.subscribeMessage(message -> {
log.info("消息主题订阅收到消息userIds={} message={}",
message.getUserIds(),
message.getPayload() == null ? null : message.getPayload().getMessage());
if (message.getPayload() == null) {
return;
}
if (CollUtil.isNotEmpty(message.getUserIds())) {
message.getUserIds().forEach(userId -> pushSessionManager.sendMessage(userId, message.getPayload()));
} else {
pushSessionManager.sendMessage(message.getPayload());
}
});
log.info("初始化消息主题订阅监听器成功");
}
@Override
public int getOrder() {
return -1;
}
}
ruoyi-common/ruoyi-common-push/src/main/java/org/dromara/common/push/properties/MessageProperties.java
+35
View File
@@ -0,0 +1,35 @@
package org.dromara.common.push.properties;
import lombok.Data;
import org.dromara.common.push.enums.MessageTransportEnum;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* 统一消息推送配置。
*
* @author Lion Li
*/
@Data
@ConfigurationProperties("message")
public class MessageProperties {
/**
* 是否启用消息推送。
*/
private Boolean enabled = true;
/**
* 传输方式：sse / websocket。
*/
private String transport = MessageTransportEnum.SSE.getCode();
/**
* 统一访问路径。
*/
private String path = "/resource/message";
/**
* WebSocket 允许的跨域来源。
*/
private String allowedOrigins = "*";
}
ruoyi-common/ruoyi-common-push/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
+3
View File
@@ -0,0 +1,3 @@
org.dromara.common.push.config.MessageAutoConfiguration
org.dromara.common.push.config.MessageSseConfiguration
org.dromara.common.push.config.MessageWebSocketConfiguration
ruoyi-common/ruoyi-common-sse/pom.xml
-40
View File
@@ -1,40 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>org.dromara</groupId>
<artifactId>ruoyi-common</artifactId>
<version>${revision}</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>ruoyi-common-sse</artifactId>
<description>
ruoyi-common-sse 模块
</description>
<dependencies>
<dependency>
<groupId>org.dromara</groupId>
<artifactId>ruoyi-common-core</artifactId>
</dependency>
<dependency>
<groupId>org.dromara</groupId>
<artifactId>ruoyi-common-redis</artifactId>
</dependency>
<dependency>
<groupId>org.dromara</groupId>
<artifactId>ruoyi-common-satoken</artifactId>
</dependency>
<dependency>
<groupId>org.dromara</groupId>
<artifactId>ruoyi-common-json</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-webmvc</artifactId>
</dependency>
</dependencies>
</project>
ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/config/SseAutoConfiguration.java
-36
View File
@@ -1,36 +0,0 @@
package org.dromara.common.sse.config;
import org.dromara.common.sse.controller.SseController;
import org.dromara.common.sse.core.SseEmitterManager;
import org.dromara.common.sse.listener.SseTopicListener;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
/**
* SSE 自动装配
*
* @author Lion Li
*/
@AutoConfiguration
@ConditionalOnProperty(value = "sse.enabled", havingValue = "true")
@EnableConfigurationProperties(SseProperties.class)
public class SseAutoConfiguration {
@Bean
public SseEmitterManager sseEmitterManager() {
return new SseEmitterManager();
}
@Bean
public SseTopicListener sseTopicListener() {
return new SseTopicListener();
}
@Bean
public SseController sseController(SseEmitterManager sseEmitterManager) {
return new SseController(sseEmitterManager);
}
}
ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/config/SseProperties.java
-21
View File
@@ -1,21 +0,0 @@
package org.dromara.common.sse.config;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* SSE 配置项
*
* @author Lion Li
*/
@Data
@ConfigurationProperties("sse")
public class SseProperties {
private Boolean enabled;
/**
* 路径
*/
private String path;
}
ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/listener/SseTopicListener.java
-48
View File
@@ -1,48 +0,0 @@
package org.dromara.common.sse.listener;
import cn.hutool.core.collection.CollUtil;
import lombok.extern.slf4j.Slf4j;
import org.dromara.common.sse.core.SseEmitterManager;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.Ordered;
/**
* SSE 主题订阅监听器
*
* @author Lion Li
*/
@Slf4j
public class SseTopicListener implements ApplicationRunner, Ordered {
@Autowired
private SseEmitterManager sseEmitterManager;
/**
* 在Spring Boot应用程序启动时初始化SSE主题订阅监听器
*
* @param args 应用程序参数
* @throws Exception 初始化过程中可能抛出的异常
*/
@Override
public void run(ApplicationArguments args) throws Exception {
sseEmitterManager.subscribeMessage((message) -> {
log.info("SSE主题订阅收到消息session keys={} message={}", message.getUserIds(), message.getMessage());
// 如果key不为空就按照key发消息 如果为空就群发
if (CollUtil.isNotEmpty(message.getUserIds())) {
message.getUserIds().forEach(key -> {
sseEmitterManager.sendMessage(key, message.getMessage());
});
} else {
sseEmitterManager.sendMessage(message.getMessage());
}
});
log.info("初始化SSE主题订阅监听器成功");
}
@Override
public int getOrder() {
return -1;
}
}
ruoyi-common/ruoyi-common-sse/src/main/java/org/dromara/common/sse/utils/SseMessageUtils.java
-84
View File
@@ -1,84 +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);
}
/**
* 本机全用户会话发送消息
*
* @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);
}
/**
* 向所有的用户发布订阅的消息(群发)
*
* @param message 要发布的消息内容
*/
public static void publishAll(String message) {
if (!isEnable()) {
return;
}
MANAGER.publishAll(message);
}
/**
* 是否开启
*/
public static Boolean isEnable() {
return SSE_ENABLE;
}
}
ruoyi-common/ruoyi-common-sse/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
-1
View File
@@ -1 +0,0 @@
org.dromara.common.sse.config.SseAutoConfiguration
ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/config/WebSocketConfig.java
-63
View File
@@ -1,63 +0,0 @@
package org.dromara.common.websocket.config;
import cn.hutool.core.util.StrUtil;
import org.dromara.common.websocket.config.properties.WebSocketProperties;
import org.dromara.common.websocket.handler.PlusWebSocketHandler;
import org.dromara.common.websocket.interceptor.PlusWebSocketInterceptor;
import org.dromara.common.websocket.listener.WebSocketTopicListener;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
import org.springframework.web.socket.server.HandshakeInterceptor;
/**
* WebSocket 配置
*
* @author zendwang
*/
@AutoConfiguration
@ConditionalOnProperty(value = "websocket.enabled", havingValue = "true")
@EnableConfigurationProperties(WebSocketProperties.class)
@EnableWebSocket
public class WebSocketConfig {
@Bean
public WebSocketConfigurer webSocketConfigurer(HandshakeInterceptor handshakeInterceptor,
WebSocketHandler webSocketHandler, WebSocketProperties webSocketProperties) {
// 如果WebSocket的路径为空，则设置默认路径为 "/websocket"
if (StrUtil.isBlank(webSocketProperties.getPath())) {
webSocketProperties.setPath("/websocket");
}
// 如果允许跨域访问的地址为空，则设置为 "*"，表示允许所有来源的跨域请求
if (StrUtil.isBlank(webSocketProperties.getAllowedOrigins())) {
webSocketProperties.setAllowedOrigins("*");
}
// 返回一个WebSocketConfigurer对象，用于配置WebSocket
return registry -> registry
// 添加WebSocket处理程序和拦截器到指定路径，设置允许的跨域来源
.addHandler(webSocketHandler, webSocketProperties.getPath())
.addInterceptors(handshakeInterceptor)
.setAllowedOrigins(webSocketProperties.getAllowedOrigins());
}
@Bean
public HandshakeInterceptor handshakeInterceptor() {
return new PlusWebSocketInterceptor();
}
@Bean
public WebSocketHandler webSocketHandler() {
return new PlusWebSocketHandler();
}
@Bean
public WebSocketTopicListener topicListener() {
return new WebSocketTopicListener();
}
}
ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/config/properties/WebSocketProperties.java
-26
View File
@@ -1,26 +0,0 @@
package org.dromara.common.websocket.config.properties;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* WebSocket 配置项
*
* @author zendwang
*/
@ConfigurationProperties("websocket")
@Data
public class WebSocketProperties {
private Boolean enabled;
/**
* 路径
*/
private String path;
/**
* 设置访问源地址
*/
private String allowedOrigins;
}
ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/constant/WebSocketConstants.java
-29
View File
@@ -1,29 +0,0 @@
package org.dromara.common.websocket.constant;
/**
* websocket的常量配置
*
* @author zendwang
*/
public interface WebSocketConstants {
/**
* websocketSession中的参数的key
*/
String LOGIN_USER_KEY = "loginUser";
/**
* 订阅的频道
*/
String WEB_SOCKET_TOPIC = "global:websocket";
/**
* 前端心跳检查的命令
*/
String PING = "ping";
/**
* 服务端心跳恢复的字符串
*/
String PONG = "pong";
}
ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/dto/WebSocketMessageDTO.java
-29
View File
@@ -1,29 +0,0 @@
package org.dromara.common.websocket.dto;
import lombok.Data;
import java.io.Serial;
import java.io.Serializable;
import java.util.List;
/**
* 消息的DTO
*
* @author zendwang
*/
@Data
public class WebSocketMessageDTO implements Serializable {
@Serial
private static final long serialVersionUID = 1L;
/**
* 需要推送到的session key 列表
*/
private List<Long> sessionKeys;
/**
* 需要发送的消息
*/
private String message;
}
ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/handler/PlusWebSocketHandler.java
-123
View File
@@ -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.websocket.dto.WebSocketMessageDTO;
import org.dromara.common.websocket.holder.WebSocketSessionHolder;
import org.dromara.common.websocket.utils.WebSocketUtils;
import org.dromara.system.api.model.LoginUser;
import org.springframework.web.socket.*;
import org.springframework.web.socket.handler.AbstractWebSocketHandler;
import org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator;
import java.io.IOException;
import java.util.List;
import static org.dromara.common.websocket.constant.WebSocketConstants.LOGIN_USER_KEY;
/**
* WebSocketHandler 实现类
*
* @author zendwang
*/
@Slf4j
public class PlusWebSocketHandler extends AbstractWebSocketHandler {
/**
* 连接成功后
*/
@Override
public void afterConnectionEstablished(WebSocketSession session) throws IOException {
LoginUser loginUser = (LoginUser) session.getAttributes().get(LOGIN_USER_KEY);
if (ObjectUtil.isNull(loginUser)) {
session.close(CloseStatus.BAD_DATA);
log.info("[connect] invalid token received. sessionId: {}", session.getId());
return;
}
WebSocketSessionHolder.addSession(loginUser.getUserId(), new ConcurrentWebSocketSessionDecorator(session, 10 * 1000, 64000));
log.info("[connect] sessionId: {},userId:{},userType:{}", session.getId(), loginUser.getUserId(), loginUser.getUserType());
}
/**
* 处理接收到的文本消息
*
* @param session WebSocket会话
* @param message 接收到的文本消息
* @throws Exception 处理消息过程中可能抛出的异常
*/
@Override
protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
// 从WebSocket会话中获取登录用户信息
LoginUser loginUser = (LoginUser) session.getAttributes().get(LOGIN_USER_KEY);
// 创建WebSocket消息DTO对象
WebSocketMessageDTO messageDTO = new WebSocketMessageDTO();
messageDTO.setSessionKeys(List.of(loginUser.getUserId()));
messageDTO.setMessage(message.getPayload());
WebSocketUtils.publishMessage(messageDTO);
}
/**
* 处理接收到的二进制消息
*
* @param session WebSocket会话
* @param message 接收到的二进制消息
* @throws Exception 处理消息过程中可能抛出的异常
*/
@Override
protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) throws Exception {
super.handleBinaryMessage(session, message);
}
/**
* 处理接收到的Pong消息（心跳监测）
*
* @param session WebSocket会话
* @param message 接收到的Pong消息
* @throws Exception 处理消息过程中可能抛出的异常
*/
@Override
protected void handlePongMessage(WebSocketSession session, PongMessage message) throws Exception {
WebSocketUtils.sendPongMessage(session);
}
/**
* 处理WebSocket传输错误
*
* @param session WebSocket会话
* @param exception 发生的异常
* @throws Exception 处理过程中可能抛出的异常
*/
@Override
public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
log.error("[transport error] sessionId: {} , exception:{}", session.getId(), exception.getMessage());
}
/**
* 在WebSocket连接关闭后执行清理操作
*
* @param session WebSocket会话
* @param status 关闭状态信息
*/
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
LoginUser loginUser = (LoginUser) session.getAttributes().get(LOGIN_USER_KEY);
if (ObjectUtil.isNull(loginUser)) {
log.info("[disconnect] invalid token received. sessionId: {}", session.getId());
return;
}
WebSocketSessionHolder.removeSession(loginUser.getUserId());
log.info("[disconnect] sessionId: {},userId:{},userType:{}", session.getId(), loginUser.getUserId(), loginUser.getUserType());
}
/**
* 指示处理程序是否支持接收部分消息
*
* @return 如果支持接收部分消息，则返回true；否则返回false
*/
@Override
public boolean supportsPartialMessages() {
return false;
}
}
ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/holder/WebSocketSessionHolder.java
-74
View File
@@ -1,74 +0,0 @@
package org.dromara.common.websocket.holder;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.WebSocketSession;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
/**
* WebSocketSession 用于保存当前所有在线的会话信息
*
* @author zendwang
*/
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class WebSocketSessionHolder {
private static final Map<Long, WebSocketSession> USER_SESSION_MAP = new ConcurrentHashMap<>();
/**
* 将WebSocket会话添加到用户会话Map中
*
* @param sessionKey 会话键，用于检索会话
* @param session 要添加的WebSocket会话
*/
public static void addSession(Long sessionKey, WebSocketSession session) {
removeSession(sessionKey);
USER_SESSION_MAP.put(sessionKey, session);
}
/**
* 从用户会话Map中移除指定会话键对应的WebSocket会话
*
* @param sessionKey 要移除的会话键
*/
public static void removeSession(Long sessionKey) {
WebSocketSession session = USER_SESSION_MAP.remove(sessionKey);
try {
session.close(CloseStatus.BAD_DATA);
} catch (Exception ignored) {
}
}
/**
* 根据会话键从用户会话Map中获取WebSocket会话
*
* @param sessionKey 要获取的会话键
* @return 与给定会话键对应的WebSocket会话，如果不存在则返回null
*/
public static WebSocketSession getSessions(Long sessionKey) {
return USER_SESSION_MAP.get(sessionKey);
}
/**
* 获取存储在用户会话Map中所有WebSocket会话的会话键集合
*
* @return 所有WebSocket会话的会话键集合
*/
public static Set<Long> getSessionsAll() {
return USER_SESSION_MAP.keySet();
}
/**
* 检查给定的会话键是否存在于用户会话Map中
*
* @param sessionKey 要检查的会话键
* @return 如果存在对应的会话键，则返回true；否则返回false
*/
public static Boolean existSession(Long sessionKey) {
return USER_SESSION_MAP.containsKey(sessionKey);
}
}
ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/interceptor/PlusWebSocketInterceptor.java
-58
View File
@@ -1,58 +0,0 @@
package org.dromara.common.websocket.interceptor;
import cn.dev33.satoken.exception.NotLoginException;
import lombok.extern.slf4j.Slf4j;
import org.dromara.common.satoken.utils.LoginHelper;
import org.dromara.system.api.model.LoginUser;
import org.springframework.http.server.ServerHttpRequest;
import org.springframework.http.server.ServerHttpResponse;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.server.HandshakeInterceptor;
import java.util.Map;
import static org.dromara.common.websocket.constant.WebSocketConstants.LOGIN_USER_KEY;
/**
* WebSocket握手请求的拦截器
*
* @author zendwang
*/
@Slf4j
public class PlusWebSocketInterceptor implements HandshakeInterceptor {
/**
* WebSocket握手之前执行的前置处理方法
*
* @param request WebSocket握手请求
* @param response WebSocket握手响应
* @param wsHandler WebSocket处理程序
* @param attributes 与WebSocket会话关联的属性
* @return 如果允许握手继续进行，则返回true；否则返回false
*/
@Override
public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Map<String, Object> attributes) {
try {
LoginUser loginUser = LoginHelper.getLoginUser();
attributes.put(LOGIN_USER_KEY, loginUser);
return true;
} catch (NotLoginException e) {
log.error("WebSocket 认证失败'{}',无法访问系统资源", e.getMessage());
return false;
}
}
/**
* WebSocket握手成功后执行的后置处理方法
*
* @param request WebSocket握手请求
* @param response WebSocket握手响应
* @param wsHandler WebSocket处理程序
* @param exception 握手过程中可能出现的异常
*/
@Override
public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Exception exception) {
// 在这个方法中可以执行一些握手成功后的后续处理逻辑，比如记录日志或者其他操作
}
}
ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/listener/WebSocketTopicListener.java
-50
View File
@@ -1,50 +0,0 @@
package org.dromara.common.websocket.listener;
import cn.hutool.core.collection.CollUtil;
import lombok.extern.slf4j.Slf4j;
import org.dromara.common.websocket.holder.WebSocketSessionHolder;
import org.dromara.common.websocket.utils.WebSocketUtils;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.Ordered;
/**
* WebSocket 主题订阅监听器
*
* @author zendwang
*/
@Slf4j
public class WebSocketTopicListener implements ApplicationRunner, Ordered {
/**
* 在Spring Boot应用程序启动时初始化WebSocket主题订阅监听器
*
* @param args 应用程序参数
* @throws Exception 初始化过程中可能抛出的异常
*/
@Override
public void run(ApplicationArguments args) throws Exception {
// 订阅WebSocket消息
WebSocketUtils.subscribeMessage((message) -> {
log.info("WebSocket主题订阅收到消息session keys={} message={}", message.getSessionKeys(), message.getMessage());
// 如果key不为空就按照key发消息 如果为空就群发
if (CollUtil.isNotEmpty(message.getSessionKeys())) {
message.getSessionKeys().forEach(key -> {
if (WebSocketSessionHolder.existSession(key)) {
WebSocketUtils.sendMessage(key, message.getMessage());
}
});
} else {
WebSocketSessionHolder.getSessionsAll().forEach(key -> {
WebSocketUtils.sendMessage(key, message.getMessage());
});
}
});
log.info("初始化WebSocket主题订阅监听器成功");
}
@Override
public int getOrder() {
return -1;
}
}
ruoyi-common/ruoyi-common-websocket/src/main/java/org/dromara/common/websocket/utils/WebSocketUtils.java
-127
View File
@@ -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;
/**
* WebSocket工具类
*
* @author zendwang
*/
@Slf4j
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class WebSocketUtils {
/**
* 向指定的WebSocket会话发送消息
*
* @param sessionKey 要发送消息的用户id
* @param message 要发送的消息内容
*/
public static void sendMessage(Long sessionKey, String message) {
WebSocketSession session = WebSocketSessionHolder.getSessions(sessionKey);
sendMessage(session, message);
}
/**
* 订阅WebSocket消息主题，并提供一个消费者函数来处理接收到的消息
*
* @param consumer 处理WebSocket消息的消费者函数
*/
public static void subscribeMessage(Consumer<WebSocketMessageDTO> consumer) {
RedisUtils.subscribe(WEB_SOCKET_TOPIC, WebSocketMessageDTO.class, consumer);
}
/**
* 发布WebSocket订阅消息
*
* @param webSocketMessage 要发布的WebSocket消息对象
*/
public static void publishMessage(WebSocketMessageDTO webSocketMessage) {
List<Long> unsentSessionKeys = new ArrayList<>();
// 当前服务内session,直接发送消息
for (Long sessionKey : webSocketMessage.getSessionKeys()) {
if (WebSocketSessionHolder.existSession(sessionKey)) {
WebSocketUtils.sendMessage(sessionKey, webSocketMessage.getMessage());
continue;
}
unsentSessionKeys.add(sessionKey);
}
// 不在当前服务内session,发布订阅消息
if (CollUtil.isNotEmpty(unsentSessionKeys)) {
WebSocketMessageDTO broadcastMessage = new WebSocketMessageDTO();
broadcastMessage.setMessage(webSocketMessage.getMessage());
broadcastMessage.setSessionKeys(unsentSessionKeys);
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);
});
}
/**
* 向指定的WebSocket会话发送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会话发送WebSocket消息对象
*
* @param session WebSocket会话
* @param message 要发送的WebSocket消息对象
*/
private static void sendMessage(WebSocketSession session, WebSocketMessage<?> message) {
if (session == null || !session.isOpen()) {
log.warn("[send] session会话已经关闭");
} else {
try {
session.sendMessage(message);
} catch (IOException e) {
log.error("[send] session({}) 发送消息({}) 异常", session, message, e);
}
}
}
}
ruoyi-common/ruoyi-common-websocket/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
-1
View File
@@ -1 +0,0 @@
org.dromara.common.websocket.config.WebSocketConfig
ruoyi-modules/ruoyi-resource/pom.xml
+1 -6
View File
@@ -85,12 +85,7 @@
<dependency>
<groupId>org.dromara</groupId>
<artifactId>ruoyi-common-websocket</artifactId>
</dependency>
<dependency>
<groupId>org.dromara</groupId>
<artifactId>ruoyi-common-sse</artifactId>
<artifactId>ruoyi-common-push</artifactId>
</dependency>
<dependency>
ruoyi-modules/ruoyi-resource/src/main/java/org/dromara/resource/controller/SysMessageController.java
+29
View File
@@ -0,0 +1,29 @@
package org.dromara.resource.controller;
import lombok.RequiredArgsConstructor;
import org.dromara.common.core.domain.R;
import org.dromara.common.satoken.utils.LoginHelper;
import org.dromara.common.web.core.BaseController;
import org.dromara.resource.domain.vo.SysMessageBoxVo;
import org.dromara.resource.service.ISysMessageService;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* 消息记录控制器
*
* @author Lion Li
*/
@RequiredArgsConstructor
@RestController
@RequestMapping("/message")
public class SysMessageController extends BaseController {
private final ISysMessageService messageService;
@GetMapping("/box")
public R<SysMessageBoxVo> getBox() {
return R.ok(messageService.queryMessageBox(LoginHelper.getUserId()));
}
}
ruoyi-modules/ruoyi-resource/src/main/java/org/dromara/resource/domain/SysMessage.java
+39
View File
@@ -0,0 +1,39 @@
package org.dromara.resource.domain;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;
import lombok.EqualsAndHashCode;
import org.dromara.common.mybatis.core.domain.BaseEntity;
/**
* 消息记录表 sys_message
*
* @author Lion Li
*/
@Data
@EqualsAndHashCode(callSuper = true)
@TableName("sys_message")
public class SysMessage extends BaseEntity {
@TableId(value = "message_id")
private Long messageId;
private String category;
private String type;
private String source;
private String title;
private String message;
private String content;
private String dataJson;
private String path;
private String sendUserIds;
}
ruoyi-modules/ruoyi-resource/src/main/java/org/dromara/resource/domain/vo/SysMessageBoxVo.java
+26
View File
@@ -0,0 +1,26 @@
package org.dromara.resource.domain.vo;
import lombok.Data;
import java.io.Serial;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
/**
* 消息盒子视图对象
*
* @author Lion Li
*/
@Data
public class SysMessageBoxVo implements Serializable {
@Serial
private static final long serialVersionUID = 1L;
private List<SysMessageVo> systemList = new ArrayList<>();
private List<SysMessageVo> noticeList = new ArrayList<>();
private List<SysMessageVo> workflowList = new ArrayList<>();
}
ruoyi-modules/ruoyi-resource/src/main/java/org/dromara/resource/domain/vo/SysMessageVo.java
+39
View File
@@ -0,0 +1,39 @@
package org.dromara.resource.domain.vo;
import lombok.Data;
import java.io.Serial;
import java.io.Serializable;
import java.util.Date;
/**
* 消息记录视图对象
*
* @author Lion Li
*/
@Data
public class SysMessageVo implements Serializable {
@Serial
private static final long serialVersionUID = 1L;
private Long messageId;
private String category;
private String type;
private String source;
private String title;
private String message;
private String content;
private Object data;
private String path;
private Date createTime;
}
ruoyi-modules/ruoyi-resource/src/main/java/org/dromara/resource/dubbo/RemoteMessageServiceImpl.java
+23 -7
View File
@@ -1,11 +1,16 @@
package org.dromara.resource.dubbo;
import cn.hutool.core.bean.BeanUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboService;
import org.dromara.common.sse.dto.SseMessageDTO;
import org.dromara.common.sse.utils.SseMessageUtils;
import org.dromara.common.core.enums.PushSourceEnum;
import org.dromara.common.core.enums.PushTypeEnum;
import org.dromara.common.push.dto.PushPayloadDTO;
import org.dromara.common.push.helper.PushHelper;
import org.dromara.resource.api.RemoteMessageService;
import org.dromara.resource.api.domain.dto.RemotePushPayLoad;
import org.dromara.resource.service.ISysMessageService;
import org.springframework.stereotype.Service;
import java.util.List;
@@ -21,6 +26,8 @@ import java.util.List;
@DubboService
public class RemoteMessageServiceImpl implements RemoteMessageService {
private final ISysMessageService sysMessageService;
/**
* 发送消息
*
@@ -29,10 +36,13 @@ public class RemoteMessageServiceImpl implements RemoteMessageService {
*/
@Override
public void publishMessage(List<Long> sessionKey, String message) {
SseMessageDTO dto = new SseMessageDTO();
dto.setMessage(message);
dto.setUserIds(sessionKey);
SseMessageUtils.publishMessage(dto);
publishMessagePayload(sessionKey, RemotePushPayLoad.of(PushTypeEnum.MESSAGE, PushSourceEnum.BACKEND, message, null));
}
@Override
public void publishMessagePayload(List<Long> userIds, RemotePushPayLoad payload) {
PushPayloadDTO pushPayload = BeanUtil.copyProperties(payload, PushPayloadDTO.class);
PushHelper.publishMessage(userIds, sysMessageService.storeUsers(userIds, pushPayload));
}
/**
@@ -42,7 +52,13 @@ public class RemoteMessageServiceImpl implements RemoteMessageService {
*/
@Override
public void publishAll(String message) {
SseMessageUtils.publishAll(message);
publishAllPayload(RemotePushPayLoad.of(PushTypeEnum.MESSAGE, PushSourceEnum.BACKEND, message, null));
}
@Override
public void publishAllPayload(RemotePushPayLoad payload) {
PushPayloadDTO pushPayload = BeanUtil.copyProperties(payload, PushPayloadDTO.class);
PushHelper.publishAll(sysMessageService.storeAll(pushPayload));
}
}
ruoyi-modules/ruoyi-resource/src/main/java/org/dromara/resource/mapper/SysMessageMapper.java
+13
View File
@@ -0,0 +1,13 @@
package org.dromara.resource.mapper;
import org.dromara.common.mybatis.core.mapper.BaseMapperPlus;
import org.dromara.resource.domain.SysMessage;
import org.dromara.resource.domain.vo.SysMessageVo;
/**
* 消息记录Mapper接口
*
* @author Lion Li
*/
public interface SysMessageMapper extends BaseMapperPlus<SysMessage, SysMessageVo> {
}
ruoyi-modules/ruoyi-resource/src/main/java/org/dromara/resource/service/ISysMessageService.java
+20
View File
@@ -0,0 +1,20 @@
package org.dromara.resource.service;
import org.dromara.common.push.dto.PushPayloadDTO;
import org.dromara.resource.domain.vo.SysMessageBoxVo;
import java.util.List;
/**
* 消息记录服务接口
*
* @author Lion Li
*/
public interface ISysMessageService {
SysMessageBoxVo queryMessageBox(Long userId);
PushPayloadDTO storeAll(PushPayloadDTO payload);
PushPayloadDTO storeUsers(List<Long> userIds, PushPayloadDTO payload);
}
ruoyi-modules/ruoyi-resource/src/main/java/org/dromara/resource/service/impl/SysMessageServiceImpl.java
+161
View File
@@ -0,0 +1,161 @@
package org.dromara.resource.service.impl;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.convert.Convert;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import lombok.RequiredArgsConstructor;
import org.dromara.common.core.enums.PushSourceEnum;
import org.dromara.common.core.enums.PushTypeEnum;
import org.dromara.common.core.utils.StringUtils;
import org.dromara.common.json.utils.JsonUtils;
import org.dromara.common.mybatis.helper.DataBaseHelper;
import org.dromara.common.mybatis.utils.IdGeneratorUtil;
import org.dromara.common.push.dto.PushPayloadDTO;
import org.dromara.resource.domain.SysMessage;
import org.dromara.resource.domain.vo.SysMessageBoxVo;
import org.dromara.resource.domain.vo.SysMessageVo;
import org.dromara.resource.mapper.SysMessageMapper;
import org.dromara.resource.service.ISysMessageService;
import org.springframework.stereotype.Service;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
/**
* 消息记录服务实现
*
* @author Lion Li
*/
@RequiredArgsConstructor
@Service
public class SysMessageServiceImpl implements ISysMessageService {
private static final String GLOBAL_USER_IDS = "0";
private static final String CATEGORY_SYSTEM = "system";
private static final String CATEGORY_NOTICE = "notice";
private static final String CATEGORY_WORKFLOW = "workflow";
private static final int BOX_LIMIT = 100;
private static final long BOX_DAYS = 30L;
private final SysMessageMapper baseMapper;
@Override
public SysMessageBoxVo queryMessageBox(Long userId) {
SysMessageBoxVo box = new SysMessageBoxVo();
box.setSystemList(selectMessageList(CATEGORY_SYSTEM, userId));
box.setNoticeList(selectMessageList(CATEGORY_NOTICE, userId));
box.setWorkflowList(selectMessageList(CATEGORY_WORKFLOW, userId));
return box;
}
@Override
public PushPayloadDTO storeAll(PushPayloadDTO payload) {
return storeMessage(null, payload);
}
@Override
public PushPayloadDTO storeUsers(List<Long> userIds, PushPayloadDTO payload) {
return storeMessage(userIds, payload);
}
private PushPayloadDTO storeMessage(List<Long> userIds, PushPayloadDTO payload) {
if (!supportsMessageBox(payload)) {
return payload;
}
SysMessage message = buildMessage(userIds, payload);
baseMapper.insert(message);
payload.setMessageId(message.getMessageId());
return payload;
}
private List<SysMessageVo> selectMessageList(String category, Long userId) {
LambdaQueryWrapper<SysMessage> lqw = Wrappers.lambdaQuery();
lqw.eq(SysMessage::getCategory, category);
lqw.ge(SysMessage::getCreateTime, new Date(System.currentTimeMillis() - TimeUnit.DAYS.toMillis(BOX_DAYS)));
lqw.and(wrapper -> wrapper.eq(SysMessage::getSendUserIds, GLOBAL_USER_IDS)
.or()
.apply(DataBaseHelper.findInSet(userId, "send_user_ids")));
lqw.orderByDesc(SysMessage::getCreateTime, SysMessage::getMessageId);
List<SysMessage> list = baseMapper.selectList(new Page<>(1, BOX_LIMIT, false), lqw);
return list.stream().map(this::buildVo).toList();
}
private SysMessage buildMessage(List<Long> userIds, PushPayloadDTO payload) {
SysMessage message = new SysMessage();
message.setMessageId(payload.getMessageId() == null ? IdGeneratorUtil.nextLongId() : payload.getMessageId());
message.setCategory(resolveCategory(payload));
message.setType(payload.getType());
message.setSource(payload.getSource());
message.setTitle(resolveTitle(payload));
message.setMessage(payload.getMessage());
message.setContent(resolveContent(payload));
message.setDataJson(JsonUtils.toJsonString(payload.getData()));
message.setPath(payload.getPath());
message.setSendUserIds(CollUtil.isEmpty(userIds) ? GLOBAL_USER_IDS : StringUtils.joinComma(userIds));
return message;
}
private SysMessageVo buildVo(SysMessage entity) {
SysMessageVo vo = new SysMessageVo();
vo.setMessageId(entity.getMessageId());
vo.setCategory(entity.getCategory());
vo.setType(entity.getType());
vo.setSource(entity.getSource());
vo.setTitle(entity.getTitle());
vo.setMessage(entity.getMessage());
vo.setContent(entity.getContent());
vo.setPath(entity.getPath());
vo.setCreateTime(entity.getCreateTime());
vo.setData(parseData(entity.getDataJson()));
return vo;
}
private boolean supportsMessageBox(PushPayloadDTO payload) {
if (payload == null) {
return false;
}
if (StringUtils.equalsAny(payload.getType(), PushTypeEnum.MESSAGE.getType(), PushTypeEnum.NOTICE.getType())) {
return !StringUtils.equalsAny(payload.getType(), PushTypeEnum.LLM.getType())
&& !StringUtils.equalsAny(payload.getSource(), PushSourceEnum.LLM.getSource());
}
return false;
}
private String resolveCategory(PushPayloadDTO payload) {
if (StringUtils.equalsAny(payload.getType(), PushTypeEnum.NOTICE.getType())
|| StringUtils.equalsAny(payload.getSource(), PushSourceEnum.NOTICE.getSource())) {
return CATEGORY_NOTICE;
}
if (StringUtils.equalsAny(payload.getSource(), PushSourceEnum.WORKFLOW.getSource())) {
return CATEGORY_WORKFLOW;
}
return CATEGORY_SYSTEM;
}
private String resolveTitle(PushPayloadDTO payload) {
return switch (resolveCategory(payload)) {
case CATEGORY_NOTICE -> "通知公告消息";
case CATEGORY_WORKFLOW -> "工作流消息";
default -> "系统消息";
};
}
private String resolveContent(PushPayloadDTO payload) {
Object data = payload.getData();
if (data instanceof Map<?, ?> map) {
return Convert.toStr(map.get("noticeContent"));
}
return null;
}
private Object parseData(String dataJson) {
if (StringUtils.isBlank(dataJson)) {
return null;
}
return JsonUtils.parseObject(dataJson, Object.class);
}
}
ruoyi-modules/ruoyi-resource/src/main/resources/mapper/resource/SysMessageMapper.xml
+5
View File
@@ -0,0 +1,5 @@
<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="org.dromara.resource.mapper.SysMessageMapper">
</mapper>
ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/controller/system/SysNoticeController.java
+20 -1
View File
@@ -4,6 +4,8 @@ import cn.dev33.satoken.annotation.SaCheckPermission;
import lombok.RequiredArgsConstructor;
import org.apache.dubbo.config.annotation.DubboReference;
import org.dromara.common.core.domain.R;
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.redis.annotation.RepeatSubmit;
import org.dromara.common.web.core.BaseController;
@@ -12,12 +14,16 @@ import org.dromara.common.log.enums.BusinessType;
import org.dromara.common.mybatis.core.page.PageQuery;
import org.dromara.common.core.domain.PageResult;
import org.dromara.resource.api.RemoteMessageService;
import org.dromara.resource.api.domain.dto.RemotePushPayLoad;
import org.dromara.system.domain.bo.SysNoticeBo;
import org.dromara.system.domain.vo.SysNoticeVo;
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;
/**
* 公告 信息操作处理
*
@@ -68,7 +74,20 @@ public class SysNoticeController extends BaseController {
return R.fail();
}
String type = dictService.getDictLabel("sys_notice_type", notice.getNoticeType());
remoteMessageService.publishAll("[" + type + "] " + notice.getNoticeTitle());
Map<String, Object> data = new HashMap<>(6);
data.put("noticeType", notice.getNoticeType());
data.put("noticeTypeLabel", type);
data.put("noticeTitle", notice.getNoticeTitle());
data.put("noticeId", notice.getNoticeId());
data.put("noticeContent", notice.getNoticeContent());
data.put("status", notice.getStatus());
remoteMessageService.publishAllPayload(RemotePushPayLoad.of(
PushTypeEnum.NOTICE,
PushSourceEnum.NOTICE,
"[" + type + "] " + notice.getNoticeTitle(),
data,
"/system/notice?noticeId=" + notice.getNoticeId()
));
return R.ok();
}
ruoyi-modules/ruoyi-system/src/main/java/org/dromara/system/service/impl/SysNoticeServiceImpl.java
+3 -1
View File
@@ -92,7 +92,9 @@ public class SysNoticeServiceImpl implements ISysNoticeService {
@Override
public int insertNotice(SysNoticeBo bo) {
SysNotice notice = MapstructUtils.convert(bo, SysNotice.class);
return baseMapper.insert(notice);
int rows = baseMapper.insert(notice);
bo.setNoticeId(notice.getNoticeId());
return rows;
}
/**
ruoyi-modules/ruoyi-workflow/src/main/java/org/dromara/workflow/common/constant/FlowConstant.java
+20
View File
@@ -58,6 +58,26 @@ public interface FlowConstant {
*/
String MESSAGE_NOTICE = "messageNotice";
/**
* 我的发起页面路径
*/
String PATH_MY_DOCUMENT = "/task/myDocument";
/**
* 我的待办页面路径
*/
String PATH_TASK_WAITING = "/task/taskWaiting";
/**
* 我的已办页面路径
*/
String PATH_TASK_FINISH = "/task/taskFinish";
/**
* 我的抄送页面路径
*/
String PATH_TASK_COPY = "/task/taskCopyList";
/**
* 任务状态
*/
ruoyi-modules/ruoyi-workflow/src/main/java/org/dromara/workflow/listener/WorkflowGlobalListener.java
+44
View File
@@ -12,6 +12,7 @@ import org.apache.dubbo.config.annotation.DubboReference;
import org.dromara.common.core.enums.BusinessStatusEnum;
import org.dromara.common.core.utils.StreamUtils;
import org.dromara.common.core.utils.StringUtils;
import org.dromara.system.api.domain.vo.RemoteUserVo;
import org.dromara.system.api.RemoteUserService;
import org.dromara.warm.flow.core.FlowEngine;
import org.dromara.warm.flow.core.dto.FlowParams;
@@ -22,6 +23,7 @@ import org.dromara.warm.flow.core.listener.GlobalListener;
import org.dromara.warm.flow.core.listener.ListenerVariable;
import org.dromara.workflow.common.ConditionalOnEnable;
import org.dromara.workflow.common.constant.FlowConstant;
import org.dromara.workflow.common.enums.MessageTypeEnum;
import org.dromara.workflow.common.enums.TaskStatusEnum;
import org.dromara.workflow.domain.bo.FlowCopyBo;
import org.dromara.workflow.domain.vo.NodeExtVo;
@@ -32,6 +34,7 @@ import org.dromara.workflow.service.IFlwNodeExtService;
import org.dromara.workflow.service.IFlwTaskService;
import org.springframework.stereotype.Component;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -201,12 +204,14 @@ public class WorkflowGlobalListener implements GlobalListener {
String status = determineFlowStatus(instance);
if (StringUtils.isNotBlank(status)) {
flowProcessEventHandler.processHandler(definition.getFlowCode(), instance, status, params, false);
notifyInitiatorIfNeeded(definition, instance, status, variable);
}
if (!BusinessStatusEnum.initialState(instance.getFlowStatus())) {
if (task != null && CollUtil.isNotEmpty(nextTasks) && nextTasks.size() == 1
&& flwCommonService.applyNodeCode(definition.getId()).equals(nextTasks.get(0).getNodeCode())) {
// 如果为画线指定驳回 线条指定为驳回 驳回得节点为申请人节点 则修改流程状态为退回
flowProcessEventHandler.processHandler(definition.getFlowCode(), instance, BusinessStatusEnum.BACK.getStatus(), params, false);
notifyInitiatorIfNeeded(definition, instance, BusinessStatusEnum.BACK.getStatus(), variable);
// 修改流程实例状态
instance.setFlowStatus(BusinessStatusEnum.BACK.getStatus());
FlowEngine.insService().updateById(instance);
@@ -238,8 +243,10 @@ public class WorkflowGlobalListener implements GlobalListener {
if (variable.containsKey(FlowConstant.MESSAGE_TYPE)) {
List<String> messageType = MapUtil.get(variable, FlowConstant.MESSAGE_TYPE, new TypeReference<>() {});
String notice = MapUtil.getStr(variable, FlowConstant.MESSAGE_NOTICE);
if (shouldSendTaskMessage(flowParams, definition, nextTasks)) {
flwCommonService.sendMessage(definition.getFlowName(), instance.getId(), messageType, notice);
}
}
FlowEngine.insService().removeVariables(instance.getId(),
FlowConstant.FLOW_COPY_LIST,
FlowConstant.MESSAGE_TYPE,
@@ -248,6 +255,43 @@ public class WorkflowGlobalListener implements GlobalListener {
);
}
private boolean shouldSendTaskMessage(FlowParams flowParams, Definition definition, List<Task> nextTasks) {
if (flowParams == null || !TaskStatusEnum.BACK.getStatus().equals(flowParams.getHisStatus())) {
return true;
}
if (CollUtil.isEmpty(nextTasks) || nextTasks.size() != 1) {
return true;
}
String applyNodeCode = flwCommonService.applyNodeCode(definition.getId());
return !StringUtils.equals(applyNodeCode, nextTasks.get(0).getNodeCode());
}
private void notifyInitiatorIfNeeded(Definition definition, Instance instance, String status, Map<String, Object> variable) {
if (!StringUtils.equalsAny(status, BusinessStatusEnum.FINISH.getStatus(), BusinessStatusEnum.BACK.getStatus())) {
return;
}
if (StringUtils.isBlank(instance.getCreateBy())) {
return;
}
Long createBy = Convert.toLong(instance.getCreateBy(), null);
if (createBy == null) {
return;
}
BusinessStatusEnum statusEnum = BusinessStatusEnum.getByStatus(status);
if (statusEnum == null) {
return;
}
List<RemoteUserVo> initiators = remoteUserService.selectListByIds(Collections.singletonList(createBy));
if (CollUtil.isEmpty(initiators)) {
return;
}
List<String> messageType = Collections.singletonList(MessageTypeEnum.SYSTEM_MESSAGE.getCode());
if (MapUtil.isNotEmpty(variable) && variable.containsKey(FlowConstant.MESSAGE_TYPE)) {
messageType = MapUtil.get(variable, FlowConstant.MESSAGE_TYPE, new TypeReference<>() {});
}
flwCommonService.sendResultMessage(definition.getFlowName(), statusEnum, messageType, initiators);
}
/**
* 根据流程实例确定最终状态
*
ruoyi-modules/ruoyi-workflow/src/main/java/org/dromara/workflow/service/IFlwCommonService.java
+22
View File
@@ -1,5 +1,6 @@
package org.dromara.workflow.service;
import org.dromara.common.core.enums.BusinessStatusEnum;
import org.dromara.system.api.domain.vo.RemoteUserVo;
import java.util.List;
@@ -30,6 +31,27 @@ public interface IFlwCommonService {
*/
void sendMessage(List<String> messageType, String message, String subject, List<RemoteUserVo> userList);
/**
* 发送带跳转路径的消息
*
* @param messageType 消息类型
* @param message 消息内容
* @param subject 邮件标题
* @param userList 接收用户
* @param path 前端跳转路径
*/
void sendMessage(List<String> messageType, String message, String subject, List<RemoteUserVo> userList, String path);
/**
* 发送流程结果消息
*
* @param flowName 流程名称
* @param status 流程状态
* @param messageType 消息类型
* @param userList 接收用户
*/
void sendResultMessage(String flowName, BusinessStatusEnum status, List<String> messageType, List<RemoteUserVo> userList);
/**
* 申请人节点编码
*
ruoyi-modules/ruoyi-workflow/src/main/java/org/dromara/workflow/service/impl/FlwCommonServiceImpl.java
+28 -3
View File
@@ -5,6 +5,9 @@ import cn.hutool.core.util.ObjectUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboReference;
import org.dromara.common.core.enums.BusinessStatusEnum;
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;
@@ -12,6 +15,7 @@ import org.dromara.common.core.utils.StringUtils;
import org.dromara.resource.api.RemoteMailService;
import org.dromara.resource.api.RemoteMessageService;
import org.dromara.resource.api.RemoteSmsService;
import org.dromara.resource.api.domain.dto.RemotePushPayLoad;
import org.dromara.system.api.domain.vo.RemoteUserVo;
import org.dromara.warm.flow.core.FlowEngine;
import org.dromara.warm.flow.core.entity.Node;
@@ -24,9 +28,11 @@ import org.springframework.stereotype.Service;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import static org.dromara.workflow.common.constant.FlowConstant.PATH_MY_DOCUMENT;
import static org.dromara.workflow.common.constant.FlowConstant.PATH_TASK_WAITING;
/**
* 工作流工具
@@ -73,7 +79,7 @@ public class FlwCommonServiceImpl implements IFlwCommonService {
if (CollUtil.isEmpty(userList)) {
return;
}
sendMessage(messageType, message, DEFAULT_SUBJECT, userList);
sendMessage(messageType, message, DEFAULT_SUBJECT, userList, PATH_TASK_WAITING);
}
/**
@@ -86,6 +92,20 @@ public class FlwCommonServiceImpl implements IFlwCommonService {
*/
@Override
public void sendMessage(List<String> messageType, String message, String subject, List<RemoteUserVo> userList) {
sendMessage(messageType, message, subject, userList, null);
}
@Override
public void sendResultMessage(String flowName, BusinessStatusEnum status, List<String> messageType, List<RemoteUserVo> userList) {
if (status == null || CollUtil.isEmpty(messageType) || CollUtil.isEmpty(userList)) {
return;
}
String message = "您发起的【" + flowName + "】单据审批已" + status.getDesc() + "。";
sendMessage(messageType, message, DEFAULT_SUBJECT, userList, PATH_MY_DOCUMENT);
}
@Override
public void sendMessage(List<String> messageType, String message, String subject, List<RemoteUserVo> userList, String path) {
if (CollUtil.isEmpty(messageType) || CollUtil.isEmpty(userList)) {
return;
}
@@ -100,7 +120,12 @@ public class FlwCommonServiceImpl implements IFlwCommonService {
try {
switch (messageTypeEnum) {
case SYSTEM_MESSAGE -> {
remoteMessageService.publishMessage(userIds, message);
RemotePushPayLoad payload = RemotePushPayLoad.of(
PushTypeEnum.MESSAGE,
PushSourceEnum.WORKFLOW,
message, null, path
);
remoteMessageService.publishMessagePayload(userIds, payload);
}
case EMAIL_MESSAGE -> {
remoteMailService.send(emails, subject, message);
ruoyi-modules/ruoyi-workflow/src/main/java/org/dromara/workflow/service/impl/FlwTaskServiceImpl.java
+11 -2
View File
@@ -44,6 +44,7 @@ import org.dromara.warm.flow.orm.mapper.FlowTaskMapper;
import org.dromara.workflow.api.domain.RemoteStartProcessReturn;
import org.dromara.workflow.common.ConditionalOnEnable;
import org.dromara.workflow.common.constant.FlowConstant;
import org.dromara.workflow.common.enums.MessageTypeEnum;
import org.dromara.workflow.common.enums.TaskAssigneeType;
import org.dromara.workflow.common.enums.TaskOperationEnum;
import org.dromara.workflow.common.enums.TaskStatusEnum;
@@ -367,6 +368,13 @@ public class FlwTaskServiceImpl implements IFlwTaskService {
.setAssociated(taskId));
// 批量保存抄送人员
FlowEngine.userService().saveBatch(userList);
flwCommonService.sendMessage(
Collections.singletonList(MessageTypeEnum.SYSTEM_MESSAGE.getCode()),
"您收到一条新的流程抄送，请及时查看。",
"单据抄送提醒",
remoteUserService.selectListByIds(StreamUtils.toList(flowCopyList, FlowCopyBo::getUserId)),
PATH_TASK_COPY
);
}
/**
@@ -812,7 +820,8 @@ public class FlwTaskServiceImpl implements IFlwTaskService {
bo.getMessageType(),
StringUtils.isNotBlank(bo.getMessage()) ? bo.getMessage() : "单据「" + op.getDesc() + "」通知",
"单据「" + op.getDesc() + "」提醒",
remoteUserService.selectListByIds(userIdList)
remoteUserService.selectListByIds(userIdList),
PATH_TASK_WAITING
);
}
}
@@ -891,7 +900,7 @@ public class FlwTaskServiceImpl implements IFlwTaskService {
}
List<String> messageType = bo.getMessageType();
String message = bo.getMessage();
flwCommonService.sendMessage(messageType, message, "单据审批提醒", userList);
flwCommonService.sendMessage(messageType, message, "单据审批提醒", userList, PATH_TASK_WAITING);
return true;
}
script/config/nacos/ruoyi-resource.yml
+7 -10
View File
@@ -24,17 +24,14 @@ spring:
# username: ${datasource.system-postgres.username}
# password: ${datasource.system-postgres.password}
# 默认/推荐使用sse推送
sse:
# 统一消息推送配置
message:
enabled: true
path: /sse
websocket:
# 如果关闭 需要和前端开关一起关闭
enabled: false
# 路径
path: /websocket
# 设置访问源地址
# sse / websocket
transport: sse
# 统一访问路径
path: /resource/message
# websocket 允许的跨域来源
allowedOrigins: '*'
mail:
script/sql/oracle/oracle_ry_cloud.sql
+42 -1
View File
@@ -1017,9 +1017,50 @@ comment on column sys_notice.remark is '备注';
insert into sys_notice values('1', '温馨提醒：2018-07-01 新版本发布啦', '2', '新版本内容', '0', 103, 1, sysdate, null, null, '管理员');
insert into sys_notice values('2', '维护通知：2018-07-01 系统凌晨维护', '1', '维护内容', '0', 103, 1, sysdate, null, null, '管理员');
-- ----------------------------
-- 18、消息记录表
-- ----------------------------
create table sys_message (
message_id number(20) not null,
category varchar2(32) not null,
type varchar2(32) not null,
source varchar2(32) not null,
title varchar2(100) not null,
message varchar2(500) default null,
content varchar2(2000) default null,
data_json clob default null,
path varchar2(255) default null,
send_user_ids varchar2(1000) not null,
create_dept number(20) default null,
create_by number(20) default null,
create_time date,
update_by number(20) default null,
update_time date
);
alter table sys_message add constraint pk_sys_message primary key (message_id);
create index idx_sys_message_category_time on sys_message(category, create_time);
comment on table sys_message is '消息记录表';
comment on column sys_message.message_id is '消息ID';
comment on column sys_message.category is '消息分组(system/notice/workflow)';
comment on column sys_message.type is '消息类型';
comment on column sys_message.source is '消息来源';
comment on column sys_message.title is '标题';
comment on column sys_message.message is '摘要消息';
comment on column sys_message.content is '详细内容';
comment on column sys_message.data_json is '扩展数据JSON';
comment on column sys_message.path is '前端跳转路径';
comment on column sys_message.send_user_ids is '目标用户ID串，0表示全局';
comment on column sys_message.create_dept is '创建部门';
comment on column sys_message.create_by is '创建者';
comment on column sys_message.create_time is '创建时间';
comment on column sys_message.update_by is '更新者';
comment on column sys_message.update_time is '更新时间';
-- ----------------------------
-- 18、代码生成业务表
-- 19、代码生成业务表
-- ----------------------------
create table gen_table (
table_id number(20) not null,
script/sql/postgres/postgres_ry_cloud.sql
+43 -1
View File
@@ -1015,9 +1015,51 @@ comment on column sys_notice.remark is '备注';
insert into sys_notice values('1', '温馨提醒：2018-07-01 新版本发布啦', '2', '新版本内容', '0', 103, 1, now(), null, null, '管理员');
insert into sys_notice values('2', '维护通知：2018-07-01 系统凌晨维护', '1', '维护内容', '0', 103, 1, now(), null, null, '管理员');
-- ----------------------------
-- 18、消息记录表
-- ----------------------------
create table if not exists sys_message
(
message_id int8,
category varchar(32) not null,
type varchar(32) not null,
source varchar(32) not null,
title varchar(100) not null,
message varchar(500) default null::varchar,
content varchar(2000) default null::varchar,
data_json text,
path varchar(255) default null::varchar,
send_user_ids varchar(1000) not null,
create_dept int8,
create_by int8,
create_time timestamp,
update_by int8,
update_time timestamp,
constraint sys_message_pk primary key (message_id)
);
create index if not exists idx_sys_message_category_time on sys_message (category, create_time);
comment on table sys_message is '消息记录表';
comment on column sys_message.message_id is '消息ID';
comment on column sys_message.category is '消息分组(system/notice/workflow)';
comment on column sys_message.type is '消息类型';
comment on column sys_message.source is '消息来源';
comment on column sys_message.title is '标题';
comment on column sys_message.message is '摘要消息';
comment on column sys_message.content is '详细内容';
comment on column sys_message.data_json is '扩展数据JSON';
comment on column sys_message.path is '前端跳转路径';
comment on column sys_message.send_user_ids is '目标用户ID串，0表示全局';
comment on column sys_message.create_dept is '创建部门';
comment on column sys_message.create_by is '创建者';
comment on column sys_message.create_time is '创建时间';
comment on column sys_message.update_by is '更新者';
comment on column sys_message.update_time is '更新时间';
-- ----------------------------
-- 18、代码生成业务表
-- 19、代码生成业务表
-- ----------------------------
create table if not exists gen_table
(
script/sql/ry-cloud.sql
+24 -1
View File
@@ -763,9 +763,32 @@ create table sys_notice (
insert into sys_notice values('1', '温馨提醒：2018-07-01 新版本发布啦', '2', '新版本内容', '0', 103, 1, sysdate(), null, null, '管理员');
insert into sys_notice values('2', '维护通知：2018-07-01 系统凌晨维护', '1', '维护内容', '0', 103, 1, sysdate(), null, null, '管理员');
-- ----------------------------
-- 18、消息记录表
-- ----------------------------
create table sys_message (
message_id bigint(20) not null comment '消息ID',
category varchar(32) not null comment '消息分组(system/notice/workflow)',
type varchar(32) not null comment '消息类型',
source varchar(32) not null comment '消息来源',
title varchar(100) not null comment '标题',
message varchar(500) default null comment '摘要消息',
content varchar(2000) default null comment '详细内容',
data_json text comment '扩展数据JSON',
path varchar(255) default null comment '前端跳转路径',
send_user_ids varchar(1000) not null comment '目标用户ID串，0表示全局',
create_dept bigint(20) default null comment '创建部门',
create_by bigint(20) default null comment '创建者',
create_time datetime comment '创建时间',
update_by bigint(20) default null comment '更新者',
update_time datetime comment '更新时间',
primary key (message_id),
key idx_sys_message_category_time (category, create_time)
) engine=innodb comment = '消息记录表';
-- ----------------------------
-- 18、代码生成业务表
-- 19、代码生成业务表
-- ----------------------------
create table gen_table (
table_id bigint(20) not null comment '编号',