# 消息路由流程

本文档详细说明 JRSoft Subway 系统中的消息路由机制和新架构设计。

## 系统架构概览

### 核心组件

```
┌─────────────┐     ┌─────────────┐     ┌──────────┐     ┌──────────┐
│   Backend   │────▶│   Gateway   │────▶│   Edge   │────▶│  Device  │
│  (FastAPI)  │     │(API + WS)   │     │  (Proxy) │     │(Client)  │
│             │◀────│             │◀────│          │◀────│          │
└─────────────┘     └─────────────┘     └──────────┘     └──────────┘
     18082              18081           Dynamic          Dynamic
      ↑                   ↑
      │                   │
   Business         Protocol API
   Management       & WebSocket
```

### 职责分离

#### Backend (业务管理层)
- **端口**: 18082
- **职责**: 
  - 节目内容管理和编排
  - 任务调度和时间管理
  - 程序发布和分发控制
  - 设备状态监控和统计
  - 提供管理 API 给前端应用
- **技术栈**: Python FastAPI, PostgreSQL, Redis


#### Gateway (API服务 + 协议路由层)
- **端口**: 18081
- **职责**:
  - 提供 HTTP API 接收命令请求
  - WebSocket 服务器，管理连接
  - 消息路由和转发
  - 连接状态管理
  - 协议版本控制
  - 命令执行状态追踪
  - 异步回调处理
- **技术栈**: Node.js, Express, WebSocket Server

#### Edge (设备代理层)
- **端口**: 动态分配
- **职责**:
  - 连接 Gateway 和本地设备
  - 协议转换和适配
  - 本地设备管理
  - 离线缓存和恢复

#### Device (设备执行层)
- **端口**: 动态分配
- **职责**:
  - 执行具体的硬件操作
  - 状态上报
  - 程序存储和播放

## 业务流程示例

### Command 流程（直接命令执行）

适用于实时控制命令，如开关灯、调整亮度等即时操作。

```
EUDI              Gateway              Edge              Device
  |                   |                   |                  |
  | 1. HTTP POST      |                   |                  |
  | /api/command      |                   |                  |
  |------------------>|                   |                  |
  |                   |                   |                  |
  |                   | 2. 转换为WS消息   |                  |
  |                   | 记录callback      |                  |
  |                   |                   |                  |
  |                   | 3. 路由到Edge     |                  |
  |                   |------------------>|                  |
  |                   |                   |                  |
  |                   |                   | 4. 转发到设备    |
  |                   |                   |----------------->|
  |                   |                   |                  |
  |                   |                   |                  | 5. 执行命令
  |                   |                   |                  |
  |                   |                   | 6. 返回结果      |
  |                   |                   |<-----------------|
  |                   |                   |                  |
  |                   | 7. 接收响应       |                  |
  |                   |<------------------|                  |
  |                   |                   |                  |
  | 8. Callback通知   |                   |                  |
  |<------------------|                   |                  |
  |                   |                   |                  |
```

### Program 流程（节目发布）

适用于节目发布、批量更新等需要调度的任务。

```
EUDI           Backend            Gateway         Edge          Device
  |               |                   |              |              |
  | 1. 创建任务   |                   |              |              |
  |-------------->|                   |              |              |
  |               |                   |              |              |
  |               | 2. 持久化任务     |              |              |
  |               | 保存callback      |              |              |
  |               |                   |              |              |
  |               | 3. 任务调度       |              |              |
  |               | (满足条件后)      |              |              |
  |               |                   |              |              |
  |               | 4. WebSocket发送  |              |              |
  |               | program命令       |              |              |
  |               |------------------>|              |              |
  |               |        ↓          |              |              |
  |               |   记录callback    |              |              |
  |               |                   |              |              |
  |               |                   | 5. 路由转发  |              |
  |               |                   |------------->|              |
  |               |                   |              |              |
  |               |                   |              | 6. 转发命令  |
  |               |                   |              |------------->|
  |               |                   |              |              |
  |               |                   |              |              | 7. 执行
  |               |                   |              |              | 下载节目
  |               |                   |              |              | 安装部署
  |               |                   |              |              |
  |               |                   |              | 8. 进度更新  |
  |               |                   |              |<-------------|
  |               |                   |              |              |
  |               |                   | 9. 转发进度  |              |
  |               |                   |<-------------|              |
  |               |                   |              |              |
  |               | 10. WS通知        | 11. Callback |              |
  |               |<------------------|───────────┐  |              |
  |               |                   |           ↓  |              |
  |               | 12. 更新任务状态  |      EUDI回调|              |
  |               |                   |              |              |
```

