# Edge 代理模式指南

本指南说明 Edge 如何作为设备代理与 Gateway 通信。

## 架构概述

```
设备端层               Edge 层                    Gateway 层              Backend 层
(隧道媒体系统)                                                  
┌─────────┐   WS    ┌─────────────┐    WS    ┌──────────────┐       ┌──────────┐
│ TD-01   ├─────────┤             ├──────────┤              ├───────┤          │
│ TD-02   ├─────────┤   Edge A    │          │              │       │ Backend  │
│ TD-03   ├─────────┤             │          │              │       │          │
└─────────┘         └─────────────┘          │              │       └──────────┘
                           ↕                  │   Gateway    │
┌─────────┐   WS    ┌─────────────┐    WS    │              │       
│ TD-04   ├─────────┤             ├──────────┤              │       
│ TD-05   ├─────────┤   Edge B    │          │              │       
└─────────┘         └─────────────┘          └──────────────┘       

WS = WebSocket 连接
```

## Edge 的职责

1. **设备管理** - 管理隧道媒体广告播放系统的设备端
2. **WebSocket 服务** - 为设备端提供 WebSocket 连接服务
3. **心跳维护** - 维护与设备端的心跳（30秒间隔，90秒超时）
4. **命令路由** - 将 Gateway 命令路由到正确的设备端
5. **状态聚合** - 收集并上报设备状态
6. **离线缓存** - 在网络中断时缓存命令和数据
7. **设备注册代理** - 为连接的设备向 Gateway 注册

## 实现 Edge 代理

### 1. Edge 初始化和注册

```typescript
import { 
  EdgeProxy, 
  EdgeProxyUtils,
  ManagedDevice 
} from '@jrsoft/subway-protocol';

class MyEdgeProxy extends EdgeProxy {
  constructor() {
    super('edge-001');  // Edge ID
  }

  async initialize() {
    // 1. 启动 WebSocket 服务器，接受设备端连接
    await this.startDeviceServer(8080);
    
    // 2. 连接到 Gateway
    await this.connectToGateway('ws://gateway:18081');
    
    // 3. 注册 Edge 自身
    await this.registerToGateway();
    
    // 4. 为已连接的设备端注册（场景1：Edge 先启动）
    for (const [deviceId, device] of this.connectedDevices) {
      await this.registerDeviceToGateway(deviceId);
    }
  }

  // 处理设备端 WebSocket 连接
  private async handleDeviceConnection(ws: WebSocket, deviceId: string) {
    // 设备端注册消息
    ws.on('message', async (data) => {
      const msg = JSON.parse(data.toString());
      
      if (msg.type === 'register') {
        // 记录设备信息
        this.connectedDevices.set(msg.clientId, {
          ws,
          deviceId: msg.clientId,
          deviceType: msg.clientInfo?.deviceType || 'media_player',
          capabilities: msg.clientInfo?.capabilities || [],
          status: 'online',
          lastSeen: new Date().toISOString()
        });
        
        // 向设备端确认注册
        ws.send(JSON.stringify({
          type: 'register_ack',
          clientId: msg.clientId,
          success: true,
          timestamp: new Date().toISOString(),
          version: '1.0'
        }));
        
        // 场景2：Gateway 在线，立即注册新设备
        if (this.isConnectedToGateway()) {
          await this.registerDeviceToGateway(msg.clientId);
        }
      }
      
      if (msg.type === 'heartbeat') {
        // 响应心跳
        ws.send(JSON.stringify({
          type: 'heartbeat_ack',
          clientId: msg.clientId,
          sequence: msg.sequence,
          timestamp: new Date().toISOString(),
          version: '1.0'
        }));
        
        // 更新最后活跃时间
        const device = this.connectedDevices.get(msg.clientId);
        if (device) {
          device.lastSeen = new Date().toISOString();
        }
      }
    });
  }
}
```

### 2. 注册流程

#### 2.1 Edge 自身注册

Edge 先注册自己到 Gateway：

