@Slf4j
@Component
public class SseEmitterUtil {
    //保存客戶連接
    public static Map<String, SseEmitter> userSseEmitters = new HashMap<>();
    /**
     * 用戶上線，開啟一個(gè)SSE連接
     * @param userId
     * @return
     */
    public SseEmitter subscribe(String userId){
        //創(chuàng)建一個(gè)超時(shí)時(shí)間為30秒的SseEmitter對象
        SseEmitter sseEmitter = new SseEmitter(30*1000L);
        //設(shè)置回調(diào)函數(shù)
        sseEmitter.onCompletion(completionCallBack(userId));
        sseEmitter.onError(errorCallBack(userId));
        sseEmitter.onTimeout(timeoutCallBack(userId));
        //緩存起來
        userSseEmitters.put(userId,sseEmitter);
        return sseEmitter;
    }
    /**
     * 推送消息給某個(gè)客戶端
     * @param userId  用戶ID
     * @param content 消息體
     */
    public static void push(String userId,String content){
        SseEmitter sseEmitter = userSseEmitters.get(userId);
        if (sseEmitter != null) {
            try {
                sseEmitter.send(content);
            }catch (Exception e){
                log.error("用戶-{} 推送消息異常：{}",userId,e);
            }
        }else {
            log.info("用戶-{} 連接不存在",userId);
        }
    }
    /**
     * 推送消息給所有客戶端
     * @param content 消息體
     */
    public static void pushAllUser(String content){
        for (SseEmitter emitter : userSseEmitters.values()) {
            try {
                emitter.send(SseEmitter.event().name("全局消息").data(content));
            } catch (Exception e) {
                log.error("推送消息異常：{}",e);
            }
        }
    }
    /**
     * 刪除某個(gè)客戶連接
     * @param userId
     */
    public static void removeEmitter(String userId) {
        SseEmitter sseEmitter = userSseEmitters.get(userId);
        if (sseEmitter != null) {
            sseEmitter.complete();
            userSseEmitters.remove(userId);
        }else {
            log.info("用戶-{} 連接不存在",userId);
        }
    }
    private Runnable completionCallBack(String userId) {
        return () -> {
            log.info("用戶-{} 連接成功", userId);
        };
    }
    /**
     * 出現(xiàn)超時(shí)，將當(dāng)前用戶緩存刪除
     * @param userId
     * @return
     */
    private Runnable timeoutCallBack(String userId) {
        return () -> {
            log.info("用戶-{} 連接超時(shí)", userId);
            userSseEmitters.remove(userId);
        };
    }
    /**
     * 出現(xiàn)異常，將當(dāng)前用戶緩存刪除
     * @param userId
     * @return
     */
    private Consumer<Throwable> errorCallBack(String userId) {
        return throwable -> {
            log.info("用戶-{} 連接異常", userId);
            userSseEmitters.remove(userId);
        };
    }
}

@Component
public class UserLoginListener {
   private final SseEmitterUtil sseService;
    @Autowired
    public UserLoginListener(SseEmitterUtil sseService) {
        this.sseService = sseService;
    }
    @EventListener
    public void onUserLogin(UserLoginEvent event) {
        String message = "用戶 " + event.getUserId() + " 上線了";
        sseService.pushAllUser(message);
    }
}

@Service
public class UserService {
    private final ApplicationEventPublisher publisher;
    @Autowired
    public UserService(ApplicationEventPublisher publisher) {
        this.publisher = publisher;
    }
    public void login(String userId) {
        // 登錄邏輯
        publisher.publishEvent(new UserLoginEvent(userId));
    }
}