ssp-server/src/main/java/com/smcic/api/conference/service/impl/SSEService.java

package com.smcic.api.conference.service.impl;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Service
@Slf4j
public class SSEService {

    private Map<Long, Map<String,SseEmitter>>  sseDataMap = new ConcurrentHashMap<>();

    public void monitor(Long cid, String uuid, SseEmitter sseEmitter) {
        if(!sseDataMap.containsKey(cid)){
            sseDataMap.put(cid, new ConcurrentHashMap<>());
        }
        sseDataMap.get(cid).put(uuid,sseEmitter);
    }

    public void sendData(Long cid, Object data) {
        try {

            if (sseDataMap != null &&  sseDataMap.containsKey(cid)) {
                Map<String, SseEmitter> stringSseEmitterMap = sseDataMap.get(cid);
                for (String uuid : stringSseEmitterMap.keySet()){
                    stringSseEmitterMap.get(uuid).send(SseEmitter.event().name("data").data(data));
                }
            }
        }catch (IOException e){
            log.error("发送数据失败",e);
        }

    }

    public void complate(Long cid, String uuid) {
        if (sseDataMap.containsKey(cid) && sseDataMap.get(cid).containsKey(uuid)) {
            Map<String, SseEmitter> stringSseEmitterMap = sseDataMap.get(cid);
            SseEmitter sseEmitter = stringSseEmitterMap.get(uuid);
            sseEmitter.complete();
            log.info("关闭连接");
            stringSseEmitterMap.remove(uuid);
            if(stringSseEmitterMap.isEmpty()){
                sseDataMap.remove(cid);
            }
        }
    }
}