## 消息路由详解

### 客户端标识符格式

#### 1. Backend Client ID
- 格式：`backend-{instance}`
- 示例：`backend-001`, `backend-primary`
- 用途：标识 Backend 实例

#### 2. Edge ID
- 格式：`edge-{location}`
- 示例：`edge-001`, `edge-tunnel-exit-1`
- 用途：标识 Edge 节点位置

#### 4. Device ID
- 格式：`{type}-{number}`
- 示例：`td-01`, `screen-05`, `pillar-100`
- 用途：标识具体设备

#### 5. Target Client ID
- 格式：直接使用设备ID
- 示例：`td-01`
- 用途：Gateway根据路由表自动找到对应Edge

### 命令路由流程

#### Simple 命令路由

```
Backend            Gateway              Edge              Device
   |                  |                   |                  |
   | 1. HTTP POST     |                   |                  |
   | /api/command     |                   |                  |
   | targetClientId:  |                   |                  |
   | "td-01"          |                   |                  |
   |----------------->|                   |                  |
   |                  |                   |                  |
   |                  | 2. 查找路由表     |                  |
   |                  | td-01 -> edge-001 |                  |
   |                  |                   |                  |
   |                  | 3. 路由到edge     |                  |
   |                  |------------------>|                  |
   |                  |                   |                  |
   |                  |                   | 4. 转发到device  |
   |                  |                   |----------------->|
   |                  |                   |                  |
   |                  |                   |                  | 5. 执行命令
   |                  |                   |                  |
   |                  |                   | 6. RESPONSE      |
   |                  |                   |<-----------------|
   |                  |                   |                  |
   |                  | 7. 路由响应       |                  |
   |                  |<------------------|                  |
   |                  |                   |                  |
   | 8. RESPONSE      |                   |                  |
   |<-----------------|                   |                  |
```

### Gateway 路由逻辑

```typescript
class GatewayRouter {
  private connections = new Map<string, WebSocket>();
  private routeMap = new Map<string, string>(); // deviceId -> edgeId
  
  route(message: CommandMessage): void {
    const { targetClientId } = message;
    
    // 1. targetClientId 直接就是设备ID
    const deviceId = targetClientId;
    
    // 2. 从路由表查找对应的 Edge
    const edgeId = this.routeMap.get(deviceId);
    if (!edgeId) {
      throw new Error(`No route found for device ${deviceId}`);
    }
    
    // 3. 查找 Edge 连接
    const edgeConnection = this.connections.get(edgeId);
    if (!edgeConnection) {
      throw new Error(`Edge ${edgeId} not connected`);
    }
    
    // 4. 转发到 Edge
    edgeConnection.send(JSON.stringify(message));
    
    // 5. 记录路由日志
    this.logRoute(message.requestRef, 'gateway', edgeId, 'command');
  }
  
  // 设备注册时更新路由表
  registerDevice(deviceId: string, edgeId: string): void {
    this.routeMap.set(deviceId, edgeId);
    console.log(`Route registered: ${deviceId} -> ${edgeId}`);
  }
  
  routeResponse(response: CommandResponseMessage): void {
    const { requestRef } = response;
    
    // 查找原始请求的来源
    const sourceConnection = this.findSourceConnection(requestRef);
    if (sourceConnection) {
      sourceConnection.send(JSON.stringify(response));
      this.logRoute(requestRef, 'gateway', 'source', 'response');
    }
  }
}
```

### Edge 转发逻辑

