[BE] 알림 기능 리팩토링 - 단일 책임 원칙

단일 책임의 원칙(Single Responsiblity Principle, SRP) - 하나의 클래스는 단 하나의 책임만 가져야 한다.

 

SRP는 클래스의 '변경'에 효과적으로 대응하기 위해 만들어진 원칙이다.

하나의 클래스가 위의 로직처럼 여러 가지의 책임을 갖고 있으면 다음과 같은 문제가 발생할 수 있다.

높은 결합도

 - 한 책임의 변경이 다른 책임에 영향을 미칠 수 있다. 

낮은 응집도

 - 서로 관련 없는 코드들이 섞여 있어 코드를 이해하기 어렵다. 

어려운 테스트

 - 하나의 기능을 테스트하기 위해 다른 기능과 관련된 불필요한 설정이 필요하다.

코드 재사용성 낮음

 - 여러 책임이 섞여 있어 원하는 기능만 떼어내어 사용하기 어렵다.

팀 협업이 어려움

 - 여러 개발자가 동일한 코드를 수정하면 충돌(Merge Comflict)이 자주 발생한다.

 

즉, 단일 책임 원칙(SRP)은 클래스를 작고, 응집도 높게, 이해하기 쉽게 만들어 유연하고 유지보수하기 좋은 소프트웨어를 만드는 핵심 원칙이다.

 

---

만들고 있던 포트폴리오에서 간단한 알림 기능이 필요했다.

동영상 파일이 업로드되었을 때, 그리고 후처리 기능들이 완료될 때마다 사용자에게 알림을 보내주고 싶었다.

 

그래서 정말 간단하게 만들었다.

/**
 * 컨트롤러보단 Service 단으로 내려서 객체 관리 필요
 * videoProcess 역시 대용량 알람 등의 문제 대비를 위해 pub/sub 구조에서 사용되는 서비스로 전환
 * 알람 필요 -> 메시지 발행 -> 메시지 수신 -> 알맞는 알람 제공
 * 다중화된 서버에서 알람 필요 -> 유저별 Emitter가 저장된 서버를 Redis 등에 기록 -> pub의 서버별 전용 큐에 메시지 발행
 */
@Slf4j
@RestController
@RequestMapping("/api/notification")
public class NotiRestController {

    private final Map<Long, SseEmitter> emitters = new ConcurrentHashMap<>();
    private static final Long SSE_CONNECTION_TIMEOUT = 60 * 60 * 1000L; // 1시간

    @GetMapping(path = "/subscribe/{userId}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter subscribeToVideo(@PathVariable(name = "userId") Long userId) {
        SseEmitter emitter = new SseEmitter(SSE_CONNECTION_TIMEOUT);
        emitters.put(userId, emitter);

        // 타임아웃 또는 완료 시 리스트에서 emitter 제거
        emitter.onTimeout(() -> {
            log.info("SSE Emitter 시간 제한: {}", userId);
            emitters.remove(userId);
        });
        emitter.onCompletion(() -> {
            log.info("SSE Emitter 완료: {}", userId);
            emitters.remove(userId);
        });
        emitter.onError(e -> {
            log.error("SSE Emitter 에러: {}: {}", userId, e.getMessage());
            emitters.remove(userId);
        });

        try {
            emitter.send(SseEmitter.event()
                    .id(String.valueOf(System.currentTimeMillis()))
                    .name("connection") // 이벤트 이름
                    .data("SSE 연결 성공 " + userId)
                    .reconnectTime(10000)); // 재연결 시간 (ms)
        } catch (IOException e) {
            log.error("ERROR {}: {}", userId, e.getMessage());
            emitter.completeWithError(e);
            emitters.remove(userId);
        }

        return emitter;
    }

    public void videoProcess(VideoDto videoDto) {
        long videoId = videoDto.getId();
        long userId = videoDto.getUploader().getId();
        String status = videoDto.getStatus().name();
        String message = String.format("VideoID %d 의 처리 상태는 %s 입니다.", videoId, status);

        SseEmitter emitter = emitters.get(userId);

        log.info("메시지 [{}]를 사용자 [{}]에게 전송합니다.", message, userId);
        if(emitter == null) {
            log.info("Sse 연결이 없습니다.");
            return;
        }
        try {
            emitter.send(SseEmitter.event()
                    .id(String.valueOf(System.currentTimeMillis()))
                    .name("upload_notification") // 이벤트 이름
                    .data(message)
                    .reconnectTime(10000));
            log.info("전송 성공: {}", userId);
        } catch (IOException e) {
            log.error("전송 실패 {}: {}, removing emitter.", videoId, e.getMessage());
        }
    }

 

---

컨트롤러 하나에서 SSE의 연결, 변수 관리, 응답 로직을 모두 갖고 있다.

위의 기능들을 하나씩 책임을 나눈 클래스로 분리하는 것이 좋아보인다.

그리고 이름도 마음에 안든다. 길긴 하지만 NotificationRestController로 변경해야겠다.

 

1) NotificationRestController : 사용자의 SSE 연결 요청을 수신하는 클래스

