直播

fs-live-app/src/main/java/com/fs/live/websocket/auth/TenantChannelContext.java

@@ -110,4 +110,101 @@ public class TenantChannelContext {
         DynamicDataSourceContextHolder.clearDataSourceType();
         SecurityContextHolder.clearContext();
     }
+
+    /**
+     * javax.websocket 握手阶段专用：根据 tenantCode 查库，将 tenantId/dsKey 存入 userProperties。
+     * 在 WebSocketConfigurator.modifyHandshake() 中调用一次，后续 onOpen/onMessage/onClose 无需再查库。
+     */
+    public void resolveAndStore(String tenantCode, java.util.Map<String, Object> userProperties) {
+        if (!saasEnabled || StringUtils.isBlank(tenantCode)) return;
+        try {
+            DynamicDataSourceContextHolder.setDataSourceType(DataSourceType.MASTER.name());
+            TenantInfo query = new TenantInfo();
+            query.setTenantCode(tenantCode);
+            List<TenantInfo> list = tenantInfoService.selectTenantInfoList(query);
+            if (list == null || list.isEmpty()) {
+                log.warn("[SaaS WS] resolveAndStore: tenantCode={} 未找到租户", tenantCode);
+                return;
+            }
+            TenantInfo tenant = list.get(0);
+            if (!Integer.valueOf(1).equals(tenant.getStatus())) return;
+            if (tenant.getExpireTime() != null && tenant.getExpireTime().before(new Date())) return;
+            // 确保数据源已注册
+            tenantDataSourceManager.switchTenant(tenant);
+            userProperties.put("_tenantId", tenant.getId());
+            userProperties.put("_dsKey", "tenant:" + tenant.getId());
+            log.info("[SaaS WS] 握手阶段绑定租户 tenantId={}, tenantCode={}", tenant.getId(), tenantCode);
+        } finally {
+            DynamicDataSourceContextHolder.clearDataSourceType();
+        }
+    }
+
+    /**
+     * javax.websocket 业务处理前专用：从 userProperties 取 tenantId/dsKey 直接激活，无需查库。
+     * 在 onOpen/onMessage/onClose 开头调用。
+     */
+    public void activateBySession(java.util.Map<String, Object> userProperties) {
+        if (!saasEnabled) return;
+        Long tenantId = (Long) userProperties.get("_tenantId");
+        String dsKey = (String) userProperties.get("_dsKey");
+        if (tenantId == null || StringUtils.isBlank(dsKey)) return;
+        DynamicDataSourceContextHolder.setDataSourceType(dsKey);
+        SecurityContextHolder.getContext().setAuthentication(
+                new UsernamePasswordAuthenticationToken(
+                        new TenantPrincipal(tenantId), null, Collections.emptyList()));
+        try {
+            SysConfig cfg = sysConfigMapper.selectConfigByConfigKey("projectConfig");
+            if (cfg != null && StringUtils.isNotBlank(cfg.getConfigValue())) {
+                TenantConfigContext.set(JSONObject.parseObject(cfg.getConfigValue()));
+            } else {
+                TenantConfigContext.set(null);
+            }
+            ProjectConfig.loadTenantConfigsFromContext();
+        } catch (Exception e) {
+            log.warn("[SaaS WS] 加载租户项目配置失败 tenantId={}", tenantId, e);
+        }
+    }
+
+    /**
+     * javax.websocket 专用：根据 tenantCode 直接切换数据源 + SecurityContext。
+     * 与 bindTenant(Channel) 不同，此方法每次调用都会查库并激活，无需提前绑定到 Channel。
+     * 若未启用 SaaS 或 tenantCode 为空，不做任何处理。
+     * @deprecated 改用 resolveAndStore + activateBySession 避免每次查库
+     */
+    public void bindTenantByCode(String tenantCode) {
+        if (!saasEnabled || StringUtils.isBlank(tenantCode)) return;
+        try {
+            DynamicDataSourceContextHolder.setDataSourceType(DataSourceType.MASTER.name());
+            TenantInfo query = new TenantInfo();
+            query.setTenantCode(tenantCode);
+            List<TenantInfo> list = tenantInfoService.selectTenantInfoList(query);
+            if (list == null || list.isEmpty()) {
+                log.warn("[SaaS WS] bindTenantByCode: tenantCode={} 未找到租户", tenantCode);
+                return;
+            }
+            TenantInfo tenant = list.get(0);
+            if (!Integer.valueOf(1).equals(tenant.getStatus())) return;
+            if (tenant.getExpireTime() != null && tenant.getExpireTime().before(new Date())) return;
+            tenantDataSourceManager.switchTenant(tenant);
+            String dsKey = "tenant:" + tenant.getId();
+            DynamicDataSourceContextHolder.setDataSourceType(dsKey);
+            SecurityContextHolder.getContext().setAuthentication(
+                    new UsernamePasswordAuthenticationToken(
+                            new TenantPrincipal(tenant.getId()), null, Collections.emptyList()));
+            try {
+                SysConfig cfg = sysConfigMapper.selectConfigByConfigKey("projectConfig");
+                if (cfg != null && StringUtils.isNotBlank(cfg.getConfigValue())) {
+                    TenantConfigContext.set(JSONObject.parseObject(cfg.getConfigValue()));
+                } else {
+                    TenantConfigContext.set(null);
+                }
+                ProjectConfig.loadTenantConfigsFromContext();
+            } catch (Exception e) {
+                log.warn("[SaaS WS] 加载租户项目配置失败 tenantId={}", tenant.getId(), e);
+            }
+        } catch (Exception e) {
+            log.error("[SaaS WS] bindTenantByCode 失败 tenantCode={}", tenantCode, e);
+            DynamicDataSourceContextHolder.clearDataSourceType();
+        }
+    }
 }