```typescript
class EdgeProxy {
  private gatewayConnection: WebSocket;
  private deviceConnections = new Map<string, WebSocket>();
  
  handleCommand(message: CommandMessage): void {
    const { targetClientId } = message;
    // targetClientId 已经是设备ID
    const deviceId = targetClientId;
    
    // 1. 查找设备连接
    const deviceConnection = this.deviceConnections.get(deviceId);
    if (!deviceConnection) {
      this.sendError('DEVICE_NOT_FOUND', `Device ${deviceId} not connected`);
      return;
    }
    
    // 2. 直接转发到设备
    deviceConnection.send(JSON.stringify(message));
    
    // 3. 记录转发日志
    this.logForward(message.requestRef, deviceId);
  }
  
  handleDeviceResponse(response: CommandResponseMessage): void {
    // 转发设备响应到 Gateway
    this.gatewayConnection.send(JSON.stringify(response));
  }
  
  // 设备连接时注册到 Gateway
  onDeviceConnect(deviceId: string): void {
    const registerMessage = {
      type: 'REGISTER',
      clientId: deviceId,
      clientType: 'DEVICE',
      edgeInfo: {
        edgeId: this.edgeId,
        edgeVersion: this.version
      },
      timestamp: new Date().toISOString(),
      version: '1.0'
    };
    
    this.gatewayConnection.send(JSON.stringify(registerMessage));
  }
}
```

## Gateway API 到 WebSocket 的转换

### API 请求处理流程

```typescript
class GatewayAPIHandler {
  // 处理来自 Backend 的 API 请求
  async handleCommandRequest(req: Request, res: Response) {
    const { target_client_id, command, callback_url, priority, timeout } = req.body;
    
    // 1. 生成唯一的 requestRef
    const requestRef = `cmd-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;
    
    // 2. 构建 WebSocket 命令消息
    const wsMessage = {
      type: 'COMMAND',
      requestRef,
      targetClientId: target_client_id,
      command,
      priority: priority || 'NORMAL',
      timeout: timeout || 30000,
      callback: callback_url,
      timestamp: new Date().toISOString(),
      version: '1.0'
    };
    
    // 3. 记录请求信息，用于回调
    this.pendingRequests.set(requestRef, {
      callback: callback_url,
      startTime: Date.now(),
      timeout
    });
    
    // 4. 通过 WebSocket 路由系统发送命令
    try {
      await this.router.route(wsMessage);
      
      // 5. 返回请求受理响应
      res.json({
        success: true,
        requestRef,
        message: 'Command accepted for processing'
      });
    } catch (error) {
      res.status(500).json({
        success: false,
        error: error.message
      });
    }
  }
  
  // 统一处理所有响应的回调（Command 和 Program）
  async handleResponse(response: CommandResponseMessage | ProgressUpdateMessage) {
    // 1. 查找请求信息
    const requestInfo = this.pendingRequests.get(response.requestRef);
    
    // 2. 如果是通过 WebSocket 发送的 Program 命令
    if (!requestInfo && response.callback) {
      // Program 命令携带 callback 在消息中
      await this.executeCallback(response.callback, response);
      
      // 同时通过 WebSocket 通知 Backend
      if (response.sourceClientId && response.sourceClientId.startsWith('backend')) {
        this.notifyBackend(response.sourceClientId, response);
      }
      return;
    }
    
    // 3. 处理通过 API 发送的 Command
    if (requestInfo && requestInfo.callback) {
      await this.executeCallback(requestInfo.callback, response);
      
      // 如果是最终响应，清理记录
      if (response.type === 'COMMAND_RESPONSE') {
        this.pendingRequests.delete(response.requestRef);
      }
    }
  }
  
  // 通知 Backend（用于 Program 命令）
  private notifyBackend(backendClientId: string, response: any) {
    const backendConnection = this.connections.get(backendClientId);
    if (backendConnection) {
      backendConnection.send(JSON.stringify(response));
    }
  }
}
```

## Callback 机制

Gateway 统一负责所有类型命令的 callback 处理：

### Command 类型（通过 API 发送）
1. EUDI 调用 Gateway API 时提供 callback URL
2. Gateway 保存 callback 信息在内存中
3. 收到响应后，Gateway 直接调用 callback URL
4. 适用于实时性要求高的命令

### Program 类型（通过 WebSocket 发送）
1. Backend 通过 WebSocket 发送命令时，在消息中包含 callback URL
2. Gateway 从消息中提取 callback 信息
3. 收到响应后，Gateway：
   - 调用 callback URL 通知 EUDI
   - 通过 WebSocket 将响应发回 Backend
4. Backend 可以更新任务状态，但不负责 callback

### Callback 执行策略

```typescript
class CallbackExecutor {
  async executeCallback(callback: string, response: any) {
    const maxRetries = 3;
    let lastError;
    
    for (let i = 0; i < maxRetries; i++) {
      try {
        const result = await fetch(callback, {
          method: 'POST',
          headers: {
            'Content-Type': 'application/json',
            'X-Callback-Attempt': `${i + 1}/${maxRetries}`
          },
          body: JSON.stringify({
            requestRef: response.requestRef,
            status: response.status,
            result: response.result,
            timestamp: response.timestamp,
            executionTime: response.executionTime
          })
        });
        
        if (result.ok) {
          console.log(`Callback success: ${callback}`);
          return;
        }
        
        lastError = new Error(`HTTP ${result.status}`);
      } catch (error) {
        lastError = error;
      }
      
      // 指数退避
      if (i < maxRetries - 1) {
        await sleep(Math.pow(2, i) * 1000);
      }
    }
    
    console.error(`Callback failed after ${maxRetries} attempts:`, lastError);
  }
}
```

## Backend 与 Gateway 交互

### 1. HTTP API 接口

```python
# Backend 调用 Gateway API
class GatewayClient:
    def __init__(self, base_url: str = "http://gateway:18081"):
        self.base_url = base_url
        self.client = httpx.AsyncClient()
    
    async def send_command(
        self,
        target_client_id: str,
        command: dict,
        callback_url: str,
        priority: str = "NORMAL",
        timeout: int = 30000
    ) -> dict:
        """发送命令到设备"""
        payload = {
            "target_client_id": target_client_id,
            "command": command,
            "callback_url": callback_url,
            "priority": priority,
            "timeout": timeout
        }
        
        response = await self.client.post(
            f"{self.base_url}/api/commands",
            json=payload
        )
        
        return response.json()
    
    async def get_command_status(self, request_ref: str) -> dict:
        """查询命令执行状态"""
        response = await self.client.get(
            f"{self.base_url}/api/commands/{request_ref}"
        )
        
        return response.json()
    
    async def cancel_command(self, request_ref: str) -> dict:
        """取消命令执行"""
        response = await self.client.delete(
            f"{self.base_url}/api/commands/{request_ref}"
        )
        
        return response.json()
