easyfk-websocket WebSocket

dependencies {
    api project(':component-cache:cache-caffeine')
    api project(':component-websocket:websocket-api')
    api('com.mcst:queue-disruptor')
    api('com.mcst:easyfk-authority')
    api('com.mcst:web-base')
    api('io.netty:netty-all')
}

websocket-server/
├── build.gradle
└── src/main/
    ├── java/com/mcst/easyfk/websocket/
    │   ├── config/
    │   │   └── ServerConfig.java              # Spring Boot 自动配置类
    │   ├── enums/
    │   │   └── EventType.java                 # 事件类型枚举
    │   ├── handler/
    │   │   ├── UserEventHandler.java          # 用户事件处理器接口（业务方实现）
    │   │   └── WebSocketHandler.java          # WebSocket 核心处理器
    │   ├── interceptor/
    │   │   ├── DefaultAuthInterceptor.java    # 默认认证拦截器
    │   │   ├── HandshakeResult.java           # 握手结果对象
    │   │   └── WebSocketInterceptor.java      # 拦截器接口
    │   ├── manager/
    │   │   ├── impl/
    │   │   │   └── SocketManagerImpl.java     # Socket 管理器实现
    │   │   ├── ISocketManager.java            # Socket 管理器接口
    │   │   ├── LocalSubManager.java           # 本地订阅管理器
    │   │   └── WsUserDataManager.java         # 用户会话数据管理器
    │   ├── properties/
    │   │   └── WebSocketProperties.java       # 配置属性类
    │   ├── sender/
    │   │   ├── SessionSendDispatcher.java     # 消息发送分发器（Disruptor 分片）
    │   │   ├── SessionSendEvent.java          # 发送事件对象
    │   │   ├── SessionSendWorker.java         # 实际发送工作器
    │   │   ├── WebSocketMessageSender.java    # 消息发送器（核心）
    │   │   └── WsSendProcessor.java           # Disruptor 消费处理器
    │   ├── server/
    │   │   ├── PushMessageServer.java         # 消息推送服务入口
    │   │   └── WebSocketServer.java           # Netty WebSocket 服务器
    │   ├── strategy/
    │   │   ├── impl/
    │   │   │   ├── ChannelCategoryPushStrategy.java  # 频道+类别推送策略
    │   │   │   ├── ChannelPushStrategy.java           # 频道推送策略
    │   │   │   ├── NotificationPushStrategy.java      # 通知推送策略
    │   │   │   └── UserPushStrategy.java              # 用户推送策略
    │   │   ├── MessagePushStrategy.java       # 推送策略接口
    │   │   └── MessagePushStrategySelector.java  # 策略选择器
    │   ├── util/
    │   │   └── SecuritySignUtil.java          # 安全签名工具类
    │   └── vo/
    │       ├── PushMessage.java               # 推送消息对象
    │       └── ResponseMessage.java           # 响应消息对象
    └── resources/META-INF/spring/
        └── org.springframework.boot.autoconfigure.AutoConfiguration.imports

客户端发起 HTTP 升级请求
    ↓
handleHttpRequest()
    ↓
执行拦截器链 beforeHandshake()（按 order 排序，正序执行）
    ↓ 全部通过
从拦截器属性获取 userSessionId
    ↓
socketManager.connectUser() 建立连接
    ↓
更新会话活跃时间
    ↓
WebSocket 协议握手
    ↓
发送欢迎消息
    ↓
执行拦截器链 afterHandshake()（逆序执行）

public interface WebSocketInterceptor {
    HandshakeResult beforeHandshake(ChannelHandlerContext ctx, FullHttpRequest request, 
                                     Map<String, Object> attributes);
    default void afterHandshake(...) {}
    default int getOrder() { return 0; }
}

userSessionId  ←→  Channel（双向映射）

PushMessageServer
    ↓ pushMessage(PushWsMessageParam)
MessagePushStrategySelector
    ↓ select() 选择策略
MessagePushStrategy（4种实现）
    ↓ push()
WebSocketMessageSender
    ├─ sendMessageToUser()       → SessionSendDispatcher → Disruptor → SessionSendWorker
    ├─ sendMessageToUserByUserId() → SessionSendDispatcher → Disruptor → SessionSendWorker
    ├─ sendToSubscribers()       → batchBroadcast()（直接批量写入，跳过 Disruptor）
    ├─ sendToChannelSubscribers() → batchBroadcast()
    └─ broadcastToAll()          → batchBroadcast()