```json
{
  "type": "REGISTER",
  "clientId": "edge-001",
  "clientType": "EDGE",
  "clientInfo": {
    "version": "1.0.0",
    "platform": "edge-proxy",
    "capabilities": ["device_proxy", "batch_command", "status_report"]
  },
  "timestamp": "2024-01-20T10:00:00Z",
  "version": "1.0"
}
```

#### 2.2 设备端通过 Edge 注册

Edge 为每个连接的设备端发送注册消息到 Gateway：

```json
{
  "type": "REGISTER",
  "clientId": "td-01",  // 设备端 ID
  "clientType": "DEVICE",
  "edgeInfo": {  // Edge 信息
    "edgeId": "edge-001",
    "edgeVersion": "1.0.0",
    "connectionTime": "2024-01-20T09:59:00Z"
  },
  "clientInfo": {
    "version": "1.0.0",
    "deviceType": "media_player",
    "capabilities": ["play", "stop", "status_report", "program_upload"]
  },
  "timestamp": "2024-01-20T10:00:01Z",
  "version": "1.0"
}
```

### 3. 设备寻址

Backend 直接使用设备ID，Gateway 根据路由表自动找到对应的 Edge：

```typescript
// Backend 发送命令，只需指定设备ID
const command = MessageFactory.createCommandMessage(
  'cmd-123',
  'td-01',  // 目标设备ID，Gateway会自动路由到管理该设备的Edge
  {
    commandType: CommandType.SIMPLE,
    commandCode: 'ProgramUpload',
    deviceType: 'media_player',
    deviceId: 1,
    operationType: 'write',
    parameters: {
      taskId: '1234567890123456789',
      programId: '1234567890123456788',
      programName: '春节活动广告',
      // ... 其他参数
    }
  }
);
```

### 4. 命令处理流程

```typescript
class MyEdgeProxy extends EdgeProxy {
  // 处理来自 Gateway 的命令
  protected async handleGatewayCommand(message: CommandMessage) {
    // targetClientId 直接就是设备ID
    const deviceId = message.targetClientId;
    
    const device = this.connectedDevices.get(deviceId);
    if (!device) {
      throw new Error(`Device ${deviceId} not found`);
    }
    
    // 转发命令到设备端（通过 WebSocket）
    // 直接转发 command 消息，保持消息类型不变
    device.ws.send(JSON.stringify(message));
    
    // 等待设备响应
    return await this.waitForDeviceResponse(message.requestRef, message.timeout);
  }

  // 处理 Complex 类型命令的持续响应
  private async handleComplexCommand(device: any, command: CommandMessage) {
    // 直接转发命令
    device.ws.send(JSON.stringify(command));
    
    // Complex 命令会收到多个响应
    device.ws.on('message', (data) => {
      const response = JSON.parse(data.toString());
      
      if (response.requestRef === command.requestRef) {
        // 转发进度更新到 Gateway
        if (response.type === 'progress_update') {
          this.forwardToGateway(response);
        }
        
        // 最终响应
        if (response.type === 'command_response') {
          this.forwardToGateway(response);
        }
      }
    });
  }
}
```

### 5. 批量命令支持

批量命令是向同一设备端的多个同类型子硬件发送相同命令：

```typescript
// Backend 发送批量命令到光柱
const batchCommand = MessageFactory.createCommandMessage(
  'batch-001',
  'td-01',  // 目标设备ID
  {
    commandType: CommandType.BATCH,
    commandCode: 'TRAIN_LENGTH',
    deviceType: 'pillar',
    deviceId: '1-10,30-40,51,52,53,54',  // 批量目标：范围表示法
    operationType: 'write',
    parameters: {
      switch: 'ON'
    }
  },
  {
    priority: Priority.HIGH,
    timeout: 10000
  }
);

// Edge 处理批量命令
class MyEdgeProxy extends EdgeProxy {
  private async handleBatchCommand(device: any, command: CommandMessage) {
    // 解析批量目标
    const targets = this.parseTargetRange(command.command.deviceId);
    
    // 转发到设备端，由设备端执行批量操作
    // 将解析后的目标列表添加到命令参数中
    const deviceCommand = {
      ...command,
      command: {
        ...command.command,
        parameters: {
          ...command.command.parameters,
          targets  // 添加解析后的目标列表
        }
      }
    };
    
    device.ws.send(JSON.stringify(deviceCommand));
  }
  
  // 解析范围表示法
  private parseTargetRange(rangeStr: string): number[] {
    if (rangeStr === '0') return [0]; // 0 表示所有
    
    const targets: number[] = [];
    const parts = rangeStr.split(',');
    
    for (const part of parts) {
      if (part.includes('-')) {
        const [start, end] = part.split('-').map(Number);
        for (let i = start; i <= end; i++) {
          targets.push(i);
        }
      } else {
        targets.push(Number(part));
      }
    }
    
    return targets;
  }
}
```