```

### 2. 回调处理

```python
# Backend 接收 Gateway 的回调
@app.post("/callback/command/{request_ref}")
async def handle_command_callback(request_ref: str, callback_data: dict):
    """处理命令执行回调"""
    
    # 1. 验证回调数据
    if not validate_callback(callback_data):
        raise HTTPException(status_code=400, detail="Invalid callback data")
    
    # 2. 更新业务状态
    await update_publication_status(request_ref, callback_data)
    
    # 3. 触发后续业务逻辑
    if callback_data["status"] == "COMPLETED":
        await handle_command_success(request_ref, callback_data)
    elif callback_data["status"] == "FAILED":
        await handle_command_failure(request_ref, callback_data)
    
    # 4. 发送前端通知
    await notify_frontend(request_ref, callback_data)
    
    return {"status": "received"}

@app.post("/callback/progress/{request_ref}")
async def handle_progress_callback(request_ref: str, progress_data: dict):
    """处理进度更新回调"""
    
    # 更新进度状态
    await update_command_progress(request_ref, progress_data)
    
    # 实时推送到前端
    await websocket_manager.broadcast({
        "type": "progress_update",
        "request_ref": request_ref,
        "progress": progress_data
    })
    
    return {"status": "received"}
```

## 错误处理和容错机制

### 1. 连接故障处理

```typescript
class ConnectionManager {
  private reconnectDelay = 5000;
  private maxReconnectAttempts = 5;
  
  async handleConnectionLost(clientId: string): Promise<void> {
    console.log(`Connection lost: ${clientId}`);
    
    // 1. 标记连接状态
    this.markConnectionLost(clientId);
    
    // 2. 通知相关服务
    await this.notifyConnectionLost(clientId);
    
    // 3. 启动重连流程
    this.scheduleReconnect(clientId);
    
    // 4. 处理待处理的消息
    this.handlePendingMessages(clientId);
  }
  