签名数据 = ClientType=xxx&ClientId=xxx&Nonce=xxx&Param=xxx&Secret=xxx
签名结果 = MD5(签名数据)

签名数据 = ClientType=xxx&ClientId=xxx&Nonce=xxx&Param=xxx
签名结果 = HMAC-SHA256(签名数据, secret)

easyfk:
  config:
    websocket:
      path: /ws
      port: 8081
      mustLogin: false
      enableSecurity: true
      securitySecret: "your-production-secret-key"
      sessionTimeoutSeconds: 1800
      heartbeatIntervalSeconds: 30
      performance:
        sendShardCount: 64
        sendQueueWorkersPerShard: 4
        sendQueueBufferSize: 262144
        sendQueueWaitStrategy: YIELDING

@Component
public class MyUserEventHandler implements UserEventHandler {

    @Override
    public BaseResult<?> handleUserEvent(WsUserOptParam param) {
        String eventType = param.getEventType();
        Map<String, Object> eventData = param.getEventData();
        
        switch (eventType) {
            case "subscribe":
                // 处理订阅逻辑
                return BRBuilder.successResult();
            case "unsubscribe":
                // 处理取消订阅逻辑
                return BRBuilder.successResult();
            default:
                // 处理其他自定义事件
                return BRBuilder.successResult();
        }
    }

    @Override
    public void clearUserSubscriptions(String clientId) {
        // 用户断开连接时清理业务层订阅数据
    }
}

@Component
public class IpWhitelistInterceptor implements WebSocketInterceptor {

    @Override
    public HandshakeResult beforeHandshake(ChannelHandlerContext ctx, 
                                           FullHttpRequest request, 
                                           Map<String, Object> attributes) {
        String ip = getClientIp(ctx);
        if (!isAllowed(ip)) {
            return HandshakeResult.forbidden("IP_BLOCKED", "IP not in whitelist");
        }
        return HandshakeResult.success();
    }

    @Override
    public int getOrder() {
        return -200; // 在默认认证拦截器之前执行
    }
}

@Resource
private PushMessageServer pushMessageServer;

// 推送给指定用户
PushWsMessageParam param = new PushWsMessageParam()
    .setUserSessionId("user-session-id")
    .setChannel("order")
    .setCategory("fill")
    .setData(Map.of("orderId", "12345", "status", "filled"));
pushMessageServer.pushMessage(param);

// 推送给频道+类别订阅者
PushWsMessageParam param2 = new PushWsMessageParam()
    .setChannel("quote")
    .setCategory("BTCUSDT")
    .setData(Map.of("price", "50000.00", "volume", "123.45"));
pushMessageServer.pushMessage(param2);

// 发送系统通知
PushWsMessageParam param3 = new PushWsMessageParam()
    .setChannel("notification")
    .setData(Map.of("title", "系统维护通知", "content", "系统将于凌晨2点维护", "level", "IMPORTANT"));
pushMessageServer.pushMessage(param3);

public interface IPushWsMsgApi {
    BaseResult<?> pushMessage(PushWsMessageParam param);
}

┌─────────────────────────────────────────────────────────────────┐
│                    Netty Thread Model                           │
│  BossGroup (1 thread)     →  接受连接                           │
│  WorkerGroup (N threads)  →  I/O 读写、编解码                    │
├─────────────────────────────────────────────────────────────────┤
│                    业务线程池                                     │
│  messageHandleExecutor    →  消息业务处理（WebSocketHandler）     │
│  heartbeatSendExecutor    →  心跳 Ping 发送                     │
│  heartbeatExecutor        →  心跳定时检测调度（单线程）            │
├─────────────────────────────────────────────────────────────────┤
│                    Disruptor 发送队列                             │
│  shardQueues[0..N-1]      →  单用户消息异步发送                   │
│  每个分片有独立的 Worker 线程                                     │
├─────────────────────────────────────────────────────────────────┤
│                    直接广播                                       │
│  WebSocketMessageSender   →  订阅推送/全员广播（跳过 Disruptor）  │
│  批量 write + 统一 flush                                        │
├─────────────────────────────────────────────────────────────────┤
│                    监控线程                                       │
│  ws-send-monitor          →  每10秒输出发送统计                   │
└─────────────────────────────────────────────────────────────────┘