### 6. 设备状态监控和上报

```typescript
class MyEdgeProxy extends EdgeProxy {
  private heartbeatTimeouts = new Map<string, NodeJS.Timeout>();
  
  startMonitoring() {
    // 监控设备端心跳（90秒超时）
    this.monitorDeviceHeartbeats();
    
    // 定期向 Gateway 发送心跳
    setInterval(() => {
      this.sendHeartbeatToGateway();
    }, 30000); // 30秒
    
    // 定期发送设备状态汇总报告
    setInterval(() => {
      this.sendStatusReport();
    }, 60000); // 60秒
  }

  private monitorDeviceHeartbeats() {
    // 每次收到设备心跳时重置超时计时器
    this.on('device_heartbeat', (deviceId: string) => {
      // 清除旧的超时计时器
      const oldTimeout = this.heartbeatTimeouts.get(deviceId);
      if (oldTimeout) {
        clearTimeout(oldTimeout);
      }
      
      // 设置新的90秒超时计时器
      const timeout = setTimeout(() => {
        this.handleDeviceTimeout(deviceId);
      }, 90000);
      
      this.heartbeatTimeouts.set(deviceId, timeout);
    });
  }

  private handleDeviceTimeout(deviceId: string) {
    const device = this.connectedDevices.get(deviceId);
    if (device && device.status === 'online') {
      // 标记设备离线
      device.status = 'offline';
      
      // 通知 Gateway 设备离线
      const notification = {
        type: 'device_status_change',
        edgeId: this.edgeId,
        deviceId: deviceId,
        previousStatus: 'online',
        currentStatus: 'offline',
        reason: 'heartbeat_timeout',
        timestamp: new Date().toISOString(),
        version: '1.0'
      };
      
      this.sendToGateway(notification);
      
      // 尝试关闭 WebSocket 连接
      if (device.ws) {
        device.ws.close();
      }
    }
  }

  private sendStatusReport() {
    const report = {
      type: 'edge_status_report',
      edgeId: this.edgeId,
      timestamp: new Date().toISOString(),
      deviceCount: this.connectedDevices.size,
      devices: Array.from(this.connectedDevices.entries()).map(([id, device]) => ({
        deviceId: id,
        status: device.status,
        lastSeen: device.lastSeen,
        deviceType: device.deviceType
      })),
      metrics: {
        cpuUsage: process.cpuUsage(),
        memoryUsage: process.memoryUsage(),
        uptime: process.uptime()
      },
      version: '1.0'
    };
    
    this.sendToGateway(report);
  }
}
```

### 7. 进度更新处理

Edge 需要正确处理和转发设备的进度更新消息：

