Refactor: 거래소 확장을 위한 커넥터 인터페이스 및 설정 구조 추가

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: SimoKii

src/main/java/com/rabbittick/streamer/MarketIngestStreamerApplication.java

@@ -2,10 +2,12 @@
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.context.properties.ConfigurationPropertiesScan;
 import org.springframework.scheduling.annotation.EnableScheduling;
 
 @SpringBootApplication
 @EnableScheduling
+@ConfigurationPropertiesScan
 public class MarketIngestStreamerApplication {
 
 	public static void main(String[] args) {

src/main/java/com/rabbittick/streamer/config/ExchangeProperties.java

@@ -0,0 +1,164 @@
+package com.rabbittick.streamer.config;
+
+import java.net.URI;
+import java.time.Duration;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * 다중 거래소 WebSocket 커넥터 설정을 바인딩하는 프로퍼티 클래스.
+ *
+ * <p>{@code application.yml}의 {@code exchanges.configs} 하위 설정을 거래소 식별자를
+ * 키로, {@link ExchangeConfig}를 값으로 하는 맵에 바인딩한다.
+ * 거래소 식별자는 YAML 키(소문자)이며, 커넥터 내부에서 대문자로 변환하여 사용한다.
+ *
+ * <p>YAML 설정 예시:
+ * <pre>{@code
+ * exchanges:
+ *   configs:
+ *     upbit:
+ *       enabled: true
+ *       websocket-uri: wss://api.upbit.com/websocket/v1
+ *       ping-interval-seconds: 60
+ *       max-markets-per-connection: 100
+ *       processing-concurrency: 32
+ *       environment: fetched
+ *       market-prefixes: [KRW, BTC, USDT]
+ *       data-types:
+ *         ticker: true
+ *         trade: true
+ *         orderbook: true
+ *     binance:
+ *       enabled: false
+ *       websocket-uri: wss://stream.binance.com:9443/ws
+ *       market-prefixes: [USDT]
+ *       data-types:
+ *         ticker: true
+ *         trade: false
+ *         orderbook: false
+ * }</pre>
+ *
+ * <p>새로운 거래소를 추가하려면 {@code exchanges.configs} 하위에 거래소 식별자를 키로
+ * 새 항목을 추가하고, 해당 {@link com.rabbittick.streamer.connector.ExchangeConnector}
+ * 구현체를 작성한다.
+ */
+@Getter
+@Setter
+@ConfigurationProperties("exchanges")
+public class ExchangeProperties {
+
+    /**
+     * 거래소 식별자(소문자)를 키로, 개별 거래소 설정을 값으로 가지는 맵.
+     *
+     * <p>키는 YAML에 선언된 소문자 거래소 이름이다.
+     * 커넥터에서 라우팅 키 생성 시 {@code String.toUpperCase()}로 변환하여 사용한다.
+     */
+    private Map<String, ExchangeConfig> configs = new LinkedHashMap<>();
+
+    /**
+     * 단일 거래소의 WebSocket 연결 및 수집 설정을 담는 클래스.
+     */
+    @Getter
+    @Setter
+    public static class ExchangeConfig {
+
+        /**
+         * 이 커넥터의 활성화 여부.
+         *
+         * <p>{@code false}이면 {@code ConnectorManager}가 {@code start()} 호출 없이
+         * 건너뛴다. 기본값은 {@code false}로, 명시적으로 활성화해야 연결이 시작된다.
+         */
+        private boolean enabled = false;
+
+        /**
+         * 거래소 WebSocket 서버 URI.
+         *
+         * <p>예: {@code wss://api.upbit.com/websocket/v1}
+         */
+        private URI websocketUri;
+
+        /**
+         * 연결 유지를 위한 Ping 전송 주기 (초).
+         *
+         * <p>거래소별 권장 주기에 맞게 설정한다. 기본값은 60초.
+         */
+        private int pingIntervalSeconds = 60;
+
+        /**
+         * 단일 WebSocket 연결당 최대 구독 마켓코드 수.
+         *
+         * <p>이 값을 초과하면 마켓코드를 청크로 나눠 복수의 연결을 유지한다.
+         * 거래소 API 제약에 맞게 조정한다. 기본값은 100.
+         */
+        private int maxMarketsPerConnection = 100;
+
+        /**
+         * 메시지 처리 시 동시에 실행할 내부 Mono의 최대 개수.
+         *
+         * <p>파싱·변환·발행 작업은 {@code boundedElastic} 스케줄러에서
+         * 이 값을 상한으로 병렬 처리된다. 기본값은 32.
+         */
+        private int processingConcurrency = 32;
+
+        /**
+         * 사용할 마켓코드 환경 이름.
+         *
+         * <p>마켓코드 YAML 파일 내 {@code env} 섹션의 키와 일치해야 한다.
+         * 예: {@code development}, {@code production}, {@code fetched}.
+         * 기본값은 {@code development}.
+         */
+        private String environment = "development";
+
+        /**
+         * 구독할 마켓 prefix 목록.
+         *
+         * <p>각 prefix에 해당하는 마켓코드를 {@code loadMarketCodes()}에서 로드한다.
+         * 예: {@code [KRW, BTC, USDT]}.
+         * 기본값은 {@code [KRW]}.
+         */
+        private List<String> marketPrefixes = List.of("KRW");
+
+        /**
+         * 수집할 데이터 타입별 활성화 설정.
+         *
+         * <p>비활성화된 타입의 메시지는 수신하더라도 무시한다.
+         * 최소 하나의 타입이 활성화되어 있어야 한다.
+         */
+        private DataTypesConfig dataTypes = new DataTypesConfig();
+
+        /**
+         * {@code pingIntervalSeconds}를 {@link Duration}으로 변환하여 반환한다.
+         *
+         * <p>{@link com.rabbittick.streamer.connector.ExchangeConnector#getPingInterval()}
+         * 구현체에서 이 메서드를 사용하여 Ping 주기를 가져온다.
+         *
+         * @return Ping 전송 주기
+         */
+        public Duration getPingInterval() {
+            return Duration.ofSeconds(pingIntervalSeconds);
+        }
+    }
+
+    /**
+     * 수집할 데이터 타입별 활성화 여부를 담는 클래스.
+     */
+    @Getter
+    @Setter
+    public static class DataTypesConfig {
+
+        /** 현재가(ticker) 데이터 수집 활성화 여부. 기본값은 {@code true}. */
+        private boolean ticker = true;
+
+        /** 거래 체결(trade) 데이터 수집 활성화 여부. 기본값은 {@code false}. */
+        private boolean trade = false;
+
+        /** 호가(orderbook) 데이터 수집 활성화 여부. 기본값은 {@code false}. */
+        private boolean orderbook = false;
+    }
+}

src/main/java/com/rabbittick/streamer/connector/ExchangeConnector.java

@@ -0,0 +1,111 @@
+package com.rabbittick.streamer.connector;
+
+import java.net.URI;
+import java.time.Duration;
+import java.util.List;
+
+import org.springframework.web.reactive.socket.WebSocketSession;
+
+import reactor.core.publisher.Mono;
+
+/**
+ * 거래소 WebSocket 커넥터의 표준 계약을 정의하는 인터페이스.
+ *
+ * <p>주요 책임:
+ * <ul>
+ *   <li>거래소별 연결 정보(URI, Ping 주기) 제공</li>
+ *   <li>WebSocket 세션에 거래소 전용 구독 메시지 전송</li>
+ *   <li>수신된 원시 JSON 메시지 파싱 및 서비스 계층 전달</li>
+ *   <li>구독 대상 마켓코드 목록 로드</li>
+ * </ul>
+ *
+ * <p>각 거래소 구현체는 이 인터페이스를 통해 거래소별 프로토콜 차이를
+ * 캡슐화하고, {@code AbstractExchangeConnector}가 제공하는 공통 연결
+ * 생명주기(연결 시작, 청크 분할, Ping, 재연결) 위에서 동작한다.
+ *
+ * <p>새로운 거래소를 추가하려면 이 인터페이스와
+ * {@code AbstractExchangeConnector}를 함께 구현한다.
+ */
+public interface ExchangeConnector {
+
+    /**
+     * 거래소 식별자를 반환한다.
+     *
+     * <p>반환값은 RabbitMQ 라우팅 키의 첫 번째 세그먼트로 사용된다.
+     * 예: {@code "UPBIT"} → 라우팅 키 {@code upbit.ticker.KRW-BTC}
+     *
+     * @return 대문자 거래소 식별자 (예: "UPBIT", "BINANCE")
+     */
+    String getExchangeName();
+
+    /**
+     * 거래소 WebSocket 서버 URI를 반환한다.
+     *
+     * @return WebSocket 서버 URI
+     */
+    URI getWebSocketUri();
+
+    /**
+     * 연결 유지를 위한 Ping 전송 주기를 반환한다.
+     *
+     * <p>거래소마다 권장 Ping 주기가 다를 수 있으므로 구현체가 직접 정의한다.
+     *
+     * @return Ping 전송 주기
+     */
+    Duration getPingInterval();
+
+    /**
+     * 이 커넥터의 활성화 여부를 반환한다.
+     *
+     * <p>{@code false}를 반환하는 커넥터는 {@code ConnectorManager}에 의해
+     * {@link #start()} 호출 없이 건너뛴다.
+     *
+     * @return 활성화 여부
+     */
+    boolean isEnabled();
+
+    /**
+     * WebSocket 연결을 시작하고 데이터 스트리밍을 개시한다.
+     *
+     * <p>{@code ConnectorManager}가 {@code ApplicationReadyEvent} 수신 후
+     * 활성화된 커넥터에 대해 호출한다. 연결 실패 시 지수 백오프 방식으로
+     * 재연결을 시도하는 것은 {@code AbstractExchangeConnector}의 공통 구현에서 처리한다.
+     */
+    void start();
+
+    /**
+     * WebSocket 세션에 거래소 전용 구독 메시지를 전송한다.
+     *
+     * <p>거래소마다 구독 메시지 포맷이 다르므로 구현체가 직접 정의한다.
+     * 예: Upbit는 {@code [{"ticket":…}, {"type":"ticker","codes":[…]}]} 형태의 JSON을 전송한다.
+     *
+     * @param session 데이터를 수신할 WebSocket 세션
+     * @param marketCodes 구독할 마켓코드 목록
+     * @return 구독 메시지 전송 완료를 나타내는 Mono
+     */
+    Mono<Void> sendSubscription(WebSocketSession session, List<String> marketCodes);
+
+    /**
+     * 수신된 원시 JSON 메시지를 파싱하고 처리한다.
+     *
+     * <p>메시지 타입 식별 방식은 거래소마다 다르므로 구현체가 직접 정의한다.
+     * 예: Upbit는 {@code "ty"} 필드로 {@code ticker} / {@code trade} / {@code orderbook}을 구분한다.
+     *
+     * <p>파싱할 수 없는 메시지(연결 수락 응답 등)는 빈 Mono를 반환하여 무시한다.
+     *
+     * @param rawJson WebSocket으로부터 수신된 원시 JSON 문자열
+     * @return 처리 완료를 나타내는 Mono. 파싱 불가 메시지의 경우 빈 Mono
+     */
+    Mono<Void> handleMessage(String rawJson);
+
+    /**
+     * 구독할 마켓코드 목록을 로드한다.
+     *
+     * <p>설정 파일 또는 외부 API에서 마켓코드를 조회하는 방식은
+     * 구현체가 직접 정의한다. 로드된 목록은 {@code AbstractExchangeConnector}의
+     * 청크 분할 로직에 의해 연결당 최대 구독 수 이하로 나뉘어 처리된다.
+     *
+     * @return 구독할 마켓코드 목록
+     */
+    List<String> loadMarketCodes();
+}

src/main/resources/application.yml

@@ -31,6 +31,24 @@ upbit:
       orderbook:
         enabled: true
 
+exchanges:
+  configs:
+    upbit:
+      enabled: true
+      websocket-uri: wss://api.upbit.com/websocket/v1
+      ping-interval-seconds: 60
+      max-markets-per-connection: 100
+      processing-concurrency: 32
+      environment: fetched
+      market-prefixes:
+        - KRW
+        - BTC
+        - USDT
+      data-types:
+        ticker: true
+        trade: true
+        orderbook: true
+
 rabbitmq:
   exchange:
     market-data: market-data.exchange