import { config, dashboardEndpoints } from '../config'; import { getMockData } from '../mock'; import type { DashboardData, Station } from '../types'; type ChangeListener = (data: DashboardData | null) => void; type ModeListener = (mode: 'sse' | 'polling') => void; type ErrorListener = (message: string) => void; /** MES 各接口返回片段 */ interface EndpointOverview { production?: DashboardData['production']; warehouse?: DashboardData['warehouse']; } interface EndpointEquipment { stations?: Station[]; } interface EndpointAlarms { items?: DashboardData['alarms']; } /** * 数据源:优先尝试 SSE(经 fetch 实现,可携带 X-API-TOKEN 请求头); * SSE 建立或读取失败时降级为 10s HTTP 轮询。 */ export class DashboardDataService { private baseUrl = config.mesBaseUrl; private token = config.internalToken; private useMock = config.useMock; private controller: AbortController | null = null; private timer: ReturnType | null = null; private changeListeners = new Set(); private modeListeners = new Set(); private errorListeners = new Set(); mode: 'sse' | 'polling' | 'init' = 'init'; private running = false; onChange(cb: ChangeListener): () => void { this.changeListeners.add(cb); return () => this.changeListeners.delete(cb); } onModeChange(cb: ModeListener): () => void { this.modeListeners.add(cb); return () => this.modeListeners.delete(cb); } onError(cb: ErrorListener): () => void { this.errorListeners.add(cb); return () => this.errorListeners.delete(cb); } private emit(data: DashboardData | null) { this.changeListeners.forEach((l) => l(data)); } private setMode(m: 'sse' | 'polling') { this.mode = m; this.modeListeners.forEach((l) => l(m)); } private emitError(msg: string) { this.errorListeners.forEach((l) => l(msg)); } /** 启动数据流(幂等,可安全重复调用) */ start() { if (this.running) return; this.running = true; if (this.useMock) { // 演示模式:直接给出数据 this.setMode('polling'); this.emit(getMockData()); return; } void this.openSse(); } /** 停止数据流并清理资源 */ stop() { this.running = false; this.closeStream(); this.clearTimer(); this.changeListeners.clear(); this.modeListeners.clear(); this.errorListeners.clear(); } /** 手动重试:恢复默认时先重新建立 SSE,失败则轮询 */ retry() { if (!this.running) this.running = true; if (this.useMock) { this.emit(getMockData()); return; } this.clearTimer(); void this.openSse(); } // ---------- SSE ---------- private async openSse() { this.closeStream(); const url = `${this.baseUrl}/api/internal/dashboard/stream`; const controller = new AbortController(); this.controller = controller; try { const res = await fetch(url, { headers: { 'X-API-TOKEN': this.token }, signal: controller.signal, }); if (!res.ok || !res.body) throw new Error(`SSE 连接失败:HTTP ${res.status}`); this.setMode('sse'); const reader = res.body.getReader(); const decoder = new TextDecoder(); let buffer = ''; for (;;) { const { done, value } = await reader.read(); if (done) break; buffer += decoder.decode(value, { stream: true }); // 按空行切分 SSE 事件块 const blocks = buffer.split('\n\n'); buffer = blocks.pop() ?? ''; for (const block of blocks) { const dataLine = block .split('\n') .find((l) => l.startsWith('data:')); if (!dataLine) continue; const payload = dataLine.replace(/^data:\s*/, '').trim(); try { const parsed = JSON.parse(payload) as DashboardData; if (parsed) this.emit(parsed); } catch { // 忽略无法解析的单条事件,继续读取 } } } } catch (err) { if (!this.running) return; if (err instanceof Error && err.name === 'AbortError') return; this.emitError('SSE 连接失败,已降级为 HTTP 轮询(每 10s 刷新)'); this.fallbackToPolling(); } } private closeStream() { this.controller?.abort(); this.controller = null; } // ---------- HTTP 轮询(降级) ---------- private fallbackToPolling() { this.clearTimer(); void this.pollOnce(); this.timer = setInterval(() => void this.pollOnce(), config.pollIntervalMs); } private async pollOnce() { try { const headers = { 'X-API-TOKEN': this.token }; const fetchJson = async (path: string): Promise => { const res = await fetch(`${this.baseUrl}${path}`, { headers }); if (!res.ok) throw new Error(`HTTP ${res.status}`); return (await res.json()) as T; }; const [overview, equipment, progress, alarms, trends] = await Promise.all([ fetchJson(dashboardEndpoints.overview), fetchJson(dashboardEndpoints.equipment), fetchJson(dashboardEndpoints.progress), fetchJson(dashboardEndpoints.alarms), fetchJson(dashboardEndpoints.trends), ]); const assembled: DashboardData = { production: overview.production ?? this.emptyProduction(), warehouse: overview.warehouse ?? this.emptyWarehouse(), equipment: equipment.stations ?? [], progress: progress ?? this.emptyProgress(), alarms: alarms.items ?? [], trends: trends ?? { production: [] }, }; this.setMode('polling'); this.emit(assembled); } catch (err) { this.emit(null); this.emitError( `数据加载失败:${err instanceof Error ? err.message : '网络异常或无 MES 服务'}`, ); } } private clearTimer() { if (this.timer) { clearInterval(this.timer); this.timer = null; } } // ---------- 空值兜底 ---------- private emptyProduction(): DashboardData['production'] { return { outputToday: 0, targetToday: 0, qualifiedRate: 0, qualityOk: 0, qualityNg: 0, stationSummary: { running: 0, idle: 0, offline: 0, alarm: 0 }, }; } private emptyWarehouse(): DashboardData['warehouse'] { return { totalStock: 0, materialTypes: 0, inboundToday: 0, outboundToday: 0, movements: [], }; } private emptyProgress(): DashboardData['progress'] { return { orderNo: '-', productName: '-', totalQty: 0, doneQty: 0, completedQty: 0, currentProcess: '-', status: '-', }; } }