-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
2 changed files
with
150 additions
and
0 deletions.
There are no files selected for viewing
111 changes: 111 additions & 0 deletions
111
batch/src/main/kotlin/com/oksusu/susu/batch/discord/job/ResendFailedSentDiscordMessageJob.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,111 @@ | ||
package com.oksusu.susu.batch.discord.job | ||
|
||
import com.oksusu.susu.cache.key.Cache | ||
import com.oksusu.susu.cache.model.FailedSentDiscordMessageCache | ||
import com.oksusu.susu.cache.service.CacheService | ||
import com.oksusu.susu.client.discord.DiscordClient | ||
import com.oksusu.susu.client.discord.model.DiscordMessageModel | ||
import io.github.oshai.kotlinlogging.KotlinLogging | ||
import kotlinx.coroutines.* | ||
import org.springframework.stereotype.Component | ||
import java.time.LocalDateTime | ||
|
||
@Component | ||
class ResendFailedSentDiscordMessageJob( | ||
private val cacheService: CacheService, | ||
private val discordClient: DiscordClient, | ||
) { | ||
private val logger = KotlinLogging.logger { } | ||
|
||
companion object { | ||
private const val RESEND_BEFORE_MINUTES = 1L | ||
} | ||
|
||
suspend fun resendFailedSentDiscordMessage() { | ||
logger.info { "start resend failed sent discord message" } | ||
|
||
// 1분 전에 실패한 것이 타겟 (현재가 24분이면 23분을 말하는 것) | ||
val targetTime = LocalDateTime.now().minusMinutes(RESEND_BEFORE_MINUTES) | ||
|
||
// 실패 메세지 조회 및 삭제 | ||
val failedMessages = withContext(Dispatchers.IO) { | ||
cacheService.sGetMembers(Cache.getFailedSentDiscordMessageCache(targetTime)) | ||
} | ||
|
||
withContext(Dispatchers.IO) { | ||
cacheService.sDelete(Cache.getFailedSentDiscordMessageCache(targetTime)) | ||
} | ||
|
||
// 다수 메세지 token 별로 하나의 메세지로 병합 | ||
val message = mergeFailedMessage(failedMessages) | ||
|
||
// 재전송 | ||
runCatching { | ||
coroutineScope { | ||
val sendDeferreds = message.map { (token, message) -> | ||
val discordMessageModel = DiscordMessageModel(content = message) | ||
|
||
async(Dispatchers.IO) { | ||
discordClient.sendMessage( | ||
message = discordMessageModel, | ||
token = token, | ||
withRecover = false | ||
) | ||
} | ||
}.toTypedArray() | ||
|
||
awaitAll(*sendDeferreds) | ||
} | ||
}.onFailure { | ||
// 재전송 실패시 1분 뒤에 다시 보낼 수 있게, 1분 뒤에 보내는 메세지 목록에 추가 | ||
logger.warn { "postpone resend discord message" } | ||
|
||
postponeResendTimeOfFailedMessage(targetTime, message) | ||
} | ||
|
||
logger.info { "finish resend failed sent discord message" } | ||
} | ||
|
||
private suspend fun mergeFailedMessage(failedMessages: List<FailedSentDiscordMessageCache>): Map<String, String> { | ||
val message = mutableMapOf<String, String>() | ||
|
||
failedMessages.forEach { model -> | ||
val recoverMsg = if (model.isStacked) { | ||
model.message | ||
} else { | ||
"[RECOVER - ${model.failedAt} discord failure] ${model.message}" | ||
} | ||
|
||
val stackedMessage = message[model.token] | ||
|
||
message[model.token] = if (stackedMessage == null) { | ||
recoverMsg | ||
} else { | ||
"$stackedMessage\n$recoverMsg" | ||
} | ||
} | ||
|
||
return message | ||
} | ||
|
||
private suspend fun postponeResendTimeOfFailedMessage(targetTime: LocalDateTime, message: Map<String, String>) { | ||
val nextTime = targetTime.plusMinutes(RESEND_BEFORE_MINUTES) | ||
|
||
coroutineScope { | ||
message.map { (token, message) -> | ||
val model = FailedSentDiscordMessageCache( | ||
token = token, | ||
message = message, | ||
isStacked = true | ||
) | ||
|
||
async(Dispatchers.IO) { | ||
cacheService.sSet( | ||
cache = Cache.getFailedSentDiscordMessageCache(nextTime), | ||
value = model | ||
) | ||
} | ||
} | ||
} | ||
} | ||
} |
39 changes: 39 additions & 0 deletions
39
...kotlin/com/oksusu/susu/batch/discord/scheduler/ResendFailedSentDiscordMessageScheduler.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,39 @@ | ||
package com.oksusu.susu.batch.discord.scheduler | ||
|
||
import com.oksusu.susu.batch.discord.job.ResendFailedSentDiscordMessageJob | ||
import com.oksusu.susu.batch.slack.job.ResendFailedSentSlackMessageJob | ||
import com.oksusu.susu.client.common.coroutine.ErrorPublishingCoroutineExceptionHandler | ||
import com.oksusu.susu.common.extension.isProd | ||
import com.oksusu.susu.common.extension.resolveCancellation | ||
import io.github.oshai.kotlinlogging.KotlinLogging | ||
import kotlinx.coroutines.CoroutineScope | ||
import kotlinx.coroutines.Dispatchers | ||
import kotlinx.coroutines.launch | ||
import org.springframework.core.env.Environment | ||
import org.springframework.scheduling.annotation.Scheduled | ||
import org.springframework.stereotype.Component | ||
|
||
@Component | ||
class ResendFailedSentDiscordMessageScheduler( | ||
private val environment: Environment, | ||
private val resendFailedSentDiscordMessageJob: ResendFailedSentDiscordMessageJob, | ||
private val coroutineExceptionHandler: ErrorPublishingCoroutineExceptionHandler, | ||
) { | ||
private val logger = KotlinLogging.logger { } | ||
|
||
@Scheduled( | ||
fixedRate = 1000 * 60, | ||
initialDelayString = "\${oksusu.scheduled-tasks.resend-failed-sent-discord-message.initial-delay:100}" | ||
) | ||
fun resendFailedSentDiscordMessageJob() { | ||
if (environment.isProd()) { | ||
CoroutineScope(Dispatchers.IO + coroutineExceptionHandler.handler).launch { | ||
runCatching { | ||
resendFailedSentDiscordMessageJob.resendFailedSentDiscordMessage() | ||
}.onFailure { e -> | ||
logger.resolveCancellation("[BATCH] fail to run resendFailedSentDiscordMessageJob", e) | ||
} | ||
} | ||
} | ||
} | ||
} |