저번 포스트에서 이어집니다.
구현
▶ 구현 순서
- 시세 정보 수집 기능
- 알림 조건 검사 기능
- 실시간 알림 발송 기능
- 이력 관리 기능
1. 시세 정보 수집 기능
1-1. Feign Client 구성
시세 정보 수집을 위해 가상화폐 거래소 API를 조사한 결과, 대부분의 거래소가 자산 조회와 주문 등의 거래 관련 서비스에만 인증 수단을 요구하며, 시세 조회는 별도의 인증 없이 가능함을 확인했습니다.
따라서, Configuration 클래스에 RequestInterceptor를 별도로 구성하지 않았습니다.
시세 조회 서비스의 종류가 많았지만, 우선 거래 가능한 마켓 목록 조회 / 분(Minute) 단위 캔들 시세 조회 두 가지를 활용해 보기로 하고, 이에 대한 요청 메서드를 작성했습니다.
@FeignClient(name = "UpbitCurrency", url = "https://api.upbit.com/v1")
public interface UpbitFeignClient {
@GetMapping("/market/all")
List<MarketInfoDto> getMarketAll(@RequestParam("isDetails") boolean isDetails);
@GetMapping("/candles/minutes/{unit}")
List<MarketPriceDto> getCandlesMinutes(
@PathVariable("unit") int unit,
@RequestParam("market") String market,
@RequestParam("count") int count
);
}
1-2. Configuration 구성
@Configuration
@EnableFeignClients(basePackages = "com.example.demo")
public class FeignConfig {
@Bean
public Logger.Level feignLoggerLevel() {
return Logger.Level.FULL;
}
@Bean
public ErrorDecoder errorDecoder() {
return new CustomFeignErrorDecorder();
}
@Bean
public Request.Options options() {
return new Request.Options(Duration.ofSeconds(5), Duration.ofSeconds(10), true);
}
}
Feign Client의 기본 Error Decoder는 주로 Feign 표준 로직 처리와, 요청 실패 시 재시도 관련 로직으로 구성되어 있습니다. 서버 오류와 클라이언트 오류를 명확히 구분하기 위해, CustomErrorDecoder를 구성했습니다.
이 CustomErrorDecoder는 응답의 상태 코드를 확인하여 서버 측 오류인 경우,
Retry-After 헤더의 유무에 따라 RetryableException* 또는 ServiceFailedException을 던집니다.
(*RetryableException가 발생하면, 재시도 정책에 따라 Retryer가 HTTP 요청을 재시도합니다.)
또한, 클라이언트측 오류나 예상치 못한 상태 코드 범위를 벗어난 응답에 대해서도 예외를 던지며, 상태 메시지를 통해 구체적인 오류 정보를 제공하도록 구성했습니다.
(CustomErrorDecoder 소스 코드 보기)
public class CustomFeignErrorDecorder implements ErrorDecoder {
@Override
public Exception decode(String methodKey, Response response) {
if (isServerError(response)) {
handleRetryAfterHeader(response);
throw new ServiceFailedException(EXTERNAL_API_ERROR);
}
if (isClientError(response)) {
throw new ServiceFailedException(FAILED_HTTP_REQUEST);
}
throw new ServiceFailedException(UNKNOWN_ERROR);
}
private boolean isClientError(Response response) {
return response.status() >= 400 && response.status() <= 499;
}
private boolean isServerError(Response response) {
return response.status() >= 500 && response.status() <= 599;
}
private void handleRetryAfterHeader(Response response) {
if (response.headers().containsKey("Retry-After")) {
String retryAfter = response.headers().get("Retry-After").iterator().next();
throw new RetryableException(response.status(), response.reason(), response.request().httpMethod(), Long.parseLong(retryAfter), response.request());
}
}
}
1-3. 서비스 로직과 DTO 구성
시세 정보를 저장할 DTO 객체와, 서비스 로직을 구현합니다.
API 문서의 응답 객체 예시를 참고해 DTO를 구성했습니다.
@Value
public class MarketPriceDto {
// 마켓명
@JsonProperty("market")
String market;
// 시가
@JsonProperty("opening_price")
BigDecimal openingPrice;
// 고가
@JsonProperty("high_price")
BigDecimal highPrice;
// 저가
@JsonProperty("low_price")
BigDecimal lowPrice;
// 종가
@JsonProperty("trade_price")
BigDecimal tradePrice;
// 이하 생략
}
일정 간격마다 작업을 실행하게 하기 위해 Spring의 Scheduling 기능을 활용했습니다.
@Service
@Slf4j
@RequiredArgsConstructor
public class CryptoScheduledService {
private final UpbitFeignClient upbitClient;
@Scheduled(fixedRate = 1000 * 5) // 5초마다
public void fetchCurrencyInfo() {
List<MarketPriceDto> response = upbitClient.getCandlesMinutes(1, "KRW-BTC", 1);
BigDecimal tradePrice = response.get(0).getTradePrice();
String message = "현재가: " + tradePrice;
log.info(message);
}
}혹시, 이 글을 참고하며 코드를 작성 중이신데, 스케줄링 작업이 실행이 안된다면
메인 클래스에 @EnableScheduling 어노테이션을 추가하셨는지 확인해 주세요!
구성이 올바르게 되었는지 확인하기 위해 5초 간격으로 시세 정보를 가져오도록 하고 테스트해 본 결과, 잘 불러오는 걸 확인할 수 있었습니다.
2. 알림 조건 검사 기능
스케줄링을 사용하여 주기적으로 시세 정보를 확인하고, 사용자가 설정한 목표 가격에 도달했는지 확인해야하지만,
Spring의 스케줄링은 독립적으로 실행되고, 이전 작업의 결과를 다음 실행에 전달하지 않기 때문에, 목표 가격은 REDIS에 저장하고, 조건 충족 시 알림 발송은 Spring의 Event Publishing / Listening 기능을 활용하였습니다.
private void checkIfReachedTargetPrice(BigDecimal tradePrice) {
String data = redisService.getData(CommandType.ALARM.getPrefix());
String targetPrice = parseAlarmData(data);
boolean isReached = tradePrice.compareTo(new BigDecimal(targetPrice)) >= 0;
if (isReached) { // 목표가 도달 시, 알림 이벤트 발행
String message = "목표가 도달! 현재가: " + tradePrice + ", 목표가: " + targetPrice;
publishNotificationEvent(ALARM, message);
}
}
- SlackNotificationEvent와 발행된 이벤트를 처리할 EventListener (상세 구현은 3-2)
@Getter
public class SlackNotificationEvent extends ApplicationEvent {
private final CommandType type;
private final String message;
public SlackNotificationEvent(Object source, CommandType type, String message) {
super(source);
this.type = type;
this.message = message;
}
}
// 이벤트 리스너
@Component
@RequiredArgsConstructor
public class AppEventListener {
private final SlackFeignClient slackClient;
@EventListener
public void handleSlackNotificationEventEvent(SlackNotificationEvent event) {
// Slack에 알림 발송
}
}
3. 실시간 알림 발송 기능
3-1. Slack Feign Client 구성
Slack API는 Bolt 클라이언트를 사용해 Java 코드로도 구성할 수 있지만,
OpennFeign을 최대한 활용해 보기 위해 REST API를 사용하기로 하고, FeignClient를 작성했습니다.
@FeignClient(name = "slack", url = "https://slack.com/api", configuration = SlackFeignConfig.class)
public interface SlackFeignClient {
@PostMapping("/chat.postMessage")
void postMessage(@RequestBody SlackMessageRequestDto request);
@PostMapping("/conversations.join")
void joinChannel(@RequestBody SlackMessageRequestDto request);
}
Slack REST API는 요청에 OAuthToken을 요구해 RequestInterceptor를 추가로 구성해 주었습니다.
public class SlackFeignConfig {
@Value("${slack.oauth.token}")
private String slackOauthToken;
@Bean
public RequestInterceptor requestInterceptor() {
return requestTemplate
-> requestTemplate
.header("Content-Type", "application/json")
.header("Authorization", "Bearer " + slackOauthToken);
}
}
※ 주의할 점!
전역 설정이 아닌, 상세 설정 클래스에 @Configuration 어노테이션을 사용하면 빈이 중복 생성되어 애플리케이션이 정상적으로 실행되지 않을 수도 있습니다.
추가로 적용해야 할 설정이 있다면, @Configuration 어노테이션 없이, @Bean 만 정의하고, @FeignClient의 configuration 속성을 사용해 적용해야 합니다.
▶ Slack App 등록 & 토큰 발급받기
Slack API 페이지에서 Create New App으로 새 앱을 생성하고, From Scratch 선택, 앱 이름과 설치할 워크스페이스를 선택해 줍니다.
밑으로 스크롤해 Scopes에서 사용할 권한을 선택합니다.
권한을 추가해 주고 페이지를 위로 스크롤 해보면 토큰이 생성되어 있습니다. 이 토큰을 요청에 사용하면 됩니다.
3-2. 알림 이벤트 리스너 구현
@Component
@RequiredArgsConstructor
public class SlackEventListener {
private final SlackFeignClient slackClient;
private final RedisService redisService;
@EventListener
public void handleNotificationEvent(SlackNotificationEvent event) {
String channelId = getChannelId(event);
SlackMessageRequestDto request = new SlackMessageRequestDto(channelId, event.getMessage());
slackClient.postMessage(request);
}
private String getChannelId(SlackNotificationEvent event) {
// Redis에서 채널ID 값을 가져온다.
}
}
(SlackNotificationEvent 코드)
@Getter
public class SlackNotificationEvent extends ApplicationEvent {
private final CommandType type;
private final String message;
public SlackNotificationEvent(Object source, CommandType type, String message) {
super(source);
this.type = type;
this.message = message;
}
}
테스트해 보니 1분마다 정상적으로 알림 & 목표가 도달 알림이 발송됐습니다.
4. 이력 관리 기능
4-1. 이력 관리 엔티티 구성
@Entity
@Getter
@Builder
@AllArgsConstructor
@NoArgsConstructor(access = PROTECTED)
@Table(name = "slack_notification_history")
public class SlackNotificationHistory extends BaseEntity {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;
@Comment("알림을 발송한 채널 ID")
private String channelId;
@Comment("알림 발송 구분(알람, 시세 정보)")
private String type;
@Comment("발송한 메시지")
private String message;
}
어떤 채널로, 어떤 메시지(구분 포함)를 발송했는지를 이력으로 남길 수 있게 구성했고, 생성 / 수정일시는 BaseEntity(MappedSuperclass)를 정의해 자동으로 기록하도록 했습니다.
@Getter
@MappedSuperclass
@EntityListeners(AuditingEntityListener.class)
public abstract class BaseEntity {
@CreatedDate
@Column(nullable = false, updatable = false)
private LocalDateTime createdAt;
@LastModifiedDate
@Column(nullable = false)
private LocalDateTime updatedAt;
}
EventListener에 이력 저장 로직을 추가해, 메시지 발송 전에 이력을 남기도록 리스너를 수정했습니다.
@Component
@RequiredArgsConstructor
public class SlackEventListener {
private final SlackFeignClient slackClient;
private final RedisService redisService;
private final SlackNotificationHistoryRepository historyRepository;
@EventListener
public void handleNotificationEvent(SlackNotificationEvent event) {
String channelId = getChannelId(event);
SlackMessageRequestDto request = // 요청 DTO 생성
saveNotificationHistory(event, channelId);
slackClient.postMessage(request);
}
private String getChannelId(SlackNotificationEvent event) {
// Redis에서 채널ID 값을 가져온다.
}
private void saveNotificationHistory(SlackNotificationEvent event, String channelId) {
SlackNotificationHistory history = SlackNotificationHistory.builder()
.channelId(channelId)
.type(event.getType().name())
.message(event.getMessage())
.build();
historyRepository.save(history);
}
}
* handle 메서드에 @Transactional을 추가해야 할지 여부?
보통 DB 작업이 포함되면, 트랜잭션으로 관리하는 것이 일반적이라 생각했는데,
이 경우에는 이력 저장과 메시지 발송을 하나의 트랜잭션으로 묶어 일관되게 관리하기보다
메시지 발송이 이력 저장 성공 여부와 무관하게 중요하다고 생각해 트랜잭션을 적용하지 않았습니다.
'Framework > Spring' 카테고리의 다른 글
| [Spring] Dispatcher-Servlet 알아보기 (0) | 2024.02.02 |
|---|---|
| [Spring] 혼자서 해보는 가상화폐 시세 알림 API - 확장 (0) | 2024.01.17 |
| [Spring] 혼자서 해보는 가상화폐 시세 알림 API - 설계 (0) | 2024.01.11 |
| [Spring] OpenFeign 알아보기 (0) | 2024.01.09 |
| [Spring] 스프링 프레임워크 핵심 원리 (1) | 2023.11.27 |