diff --git a/src/main/java/com/example/scoi/domain/websocket/WebsocketConnect.java b/src/main/java/com/example/scoi/domain/websocket/WebsocketConnect.java index 3bebb0b..b2a338a 100644 --- a/src/main/java/com/example/scoi/domain/websocket/WebsocketConnect.java +++ b/src/main/java/com/example/scoi/domain/websocket/WebsocketConnect.java @@ -1,13 +1,28 @@ 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 @@ -15,13 +30,78 @@ public class WebsocketConnect { private final WebSocketClient webSocketClient; private final UpbitTickerHandler upbitTickerHandler; + private final ObjectProvider 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(); } } diff --git a/src/main/java/com/example/scoi/domain/websocket/code/WebsocketErrorCode.java b/src/main/java/com/example/scoi/domain/websocket/code/WebsocketErrorCode.java new file mode 100644 index 0000000..acacb6d --- /dev/null +++ b/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; +} diff --git a/src/main/java/com/example/scoi/domain/websocket/event/WebsocketConnectionEvent.java b/src/main/java/com/example/scoi/domain/websocket/event/WebsocketConnectionEvent.java new file mode 100644 index 0000000..bca93e0 --- /dev/null +++ b/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 {} +} diff --git a/src/main/java/com/example/scoi/domain/websocket/exception/WebsocketException.java b/src/main/java/com/example/scoi/domain/websocket/exception/WebsocketException.java new file mode 100644 index 0000000..19a64d9 --- /dev/null +++ b/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); + } +} diff --git a/src/main/java/com/example/scoi/domain/websocket/handler/UpbitTickerHandler.java b/src/main/java/com/example/scoi/domain/websocket/handler/UpbitTickerHandler.java index e9a9e3c..e4e5003 100644 --- a/src/main/java/com/example/scoi/domain/websocket/handler/UpbitTickerHandler.java +++ b/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()); + } + } + } } diff --git a/src/main/java/com/example/scoi/domain/websocket/service/WebSocketService.java b/src/main/java/com/example/scoi/domain/websocket/service/WebSocketService.java index 53bc654..e4e404c 100644 --- a/src/main/java/com/example/scoi/domain/websocket/service/WebSocketService.java +++ b/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();