```typescript
class MyEdgeProxy extends EdgeProxy {
  // 处理设备端的进度更新
  private handleDeviceProgressUpdate(deviceId: string, update: ProgressUpdateMessage) {
    // 记录日志
    if (update.log) {
      this.logDeviceMessage(deviceId, update.log.level, update.log.code, update.log.data);
    }
    
    // 检查状态
    switch (update.status) {
      case 'failed':
        this.handleDeviceError(deviceId, update);
        break;
      case 'paused':
        this.handleDevicePaused(deviceId, update);
        break;
      case 'cancelled':
        this.handleDeviceCancelled(deviceId, update);
        break;
    }
    
    // 转发到 Gateway
    this.forwardToGateway(update);
  }
  
  // 处理设备错误
  private handleDeviceError(deviceId: string, update: ProgressUpdateMessage) {
    if (update.log?.level === 'critical') {
      // 严重错误，可能需要重启设备
      this.scheduleDeviceRestart(deviceId);
    }
    
    // 记录错误统计
    this.errorStats.record(deviceId, update.log?.code || 'UNKNOWN_ERROR');
  }
  
  // 汇总进度信息
  private aggregateProgress(deviceId: string, update: ProgressUpdateMessage) {
    const progress = this.deviceProgress.get(deviceId) || {};
    progress[update.requestRef] = {
      phase: update.phase,
      progress: update.progress,
      status: update.status,
      lastUpdate: update.timestamp
    };
    this.deviceProgress.set(deviceId, progress);
  }
}
```

### 8. 离线模式和数据缓存

```typescript
class OfflineCapableEdge extends EdgeProxy {
  private commandQueue: CommandMessage[] = [];
  private dataCache: Map<string, any> = new Map();

  protected async handleGatewayCommand(message: CommandMessage) {
    if (!this.isConnectedToGateway()) {
      // 离线模式：缓存命令
      this.commandQueue.push(message);
      return;
    }

    try {
      await super.handleGatewayCommand(message);
    } catch (error) {
      // 执行失败，根据策略决定是否缓存
      if (this.shouldCacheOnError(error)) {
        this.commandQueue.push(message);
      }
    }
  }

  // 重连后处理缓存的命令
  private async processCachedCommands() {
    while (this.commandQueue.length > 0) {
      const command = this.commandQueue.shift()!;
      try {
        await super.handleGatewayCommand(command);
      } catch (error) {
        console.error('Failed to process cached command:', error);
      }
    }
  }

  // 缓存设备数据供离线查询
  private cacheDeviceData(deviceId: string, data: any) {
    this.dataCache.set(
      `${deviceId}:${Date.now()}`,
      {
        deviceId,
        data,
        timestamp: new Date().toISOString()
      }
    );
  }
}
```

## 最佳实践

### 1. 设备分组管理

```typescript
// 按类型、位置或功能分组设备
const deviceGroups = {
  floor1: ['plc-01', 'plc-02', 'sensor-01'],
  floor2: ['plc-03', 'plc-04', 'sensor-02'],
  critical: ['plc-01', 'plc-03']  // 关键设备
};

// 支持按组执行命令
async function executeGroupCommand(group: string, command: any) {
  const devices = deviceGroups[group] || [];
  return await EdgeProxyUtils.createEdgeBatchCommand(
    `group-${group}`,
    'edge-01',
    devices,
    command
  );
}
```

### 2. 错误处理和恢复

```typescript
class ResilientEdge extends EdgeProxy {
  private deviceRetryCount = new Map<string, number>();

  protected async handleGatewayCommand(message: CommandMessage): Promise<any> {
    // targetClientId 直接就是设备ID
    const deviceId = message.targetClientId;
    const device = this.connectedDevices.get(deviceId);
    
    if (!device) {
      throw new Error(`Device ${deviceId} not found`);
    }

    const maxRetries = 3;
    let lastError;

    for (let i = 0; i < maxRetries; i++) {
      try {
        // 发送命令到设备
        device.ws.send(JSON.stringify(message));
        
        // 等待响应
        const result = await this.waitForDeviceResponse(message.requestRef, message.timeout);
        
        // 成功，重置重试计数
        this.deviceRetryCount.set(deviceId, 0);
        return result;
        
      } catch (error) {
        lastError = error;
        const retryCount = (this.deviceRetryCount.get(deviceId) || 0) + 1;
        this.deviceRetryCount.set(deviceId, retryCount);
        
        // 指数退避
        if (i < maxRetries - 1) {
          await this.delay(Math.pow(2, i) * 1000);
        }
      }
    }

    // 标记设备为错误状态
    device.status = 'error';
    this.notifyDeviceError(deviceId, lastError);

    throw lastError;
  }

  private async notifyDeviceError(deviceId: string, error: any) {
    const errorNotification = {
      type: 'error',
      code: 'DEVICE_ERROR',
      message: `Device ${deviceId} error: ${error.message}`,
      severity: 'high',
      category: 'device',
      context: {
        edgeId: this.edgeId,
        deviceId: deviceId,
        error: error.toString()
      },
      retryable: true,
      timestamp: new Date().toISOString(),
      version: '1.0'
    };
    
    this.sendToGateway(errorNotification);
  }
}
```