  private async scheduleReconnect(clientId: string): Promise<void> {
    let attempts = 0;
    
    while (attempts < this.maxReconnectAttempts) {
      await this.sleep(this.reconnectDelay * Math.pow(2, attempts));
      
      try {
        await this.reconnect(clientId);
        console.log(`Reconnected: ${clientId}`);
        return;
      } catch (error) {
        attempts++;
        console.log(`Reconnect failed (${attempts}/${this.maxReconnectAttempts}): ${error.message}`);
      }
    }
    
    // 重连失败，标记为永久离线
    this.markPermanentlyOffline(clientId);
  }
}
```

### 2. 消息超时处理

```typescript
class TimeoutManager {
  private pendingMessages = new Map<string, TimeoutInfo>();
  
  startTimeout(requestRef: string, timeoutMs: number): void {
    const timeoutId = setTimeout(() => {
      this.handleTimeout(requestRef);
    }, timeoutMs);
    
    this.pendingMessages.set(requestRef, {
      timeoutId,
      startTime: Date.now(),
      timeoutMs
    });
  }
  
  clearTimeout(requestRef: string): void {
    const info = this.pendingMessages.get(requestRef);
    if (info) {
      clearTimeout(info.timeoutId);
      this.pendingMessages.delete(requestRef);
    }
  }
  
  private handleTimeout(requestRef: string): void {
    const info = this.pendingMessages.get(requestRef);
    if (!info) return;
    
    // 发送超时错误响应
    const timeoutResponse = {
      type: "ERROR",
      code: "TIMEOUT",
      message: `Command timeout after ${info.timeoutMs}ms`,
      requestRef,
      timestamp: new Date().toISOString()
    };
    
    this.routeResponse(timeoutResponse);
    this.pendingMessages.delete(requestRef);
  }
}
```

### 3. 错误响应类型

```json
{
  "type": "ERROR",
  "code": "TARGET_NOT_FOUND",
  "message": "目标客户端不存在",
  "details": {
    "targetClientId": "device-99",
    "reason": "No route found for device device-99"
  },
  "requestRef": "cmd-123",
  "timestamp": "2024-01-20T10:00:00Z"
}
```

```json
{
  "type": "ERROR",
  "code": "GATEWAY_UNAVAILABLE",
  "message": "网关服务不可用",
  "details": {
    "service": "gateway",
    "lastSeen": "2024-01-20T09:55:00Z",
    "retryAfter": 30
  },
  "requestRef": "cmd-124",
  "timestamp": "2024-01-20T10:00:00Z"
}
```

## 监控和可观测性

### 1. 消息追踪

```typescript
interface MessageTrace {
  requestRef: string;
  startTime: number;
  path: Array<{
    node: string;
    timestamp: number;
    action: 'send' | 'receive' | 'forward';
  }>;
  status: 'pending' | 'completed' | 'failed' | 'timeout';
  duration?: number;
}

class MessageTracer {
  private traces = new Map<string, MessageTrace>();
  
  startTrace(requestRef: string, from: string): void {
    this.traces.set(requestRef, {
      requestRef,
      startTime: Date.now(),
      path: [{
        node: from,
        timestamp: Date.now(),
        action: 'send'
      }],
      status: 'pending'
    });
  }
  
  addHop(requestRef: string, node: string, action: 'send' | 'receive' | 'forward'): void {
    const trace = this.traces.get(requestRef);
    if (trace) {
      trace.path.push({
        node,
        timestamp: Date.now(),
        action
      });
    }
  }
  
  completeTrace(requestRef: string, status: 'completed' | 'failed' | 'timeout'): void {
    const trace = this.traces.get(requestRef);
    if (trace) {
      trace.status = status;
      trace.duration = Date.now() - trace.startTime;
      
      // 发送到监控系统
      this.reportTrace(trace);
      
      // 清理
      this.traces.delete(requestRef);
    }
  }
}
```

### 2. 性能指标

```typescript
class PerformanceMonitor {
  recordRouteLatency(from: string, to: string, duration: number): void {
    metrics.histogram('route_latency_ms', duration, {
      from,
      to
    });
  }
  