/**
 * videoProcess 역시 대용량 알람 등의 문제 대비를 위해 pub/sub 구조에서 사용되는 서비스로 전환
 * 다중화된 서버에서 알람 필요 -> 유저별 Emitter가 저장된 서버를 Redis 등에 기록 -> pub의 서버별 전용 큐에 메시지 발행
 */
@Slf4j
@RestController
@RequestMapping("/api/notification")
@RequiredArgsConstructor
public class NotificationRestController {
    private final NotificationService notificationService;

    @GetMapping(path = "/subscribe/{userId}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter subscribe(@PathVariable Long userId) {
        return notificationService.subscribe(userId);
    }
}

 

 

2) NotificationService : 실제 SSE와 관련된 로직(연결, 삭제, 알림 전송 등)

코드의 길이가 너무 길고, 재사용이 가능한 함수들이 보여서 private으로 분리했다.

@Slf4j
@Service
@RequiredArgsConstructor
public class NotificationServiceImpl implements NotificationService{

    private final EmitterRepository emitterRepository;
    private static final Long SSE_CONNECTION_TIMEOUT = 60 * 60 * 1000L; // 1시간

    public SseEmitter subscribe(Long userId) {
        SseEmitter emitter = createEmitter(userId);
        sendToClient(emitter, userId, "connection", "SSE 연결이 성공했습니다. [userId=" + userId + "]");
        return emitter;
    }

    public void sendNotification(Long userId, String message, String eventName) {
        SseEmitter emitter = emitterRepository.findByUserId(userId);
        if (emitter != null) {
            sendToClient(emitter, userId, eventName, message);
        } else {
            log.info("사용자 {}가 연결되어 있지 않습니다.", userId);
        }
    }

    public void videoProcess(VideoDto videoDto) {
        long videoId = videoDto.getId();
        long userId = videoDto.getUploader().getId();
        String status = videoDto.getStatus().name();
        String message = String.format("VideoID %d 의 처리 상태는 %s 입니다.", videoId, status);

        this.sendNotification(userId, message, "upload_notification");
    }



    private SseEmitter createEmitter(Long userId) {
        SseEmitter emitter = new SseEmitter(SSE_CONNECTION_TIMEOUT);
        emitterRepository.save(userId, emitter);

        emitter.onTimeout(() -> {
            log.info("SSE Emitter 시간 제한: {}", userId);
            emitterRepository.deleteById(userId);
        });
        emitter.onCompletion(() -> {
            log.info("SSE Emitter 완료: {}", userId);
            emitterRepository.deleteById(userId);
        });
        emitter.onError(e -> {
            log.error("SSE Emitter 에러: {}: {}", userId, e.getMessage());
            emitterRepository.deleteById(userId);
        });

        return emitter;
    }

    private void sendToClient(SseEmitter emitter, Long userId, String eventName, Object data) {
        try {
            emitter.send(SseEmitter.event()
                    .id(String.valueOf(System.currentTimeMillis())) // 고유 ID
                    .name(eventName)
                    .data(data)
                    .reconnectTime(10000));
            log.info("사용자 {} 에게 {}으로 {}가 전송되었습니다.", userId, eventName, data);
        } catch (IOException e) {
            emitterRepository.deleteById(userId);
            log.error("사용자 {}에게 알림 전송이 실패했습니다.: {}", userId, e.getMessage());
        }
    }
}

 

 

3) EmitterRepository : Emitter를 관리하는 서비스

@Slf4j
@Repository
public class EmitterRepository {
    private final Map<Long, SseEmitter> emitters = new ConcurrentHashMap<>();

    public SseEmitter save(Long userId, SseEmitter sseEmitter) {
        log.info("알림 Emitter 저장 userId :{}", userId);
        emitters.put(userId, sseEmitter);
        return sseEmitter;
    }

    public SseEmitter findByUserId(Long userId) {
        return emitters.get(userId);
    }

    public void deleteById(Long userId) {
        log.info("알림 Emitter 삭제 userId :{}", userId);
        emitters.remove(userId);
    }
}

 

 

---

속이 다 시원하다.

전체 코드량은 더 길어지긴 했지만, 만약 알림을 보내는 기능이 더 늘어나게 되면 이 분리한 날을 칭찬하게 될 것이다.

다음으로는 분산 환경을 대비해서 Pub/Sub 구조로 변경해야 겠다.