SseEmitter를 사용하여 특정 대상에게 메시지(알람) 전송

Dependency 추가

Spring Boot 2 기준으로 다음과 같이 Dependency를 추가해준다.

dependencies {  
    implementation 'org.springframework.boot:spring-boot-starter-security'  
    implementation 'org.springframework.boot:spring-boot-starter-web'  
    ...
    testImplementation 'org.springframework.boot:spring-boot-starter-test'  
    testImplementation 'org.springframework.security:spring-security-test'  
}

Security 설정

Spring Security 설정 부분은 이전에 작성한 Spring WebSocket을 사용하여 특정 대상에게 메시지(알람) 전송과 동일하다.

Repository 추가

SseEmitter를 저장하고 가져오기위해 Respository를 추가한다.

@Repository  
@NoArgsConstructor  
public class EmitterRepository {  

    private final Map<String, SseEmitter> emitterMap = new ConcurrentHashMap<>();  

    /**
    * username을 key로 하여 SseEmitter 저장
    */
    public SseEmitter save(String username, SseEmitter sseEmitter) {  
        final String key = getKey(username);  
        emitterMap.put(key, sseEmitter);  
        return sseEmitter;  
    }  

    /**
    * username로 특정 사용자의 SseEmitter를 가져온다.
    */
    public Optional<SseEmitter> get(String username) {  
        final String key = getKey(username);  
        return Optional.ofNullable(emitterMap.get(key));  
    }  

    /**
    * username로 특정 사용자의 SseEmitter를 삭제한다.
    */
    public void delete(String username) {  
        emitterMap.remove(getKey(username));  
    }  

    /**
    * 실제로 emitterMap에 저장될때 사용되는 key를 생성
    */
    private String getKey(String username) {  
        return "Emitter:Username:" + username;  
    }  
}

비지니스 로직 구현

SseEmitter를 연결(저장)하고 특정 대상에게 메시지를 전송하는 로직을 구현한다.

@Service  
@RequiredArgsConstructor  
public class AlarmService {  
    private static final Long SSE_TIMEOUT = 60L * 60 * 1000; // 1시간  
    private static final String ALARM_NAME = "alarm";  
    private final EmitterRepository emitterRepository;  

    /**  
     * receiverName의 emitter를 찾아 메시지 전송  
     */  
    public void send(String receiverName, String content) {  
        emitterRepository  
                .get(receiverName)  
                .ifPresentOrElse(  
                        sseEmitter -> {  
                            try {  
                                sseEmitter.send(SseEmitter.event()  
                                        .id(receiverName)  
                                        .name(ALARM_NAME)  
                                        .data(content));  
                            } catch (IOException e) {  
                                emitterRepository.delete(receiverName);  
                                throw new RuntimeException();  
                            }  
                        },  
                        () -> log.info("Emitter를 찾을 수 없습니다."));  
    }  

    /**  
     * SseEmitter 연결  
     */  
    public SseEmitter connectAlarm(String username) {  
        SseEmitter sseEmitter = new SseEmitter(SSE_TIMEOUT);  
        emitterRepository.save(username, sseEmitter);  
        sseEmitter.onCompletion(() -> emitterRepository.delete(username));  
        sseEmitter.onTimeout(() -> emitterRepository.delete(username));  

        try {  
        //처음에 SSE 응답을 할 때 아무런 이벤트도 보내지 않으면 재연결 요청을 보낼때나, 아니면 연결 요청 자체에서 에러가 발생한다. 따라서 임의로 데이터를 전송한다.
        sseEmitter.send(SseEmitter.event().id("").name(ALARM_NAME).data("connect completed"));  
        } catch (IOException e) {  
            throw new RuntimeException();  
        }  

        return sseEmitter;  
    }  
}

API 구현

@RestController  
@RequestMapping("/api/v1/alarm")  
@RequiredArgsConstructor  
public class AlarmApi {  
    private final AlarmService alarmService;  


    /**  
     * SSE 연결 요청  
     */  
    @GetMapping("/subscribe")  
    public ResponseEntity<SseEmitter> subscribe(@AuthenticationPrincipal MemberPrincipal memberPrincipal) {  

        return ResponseEntity.ok(alarmService.connectAlarm(memberPrincipal.username()));  
    }  

    /**  
     * 메시지 전송  
     */  
    @PostMapping("/send")  
    public ResponseEntity<Void> sendAlarm(@RequestParam String receiverName, @RequestParam String content) {  
        alarmService.send(receiverName, content);  
        return ResponseEntity.ok(null);  
    }  
}

Front

html

html 또한 Spring WebSocket을 사용하여 특정 대상에게 메시지(알람) 전송에서 사용한 html을 재사용했다.

js

var eventSource = null;  

function setConnected(connected) {  
    $("#connect").prop("disabled", connected);  
    $("#disconnect").prop("disabled", !connected);  
    if (connected) {  
        $("#conversation").show();  
    } else {  
        $("#conversation").hide();  
    }  
    $("#greetings").html("");  
}  

/**  
 * 연결  
 */  
function connect() {  
    eventSource = new EventSource("/api/v1/alarm/subscribe");  
    eventSource.addEventListener('alarm', function (event) {  
        console.log(event);  
        showGreeting(event.data);  
    });  
}  

/**  
 * 종료(닫기)  
 */function disconnect() {  
    eventSource.close();  
    setConnected(false);  
}  

/**  
 * 메시지 전송  
 */  
function sendName() {  
    var name = $("#name").val();  
    var receiverName = 'admin';  

    var formData = new FormData();  
    formData.append('content', name);  
    formData.append('receiverName', receiverName);  

    $.ajax({  
        url:"/api/v1/alarm/send",  
        data: formData,  
        processData: false,  
        contentType: false,  
        type: "POST",  
        success: function (result) {  
            console.log("success");  
        },  
        error: function (result) {  
            console.log("fail");  
        }  
    })  
}  

function showGreeting(message) {  
    $("#greetings").append("<tr><td>" + message + "</td></tr>");  
}  

$(function () {  
    $("form").on('submit', (e) => e.preventDefault());  
    $("#connect").click(() => connect());  
    $("#disconnect").click(() => disconnect());  
    $("#send").click(() => sendName());  
});

 

 

전체 코드는 https://github.com/JadeKim042386/SpringWebSocket/tree/main/MessageSseEmitter에서 확인할 수 있다.