  recordCommandSuccess(commandType: string, deviceType: string): void {
    metrics.counter('commands_success_total', 1, {
      command_type: commandType,
      device_type: deviceType
    });
  }
  
  recordCommandFailure(commandType: string, errorCode: string): void {
    metrics.counter('commands_failure_total', 1, {
      command_type: commandType,
      error_code: errorCode
    });
  }
  
  recordConnectionCount(nodeType: string, count: number): void {
    metrics.gauge('active_connections', count, {
      node_type: nodeType
    });
  }
}
```

## 最佳实践

### 1. 消息设计原则

```typescript
// ✅ 良好的消息设计
interface CommandMessage {
  type: 'COMMAND';
  requestRef: string;        // 唯一请求ID
  targetClientId: string;    // 明确的目标
  command: {
    commandType: string;     // 命令类型
    // ... 其他字段
  };
  priority: 'LOW' | 'NORMAL' | 'HIGH';
  timeout: number;
  callback?: string;         // 回调URL
  timestamp: string;
  version: string;
}

// ❌ 避免的设计
interface BadMessage {
  cmd: string;              // 不明确的字段名
  target: string;           // 缺少结构化
  data: any;                // 太宽泛的类型
}
```

### 2. 错误处理策略

```typescript
class ErrorHandler {
  handleRoutingError(error: Error, message: CommandMessage): void {
    // 1. 记录错误
    console.error(`Routing error for ${message.requestRef}:`, error);
    
    // 2. 分类错误
    const errorCode = this.classifyError(error);
    
    // 3. 发送错误响应
    const errorResponse = {
      type: 'ERROR',
      code: errorCode,
      message: error.message,
      requestRef: message.requestRef,
      timestamp: new Date().toISOString()
    };
    
    // 4. 路由错误响应
    this.routeErrorResponse(errorResponse);
    
    // 5. 更新监控指标
    metrics.counter('routing_errors_total', 1, {
      error_code: errorCode
    });
  }
}
```

### 3. 连接管理

```typescript
class ConnectionManager {
  private connections = new Map<string, Connection>();
  private healthChecks = new Map<string, NodeJS.Timeout>();
  
  addConnection(clientId: string, ws: WebSocket): void {
    const connection = new Connection(clientId, ws);
    this.connections.set(clientId, connection);
    
    // 启动健康检查
    this.startHealthCheck(clientId);
    
    // 记录连接指标
    metrics.gauge('active_connections', this.connections.size);
  }
  
  private startHealthCheck(clientId: string): void {
    const interval = setInterval(() => {
      const connection = this.connections.get(clientId);
      if (connection && !connection.isHealthy()) {
        this.handleUnhealthyConnection(clientId);
      }
    }, 30000); // 30秒检查一次
    
    this.healthChecks.set(clientId, interval);
  }
}
```

## 总结

新架构的消息路由系统具有以下特点：

### 1. 清晰的职责分离
- **Backend**: 专注业务逻辑、任务调度和数据管理
- **Gateway**: 提供API接口、协议路由、连接管理和统一callback处理
- **Edge**: 专注设备代理和适配
- **Device**: 专注硬件操作执行

### 2. 双通道命令支持
- **Command 通道**: EUDI → Gateway API → WebSocket，用于实时控制
- **Program 通道**: EUDI → Backend → WebSocket → Gateway，用于调度任务

### 3. 统一的 Callback 机制
- Gateway 负责所有类型命令的 callback 执行
- 支持重试和指数退避策略
- Program 命令同时通知 Backend 和 EUDI

### 4. 可靠的消息传递
- 层次化的路由机制
- 完整的错误处理和重试
- 消息追踪和监控
- 连接状态管理

### 5. 良好的可扩展性
- 服务独立部署和扩展
- 松耦合的服务交互
- 标准化的接口协议
- 灵活的负载均衡

### 6. 强大的可观测性
- 完整的消息链路追踪
- 丰富的性能指标
- 实时的状态监控
- 详细的错误报告

这种设计确保了系统的高可用性、高可扩展性和易维护性，为地铁显示系统提供了稳定可靠的通信基础。