package com.ruoyi.unattended.telemetry.service; import com.ruoyi.unattended.telemetry.dto.DeviceMonitorVo; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import java.io.IOException; import java.time.LocalDateTime; import java.util.Collections; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; /** * 设备报警 SSE 推送中心。 */ @Slf4j @Component public class DeviceAlarmPushService { private final List clients = new CopyOnWriteArrayList<>(); public SseEmitter subscribe() { SseEmitter emitter = new SseEmitter(0L); clients.add(emitter); emitter.onCompletion(() -> clients.remove(emitter)); emitter.onTimeout(() -> clients.remove(emitter)); emitter.onError(error -> clients.remove(emitter)); try { emitter.send(SseEmitter.event() .name("connected") .data(Collections.singletonMap("time", LocalDateTime.now().toString()))); } catch (IOException e) { clients.remove(emitter); } return emitter; } public void publish(DeviceMonitorVo alarm) { for (SseEmitter client : clients) { try { String eventId = alarm.getDeviceId() + "-" + System.currentTimeMillis(); client.send(SseEmitter.event().id(eventId).name("alarm").data(alarm)); } catch (Exception e) { clients.remove(client); } } } /** 保持代理和浏览器之间的长连接,及时清理已断开的客户端。 */ @Scheduled(fixedRate = 25000L) public void heartbeat() { for (SseEmitter client : clients) { try { client.send(SseEmitter.event() .name("heartbeat") .data(Collections.singletonMap("time", LocalDateTime.now().toString()))); } catch (Exception e) { clients.remove(client); } } } }