Merge pull request #14 from f-lab-edu/feature/12

[#12] Websocket 모듈 설정

api/build.gradle

@@ -7,6 +7,7 @@ plugins {
 dependencies {
     implementation project(':common')
     implementation project(':service')
+    implementation project(':collector')
 
     implementation 'org.springframework.boot:spring-boot-starter'
     implementation 'org.springframework.boot:spring-boot-starter-web'

application.yml

@@ -22,3 +22,7 @@ jwt:
   secret-key: "${JWT_SECRET_KEY}"
   access-token-validity-in-seconds: 600
   refresh-token-validity-in-seconds: 1209600
+
+logging:
+  level:
+    com.whalewatch.service: DEBUG

build.gradle

@@ -13,6 +13,10 @@ java {
     }
 }
 
+repositories {
+    mavenCentral()  // Maven Central 레포지토리에서 라이브러리 다운로드
+}
+
 subprojects {
     // 모든 서브프로젝트에 공통 적용할 설정
     apply plugin: 'java'

collector/build.gradle

@@ -0,0 +1,17 @@
+plugins {
+    id 'java'
+    id 'org.springframework.boot' version '3.3.4'
+    id 'io.spring.dependency-management' version '1.1.6'
+}
+
+dependencies {
+    // SpringBoot , Websocket
+    implementation 'org.springframework.boot:spring-boot-starter-websocket'
+
+    // Jackson
+    implementation 'com.fasterxml.jackson.core:jackson-databind'
+
+    // logging
+    implementation 'org.slf4j:slf4j-api:2.0.7'
+    runtimeOnly 'ch.qos.logback:logback-classic:1.4.5'
+}

collector/src/main/java/com/whalewatch/CollectorApplication.java

@@ -0,0 +1,11 @@
+package com.whalewatch;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class CollectorApplication {
+    public static void main(String[] args) {
+        SpringApplication.run(CollectorApplication.class, args);
+    }
+}

collector/src/main/java/com/whalewatch/TradeDto.java

@@ -0,0 +1,106 @@
+package com.whalewatch;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class TradeDto {
+
+    private String type;
+    private String code;
+
+    @JsonProperty("trade_price")
+    private Double tradePrice;
+
+    @JsonProperty("trade_volume")
+    private Double tradeVolume;
+
+    @JsonProperty("ask_bid")
+    private String askBid;
+
+    @JsonProperty("change_price")
+    private Double changePrice;
+
+    private Long timestamp;
+    @JsonProperty("trade_timestamp")
+    private Long tradeTimestamp;
+
+    public String getType() {
+        return type;
+    }
+
+    public void setType(String type) {
+        this.type = type;
+    }
+
+    public String getCode() {
+        return code;
+    }
+
+    public void setCode(String code) {
+        this.code = code;
+    }
+
+    public Double getTradePrice() {
+        return tradePrice;
+    }
+
+    public void setTradePrice(Double tradePrice) {
+        this.tradePrice = tradePrice;
+    }
+
+    public Double getTradeVolume() {
+        return tradeVolume;
+    }
+
+    public void setTradeVolume(Double tradeVolume) {
+        this.tradeVolume = tradeVolume;
+    }
+
+    public String getAskBid() {
+        return askBid;
+    }
+
+    public void setAskBid(String askBid) {
+        this.askBid = askBid;
+    }
+
+
+    public Double getChangePrice() {
+        return changePrice;
+    }
+
+    public void setChangePrice(Double changePrice) {
+        this.changePrice = changePrice;
+    }
+
+    public Long getTimestamp() {
+        return timestamp;
+    }
+
+    public void setTimestamp(Long timestamp) {
+        this.timestamp = timestamp;
+    }
+
+    public Long getTradeTimestamp() {
+        return tradeTimestamp;
+    }
+
+    public void setTradeTimestamp(Long tradeTimestamp) {
+        this.tradeTimestamp = tradeTimestamp;
+    }
+
+    @Override
+    public String toString() {
+        return "TradeDto{" +
+                "type='" + type + '\'' +
+                ", code='" + code + '\'' +
+                ", tradePrice=" + tradePrice +
+                ", tradeVolume=" + tradeVolume +
+                ", askBid='" + askBid + '\'' +
+                ", changePrice=" + changePrice +
+                ", timestamp=" + timestamp +
+                ", tradeTimestamp=" + tradeTimestamp +
+                '}';
+    }
+}

collector/src/main/java/com/whalewatch/service/FilteringService.java

@@ -0,0 +1,36 @@
+package com.whalewatch.service;
+
+import com.whalewatch.TradeDto;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+@Service
+public class FilteringService {
+
+    private static final Logger log = LoggerFactory.getLogger(ParsingService.class);
+    private final Map<String, Double> volumeThresholdMap = new ConcurrentHashMap<>();
+
+    public FilteringService() {
+        // 테스트용 초기값 설정
+        volumeThresholdMap.put("KRW-BTC", 0.2);
+        volumeThresholdMap.put("KRW-ETH", 5.0);
+    }
+
+    public double getVolumeThreshold(String coin) {
+        return volumeThresholdMap.get(coin);
+    }
+
+    public boolean shouldAlert(TradeDto dto) {
+        if (dto.getCode() == null || dto.getTradeVolume() == null) {
+            return false;
+        }
+        double threshold = getVolumeThreshold(dto.getCode());
+        return dto.getTradeVolume() > threshold;
+    }
+
+
+}

collector/src/main/java/com/whalewatch/service/ParsingService.java

@@ -0,0 +1,42 @@
+package com.whalewatch.service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.whalewatch.TradeDto;
+import jakarta.annotation.PostConstruct;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+@Service
+public class ParsingService {
+
+    private static final Logger log = LoggerFactory.getLogger(ParsingService.class);
+
+    private final ObjectMapper objectMapper;
+    private final FilteringService filteringService;
+
+    public ParsingService(ObjectMapper objectMapper,
+                               FilteringService filteringService) {
+        this.objectMapper = objectMapper;
+        this.filteringService = filteringService;
+    }
+
+    public void parsingMessage(String jsonMessage) {
+        try {
+            // JSON → TradeDto
+            TradeDto tradeDto = objectMapper.readValue(jsonMessage, TradeDto.class);
+
+            // 필터링
+            if (filteringService.shouldAlert(tradeDto)) {
+                log.info("[ALERT] Coin={}, volume={} exceeded threshold => {}",
+                        tradeDto.getCode(),
+                        tradeDto.getTradeVolume(),
+                        tradeDto);
+            }
+
+        } catch (Exception e) {
+            log.error("Failed to parse JSON message: {}", jsonMessage, e);  // error 로그로 변경하여 더욱 눈에 띄게 함
+        }
+    }
+
+}

collector/src/main/java/com/whalewatch/service/WebSocketListener.java

@@ -0,0 +1,65 @@
+package com.whalewatch.service;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.web.socket.BinaryMessage;
+import org.springframework.web.socket.CloseStatus;
+import org.springframework.web.socket.TextMessage;
+import org.springframework.web.socket.WebSocketSession;
+import org.springframework.web.socket.handler.BinaryWebSocketHandler;
+
+import java.nio.charset.StandardCharsets;
+
+public class WebSocketListener extends BinaryWebSocketHandler {
+    private static final Logger log = LoggerFactory.getLogger(WebSocketListener.class);
+
+    private final ParsingService parsingService;
+
+    public WebSocketListener(ParsingService parsingService) {
+        this.parsingService = parsingService;
+    }
+
+    //연결
+    @Override
+    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
+        log.info("Listener Connected: {}", session.getRemoteAddress());
+
+        String subscriptionJson = "[" +
+                "{\"ticket\":\"test\"}," +
+                "{\"type\":\"trade\",\"codes\":[\"KRW-BTC\",\"KRW-ETH\"]}," +
+                "{\"format\":\"DEFAULT\"}" +
+                "]";
+        session.sendMessage(new TextMessage(subscriptionJson));
+        log.info("message: {}", subscriptionJson);
+    }
+
+    //메시지 수신
+    @Override
+    protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) throws Exception {
+        // Binary 데이터를 String으로 변환
+        String payload = new String(message.getPayload().array(), StandardCharsets.UTF_8);
+
+        try {
+            parsingService.parsingMessage(payload);  // JSON 변환 및 필터링
+        } catch (Exception e) {
+            log.error("Error parsing WebSocket message: {}", payload, e);
+        }
+    }
+
+    //에러 발생
+    @Override
+    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
+        log.error("Error: ", exception);
+    }
+
+    //연결 종료
+    @Override
+    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
+        log.info("Closed Status: {}", status);
+    }
+
+
+
+
+
+}