fs-live-app/src/main/java/com/fs/live/websocket/auth/WebSocketConfigurator.java

@@ -62,6 +62,26 @@ public class WebSocketConfigurator extends ServerEndpointConfig.Configurator {
             userProperties.put(AttrConstant.EXTERNAL_CONTACT_ID, Long.valueOf(parameterMap.get(AttrConstant.EXTERNAL_CONTACT_ID).get(0)));
         }
 
+        // SaaS：从请求头获取 tenantCode（优先），兜底从 URL 参数取
+        String tenantCode = null;
+        Map<String, List<String>> headers = request.getHeaders();
+        if (headers.containsKey(AttrConstant.TENANT_CODE)) {
+            List<String> values = headers.get(AttrConstant.TENANT_CODE);
+            if (values != null && !values.isEmpty()) {
+                tenantCode = values.get(0);
+            }
+        }
+        if (tenantCode == null && parameterMap.containsKey(AttrConstant.TENANT_CODE)) {
+            tenantCode = parameterMap.get(AttrConstant.TENANT_CODE).get(0);
+        }
+        if (tenantCode != null) {
+            userProperties.put(AttrConstant.TENANT_CODE, tenantCode);
+        }
+
+        // SaaS：握手阶段查库，将 tenantId/dsKey 存入 userProperties，后续 onOpen/onMessage/onClose 无需再查库
+        TenantChannelContext tenantChannelContext = SpringUtils.getBean(TenantChannelContext.class);
+        tenantChannelContext.resolveAndStore((String) userProperties.get(AttrConstant.TENANT_CODE), userProperties);
+
         // 验证token
         if (parameterMap.containsKey(tokenKey)) {
             String token = parameterMap.get(tokenKey).get(0);

fs-live-app/src/main/java/com/fs/live/websocket/constant/AttrConstant.java

@@ -16,7 +16,7 @@ public class AttrConstant {
     public static final String QW_USER_ID = "qwUserId";
     public static final String EXTERNAL_CONTACT_ID = "externalContactId";
     /** SaaS: URL 参数中的租户编码 */
-    public static final String TENANT_CODE = "tenantCode";
+    public static final String TENANT_CODE = "x-tenant-code";
 
     // 定义 AttributeKey 保存必要参数
     public static final AttributeKey<Long> ATTR_LIVE_ID = AttributeKey.valueOf(LIVE_ID);

fs-live-app/src/main/java/com/fs/live/websocket/service/WebSocketServer.java

@@ -25,6 +25,7 @@ import com.fs.common.utils.spring.SpringUtils;
 import com.fs.live.domain.*;
 import com.fs.live.service.*;
 import com.fs.live.vo.LiveGoodsVo;
+import com.fs.live.websocket.constant.AttrConstant;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.time.DateUtils;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -85,6 +86,7 @@ public class WebSocketServer {
     private final ILiveWatchLogService liveWatchLogService = SpringUtils.getBean(ILiveWatchLogService.class);
     private final ILiveVideoService liveVideoService = SpringUtils.getBean(ILiveVideoService.class);
     private final ILiveCompletionPointsRecordService completionPointsRecordService = SpringUtils.getBean(ILiveCompletionPointsRecordService.class);
+    private final com.fs.live.websocket.auth.TenantChannelContext tenantChannelContext = SpringUtils.getBean(com.fs.live.websocket.auth.TenantChannelContext.class);
     private static Random random = new Random();
 
     // Redis key 前缀：用户进入直播间时间
@@ -97,8 +99,9 @@ public class WebSocketServer {
     //建立连接成功调用
     @OnOpen
     public void onOpen(Session session) {
-
         Map<String, Object> userProperties = session.getUserProperties();
+        tenantChannelContext.activateBySession(userProperties);
+        try {
         long liveId = (long) userProperties.get("liveId");
         long userId = (long) userProperties.get("userId");
         long userType = (long) userProperties.get("userType");
@@ -108,7 +111,9 @@ public class WebSocketServer {
 
         Live live = liveService.selectLiveByLiveId(liveId);
         if (live == null) {
-            throw new BaseException("未找到直播间");
+            log.warn("[WS onOpen] live not found, liveId={}", liveId);
+            try { session.close(); } catch (Exception ignore) {}
+            return;
         }
         long companyId = -1L;
         long companyUserId = -1L;
@@ -141,10 +146,11 @@ public class WebSocketServer {
                 }
             }
             if (Objects.isNull(fsUser)) {
-                throw new BaseException("用户信息错误");
+                log.warn("[WS onOpen] user not found, userId={}", userId);
+                try { session.close(); } catch (Exception ignore) {}
+                return;
             }
-
-            LiveWatchUser liveWatchUserVO = liveWatchUserService.join(fsUser,liveId, userId, location);
+            LiveWatchUser liveWatchUserVO = liveWatchUserService.join(fsUser, liveId, userId, location);
             room.put(userId, session);
 
             // 存储用户进入直播间的时间到 Redis（用于计算在线时长）
@@ -289,12 +295,17 @@ public class WebSocketServer {
         sessionLocks.putIfAbsent(session.getId(), new ReentrantLock());
         // 初始化心跳时间
         heartbeatCache.put(session.getId(), System.currentTimeMillis());
-
+        } finally {
+            tenantChannelContext.clear();
+        }
     }
 
     //关闭连接时调用
     @OnClose
     public void onClose(Session session) {
+        Map<String, Object> _closeProps = session.getUserProperties();
+        tenantChannelContext.bindTenantByCode((String) _closeProps.get(AttrConstant.TENANT_CODE));
+        try {
         Map<String, Object> userProperties = session.getUserProperties();
         // 获取公司ID和销售ID
         long companyId = -1L;
@@ -370,11 +381,15 @@ public class WebSocketServer {
         // 清理Session相关资源
         heartbeatCache.remove(session.getId());
         sessionLocks.remove(session.getId());
+        } finally {
+            tenantChannelContext.clear();
+        }
     }
 
     //收到客户端信息
     @OnMessage
     public void onMessage(Session session,String message) throws IOException {
+        tenantChannelContext.activateBySession(session.getUserProperties());
         Map<String, Object> userProperties = session.getUserProperties();
 
         long liveId = (long) userProperties.get("liveId");
@@ -586,6 +601,8 @@ public class WebSocketServer {
             }
         } catch (Exception e) {
             log.error("webSocket 消息处理失败 msg: {}", e.getMessage(), e);
+        } finally {
+            tenantChannelContext.clear();
         }
     }