✨ Introducing new features. common-websocket #3170 【重构】websocket 服务端整合

pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketHandlerConfig.java

@@ -13,6 +13,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.client.ServiceInstance;
 import org.springframework.context.annotation.Bean;
 import org.springframework.web.socket.WebSocketHandler;
 import org.springframework.web.socket.handler.TextWebSocketHandler;
@@ -30,8 +31,8 @@ public class WebSocketHandlerConfig {
 
 	@Bean
 	@ConditionalOnMissingBean(SessionKeyGenerator.class)
-	public SessionKeyGenerator sessionKeyGenerator() {
-		return new PigxSessionKeyGenerator();
+	public SessionKeyGenerator sessionKeyGenerator(ServiceInstance serviceInstance) {
+		return new PigxSessionKeyGenerator(serviceInstance);
 	}
 
 	@Bean

pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/config/WebSocketMessageSender.java

@@ -27,7 +27,7 @@ public class WebSocketMessageSender {
 	public static boolean send(Object sessionKey, String message) {
 		WebSocketSession session = WebSocketSessionHolder.getSession(sessionKey);
 		if (session == null) {
-			log.trace("[send] 当前 sessionKey：{} 对应 session 不在本服务中", sessionKey);
+			log.info("[send] 当前 sessionKey：{} 对应 session 不在本服务中", sessionKey);
 			return false;
 		}
 		else {

pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/custom/PigxSessionKeyGenerator.java

@@ -2,6 +2,8 @@ package com.pig4cloud.pigx.common.websocket.custom;
 
 import com.pig4cloud.pigx.common.security.service.PigxUser;
 import com.pig4cloud.pigx.common.websocket.holder.SessionKeyGenerator;
+import lombok.RequiredArgsConstructor;
+import org.springframework.cloud.client.ServiceInstance;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.web.socket.WebSocketSession;
 
@@ -10,8 +12,11 @@ import org.springframework.web.socket.WebSocketSession;
  * @date 2021/10/4 websocket session 标识生成规则
  */
 @Configuration
+@RequiredArgsConstructor
 public class PigxSessionKeyGenerator implements SessionKeyGenerator {
 
+	private final ServiceInstance instance;
+
 	/**
 	 * 获取当前session的唯一标识
 	 * @param webSocketSession 当前session
@@ -24,7 +29,8 @@ public class PigxSessionKeyGenerator implements SessionKeyGenerator {
 
 		if (obj instanceof PigxUser) {
 			PigxUser user = (PigxUser) obj;
-			return String.format("%s%s", user.getTenantId(), user.getUsername());
+			// IP:port:userId 作为唯一区分
+			return String.format("%s:%s:%s", instance.getHost(), instance.getPort(), user.getId());
 		}
 
 		return null;

pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/distribute/RedisMessageDistributor.java

@@ -1,9 +1,14 @@
 package com.pig4cloud.pigx.common.websocket.distribute;
 
 import cn.hutool.json.JSONUtil;
+import com.pig4cloud.pigx.common.core.util.SpringContextHolder;
 import lombok.RequiredArgsConstructor;
+import org.springframework.cloud.client.ServiceInstance;
 import org.springframework.data.redis.core.StringRedisTemplate;
 
+import java.util.List;
+import java.util.stream.Collectors;
+
 /**
  * 消息分发器
  *
@@ -21,6 +26,13 @@ public class RedisMessageDistributor implements MessageDistributor {
 	 */
 	@Override
 	public void distribute(MessageDO messageDO) {
+		// 包装 sessionKey 适配分布式多环境
+		ServiceInstance instance = SpringContextHolder.getBean(ServiceInstance.class);
+		List<Object> sessionKeyList = messageDO.getSessionKeys().stream()
+				.map(key -> String.format("%s:%s:%s", instance.getHost(), instance.getPort(), key))
+				.collect(Collectors.toList());
+		messageDO.setSessionKeys(sessionKeyList);
+
 		String str = JSONUtil.toJsonStr(messageDO);
 		stringRedisTemplate.convertAndSend(RedisWebsocketMessageListener.CHANNEL, str);
 	}

pigx-common/pigx-common-websocket/src/main/java/com/pig4cloud/pigx/common/websocket/package-info.java

@@ -1,6 +1,5 @@
 /**
- * 此包内代码 来源至 BallCat
- * 关于BallCat: 组织旨在为项目快速开发提供一系列的基础能力，方便使用者根据项目需求快速进行功能拓展。
- *  https://github.com/ballcat-projects/ballcat
+ * 此包内代码 来源至 BallCat 关于BallCat: 组织旨在为项目快速开发提供一系列的基础能力，方便使用者根据项目需求快速进行功能拓展。
+ * https://github.com/ballcat-projects/ballcat
  */
 package com.pig4cloud.pigx.common.websocket;

pigx-visual/pigx-oa-platform/src/main/java/com/pig4cloud/pigx/act/listener/LeaveProcessTaskListener.java

@@ -22,7 +22,6 @@ import com.pig4cloud.pigx.admin.api.entity.SysUser;
 import com.pig4cloud.pigx.admin.api.feign.RemoteUserService;
 import com.pig4cloud.pigx.common.core.util.R;
 import com.pig4cloud.pigx.common.core.util.SpringContextHolder;
-import com.pig4cloud.pigx.common.data.tenant.TenantContextHolder;
 import com.pig4cloud.pigx.common.security.util.SecurityUtils;
 import com.pig4cloud.pigx.common.websocket.distribute.MessageDO;
 import com.pig4cloud.pigx.common.websocket.distribute.RedisMessageDistributor;
@@ -51,25 +50,24 @@ public class LeaveProcessTaskListener implements TaskListener {
 		RemoteUserService userService = SpringContextHolder.getBean(RemoteUserService.class);
 
 		R<List<SysUser>> result = userService.ancestorUsers(SecurityUtils.getUser().getUsername());
-		List<String> remindUserList = new ArrayList<>();
+		List<Object> remindUserList = new ArrayList<>();
 
 		if (CollUtil.isEmpty(result.getData())) {
 			log.info("用户 {} 不存在上级,任务单由当前用户审批", SecurityUtils.getUser().getUsername());
 			delegateTask.addCandidateUser(SecurityUtils.getUser().getUsername());
-			remindUserList.add(SecurityUtils.getUser().getUsername());
+			remindUserList.add(SecurityUtils.getUser().getId());
 		}
 		else {
 			List<String> userList = result.getData().stream().map(SysUser::getUsername).collect(Collectors.toList());
+			List<Integer> userIdList = result.getData().stream().map(SysUser::getUserId).collect(Collectors.toList());
 			log.info("当前任务 {}，由 {}处理", delegateTask.getId(), userList);
 			delegateTask.addCandidateUsers(userList);
-			remindUserList.addAll(userList);
+			remindUserList.addAll(userIdList);
 		}
 
 		// websocket 发送消息
-		List<Object> sessionKey = remindUserList.stream().map(s -> TenantContextHolder.getTenantId() + s)
-				.collect(Collectors.toList());
 		MessageDO messageDO = new MessageDO();
-		messageDO.setSessionKeys(sessionKey);
+		messageDO.setSessionKeys(remindUserList);
 		messageDO.setMessageText(String.format("协同办公 %s 的任务需要您处理", delegateTask.getName()));
 		messageDistributor.distribute(messageDO);
 	}