### 3. 性能优化

```typescript
// 1. 命令合并
class OptimizedEdge extends EdgeProxy {
  private pendingCommands = new Map<string, CommandMessage[]>();

  // 合并相同设备的读命令
  protected optimizeReadCommands(commands: CommandMessage[]): CommandMessage[] {
    const grouped = this.groupByDevice(commands);
    
    return grouped.map(group => {
      if (group.length === 1) return group[0];
      
      // 合并多个读取请求
      return this.mergeReadCommands(group);
    });
  }
}

// 2. 数据预取
class PrefetchingEdge extends EdgeProxy {
  private prefetchPatterns = new Map<string, string[]>();

  // 根据历史模式预取数据
  protected async handleGatewayCommand(message: CommandMessage) {
    await super.handleGatewayCommand(message);
    
    // 分析命令模式
    this.analyzeCommandPattern(message);
    
    // 预取相关数据
    const relatedData = this.prefetchPatterns.get(message.command.commandCode);
    if (relatedData) {
      this.prefetchDeviceData(message.clientId, relatedData);
    }
  }
}
```

### 4. 监控和告警

```typescript
interface EdgeAlert {
  level: 'info' | 'warning' | 'error' | 'critical';
  source: string;
  message: string;
  timestamp: string;
  deviceId?: string;
}

class MonitoredEdge extends EdgeProxy {
  private alerts: EdgeAlert[] = [];

  protected sendAlert(alert: EdgeAlert) {
    this.alerts.push(alert);
    
    // 发送到 Gateway
    this.sendToGateway({
      type: 'edge_alert',
      edgeId: this.edgeId,
      alert
    });
    
    // 严重告警立即通知
    if (alert.level === 'critical') {
      this.notifyCriticalAlert(alert);
    }
  }

  // 监控指标
  getMetrics() {
    return {
      deviceCount: this.devices.size,
      onlineDevices: Array.from(this.devices.values())
        .filter(d => d.status === 'online').length,
      commandsPerMinute: this.calculateCommandRate(),
      errorRate: this.calculateErrorRate(),
      avgResponseTime: this.calculateAvgResponseTime()
    };
  }
}
```

## 故障场景处理

### 1. Gateway 连接中断

```typescript
// Edge 自动切换到离线模式
// 缓存数据和命令
// 保持设备正常运行
// 重连后同步状态
```

### 2. 设备故障

```typescript
// 标记设备离线
// 通知 Gateway
// 尝试恢复
// 提供故障诊断信息
```

### 3. 批量命令部分失败

```typescript
// 继续执行其他设备命令
// 收集所有结果
// 返回详细的执行报告
// 支持重试失败的命令
```

## 总结

Edge 代理模式在 JRSoft Subway 系统中的作用：

1. **WebSocket 代理** - 为设备端提供 WebSocket 服务，管理隧道媒体广告播放系统
2. **双向注册** - Edge 自身注册 + 代理设备端注册
3. **心跳维护** - 设备端→Edge（30秒/90秒），Edge→Gateway（30秒/90秒）
4. **命令路由** - 支持 Simple、Batch、Complex 三种命令类型
5. **状态监控** - 实时监控设备状态，及时上报异常
6. **离线缓存** - Gateway 断线时缓存数据，重连后同步

关键特性：
- 使用 `edgeId:deviceId` 格式进行设备寻址
- 支持批量命令的范围表示法（如 "1-10,30-40,51"）
- Complex 命令支持持续响应（如健康检查）
- 所有消息使用协议版本 "1.0"

通过 Edge 代理，可以将设备管理逻辑下沉到边缘，提高系统的可扩展性和可靠性。