添加websocket推送框架。

This commit is contained in:
zfjia committed 2026-11-12 14:00:31 +08:00
1 parent bacdd15f3a
commit 83ece20a00
22 files changed
+1462 -7

No files matched your search

+6
View File
@@ -41,6 +41,12 @@
<artifactId>spring-tx</artifactId>
</dependency>
<!-- 值班员实时通道:原生 WebSocket(/apt/ws),不引入 STOMP/SockJS -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
<!-- 公寓端自测(JUnit5 + Mockito):仅在 ruoyi-apt 内生效,不向下游传递 -->
<dependency>
<groupId>org.springframework.boot</groupId>
@@ -13,6 +13,7 @@ import com.ruoyi.apt.dispatch.domain.AptDutyCommand;
import com.ruoyi.apt.dispatch.domain.AptDutyCommandResult;
import com.ruoyi.apt.dispatch.mapper.AptJcxxMapper;
import com.ruoyi.apt.framework.audit.AptAuditWriter;
import com.ruoyi.apt.framework.realtime.IAptRealtimePublisher;
import com.ruoyi.apt.ops.domain.AptOperationLog;
import com.ruoyi.common.utils.SecurityUtils;
import org.springframework.dao.DuplicateKeyException;
@@ -48,13 +49,15 @@ public class AptDutyCommandServiceImpl implements IAptDutyCommandService
private final AptJcxxMapper jcxxMapper;
private final IAptDutyStateService stateService;
private final AptAuditWriter auditWriter;
private final IAptRealtimePublisher realtimePublisher;
public AptDutyCommandServiceImpl(AptJcxxMapper jcxxMapper, IAptDutyStateService stateService,
AptAuditWriter auditWriter)
AptAuditWriter auditWriter, IAptRealtimePublisher realtimePublisher)
{
this.jcxxMapper = jcxxMapper;
this.stateService = stateService;
this.auditWriter = auditWriter;
this.realtimePublisher = realtimePublisher;
}
@Override
@@ -118,6 +121,8 @@ public class AptDutyCommandServiceImpl implements IAptDutyCommandService
result.setDuplicated(false);
result.setStateAfter(afterLabel);
result.setActionId(actionId);
// 事务提交后推送(回滚不发),按 apt:jcxx:list 投递
realtimePublisher.callStateChanged(jcid, action, afterLabel, false);
return result;
}
@@ -7,6 +7,7 @@ import com.ruoyi.apt.common.AptErrorCode;
import com.ruoyi.apt.common.AptException;
import com.ruoyi.apt.dispatch.domain.AptStateSnapshot;
import com.ruoyi.apt.dispatch.mapper.AptDutyStateMapper;
import com.ruoyi.apt.framework.realtime.IAptRealtimePublisher;
import com.ruoyi.apt.job.service.IAptJobService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -39,11 +40,14 @@ public class AptDutyStateServiceImpl implements IAptDutyStateService
private final AptDutyStateMapper stateMapper;
private final IAptJobService jobService;
private final IAptRealtimePublisher realtimePublisher;
public AptDutyStateServiceImpl(AptDutyStateMapper stateMapper, IAptJobService jobService)
public AptDutyStateServiceImpl(AptDutyStateMapper stateMapper, IAptJobService jobService,
IAptRealtimePublisher realtimePublisher)
{
this.stateMapper = stateMapper;
this.jobService = jobService;
this.realtimePublisher = realtimePublisher;
}
@Override
@@ -70,7 +74,10 @@ public class AptDutyStateServiceImpl implements IAptDutyStateService
@Transactional
public int callFailed(String jcid)
{
return require(stateMapper.updateCallFailed(gyid(), jcid), jcid, "CALL_FAILED");
int rows = require(stateMapper.updateCallFailed(gyid(), jcid), jcid, "CALL_FAILED");
// 无应答 / 超时转叫乘报警:P0,值班台需立即打断
realtimePublisher.alarm(jcid, "CALL_FAILED", "无应答转叫乘报警");
return rows;
}
@Override
@@ -19,6 +19,7 @@ import com.ruoyi.apt.framework.device.AptDeviceSigner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DuplicateKeyException;
import com.ruoyi.apt.framework.realtime.IAptRealtimePublisher;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
@@ -61,11 +62,14 @@ public class AptTabletResultServiceImpl implements IAptTabletResultService
private final AptTabletMapper tabletMapper;
private final IAptDutyStateService stateService;
private final IAptRealtimePublisher realtimePublisher;
public AptTabletResultServiceImpl(AptTabletMapper tabletMapper, IAptDutyStateService stateService)
public AptTabletResultServiceImpl(AptTabletMapper tabletMapper, IAptDutyStateService stateService,
IAptRealtimePublisher realtimePublisher)
{
this.tabletMapper = tabletMapper;
this.stateService = stateService;
this.realtimePublisher = realtimePublisher;
}
@Override
@@ -136,7 +140,11 @@ public class AptTabletResultServiceImpl implements IAptTabletResultService
List<String> warnings = new ArrayList<>();
warnings.add("未刷脸:保持待确认并提示复叫(Q-22),值班员确认前不判完成");
result.put("warnings", warnings);
// 未刷脸/跳过需值班员介入:P0 报警 + 复叫提示
realtimePublisher.alarm(jcid, "FACE_UNVERIFIED", "刷脸未通过,需值班员确认");
}
// 平板结果受理完成(状态机已提交):P1 状态变更
realtimePublisher.callStateChanged(jcid, "TABLET_RESULT", appliedState, false);
return result;
}
@@ -23,6 +23,7 @@ public class AptProperties
private Dispatch dispatch = new Dispatch();
private Voice voice = new Voice();
private Report report = new Report();
private Realtime realtime = new Realtime();
public String getGyId()
{
@@ -114,6 +115,16 @@ public class AptProperties
this.report = report;
}
public Realtime getRealtime()
{
return realtime;
}
public void setRealtime(Realtime realtime)
{
this.realtime = realtime;
}
public static class Probe
{
private boolean enabled = true;
@@ -650,4 +661,62 @@ public class AptProperties
this.resumeBatchLimit = resumeBatchLimit;
}
}
/**
* 值班员实时通道(WebSocket)。
*
* <p>浏览器无法设置握手 Header,令牌走 query;权限在连接建立时快照,
* 角色调整后需重连生效。</p>
*/
public static class Realtime
{
/** 总开关;关闭后不注册 /apt/ws,也不影响任何既有 HTTP 接口 */
private boolean enabled = true;
/** 允许的来源;生产应收敛为值班台域名 */
private java.util.List<String> allowedOrigins = new java.util.ArrayList<>(java.util.List.of("*"));
/** 客户端心跳间隔(秒),服务端在 CONNECTED 中下发 */
private int heartbeatIntervalSeconds = 25;
/** 建立连接所需权限点;超管 *:*:* 直接放行 */
private String connectPermission = "apt:realtime:connect";
public boolean isEnabled()
{
return enabled;
}
public void setEnabled(boolean enabled)
{
this.enabled = enabled;
}
public java.util.List<String> getAllowedOrigins()
{
return allowedOrigins;
}
public void setAllowedOrigins(java.util.List<String> allowedOrigins)
{
this.allowedOrigins = allowedOrigins;
}
public int getHeartbeatIntervalSeconds()
{
return heartbeatIntervalSeconds;
}
public void setHeartbeatIntervalSeconds(int heartbeatIntervalSeconds)
{
this.heartbeatIntervalSeconds = heartbeatIntervalSeconds;
}
public String getConnectPermission()
{
return connectPermission;
}
public void setConnectPermission(String connectPermission)
{
this.connectPermission = connectPermission;
}
}
}
@@ -0,0 +1,43 @@
package com.ruoyi.apt.framework.realtime;
import com.ruoyi.apt.framework.config.AptProperties;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
/**
* WebSocket 注册:{@code /apt/ws?token=xxx}。
*
* <p>握手路径在 SecurityConfig 中放行,鉴权由 {@link AptRealtimeHandshakeInterceptor} 完成
* (JWT + Redis 会话 + {@code apt:realtime:connect} 权限点)。</p>
*
* @author apt-rev01
*/
@Configuration
@EnableWebSocket
@ConditionalOnProperty(prefix = "apt.realtime", name = "enabled", havingValue = "true", matchIfMissing = true)
public class AptRealtimeConfig implements WebSocketConfigurer
{
private final AptRealtimeHandler handler;
private final AptRealtimeHandshakeInterceptor interceptor;
private final AptProperties properties;
public AptRealtimeConfig(AptRealtimeHandler handler, AptRealtimeHandshakeInterceptor interceptor,
AptProperties properties)
{
this.handler = handler;
this.interceptor = interceptor;
this.properties = properties;
}
@Override
public void registerWebSocketHandlers(WebSocketHandlerRegistry registry)
{
registry.addHandler(handler, "/apt/ws")
.addInterceptors(interceptor)
.setAllowedOriginPatterns(
properties.getRealtime().getAllowedOrigins().toArray(new String[0]));
}
}
@@ -0,0 +1,123 @@
package com.ruoyi.apt.framework.realtime;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.ruoyi.apt.framework.config.AptProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;
/**
* 值班员实时通道。
*
* <p>心跳:客户端每 25s 发送 {@code {"type":"PING"}}(或纯文本 PING),服务端回
* {@code {"type":"PONG","serverTime":...}};服务端不自造业务消息。断线由客户端指数退避重连。</p>
*
* @author apt-rev01
*/
@Component
public class AptRealtimeHandler extends TextWebSocketHandler
{
private static final Logger log = LoggerFactory.getLogger(AptRealtimeHandler.class);
private final AptRealtimeSessionRegistry registry;
private final AptProperties properties;
public AptRealtimeHandler(AptRealtimeSessionRegistry registry, AptProperties properties)
{
this.registry = registry;
this.properties = properties;
}
@Override
public void afterConnectionEstablished(WebSocketSession session) throws Exception
{
Map<String, Object> attributes = session.getAttributes();
Long userId = (Long) attributes.get("userId");
String username = (String) attributes.get("username");
String gyid = (String) attributes.get("gyid");
@SuppressWarnings("unchecked")
Set<String> permissions = (Set<String>) attributes.get("permissions");
if (userId == null || gyid == null)
{
session.close(CloseStatus.NOT_ACCEPTABLE);
return;
}
registry.register(session, userId, username, gyid, permissions);
Map<String, Object> data = new LinkedHashMap<>();
data.put("sessionId", session.getId());
data.put("userId", userId);
data.put("username", username);
data.put("gyid", gyid);
data.put("serverTime", new Date());
data.put("heartbeatIntervalSec", properties.getRealtime().getHeartbeatIntervalSeconds());
session.sendMessage(new TextMessage(JSON.toJSONString(envelope(AptRealtimeTypes.CONNECTED, data))));
}
@Override
protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception
{
String payload = message.getPayload();
if (payload == null || payload.isBlank())
{
return;
}
registry.touch(session.getId());
String type = null;
try
{
JSONObject json = JSON.parseObject(payload);
type = json == null ? null : json.getString("type");
}
catch (Exception e)
{
if ("PING".equalsIgnoreCase(payload.trim()))
{
type = "PING";
}
}
if ("PING".equalsIgnoreCase(type) || "PING".equalsIgnoreCase(payload.trim()))
{
Map<String, Object> data = new LinkedHashMap<>();
data.put("serverTime", new Date());
session.sendMessage(new TextMessage(JSON.toJSONString(envelope(AptRealtimeTypes.PONG, data))));
}
}
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status)
{
registry.remove(session.getId());
}
@Override
public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception
{
log.warn("WebSocket 传输异常:sessionId={}, cause={}", session.getId(), exception.toString());
registry.remove(session.getId());
if (session.isOpen())
{
session.close(CloseStatus.SERVER_ERROR);
}
}
private Map<String, Object> envelope(String type, Map<String, Object> data)
{
Map<String, Object> message = new LinkedHashMap<>();
message.put("type", type);
message.put("category", AptRealtimeTypes.CATEGORY_SYSTEM);
message.put("priority", AptRealtimeTypes.PRIORITY_P2);
message.put("occurredAt", System.currentTimeMillis());
message.put("data", data);
return message;
}
}
@@ -0,0 +1,125 @@
package com.ruoyi.apt.framework.realtime;
import java.util.Map;
import java.util.Set;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletRequestWrapper;
import com.ruoyi.apt.framework.config.AptProperties;
import com.ruoyi.common.core.domain.model.LoginUser;
import com.ruoyi.framework.web.service.TokenService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.server.ServerHttpRequest;
import org.springframework.http.server.ServerHttpResponse;
import org.springframework.http.server.ServletServerHttpRequest;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.server.HandshakeInterceptor;
import org.springframework.web.util.UriComponentsBuilder;
/**
* WebSocket 握手鉴权。
*
* <p>浏览器无法自定义握手 Header,因此支持 {@code /apt/ws?token=xxx};同时保留
* {@code Authorization: Bearer}(非浏览器客户端)。令牌校验复用若依 {@link TokenService}
* (JWT + Redis 会话),权限在连接建立时快照,角色变更需重连生效。</p>
*
* @author apt-rev01
*/
@Component
public class AptRealtimeHandshakeInterceptor implements HandshakeInterceptor
{
private static final Logger log = LoggerFactory.getLogger(AptRealtimeHandshakeInterceptor.class);
private final TokenService tokenService;
private final AptProperties properties;
public AptRealtimeHandshakeInterceptor(TokenService tokenService, AptProperties properties)
{
this.tokenService = tokenService;
this.properties = properties;
}
@Override
public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response,
WebSocketHandler wsHandler, Map<String, Object> attributes)
{
String token = resolveToken(request);
if (!StringUtils.hasText(token))
{
response.setStatusCode(HttpStatus.UNAUTHORIZED);
return false;
}
LoginUser loginUser = tokenService.getLoginUser(withBearerToken(request, token));
if (loginUser == null)
{
log.warn("WebSocket 握手拒绝:令牌无效或已过期");
response.setStatusCode(HttpStatus.UNAUTHORIZED);
return false;
}
Set<String> permissions = loginUser.getPermissions();
String required = properties.getRealtime().getConnectPermission();
if (StringUtils.hasText(required) && !permissions.contains("*:*:*")
&& !permissions.contains(required))
{
log.warn("WebSocket 握手拒绝:userId={} 缺少权限点 {}", loginUser.getUserId(), required);
response.setStatusCode(HttpStatus.FORBIDDEN);
return false;
}
String gyid = properties.getGyId();
if (!StringUtils.hasText(gyid))
{
response.setStatusCode(HttpStatus.FORBIDDEN);
return false;
}
attributes.put("userId", loginUser.getUserId());
attributes.put("username", loginUser.getUsername());
attributes.put("gyid", gyid);
attributes.put("permissions", permissions);
return true;
}
@Override
public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response,
WebSocketHandler wsHandler, Exception exception)
{
// 无需处理
}
private String resolveToken(ServerHttpRequest request)
{
String token = UriComponentsBuilder.fromUri(request.getURI()).build()
.getQueryParams().getFirst("token");
if (StringUtils.hasText(token))
{
return token.trim();
}
String header = request.getHeaders().getFirst(HttpHeaders.AUTHORIZATION);
if (StringUtils.hasText(header) && header.startsWith("Bearer "))
{
return header.substring(7).trim();
}
return null;
}
/** TokenService 只从 Header 读取令牌,这里把 query 参数包装成 Authorization 头 */
private HttpServletRequest withBearerToken(ServerHttpRequest request, String token)
{
HttpServletRequest original = ((ServletServerHttpRequest) request).getServletRequest();
return new HttpServletRequestWrapper(original)
{
@Override
public String getHeader(String name)
{
if (HttpHeaders.AUTHORIZATION.equalsIgnoreCase(name))
{
return "Bearer " + token;
}
return super.getHeader(name);
}
};
}
}
@@ -0,0 +1,148 @@
package com.ruoyi.apt.framework.realtime;
import java.util.LinkedHashMap;
import java.util.Map;
/**
* 实时推送消息信封(公寓后端 → 值班员前端)。
*
* <p>字段稳定,便于前端统一分发:{@code type} 决定处理分支,{@code category} 决定声音/角标,
* {@code priority} 决定提示强度,{@code target} 由服务端在连接时校验后投递。</p>
*
* @author apt-rev01
*/
public class AptRealtimeMessage
{
private String id;
private String type;
private String category;
private String priority;
private String gyid;
private long occurredAt;
private String traceId;
/** PERMISSION / USER / ALL */
private String targetScope = "PERMISSION";
private String targetPermission;
private Long targetUserId;
private Map<String, Object> data = new LinkedHashMap<>();
public String getId()
{
return id;
}
public void setId(String id)
{
this.id = id;
}
public String getType()
{
return type;
}
public void setType(String type)
{
this.type = type;
}
public String getCategory()
{
return category;
}
public void setCategory(String category)
{
this.category = category;
}
public String getPriority()
{
return priority;
}
public void setPriority(String priority)
{
this.priority = priority;
}
public String getGyid()
{
return gyid;
}
public void setGyid(String gyid)
{
this.gyid = gyid;
}
public long getOccurredAt()
{
return occurredAt;
}
public void setOccurredAt(long occurredAt)
{
this.occurredAt = occurredAt;
}
public String getTraceId()
{
return traceId;
}
public void setTraceId(String traceId)
{
this.traceId = traceId;
}
public String getTargetScope()
{
return targetScope;
}
public void setTargetScope(String targetScope)
{
this.targetScope = targetScope;
}
public String getTargetPermission()
{
return targetPermission;
}
public void setTargetPermission(String targetPermission)
{
this.targetPermission = targetPermission;
}
public Long getTargetUserId()
{
return targetUserId;
}
public void setTargetUserId(Long targetUserId)
{
this.targetUserId = targetUserId;
}
public Map<String, Object> getData()
{
return data;
}
public void setData(Map<String, Object> data)
{
this.data = data == null ? new LinkedHashMap<>() : data;
}
}
@@ -0,0 +1,192 @@
package com.ruoyi.apt.framework.realtime;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.UUID;
import com.alibaba.fastjson2.JSON;
import com.ruoyi.apt.common.AptContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;
/**
* 实时推送实现。
*
* <p>铁律:推送失败绝不影响业务;事务内调用一律延迟到 afterCommit,
* 避免前端看到未提交状态(回滚后仍收到"已变更")。</p>
*
* @author apt-rev01
*/
@Component
public class AptRealtimePublisher implements IAptRealtimePublisher
{
private static final Logger log = LoggerFactory.getLogger(AptRealtimePublisher.class);
private static final String SCOPE_PERMISSION = "PERMISSION";
private static final String SCOPE_USER = "USER";
private static final String SCOPE_ALL = "ALL";
private final AptRealtimeSessionRegistry registry;
public AptRealtimePublisher(AptRealtimeSessionRegistry registry)
{
this.registry = registry;
}
@Override
public void callStateChanged(String jcid, String command, String stateAfter, boolean duplicated)
{
AptRealtimeMessage message = base(AptRealtimeTypes.CALL_STATE_CHANGED,
AptRealtimeTypes.CATEGORY_CALL);
message.getData().put("jcid", jcid);
message.getData().put("command", command);
message.getData().put("stateAfter", stateAfter);
message.getData().put("duplicated", duplicated);
dispatch(message);
}
@Override
public void alarm(String jcid, String alarmType, String detail)
{
AptRealtimeMessage message = base(AptRealtimeTypes.CALL_ALARM,
AptRealtimeTypes.CATEGORY_ALARM);
message.getData().put("jcid", jcid);
message.getData().put("alarmType", alarmType);
message.getData().put("detail", detail);
message.getData().put("sound", true);
dispatch(message);
}
@Override
public void jobChanged(String eventType, String bizType, String bizId, String roomNo, String state)
{
AptRealtimeMessage message = base(eventType, AptRealtimeTypes.CATEGORY_JOB);
message.getData().put("bizType", bizType);
message.getData().put("bizId", bizId);
message.getData().put("roomNo", roomNo);
message.getData().put("state", state);
dispatch(message);
}
@Override
public void checkinApplied(String applyId, String workNo, String name)
{
AptRealtimeMessage message = base(AptRealtimeTypes.CHECKIN_APPLIED,
AptRealtimeTypes.CATEGORY_CHECKIN);
message.getData().put("applyId", applyId);
message.getData().put("workNo", workNo);
message.getData().put("name", name);
dispatch(message);
}
@Override
public void systemNotice(String title, Map<String, Object> data)
{
AptRealtimeMessage message = base(AptRealtimeTypes.SYSTEM_NOTICE,
AptRealtimeTypes.CATEGORY_SYSTEM);
message.setTargetScope(SCOPE_ALL);
message.getData().put("title", title);
if (data != null && !data.isEmpty())
{
message.getData().putAll(data);
}
dispatch(message);
}
@Override
public void publish(AptRealtimeMessage message)
{
if (message != null)
{
dispatch(message);
}
}
private AptRealtimeMessage base(String type, String category)
{
AptRealtimeMessage message = new AptRealtimeMessage();
message.setType(type);
message.setCategory(category);
message.setPriority(AptRealtimeTypes.priorityOf(category));
message.setTargetPermission(AptRealtimeTypes.permissionOf(category));
return message;
}
private void dispatch(AptRealtimeMessage message)
{
try
{
String gyid = AptContext.getGyIdOrNull();
if (gyid == null || gyid.isBlank())
{
// 无租户上下文(如未绑定公寓的调度线程)不推送,避免跨域泄露
return;
}
message.setGyid(gyid);
if (message.getId() == null)
{
message.setId(UUID.randomUUID().toString().replace("-", ""));
}
if (message.getOccurredAt() == 0L)
{
message.setOccurredAt(System.currentTimeMillis());
}
if (message.getTraceId() == null)
{
message.setTraceId(AptContext.getTraceId());
}
Runnable task = () -> send(message);
if (TransactionSynchronizationManager.isSynchronizationActive())
{
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization()
{
@Override
public void afterCommit()
{
task.run();
}
});
}
else
{
task.run();
}
}
catch (Exception e)
{
log.warn("实时推送登记失败,type={}, cause={}", message.getType(), e.toString());
}
}
private void send(AptRealtimeMessage message)
{
try
{
String payload = JSON.toJSONString(message);
String scope = message.getTargetScope() == null ? SCOPE_PERMISSION : message.getTargetScope();
int sent;
if (SCOPE_ALL.equals(scope))
{
sent = registry.broadcast(message.getGyid(), payload);
}
else if (SCOPE_USER.equals(scope))
{
sent = registry.sendToUser(message.getGyid(), message.getTargetUserId(), payload);
}
else
{
sent = registry.sendToPermission(message.getGyid(), message.getTargetPermission(), payload);
}
if (sent == 0)
{
log.debug("实时推送无在线订阅者:type={}, scope={}", message.getType(), scope);
}
}
catch (Exception e)
{
log.warn("实时推送失败,type={}, cause={}", message.getType(), e.toString());
}
}
}
@@ -0,0 +1,196 @@
package com.ruoyi.apt.framework.realtime;
import java.io.IOException;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
/**
* 在线值班员会话注册表(单进程内存)。
*
* <p>键为 WebSocket sessionId;投递前按 gyid 与权限点过滤,避免跨公寓/越权推送。
* 发送串行化到单个 session(WebSocketSession 非线程安全),失败即摘除会话。</p>
*
* @author apt-rev01
*/
@Component
public class AptRealtimeSessionRegistry
{
private static final Logger log = LoggerFactory.getLogger(AptRealtimeSessionRegistry.class);
private final Map<String, Connection> connections = new ConcurrentHashMap<>();
public void register(WebSocketSession session, Long userId, String username, String gyid,
Set<String> permissions)
{
Connection connection = new Connection(session, userId, username, gyid, permissions);
connections.put(session.getId(), connection);
log.info("WebSocket 已连接:sessionId={}, userId={}, gyid={}, 在线={}",
session.getId(), userId, gyid, connections.size());
}
public void remove(String sessionId)
{
Connection removed = connections.remove(sessionId);
if (removed != null)
{
log.info("WebSocket 已断开:sessionId={}, userId={}, 在线={}",
sessionId, removed.userId, connections.size());
}
}
public void touch(String sessionId)
{
Connection connection = connections.get(sessionId);
if (connection != null)
{
connection.lastActiveAt.set(System.currentTimeMillis());
}
}
public int online(String gyid)
{
int count = 0;
for (Connection connection : connections.values())
{
if (connection.gyid.equals(gyid))
{
count++;
}
}
return count;
}
/** 按 gyid 全员广播 */
public int broadcast(String gyid, String payload)
{
int sent = 0;
for (Connection connection : connections.values())
{
if (connection.gyid.equals(gyid) && send(connection, payload))
{
sent++;
}
}
return sent;
}
/** 按权限点投递(超管 *:*:* 命中) */
public int sendToPermission(String gyid, String permission, String payload)
{
int sent = 0;
for (Connection connection : connections.values())
{
if (!connection.gyid.equals(gyid))
{
continue;
}
if (permission != null && !connection.permissions.contains(permission)
&& !connection.permissions.contains("*:*:*"))
{
continue;
}
if (send(connection, payload))
{
sent++;
}
}
return sent;
}
/** 定向推送(如派单给具体值班员) */
public int sendToUser(String gyid, Long userId, String payload)
{
int sent = 0;
for (Connection connection : connections.values())
{
if (connection.gyid.equals(gyid) && connection.userId.equals(userId)
&& send(connection, payload))
{
sent++;
}
}
return sent;
}
public int sendToSession(String sessionId, String payload)
{
Connection connection = connections.get(sessionId);
return connection != null && send(connection, payload) ? 1 : 0;
}
/** 关闭超过 idleMs 未发心跳的会话(供后续定时任务调用) */
public int sweepIdle(long idleMs)
{
long now = System.currentTimeMillis();
int closed = 0;
for (Connection connection : connections.values())
{
if (now - connection.lastActiveAt.get() > idleMs && connection.session.isOpen())
{
try
{
connection.session.close();
}
catch (IOException e)
{
log.debug("关闭空闲 WebSocket 失败:{}", e.toString());
}
connections.remove(connection.session.getId());
closed++;
}
}
return closed;
}
private boolean send(Connection connection, String payload)
{
WebSocketSession session = connection.session;
if (!session.isOpen())
{
connections.remove(session.getId());
return false;
}
try
{
synchronized (session)
{
session.sendMessage(new TextMessage(payload));
}
connection.lastActiveAt.set(System.currentTimeMillis());
return true;
}
catch (Exception e)
{
log.warn("WebSocket 推送失败,摘除会话:sessionId={}, cause={}", session.getId(), e.toString());
connections.remove(session.getId());
return false;
}
}
private static final class Connection
{
private final WebSocketSession session;
private final Long userId;
private final String username;
private final String gyid;
private final Set<String> permissions;
private final AtomicLong lastActiveAt = new AtomicLong(System.currentTimeMillis());
private Connection(WebSocketSession session, Long userId, String username, String gyid,
Set<String> permissions)
{
this.session = session;
this.userId = userId;
this.username = username;
this.gyid = gyid;
this.permissions = permissions == null ? Set.of() : permissions;
}
}
}
@@ -0,0 +1,87 @@
package com.ruoyi.apt.framework.realtime;
/**
* 实时消息类型 / 分类 / 优先级 / 默认投递权限。
*
* <p>优先级口径:P0 需立即打断(报警类);P1 值班业务变化(叫乘、入住审批、工单);
* P2 状态可视(设备、上报水位、系统通告),允许合并/降频。</p>
*
* @author apt-rev01
*/
public final class AptRealtimeTypes
{
private AptRealtimeTypes()
{
}
// ---------------------------------------------------------------- 消息类型
public static final String CONNECTED = "CONNECTED";
public static final String PONG = "PONG";
public static final String CALL_STATE_CHANGED = "CALL_STATE_CHANGED";
public static final String CALL_ALARM = "CALL_ALARM";
public static final String CHECKIN_APPLIED = "CHECKIN_APPLIED";
public static final String CHECKIN_REVIEWED = "CHECKIN_REVIEWED";
public static final String FACE_REG_APPLIED = "FACE_REG_APPLIED";
public static final String JOB_CREATED = "JOB_CREATED";
public static final String JOB_STATE_CHANGED = "JOB_STATE_CHANGED";
public static final String DEVICE_OFFLINE = "DEVICE_OFFLINE";
public static final String REPORT_BACKLOG = "REPORT_BACKLOG";
public static final String SYSTEM_NOTICE = "SYSTEM_NOTICE";
// ---------------------------------------------------------------- 分类
public static final String CATEGORY_CALL = "CALL";
public static final String CATEGORY_ALARM = "ALARM";
public static final String CATEGORY_CHECKIN = "CHECKIN";
public static final String CATEGORY_FACE = "FACE";
public static final String CATEGORY_JOB = "JOB";
public static final String CATEGORY_DEVICE = "DEVICE";
public static final String CATEGORY_REPORT = "REPORT";
public static final String CATEGORY_SYSTEM = "SYSTEM";
// ---------------------------------------------------------------- 优先级
public static final String PRIORITY_P0 = "P0";
public static final String PRIORITY_P1 = "P1";
public static final String PRIORITY_P2 = "P2";
/** 分类默认投递权限;返回 null 表示按 gyid 广播 */
public static String permissionOf(String category)
{
if (category == null)
{
return null;
}
switch (category)
{
case CATEGORY_CALL:
case CATEGORY_ALARM:
return "apt:jcxx:list";
case CATEGORY_CHECKIN:
return "duty:checkin:list";
case CATEGORY_FACE:
return "duty:person:reg";
case CATEGORY_JOB:
return "apt:workorder:list";
case CATEGORY_DEVICE:
return "apt:device:list";
case CATEGORY_REPORT:
return "apt:report:list";
default:
return null;
}
}
/** 分类默认优先级 */
public static String priorityOf(String category)
{
if (CATEGORY_ALARM.equals(category))
{
return PRIORITY_P0;
}
if (CATEGORY_CALL.equals(category) || CATEGORY_CHECKIN.equals(category)
|| CATEGORY_JOB.equals(category) || CATEGORY_FACE.equals(category))
{
return PRIORITY_P1;
}
return PRIORITY_P2;
}
}
@@ -0,0 +1,29 @@
package com.ruoyi.apt.framework.realtime;
import java.util.Map;
/**
* 实时推送发布器(事务提交后投递,业务无感)。
*
* @author apt-rev01
*/
public interface IAptRealtimePublisher
{
/** 叫乘状态变更(P1,按 apt:jcxx:list 投递) */
void callStateChanged(String jcid, String command, String stateAfter, boolean duplicated);
/** 叫乘/离寓报警(P0,值班台需立即打断) */
void alarm(String jcid, String alarmType, String detail);
/** 工单创建 / 状态变更(P1) */
void jobChanged(String eventType, String bizType, String bizId, String roomNo, String state);
/** 入住申请待审(P1,按 duty:checkin:list 投递) */
void checkinApplied(String applyId, String workNo, String name);
/** 通用发布(自定义 type/category/priority/目标) */
void publish(AptRealtimeMessage message);
/** 系统通告(P2,gyid 广播) */
void systemNotice(String title, Map<String, Object> data);
}
@@ -53,6 +53,13 @@ public class AptContextInterceptor implements HandlerInterceptor
{
AptContext.initTraceId();
String uri = request.getRequestURI();
// WebSocket 握手:浏览器无法带 Authorization 头,由 AptRealtimeHandshakeInterceptor
// 用 query token 完成 JWT + 权限点校验;此处只注入 gyid,不做会话断言(IT-6)
if (uri.equals("/apt/ws") || uri.startsWith("/apt/ws/"))
{
AptContext.setGyId(properties.getGyId());
return true;
}
Prefix prefix = AptPrefixResolver.resolve(uri);
boolean anonymous = isAnonymous(handler);
@@ -150,3 +150,13 @@ apt:
# 超过该秒数无心跳视为离线(不据此伪造 ACKED)
heartbeat-timeout-seconds: ${APT_REPORT_HEARTBEAT_TIMEOUT_SEC:90}
resume-batch-limit: ${APT_REPORT_RESUME_BATCH:200}
# ---------------------------------------------------------------------------
# 值班员实时通道(WebSocket /apt/ws)
# 浏览器无法设置握手 Header,令牌通过 ?token= 传入;生产应收敛 allowed-origins。
# 未订阅者不影响既有 HTTP 接口,关闭 enabled 即完全退化到轮询。
# ---------------------------------------------------------------------------
realtime:
enabled: ${APT_REALTIME_ENABLED:true}
allowed-origins: ${APT_REALTIME_ALLOWED_ORIGINS:*}
heartbeat-interval-seconds: ${APT_REALTIME_HEARTBEAT_SEC:25}
connect-permission: ${APT_REALTIME_CONNECT_PERM:apt:realtime:connect}
@@ -41,7 +41,8 @@ class AptDutyCommandServiceTest
mapper = mock(AptJcxxMapper.class);
stateService = mock(IAptDutyStateService.class);
auditWriter = mock(AptAuditWriter.class);
service = new AptDutyCommandServiceImpl(mapper, stateService, auditWriter);
service = new AptDutyCommandServiceImpl(mapper, stateService, auditWriter,
mock(com.ruoyi.apt.framework.realtime.IAptRealtimePublisher.class));
AptContext.clear();
AptContext.setGyId("APT-A");
}
@@ -35,7 +35,8 @@ class AptDutyStateServiceTest
{
mapper = mock(AptDutyStateMapper.class);
jobService = mock(IAptJobService.class);
service = new AptDutyStateServiceImpl(mapper, jobService);
service = new AptDutyStateServiceImpl(mapper, jobService,
mock(com.ruoyi.apt.framework.realtime.IAptRealtimePublisher.class));
AptContext.clear();
AptContext.setGyId("APT-A");
}
@@ -42,7 +42,8 @@ class AptTabletResultServiceTest
{
mapper = mock(AptTabletMapper.class);
stateService = mock(IAptDutyStateService.class);
service = new AptTabletResultServiceImpl(mapper, stateService);
service = new AptTabletResultServiceImpl(mapper, stateService,
mock(com.ruoyi.apt.framework.realtime.IAptRealtimePublisher.class));
AptContext.clear();
AptContext.setGyId("APT-A");
}
@@ -0,0 +1,108 @@
package com.ruoyi.apt.framework;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.contains;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.Map;
import java.util.Set;
import com.ruoyi.apt.common.AptContext;
import com.ruoyi.apt.framework.realtime.AptRealtimePublisher;
import com.ruoyi.apt.framework.realtime.AptRealtimeSessionRegistry;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
/**
* WebSocket 路由与投递回归:权限过滤、广播、报警优先级、关闭会话摘除。
*/
class AptRealtimeTest
{
@BeforeEach
void setUp()
{
AptContext.clear();
AptContext.setGyId("APT-A");
}
@AfterEach
void tearDown()
{
AptContext.clear();
}
@Test
void alarmRoutesToCallPermissionWithP0()
{
AptRealtimeSessionRegistry registry = mock(AptRealtimeSessionRegistry.class);
AptRealtimePublisher publisher = new AptRealtimePublisher(registry);
publisher.alarm("JC-1", "CALL_FAILED", "无应答");
verify(registry).sendToPermission(eq("APT-A"), eq("apt:jcxx:list"),
contains("\"CALL_ALARM\""));
verify(registry).sendToPermission(eq("APT-A"), eq("apt:jcxx:list"), contains("\"P0\""));
}
@Test
void systemNoticeBroadcastsWholeGyid()
{
AptRealtimeSessionRegistry registry = mock(AptRealtimeSessionRegistry.class);
AptRealtimePublisher publisher = new AptRealtimePublisher(registry);
publisher.systemNotice("系统维护", Map.of("window", "02:00-03:00"));
verify(registry).broadcast(eq("APT-A"), contains("SYSTEM_NOTICE"));
}
@Test
void permissionFilterBlocksUnrelatedSubscribers() throws Exception
{
AptRealtimeSessionRegistry registry = new AptRealtimeSessionRegistry();
WebSocketSession session = mock(WebSocketSession.class);
when(session.getId()).thenReturn("S-1");
when(session.isOpen()).thenReturn(true);
registry.register(session, 7L, "duty01", "APT-A", Set.of("apt:workorder:list"));
assertEquals(0, registry.sendToPermission("APT-A", "apt:jcxx:list", "{}"));
verify(session, never()).sendMessage(any(TextMessage.class));
assertEquals(1, registry.sendToPermission("APT-A", "apt:workorder:list", "{}"));
verify(session).sendMessage(any(TextMessage.class));
}
@Test
void closedSessionIsDropped() throws Exception
{
AptRealtimeSessionRegistry registry = new AptRealtimeSessionRegistry();
WebSocketSession session = mock(WebSocketSession.class);
when(session.getId()).thenReturn("S-2");
when(session.isOpen()).thenReturn(false);
registry.register(session, 8L, "duty02", "APT-A", Set.of("*:*:*"));
assertEquals(1, registry.online("APT-A"));
assertEquals(0, registry.broadcast("APT-A", "{}"));
assertEquals(0, registry.online("APT-A"));
verify(session, never()).sendMessage(any(TextMessage.class));
}
@Test
void adminWildcardReceivesEverything() throws Exception
{
AptRealtimeSessionRegistry registry = new AptRealtimeSessionRegistry();
WebSocketSession session = mock(WebSocketSession.class);
when(session.getId()).thenReturn("S-3");
when(session.isOpen()).thenReturn(true);
registry.register(session, 1L, "admin", "APT-A", Set.of("*:*:*"));
assertEquals(1, registry.sendToPermission("APT-A", "apt:report:list", "{}"));
verify(session).sendMessage(any(TextMessage.class));
}
}
@@ -121,6 +121,9 @@ public class SecurityConfig
// 时间窗 / nonce)。此处若不放行,anyRequest().authenticated() 会提前返回 401,
// 四前缀拦截器永远无法生效(IT-1 / M0)。
.requestMatchers("/open/v1/**").permitAll()
// 值班员 WebSocket 握手:浏览器无法设置 Authorization 头,令牌走 query,
// 由 AptRealtimeHandshakeInterceptor 完成 JWT + 权限点校验(IT-6)
.requestMatchers("/apt/ws", "/apt/ws/**").permitAll()
// 静态资源,可匿名访问
.requestMatchers(HttpMethod.GET, "/", "/*.html", "/**.html", "/**.css", "/**.js", "/profile/**").permitAll()
.requestMatchers("/swagger-ui.html", "/v3/api-docs/**", "/swagger-ui/**", "/druid/**").permitAll()
+28
View File
@@ -0,0 +1,28 @@
-- =============================================================================
-- IT-6 WebSocket 实时推送:连接权限点(可重放)
-- 口径:新增 1 个 F 权限点;不新增表、不改业务列。
-- 握手时由 AptRealtimeHandshakeInterceptor 校验(超管 *:*:* 直接放行)。
-- =============================================================================
SET NAMES utf8mb4;
INSERT INTO sys_menu
(menu_id, menu_name, parent_id, order_num, path, component, query, route_name,
is_frame, is_cache, menu_type, visible, status, perms, icon, create_by, create_time, remark)
VALUES
(3259, '实时通道连接', 3000, 39, '', '', '', '', 1, 0, 'F', '0', '0',
'apt:realtime:connect', '#', 'IT-6', NOW(3),
'WebSocket /apt/ws 握手权限;未授权用户不允许建立长连接')
ON DUPLICATE KEY UPDATE menu_name = VALUES(menu_name), parent_id = VALUES(parent_id),
order_num = VALUES(order_num), menu_type = VALUES(menu_type), status = VALUES(status),
perms = VALUES(perms), remark = VALUES(remark);
INSERT INTO sys_role_menu (role_id, menu_id)
SELECT 1, 3259 FROM DUAL
WHERE NOT EXISTS (SELECT 1 FROM sys_role_menu WHERE role_id = 1 AND menu_id = 3259);
-- 校验
SELECT '=== 66. 实时通道权限点(应 1)===' AS section;
SELECT menu_id, perms, menu_name, status FROM sys_menu WHERE menu_id = 3259;
SELECT '=== 67. 超管绑定(应 1)===' AS section;
SELECT COUNT(*) AS admin_bound FROM sys_role_menu WHERE role_id = 1 AND menu_id = 3259;