collector/src/main/java/com/whalewatch/service/WebsocketService.java

@@ -0,0 +1,52 @@
+package com.whalewatch.service;
+
+import jakarta.annotation.PostConstruct;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+import org.springframework.web.socket.WebSocketHttpHeaders;
+import org.springframework.web.socket.WebSocketSession;
+import org.springframework.web.socket.client.WebSocketClient;
+import org.springframework.web.socket.client.standard.StandardWebSocketClient;
+
+import java.net.URI;
+import java.util.Collections;
+import java.util.concurrent.ExecutionException;
+
+@Service
+public class WebsocketService {
+    private static final Logger log = LoggerFactory.getLogger(WebsocketService.class);
+
+    private final ParsingService parsingService;
+    private WebSocketSession session;
+
+    public WebsocketService(ParsingService parsingService) {
+        this.parsingService = parsingService;
+    }
+
+    @PostConstruct
+    public void init() {
+        startConnection();
+    }
+
+    public void startConnection() {
+        try {
+            WebSocketClient client = new StandardWebSocketClient();
+            WebSocketListener listener = new WebSocketListener(parsingService);
+            WebSocketHttpHeaders headers = new WebSocketHttpHeaders();
+
+            headers.setSecWebSocketProtocol(Collections.singletonList("json"));
+
+            URI uri = new URI("wss://api.upbit.com/websocket/v1");
+
+            session = client.doHandshake(listener, headers, uri).get();
+
+            log.info("WebSocket service : {}", uri);
+        } catch (InterruptedException | ExecutionException e) {
+            log.error("Failed to WebSocket connection", e);
+        } catch (Exception ex) {
+            log.error("Unexpected error", ex);
+        }
+    }
+
+}