✨ [Feat] 웹소켓 재연결 로직 추가

src/main/java/com/example/scoi/domain/websocket/WebsocketConnect.java

@@ -1,27 +1,107 @@
 package com.example.scoi.domain.websocket;
 
+import com.example.scoi.domain.websocket.code.WebsocketErrorCode;
+import com.example.scoi.domain.websocket.event.WebsocketConnectionEvent.ConnectedEvent;
+import com.example.scoi.domain.websocket.event.WebsocketConnectionEvent.DisconnectedEvent;
+import com.example.scoi.domain.websocket.exception.WebsocketException;
 import com.example.scoi.domain.websocket.handler.UpbitTickerHandler;
+import jakarta.annotation.PreDestroy;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.boot.context.event.ApplicationReadyEvent;
 import org.springframework.context.event.EventListener;
+import org.springframework.retry.annotation.Backoff;
+import org.springframework.retry.annotation.Recover;
+import org.springframework.retry.annotation.Retryable;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import org.springframework.web.socket.client.WebSocketClient;
 
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
 @Service
 @RequiredArgsConstructor
 @Slf4j
 public class WebsocketConnect {
 
     private final WebSocketClient webSocketClient;
     private final UpbitTickerHandler upbitTickerHandler;
+    private final ObjectProvider<WebsocketConnect> selfProvider; // AOP 우회 방지를 위한 프록시 의존성
 
     private static final String PUBLIC_URL = "wss://api.upbit.com/websocket/v1";
 
-    // 실시간 가격 변동 체크
+    // 워치독(Half-Open 헬스체크) 전용 스케줄러
+    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
+    private ScheduledFuture<?> watchdogTask;
+
     @EventListener(ApplicationReadyEvent.class)
-    public void connect(){
-        log.info("[ Websocket ]: 디페깅 알고리즘 구동 시작...");
-        webSocketClient.execute(upbitTickerHandler, PUBLIC_URL);
+    public void init() {
+        log.info("[ Websocket ]: 애플리케이션 준비 완료.");
+        selfProvider.getObject().connectWithRetry();
+    }
+
+    @Async
+    @Retryable(
+            retryFor = WebsocketException.class,
+            maxAttempts = 10,
+            backoff = @Backoff(delay = 1000, multiplier = 2, maxDelay = 60000)
+    )
+    public void connectWithRetry() {
+        log.info("[ Websocket ]: 업비트 웹소켓 연결 시도 중... URL: {}", PUBLIC_URL);
+        try {
+            // execute()의 리턴값인 Future를 대기(get)하여 타임아웃이나 연결 거부 시 예외를 발생시키도록 유도
+            webSocketClient.execute(upbitTickerHandler, PUBLIC_URL).get(5, TimeUnit.SECONDS);
+        } catch (Exception e) {
+            log.error("[ Websocket ]: 웹소켓 연결 실패: {}", e.getMessage());
+            // 예외를 던져서 @Retryable이 AOP 기반 백오프 및 재시도를 수행하도록 트리거
+            throw new WebsocketException(WebsocketErrorCode.UPBIT_WEBSOCKET_ERROR);
+        }
+    }
+
+    @Recover
+    public void recover(Exception e) {
+        log.error("[ Websocket ]: 웹소켓 재연결 최대 시도 횟수(10회) 초과");
+
+        // 2시간뒤 다시 시도
+        scheduler.schedule(() -> {
+            log.info("[ Websocket ]: 시스템 복구를 위해 웹소켓 재연결 사이클을 다시 시작합니다.");
+            selfProvider.getObject().connectWithRetry();
+        }, 2, TimeUnit.HOURS);
+    }
+
+    @EventListener
+    public void onConnected(ConnectedEvent event) {
+        log.info("[ Websocket ]: 이벤트 수신 - 웹소켓 연결 활성화 완료.");
+
+        if (watchdogTask != null && !watchdogTask.isCancelled()) {
+            watchdogTask.cancel(false);
+        }
+        // 10초마다 마지막 수신 시간 검사
+        watchdogTask = scheduler.scheduleAtFixedRate(() -> {
+            long idleTime = System.currentTimeMillis() - upbitTickerHandler.getLastMessageTime();
+            if (idleTime > 30_000) { // 30초 초과 시
+                log.error("[ Websocket ]: 30초간 업비트 응답이 없습니다...");
+                upbitTickerHandler.closeSession(); // DisconnectedEvent 유발
+            }
+        }, 10, 10, TimeUnit.SECONDS);
+    }
+
+    @EventListener
+    public void onDisconnected(DisconnectedEvent event) {
+        log.warn("[ Websocket ]: 이벤트 수신 - 웹소켓 연결 해제. 재연결 시도");
+        if (watchdogTask != null) {
+            watchdogTask.cancel(false);
+        }
+        selfProvider.getObject().connectWithRetry();
+    }
+
+    @PreDestroy
+    public void shutdown() {
+        log.info("[ Websocket ]: 워치독 스케줄러 종료 중...");
+        scheduler.shutdown();
     }
 }

src/main/java/com/example/scoi/domain/websocket/code/WebsocketErrorCode.java

@@ -0,0 +1,20 @@
+package com.example.scoi.domain.websocket.code;
+
+import com.example.scoi.global.apiPayload.code.BaseErrorCode;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import org.springframework.http.HttpStatus;
+
+@Getter
+@RequiredArgsConstructor
+public enum WebsocketErrorCode implements BaseErrorCode {
+
+    UPBIT_WEBSOCKET_ERROR(HttpStatus.INTERNAL_SERVER_ERROR,
+            "WEBSOCKET500_1",
+            "웹소켓 연결 도중 에러가 발생했습니다."),
+    ;
+
+    private final HttpStatus status;
+    private final String code;
+    private final String message;
+}

src/main/java/com/example/scoi/domain/websocket/event/WebsocketConnectionEvent.java

@@ -0,0 +1,6 @@
+package com.example.scoi.domain.websocket.event;
+
+public interface WebsocketConnectionEvent {
+    record ConnectedEvent() implements WebsocketConnectionEvent {}
+    record DisconnectedEvent() implements WebsocketConnectionEvent {}
+}

src/main/java/com/example/scoi/domain/websocket/exception/WebsocketException.java

@@ -0,0 +1,10 @@
+package com.example.scoi.domain.websocket.exception;
+
+import com.example.scoi.global.apiPayload.code.BaseErrorCode;
+import com.example.scoi.global.apiPayload.exception.ScoiException;
+
+public class WebsocketException extends ScoiException {
+    public WebsocketException(BaseErrorCode code) {
+        super(code);
+    }
+}

src/main/java/com/example/scoi/domain/websocket/handler/UpbitTickerHandler.java

@@ -2,14 +2,19 @@
 
 import com.example.scoi.domain.websocket.converter.WebSocketConverter;
 import com.example.scoi.domain.websocket.dto.UpbitResDTO;
+import com.example.scoi.domain.websocket.event.WebsocketConnectionEvent.ConnectedEvent;
+import com.example.scoi.domain.websocket.event.WebsocketConnectionEvent.DisconnectedEvent;
 import com.example.scoi.domain.websocket.service.WebSocketService;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.PropertyNamingStrategies;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.messaging.simp.SimpMessageSendingOperations;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
 import org.springframework.web.socket.BinaryMessage;
+import org.springframework.web.socket.CloseStatus;
 import org.springframework.web.socket.WebSocketSession;
 import org.springframework.web.socket.handler.BinaryWebSocketHandler;
 
@@ -19,18 +24,27 @@
 
 @Component
 @RequiredArgsConstructor
+@Slf4j
 public class UpbitTickerHandler extends BinaryWebSocketHandler {
 
     private final SimpMessageSendingOperations simpMessageSendingOperations;
     private final WebSocketService webSocketService;
+    private final ApplicationEventPublisher eventPublisher;
 
     private static final ObjectMapper objectMapper = new ObjectMapper()
             .setPropertyNamingStrategy(PropertyNamingStrategies.LOWER_CAMEL_CASE);
 
+    private volatile long lastMessageTime;
+    private volatile WebSocketSession currentSession;
+
     @Override
     public void afterConnectionEstablished(
             WebSocketSession session
     ) throws IOException {
+        this.currentSession = session;
+        this.lastMessageTime = System.currentTimeMillis();
+        log.info("[ Websocket ]: 업비트 웹소켓 연결 성공.");
+        eventPublisher.publishEvent(new ConnectedEvent());
         session.sendMessage(WebSocketConverter.toGetCoinPrice(List.of("KRW-USDT","KRW-USDC")));
     }
 
@@ -39,6 +53,8 @@ public void handleBinaryMessage(
             WebSocketSession session,
             BinaryMessage message
     ) throws IOException {
+        this.lastMessageTime = System.currentTimeMillis(); // Heartbeat 갱신
+
         String converted = new String(message.getPayload().array(), StandardCharsets.UTF_8);
         converted = converted.replace("[","").replace("]","");
         publish(converted);
@@ -50,4 +66,31 @@ public void handleBinaryMessage(
     protected void publish(String message) {
         simpMessageSendingOperations.convertAndSend("/topic/ticker", message);
     }
+
+    @Override
+    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
+        log.warn("[ Websocket ]: 업비트 웹소켓 연결이 종료되었습니다. CloseStatus: {}", status);
+        eventPublisher.publishEvent(new DisconnectedEvent());
+    }
+
+    @Override
+    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
+        log.error("[ Websocket ]: 웹소켓 에러 발생: {}", exception.getMessage());
+        eventPublisher.publishEvent(new DisconnectedEvent());
+    }
+
+    public long getLastMessageTime() {
+        return lastMessageTime;
+    }
+
+    public void closeSession() {
+        if (currentSession != null && currentSession.isOpen()) {
+            try {
+                log.warn("[ Websocket ]: 하프 오픈(Half-open) 상태 감지.");
+                currentSession.close();
+            } catch (IOException e) {
+                log.error("[ Websocket ]: 세션 강제 종료 중 에러 발생: {}", e.getMessage());
+            }
+        }
+    }
 }

src/main/java/com/example/scoi/domain/websocket/service/WebSocketService.java

@@ -65,8 +65,6 @@ public void ticker(UpbitResDTO.Ticker dto) {
             BigDecimal devNumerator = new BigDecimal(Math.abs(dto.tp() - baseline));
             BigDecimal dev = devNumerator.divide(BigDecimal.valueOf(Math.max(baseline, EPS)), RoundingMode.HALF_UP);
 
-            log.info("coin: {}, dev: {}, baseline: {}, tp: {}", code, dev, baseline, dto.tp());
-
             // duration 누적/리셋
             if (dev.compareTo(BigDecimal.valueOf(DEV_TH)) >= 0){
                 duration += deltaSec.doubleValue();