Skip to content

Commit

Permalink
πŸ› fix: Kafka Consumer μ˜ˆμ™Έ 상황 처리
Browse files Browse the repository at this point in the history
πŸ› fix: Kafka Consumer μ˜ˆμ™Έ 상황 처리
  • Loading branch information
DDonghyeo authored Jul 5, 2024
2 parents 99a3d0f + 364d294 commit 6f4711b
Showing 1 changed file with 34 additions and 5 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -127,9 +127,17 @@ public void consumeWindAlarm(KafkaDto.WeatherDto weatherDto) {

// Wind Alertλ₯Ό True둜 μ„€μ •ν•œ User Query
List<UserData> userData = userDataRepository.findAllByWindAlertIsTrue();
if (userData.isEmpty()) {
log.info("[ Kafka Listener ] 보낼 μ‚¬μš©μž μ—†μŒ.");
return;
}

//μ•Œλ¦Ό 보낼 μ‚¬μš©μž 이메일
List<String> userEmails = filterRegionAndWindAlarm(region, userData, currentHour);
if (userEmails.isEmpty()) {
log.info("[ Kafka Listener ] 보낼 μ‚¬μš©μž μ—†μŒ.");
return;
}

sb.append("ν˜„μž¬ λ°”λžŒ μ„ΈκΈ°κ°€ ").append(windStrength).append("m/s μ΄μƒμž…λ‹ˆλ‹€.");

Expand Down Expand Up @@ -165,15 +173,27 @@ public void consumeRain(KafkaDto.WeatherDto weatherDto) {
return;
}

String title = "Waither κ°•μˆ˜ 정보 μ•Œλ¦Ό";
List<UserData> userData = userDataRepository.findAllBySnowAlertIsTrue();
StringBuilder sb = new StringBuilder();

//지역
String region = weatherDto.region();
log.info("[ Kafka Listener ] κ°•μˆ˜λŸ‰ 지역 --> {}", region);
String message = weatherDto.message();

String title = "Waither κ°•μˆ˜ 정보 μ•Œλ¦Ό";
List<UserData> userData = userDataRepository.findAllBySnowAlertIsTrue();
if (userData.isEmpty()) {
log.info("[ Kafka Listener ] 보낼 μ‚¬μš©μž μ—†μŒ.");
return;
}

List<String> userEmails = filterRegionAndRainAlarm(region, userData, currentHour);
if (userEmails.isEmpty()) {
log.info("[ Kafka Listener ] 보낼 μ‚¬μš©μž μ—†μŒ.");
return;
}

StringBuilder sb = new StringBuilder();


//1μ‹œκ°„ λ’€, 2μ‹œκ°„ λ’€, 3μ‹œκ°„ λ’€, 4μ‹œκ°„ λ’€, 5μ‹œκ°„ λ’€, 6μ‹œκ°„ λ’€
List<Double> predictions = Arrays.stream(message.split(","))
.map(String::trim) //곡백 제거
Expand All @@ -183,13 +203,14 @@ public void consumeRain(KafkaDto.WeatherDto weatherDto) {
String rainMessage = WeatherMessageUtils.getRainPredictions(predictions);

if (rainMessage == null) {
log.info("[ Kafka Listener ] 6μ‹œκ°„ λ™μ•ˆ κ°•μˆ˜ 정보 μ—†μŒ.");
//6μ‹œκ°„ λ™μ•ˆ κ°•μˆ˜ 정보 μ—†μŒ
return;
}

sb.append("ν˜„μž¬ ").append(region).append(" 지역에 ").append(rainMessage);
//μ•Œλ¦Ό 보낼 μ‚¬μš©μž 이메일
List<String> userEmails = filterRegionAndRainAlarm(region, userData, currentHour);


System.out.println("[ ν‘Έμ‹œμ•Œλ¦Ό ] κ°•μˆ˜λŸ‰ μ•Œλ¦Ό");
alarmService.sendAlarms(userEmails, title, sb.toString());
Expand Down Expand Up @@ -226,9 +247,17 @@ public void consumeClimateAlarm(KafkaDto.WeatherDto weatherDto) {

// Wind Climateλ₯Ό True둜 μ„€μ •ν•œ User Query
List<UserData> userData = userDataRepository.findAllByClimateAlertIsTrue();
if (userData.isEmpty()) {
log.info("[ Kafka Listener ] 보낼 μ‚¬μš©μž μ—†μŒ.");
return;
}

// μ•Œλ¦Ό 보낼 μ‚¬μš©μž 이메일
List<String> userEmails = filterRegion(region, userData);
if (userEmails.isEmpty()) {
log.info("[ Kafka Listener ] 보낼 μ‚¬μš©μž μ—†μŒ.");
return;
}

sb.append("[기상청 기상 특보] ").append(message);

Expand Down

0 comments on commit 6f4711b

Please sign in to comment.