[fix] SSE 문제 해결

src/main/java/codesquad/fineants/domain/portfolio_holding/PortFolioHoldingRepository.java

@@ -3,6 +3,7 @@
 import java.util.List;
 
 import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Query;
 
 import codesquad.fineants.domain.portfolio.Portfolio;
 
@@ -12,4 +13,6 @@ public interface PortFolioHoldingRepository extends JpaRepository<PortfolioHoldi
 
 	List<PortfolioHolding> findAllByPortfolio(Portfolio portfolio);
 
+	@Query("select distinct s.tickerSymbol from PortfolioHolding p inner join Stock s on p.stock.tickerSymbol = s.tickerSymbol")
+	List<String> findAllTickerSymbol();
 }

src/main/java/codesquad/fineants/spring/aop/SimpleLogAop.java

@@ -10,13 +10,12 @@
 import org.aspectj.lang.annotation.Before;
 import org.aspectj.lang.annotation.Pointcut;
 import org.aspectj.lang.reflect.MethodSignature;
-import org.springframework.stereotype.Component;
 
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
 @Aspect
-@Component
+// @Component
 public class SimpleLogAop {
 
 	@Pointcut("execution(* codesquad.fineants..*.*(..))")

src/main/java/codesquad/fineants/spring/api/kis/KisService.java

@@ -1,7 +1,6 @@
 package codesquad.fineants.spring.api.kis;
 
 import java.time.LocalDateTime;
-import java.util.Collection;
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
@@ -14,12 +13,8 @@
 import org.springframework.messaging.simp.SimpMessagingTemplate;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
-import org.springframework.transaction.annotation.Transactional;
 
-import codesquad.fineants.domain.portfolio.Portfolio;
-import codesquad.fineants.domain.portfolio.PortfolioRepository;
-import codesquad.fineants.domain.portfolio_holding.PortfolioHolding;
-import codesquad.fineants.domain.stock.Stock;
+import codesquad.fineants.domain.portfolio_holding.PortFolioHoldingRepository;
 import codesquad.fineants.spring.api.kis.client.KisClient;
 import codesquad.fineants.spring.api.kis.manager.CurrentPriceManager;
 import codesquad.fineants.spring.api.kis.manager.KisAccessTokenManager;
@@ -37,10 +32,10 @@
 @Service
 public class KisService {
 	private static final String SUBSCRIBE_PORTFOLIO_HOLDING_FORMAT = "/sub/portfolio/%d";
-	private static final ScheduledExecutorService executorService = Executors.newScheduledThreadPool(5);
+	private static final ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
 
 	private final KisClient kisClient;
-	private final PortfolioRepository portfolioRepository;
+	private final PortFolioHoldingRepository portFolioHoldingRepository;
 	private final SimpMessagingTemplate messagingTemplate;
 	private final PortfolioStockService portfolioStockService;
 	private final KisAccessTokenManager manager;
@@ -53,30 +48,19 @@ public class KisService {
 		return thread;
 	});
 
-	// 제약조건 : kis 서버에 1초당 최대 5건, TR간격 0.1초 이하면 안됨
-	public CurrentPriceResponse readRealTimeCurrentPrice(String tickerSymbol) {
-		long currentPrice = kisClient.readRealTimeCurrentPrice(tickerSymbol, manager.createAuthorization());
-		log.info("tickerSymbol={}, currentPrice={}, time={}", tickerSymbol, currentPrice, LocalDateTime.now());
-		return new CurrentPriceResponse(tickerSymbol, currentPrice);
-	}
-
 	public void addPortfolioSubscription(String sessionId, PortfolioSubscription subscription) {
-		addTickerSymbols(subscription.getTickerSymbols());
-		portfolioSubscriptionManager.addPortfolioSubscription(sessionId, subscription);
-	}
-
-	public void addTickerSymbols(List<String> tickerSymbols) {
-		tickerSymbols.stream()
+		subscription.getTickerSymbols().stream()
 			.filter(tickerSymbol -> !currentPriceManager.hasKey(tickerSymbol))
 			.forEach(currentPriceManager::addKey);
+		portfolioSubscriptionManager.addPortfolioSubscription(sessionId, subscription);
 	}
 
 	public void removePortfolioSubscription(String sessionId) {
 		portfolioSubscriptionManager.removePortfolioSubscription(sessionId);
 	}
 
 	@Scheduled(fixedRate = 5, timeUnit = TimeUnit.SECONDS)
-	public void publishPortfolioDetail() {
+	protected void publishPortfolioDetail() {
 		List<CompletableFuture<PortfolioHoldingsResponse>> futures = portfolioSubscriptionManager.values()
 			.parallelStream()
 			.filter(this::hasAllCurrentPrice)
@@ -110,26 +94,50 @@ private boolean hasAllCurrentPrice(PortfolioSubscription subscription) {
 			.allMatch(currentPriceManager::hasCurrentPrice);
 	}
 
+	// 종목 가격정보 갱신
 	@Scheduled(fixedRate = 5, timeUnit = TimeUnit.SECONDS)
-	@Transactional(readOnly = true)
-	public void refreshCurrentPrice() {
-		List<String> tickerSymbols = portfolioRepository.findAll().parallelStream()
-			.map(Portfolio::getPortfolioHoldings)
-			.flatMap(Collection::stream)
-			.map(PortfolioHolding::getStock)
-			.map(Stock::getTickerSymbol)
-			.collect(Collectors.toList());
+	public void refreshStockPrice() {
+		List<String> tickerSymbols = portFolioHoldingRepository.findAllTickerSymbol();
+		refreshStockCurrentPrice(tickerSymbols);
+		refreshLastDayClosingPrice(tickerSymbols);
+	}
 
-		// 종목 현재가 갱신
-		refreshCurrentPrice(tickerSymbols);
-		log.info("{}개의 종목 현재가 갱신 완료", tickerSymbols.size());
+	// 주식 현재가 갱신
+	public void refreshStockCurrentPrice(List<String> tickerSymbols) {
+		List<CompletableFuture<CurrentPriceResponse>> futures = tickerSymbols.parallelStream()
+			.map(tickerSymbol -> {
+				CompletableFuture<CurrentPriceResponse> future = new CompletableFuture<>();
+				executorService.schedule(createCurrentPriceRequest(tickerSymbol, future), 200, TimeUnit.MILLISECONDS);
+				return future;
+			}).collect(Collectors.toList());
 
-		// 종가 갱신
-		refreshLastDayClosingPrice(tickerSymbols);
+		futures.parallelStream()
+			.map(CompletableFuture::join)
+			.filter(Objects::nonNull)
+			.forEach(currentPriceManager::addCurrentPrice);
 	}
 
-	// 종가 매니저가 가격을 갖지 않은 종목들의 직전거래일의 종가를 요청하여 저장한다
-	private void refreshLastDayClosingPrice(List<String> tickerSymbols) {
+	private Runnable createCurrentPriceRequest(final String tickerSymbol,
+		CompletableFuture<CurrentPriceResponse> future) {
+		return () -> {
+			CurrentPriceResponse response = readRealTimeCurrentPrice(tickerSymbol);
+			future.completeOnTimeout(response, 10, TimeUnit.SECONDS);
+			future.exceptionally(e -> {
+				log.info(e.getMessage(), e);
+				return null;
+			});
+		};
+	}
+
+	// 제약조건 : kis 서버에 1초당 최대 5건, TR간격 0.1초 이하면 안됨
+	public CurrentPriceResponse readRealTimeCurrentPrice(String tickerSymbol) {
+		long currentPrice = kisClient.readRealTimeCurrentPrice(tickerSymbol, manager.createAuthorization());
+		log.info("tickerSymbol={}, currentPrice={}, time={}", tickerSymbol, currentPrice, LocalDateTime.now());
+		return new CurrentPriceResponse(tickerSymbol, currentPrice);
+	}
+
+	// 종가 갱신 (매일 0시 1분 0초에 시작)
+	public void refreshLastDayClosingPrice(List<String> tickerSymbols) {
 		List<CompletableFuture<LastDayClosingPriceResponse>> futures = tickerSymbols.parallelStream()
 			.filter(tickerSymbol -> !lastDayClosingPriceManager.hasPrice(tickerSymbol))
 			.map(tickerSymbol -> {
@@ -141,6 +149,7 @@ private void refreshLastDayClosingPrice(List<String> tickerSymbols) {
 			.collect(Collectors.toList());
 		futures.parallelStream()
 			.map(CompletableFuture::join)
+			.peek(response -> log.info("종가 갱신 응답 : {}", response))
 			.filter(Objects::nonNull)
 			.forEach(response -> lastDayClosingPriceManager.addPrice(response.getTickerSymbol(), response.getPrice()));
 	}
@@ -160,30 +169,4 @@ private Runnable createLastDayClosingPriceRequest(final String tickerSymbol,
 			});
 		};
 	}
-
-	public void refreshCurrentPrice(List<String> tickerSymbols) {
-		List<CompletableFuture<CurrentPriceResponse>> futures = tickerSymbols.parallelStream()
-			.map(tickerSymbol -> {
-				CompletableFuture<CurrentPriceResponse> future = new CompletableFuture<>();
-				executorService.schedule(createCurrentPriceRequest(tickerSymbol, future), 200, TimeUnit.MILLISECONDS);
-				return future;
-			}).collect(Collectors.toList());
-
-		futures.parallelStream()
-			.map(CompletableFuture::join)
-			.filter(Objects::nonNull)
-			.forEach(currentPriceManager::addCurrentPrice);
-	}
-
-	public Runnable createCurrentPriceRequest(final String tickerSymbol,
-		CompletableFuture<CurrentPriceResponse> future) {
-		return () -> {
-			CurrentPriceResponse response = readRealTimeCurrentPrice(tickerSymbol);
-			future.completeOnTimeout(response, 10, TimeUnit.SECONDS);
-			future.exceptionally(e -> {
-				log.info(e.getMessage(), e);
-				return null;
-			});
-		};
-	}
 }

src/main/java/codesquad/fineants/spring/api/kis/aop/AccessTokenAspect.java

@@ -25,15 +25,11 @@ public class AccessTokenAspect {
 	private final KisClient client;
 	private final KisRedisService redisService;
 
-	@Pointcut("execution(* codesquad.fineants.spring.api.kis.KisService.refreshCurrentPrice())")
-	public void refreshCurrentPrice() {
+	@Pointcut("execution(* codesquad.fineants.spring.api.kis.KisService.refreshStockPrice())")
+	public void refreshStockPrice() {
 	}
 
-	@Pointcut("execution(* codesquad.fineants.spring.api.kis.KisService.readRealTimeCurrentPrice())")
-	public void readRealTimeCurrentPrice() {
-	}
-
-	@Before(value = "refreshCurrentPrice(), readRealTimeCurrentPrice()")
+	@Before(value = "refreshStockPrice()")
 	public void checkAccessTokenExpiration() {
 		if (manager.isAccessTokenExpired(LocalDateTime.now())) {
 			final Optional<Map<String, Object>> optionalMap = redisService.getAccessTokenMap();

src/main/java/codesquad/fineants/spring/api/kis/client/KisClient.java

@@ -78,7 +78,6 @@ public Map<String, Object> accessToken() {
 
 	public long readRealTimeCurrentPrice(String tickerSymbol, String authorization) {
 		MultiValueMap<String, String> headerMap = new LinkedMultiValueMap<>();
-		log.info("authorization : {}", authorization);
 		headerMap.add("authorization", authorization);
 		headerMap.add("appkey", appkey);
 		headerMap.add("appsecret", secretkey);

src/main/java/codesquad/fineants/spring/api/kis/manager/LastDayClosingPriceManager.java

@@ -15,7 +15,7 @@ public class LastDayClosingPriceManager {
 	private final RedisTemplate<String, String> redisTemplate;
 
 	public void addPrice(String tickerSymbol, long price) {
-		redisTemplate.opsForValue().set(String.format(format, tickerSymbol), String.valueOf(price), Duration.ofDays(1));
+		redisTemplate.opsForValue().set(String.format(format, tickerSymbol), String.valueOf(price), Duration.ofDays(2));
 	}
 
 	public Long getPrice(String tickerSymbol) {

src/main/java/codesquad/fineants/spring/api/portfolio/PortFolioService.java

@@ -141,7 +141,7 @@ public Portfolio findPortfolio(Long portfolioId) {
 	}
 
 	public PortfoliosResponse readMyAllPortfolio(AuthMember authMember, int size, Long nextCursor) {
-		kisService.refreshCurrentPrice(portfolioRepository.findAllByMemberId(authMember.getMemberId()).stream()
+		kisService.refreshStockCurrentPrice(portfolioRepository.findAllByMemberId(authMember.getMemberId()).stream()
 			.map(Portfolio::getPortfolioHoldings)
 			.flatMap(Collection::stream)
 			.map(PortfolioHolding::getStock)

src/main/java/codesquad/fineants/spring/api/portfolio_stock/PortfolioStockRestController.java

@@ -1,6 +1,5 @@
 package codesquad.fineants.spring.api.portfolio_stock;
 
-import java.io.IOException;
 import java.time.Duration;
 import java.time.LocalDateTime;
 import java.util.concurrent.Executors;
@@ -22,7 +21,6 @@
 
 import codesquad.fineants.domain.oauth.support.AuthMember;
 import codesquad.fineants.domain.oauth.support.AuthPrincipalMember;
-import codesquad.fineants.spring.api.errors.exception.FineAntsException;
 import codesquad.fineants.spring.api.kis.manager.LastDayClosingPriceManager;
 import codesquad.fineants.spring.api.portfolio_stock.request.PortfolioStockCreateRequest;
 import codesquad.fineants.spring.api.response.ApiResponse;
@@ -61,7 +59,7 @@ public ApiResponse<Void> deletePortfolioStock(@PathVariable Long portfolioId,
 
 	@GetMapping
 	public SseEmitter readMyPortfolioStocks(@PathVariable Long portfolioId) {
-		SseEmitter emitter = new SseEmitter(Duration.ofHours(10L).toMillis());
+		SseEmitter emitter = new SseEmitter(Duration.ofHours(10).toMillis());
 		emitter.onTimeout(emitter::complete);
 
 		// 장시간 동안에는 스케줄러를 이용하여 지속적 응답
@@ -81,12 +79,14 @@ private void scheduleSseEventTask(Long portfolioId, SseEmitter emitter, boolean
 					.name("sse event - myPortfolioStocks"));
 				log.info("send message");
 				if (isComplete) {
+					Thread.sleep(2000L); // sse event - myPortfolioStocks 메시지와 전송 간격
 					emitter.send(SseEmitter.event()
 						.data("sse complete")
 						.name("complete"));
 					emitter.complete();
+					log.info("emitter complete");
 				}
-			} catch (IOException | FineAntsException e) {
+			} catch (Exception e) {
 				log.error(e.getMessage());
 				emitter.completeWithError(e);
 			}

src/main/resources/db/mysql/data.sql

@@ -30,10 +30,27 @@ INSERT INTO fineAnts.portfolio (id, budget, maximum_loss, name, securities_firm,
                                 maximum_is_active, member_id)
 VALUES (1, 1000000, 900000, '내꿈은 워렌버핏', '토스', 1500000, false, false, 1);
 
-INSERT INTO fineAnts.portfolio_holding (id, create_at, modified_at,
-                                        portfolio_id, ticker_symbol)
+INSERT INTO fineAnts.portfolio_holding (id, create_at, modified_at, portfolio_id, ticker_symbol)
 VALUES (1, '2023-10-26 15:25:39.409612', '2023-10-26 15:25:39.409612', 1, '005930');
 
+INSERT INTO fineAnts.portfolio_holding (id, create_at, modified_at, portfolio_id, ticker_symbol)
+VALUES (2, '2023-10-26 15:25:39.409612', '2023-10-26 15:25:39.409612', 1, '000020');
+
+INSERT INTO fineAnts.portfolio_holding (id, create_at, modified_at, portfolio_id, ticker_symbol)
+VALUES (3, '2023-10-26 15:25:39.409612', '2023-10-26 15:25:39.409612', 1, '000040');
+
+INSERT INTO fineAnts.portfolio_holding (id, create_at, modified_at, portfolio_id, ticker_symbol)
+VALUES (4, '2023-10-26 15:25:39.409612', '2023-10-26 15:25:39.409612', 1, '000050');
+
+INSERT INTO fineAnts.portfolio_holding (id, create_at, modified_at, portfolio_id, ticker_symbol)
+VALUES (5, '2023-10-26 15:25:39.409612', '2023-10-26 15:25:39.409612', 1, '000070');
+
+INSERT INTO fineAnts.portfolio_holding (id, create_at, modified_at, portfolio_id, ticker_symbol)
+VALUES (6, '2023-10-26 15:25:39.409612', '2023-10-26 15:25:39.409612', 1, '000080');
+
+INSERT INTO fineAnts.portfolio_holding (id, create_at, modified_at, portfolio_id, ticker_symbol)
+VALUES (7, '2023-10-26 15:25:39.409612', '2023-10-26 15:25:39.409612', 1, '000087');
+
 INSERT INTO fineAnts.purchase_history (id, create_at, modified_at, memo, num_shares, purchase_date,
                                        purchase_price_per_share, portfolio_holding_id)
 VALUES (1, '2023-10-26 15:26:11.219793', '2023-10-26 15:26:11.219793', null, 3, '2023-10-23 13:00:00.000000', 50000, 1);