Skip to content

Commit 0a46c93

Browse files
simhani1simhani1
authored
release: 0.1.8 (#122)
2 parents 6801111 + a828a36 commit 0a46c93

17 files changed

Lines changed: 802 additions & 32 deletions

File tree

‎application/src/main/kotlin/site/techmoa/application/service/BlogService.kt‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
package site.techmoa.application.service
2+
3+
import org.springframework.stereotype.Service
4+
import site.techmoa.application.port.BlogPort
5+
import site.techmoa.domain.model.Blog
6+
import site.techmoa.domain.model.BlogStatus
7+
8+
@Service
9+
class BlogService(
10+
private val blogPort: BlogPort
11+
) {
12+
13+
fun getAllBlogs(): List<Blog> {
14+
return blogPort.findAllBy(BlogStatus.ACTIVE)
15+
}
16+
}

‎application/src/main/kotlin/site/techmoa/application/service/WebhookService.kt‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,10 @@ class WebhookService(
1313

1414
@Transactional
1515
fun save(platform: WebhookPlatform, url: String): Webhook {
16-
return saveWebhookUseCase.save(platform = platform, url = url)
16+
val newWebhook = Webhook.of(
17+
platform = platform,
18+
url = url,
19+
)
20+
return saveWebhookUseCase.save(newWebhook)
1721
}
1822
}

‎application/src/main/kotlin/site/techmoa/application/usecase/SaveWebhookUseCase.kt‎

Lines changed: 4 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -4,22 +4,16 @@ import org.springframework.stereotype.Component
44
import site.techmoa.application.port.WebhookPort
55
import site.techmoa.domain.exception.DuplicatedWebhookException
66
import site.techmoa.domain.model.Webhook
7-
import site.techmoa.domain.model.WebhookPlatform
87

98
@Component
109
class SaveWebhookUseCase(
1110
private val webhookPort: WebhookPort,
1211
) {
1312

14-
fun save(platform: WebhookPlatform, url: String): Webhook {
15-
if (webhookPort.existsBy(platform, url)) {
16-
throw DuplicatedWebhookException("Webhook already exists. platform=$platform, url=$url")
13+
fun save(webhook: Webhook): Webhook {
14+
if (webhookPort.existsBy(webhook.platform, webhook.url)) {
15+
throw DuplicatedWebhookException("Webhook already exists. platform=${webhook.platform}, url=${webhook.url}")
1716
}
18-
return webhookPort.save(
19-
Webhook.of(
20-
platform = platform,
21-
url = url,
22-
)
23-
)
17+
return webhookPort.save(webhook)
2418
}
2519
}

‎application/src/test/kotlin/site/techmoa/application/service/WebhookServiceTest.kt‎

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
package site.techmoa.application.service
2+
3+
import io.kotest.assertions.throwables.shouldThrow
4+
import io.kotest.core.spec.style.BehaviorSpec
5+
import io.kotest.matchers.shouldBe
6+
import io.mockk.every
7+
import io.mockk.mockk
8+
import io.mockk.verify
9+
import site.techmoa.application.port.WebhookPort
10+
import site.techmoa.application.usecase.SaveWebhookUseCase
11+
import site.techmoa.domain.exception.DuplicatedWebhookException
12+
import site.techmoa.domain.exception.ErrorCode
13+
import site.techmoa.domain.model.Webhook
14+
import site.techmoa.domain.model.WebhookPlatform
15+
16+
class WebhookServiceTest : BehaviorSpec({
17+
given("웹훅을 등록할 때") {
18+
`when`("이미 등록된 웹훅인 경우") {
19+
val webhookPort = mockk<WebhookPort>()
20+
val saveWebhookUseCase = SaveWebhookUseCase(webhookPort)
21+
val webhookService = WebhookService(saveWebhookUseCase)
22+
every { webhookPort.existsBy(DISCORD, WEBHOOK_URL) } returns true
23+
24+
then("DuplicatedWebhookException이 발생하고 저장은 수행되지 않는다") {
25+
val exception = shouldThrow<DuplicatedWebhookException> {
26+
webhookService.save(DISCORD, WEBHOOK_URL)
27+
}
28+
29+
exception.errorCode shouldBe ErrorCode.DUPLICATED_WEBHOOK
30+
exception.message shouldBe "Webhook already exists. platform=DISCORD, url=$WEBHOOK_URL"
31+
verify(exactly = 1) { webhookPort.existsBy(DISCORD, WEBHOOK_URL) }
32+
verify(exactly = 0) { webhookPort.save(any()) }
33+
}
34+
}
35+
36+
`when`("플랫폼과 URL로 저장을 요청하면") {
37+
val webhookPort = mockk<WebhookPort>()
38+
val saveWebhookUseCase = SaveWebhookUseCase(webhookPort)
39+
val webhookService = WebhookService(saveWebhookUseCase)
40+
val savedWebhook = Webhook.of(
41+
id = SAVED_WEBHOOK_ID,
42+
url = WEBHOOK_URL,
43+
platform = DISCORD,
44+
)
45+
every { webhookPort.existsBy(DISCORD, WEBHOOK_URL) } returns false
46+
every { webhookPort.save(any()) } returns savedWebhook
47+
48+
then("정상 저장된 웹훅 정보를 반환한다") {
49+
val result = webhookService.save(DISCORD, WEBHOOK_URL)
50+
result.id shouldBe SAVED_WEBHOOK_ID
51+
result.platform shouldBe DISCORD
52+
result.url shouldBe WEBHOOK_URL
53+
verify(exactly = 1) { webhookPort.existsBy(DISCORD, WEBHOOK_URL) }
54+
verify(exactly = 1) { webhookPort.save(any()) }
55+
}
56+
}
57+
}
58+
}) {
59+
companion object {
60+
private val DISCORD = WebhookPlatform.DISCORD
61+
private const val WEBHOOK_URL = "https://discord.com/api/webhooks"
62+
private const val SAVED_WEBHOOK_ID = 1L
63+
}
64+
}

‎application/src/test/kotlin/site/techmoa/application/usecase/SaveWebhookUseCaseTest.kt‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,12 @@ class SaveWebhookUseCaseTest : BehaviorSpec({
2525
}
2626

2727
`when`("중복이 아닌 url을 저장하면") {
28-
val result = useCase.save(WebhookPlatform.DISCORD, "https://discord.com/api/webhooks/123")
28+
val result = useCase.save(
29+
Webhook.of(
30+
platform = WebhookPlatform.DISCORD,
31+
url = "https://discord.com/api/webhooks/123",
32+
)
33+
)
2934

3035
then("저장 후 생성된 webhook을 반환한다") {
3136
result.id shouldBe 1L
@@ -45,7 +50,12 @@ class SaveWebhookUseCaseTest : BehaviorSpec({
4550
`when`("동일한 url을 저장하면") {
4651
then("DuplicatedWebhookException을 던진다") {
4752
val exception = shouldThrow<DuplicatedWebhookException> {
48-
useCase.save(WebhookPlatform.DISCORD, "https://discord.com/api/webhooks/dup")
53+
useCase.save(
54+
Webhook.of(
55+
platform = WebhookPlatform.DISCORD,
56+
url = "https://discord.com/api/webhooks/dup",
57+
)
58+
)
4959
}
5060
exception.errorCode.name shouldBe "DUPLICATED_WEBHOOK"
5161
verify(exactly = 0) { webhookPort.save(any()) }

‎batch/rss/src/main/kotlin/site/techmoa/batch/rss/config/BatchRssConfig.kt‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,9 @@ import org.slf4j.LoggerFactory
44
import org.springframework.context.annotation.Bean
55
import org.springframework.context.annotation.Configuration
66
import org.springframework.scheduling.annotation.EnableScheduling
7+
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor
78
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler
9+
import java.util.concurrent.Executor
810

911
@Configuration
1012
@EnableScheduling
@@ -15,7 +17,6 @@ class BatchRssConfig {
1517
@Bean("rssTaskScheduler")
1618
fun rssTaskScheduler(): ThreadPoolTaskScheduler {
1719
val scheduler = ThreadPoolTaskScheduler()
18-
scheduler.poolSize = 4
1920
scheduler.threadNamePrefix = "rss-scheduler-"
2021
scheduler.setWaitForTasksToCompleteOnShutdown(true)
2122
scheduler.setAwaitTerminationSeconds(30)
@@ -25,4 +26,16 @@ class BatchRssConfig {
2526
scheduler.initialize()
2627
return scheduler
2728
}
29+
30+
@Bean("rssFetchExecutor")
31+
fun rssFetchExecutor(): Executor {
32+
val executor = ThreadPoolTaskExecutor()
33+
executor.corePoolSize = 4
34+
executor.maxPoolSize = 4
35+
executor.setQueueCapacity(100)
36+
executor.setThreadNamePrefix("rss-fetch-")
37+
executor.setWaitForTasksToCompleteOnShutdown(true)
38+
executor.initialize()
39+
return executor
40+
}
2841
}

‎batch/rss/src/main/kotlin/site/techmoa/batch/rss/repository/ArticleRepository.kt‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,8 @@ package site.techmoa.batch.rss.repository
22

33
import org.springframework.jdbc.core.JdbcTemplate
44
import org.springframework.stereotype.Repository
5+
import org.springframework.transaction.annotation.Propagation
6+
import org.springframework.transaction.annotation.Transactional
57
import site.techmoa.batch.rss.domain.model.Article
68
import site.techmoa.batch.rss.port.ArticlePort
79
import java.sql.Timestamp
@@ -12,6 +14,7 @@ class ArticleRepository(
1214
private val jdbcTemplate: JdbcTemplate
1315
) : ArticlePort {
1416

17+
@Transactional(propagation = Propagation.REQUIRES_NEW)
1518
override fun saveAllIgnoringDuplicates(articles: List<Article>) {
1619
if (articles.isEmpty()) return
1720

‎batch/rss/src/main/kotlin/site/techmoa/batch/rss/support/RssReaderClient.kt‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -24,9 +24,8 @@ class RssReaderClient(
2424
private val articleLinkManager: ArticleLinkManager,
2525
) : RssClient {
2626

27-
private val rssReader = RssReader().addFeedFilter(CustomInvalidXmlCharacterFilter()) as RssReader
28-
2927
override fun fetch(blog: Blog): List<Article> {
28+
val rssReader = RssReader().addFeedFilter(CustomInvalidXmlCharacterFilter()) as RssReader
3029
val rawItems = rssReader.read(blog.rssLink)
3130
.toList()
3231
val items = rawItems.map { toItem(it, blog) }

‎batch/rss/src/main/kotlin/site/techmoa/batch/rss/trigger/RssCollectorTriggerSupport.kt‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -11,18 +11,18 @@ abstract class RssCollectorTriggerSupport(
1111

1212
protected fun runWithDuration() {
1313
val startAt = System.currentTimeMillis()
14-
log.info("[${this.javaClass.simpleName}] Starting collection")
14+
log.info("Starting collection")
1515

1616
try {
1717
collector.execute()
1818
val elapsedMs = System.currentTimeMillis() - startAt
1919
log.info(
20-
"[${this.javaClass.simpleName}] Finished collection, elapsedMs=$elapsedMs",
20+
"Finished collection, elapsedMs=$elapsedMs",
2121
)
2222
} catch (ex: Throwable) {
2323
val elapsedMs = System.currentTimeMillis() - startAt
2424
log.error(
25-
"[${this.javaClass.simpleName}] Failed collection, elapsedMs=$elapsedMs",
25+
"Failed collection, elapsedMs=$elapsedMs",
2626
ex
2727
)
2828
throw RssCollectionExecutionException(

‎batch/rss/src/main/kotlin/site/techmoa/batch/rss/usecase/CollectByRssReaderUseCase.kt‎

Lines changed: 78 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,35 +1,106 @@
11
package site.techmoa.batch.rss.usecase
22

33
import org.slf4j.LoggerFactory
4+
import org.springframework.beans.factory.annotation.Qualifier
45
import org.springframework.stereotype.Component
5-
import org.springframework.transaction.annotation.Transactional
6+
import site.techmoa.batch.rss.domain.model.Article
7+
import site.techmoa.batch.rss.domain.model.Blog
68
import site.techmoa.batch.rss.domain.model.BlogStatus
79
import site.techmoa.batch.rss.port.ArticlePort
810
import site.techmoa.batch.rss.port.BlogPort
911
import site.techmoa.batch.rss.support.RssClient
1012
import site.techmoa.batch.rss.trigger.CollectRssUseCase
13+
import java.util.concurrent.CompletableFuture
14+
import java.util.concurrent.Executor
15+
import java.util.concurrent.atomic.AtomicInteger
1116

1217
@Component
1318
class CollectByRssReaderUseCase(
1419
private val rssClient: RssClient,
1520
private val articlePort: ArticlePort,
16-
private val blogPort: BlogPort
21+
private val blogPort: BlogPort,
22+
@Qualifier("rssFetchExecutor") private val fetchExecutor: Executor,
1723
) : CollectRssUseCase {
1824

1925
private val log = LoggerFactory.getLogger(this.javaClass)
2026

21-
@Transactional
27+
private data class FetchResult(
28+
val blogId: Long,
29+
val blogName: String,
30+
val articles: List<Article>,
31+
val error: Throwable?
32+
)
33+
2234
override fun execute() {
2335
// 1. 구독 블로그 조회
24-
log.info("[${this.javaClass.simpleName}] Select Blogs List")
36+
log.info("Select Blogs List")
2537
val blogs = blogPort.findAllBy(BlogStatus.ACTIVE)
38+
if (blogs.isEmpty()) return
2639

2740
// 2. 각 블로그 RSS에서 아티클 수집
28-
log.info("[${this.javaClass.simpleName}] Start Collecting Articles from RSS")
29-
val articles = blogs.flatMap { rssClient.fetch(it) }
41+
log.info("Start Collecting Articles from RSS")
42+
val results = fetchWithLimitedConcurrency(blogs)
43+
44+
val articles = results
45+
.filter { it.error == null }
46+
.flatMap { it.articles }
47+
48+
val failedCount = results.count { it.error != null }
49+
if (failedCount > 0) {
50+
log.warn(
51+
"RSS collect partial failure. total=${results.size}, failed=$failedCount",
52+
)
53+
}
3054

3155
// 3. 아티클 저장
32-
log.info("[${this.javaClass.simpleName}] Save New Articles")
56+
log.info("Save New Articles")
3357
articlePort.saveAllIgnoringDuplicates(articles)
3458
}
59+
60+
private fun fetchWithLimitedConcurrency(blogs: List<Blog>): List<FetchResult> {
61+
val total = blogs.size
62+
val failures = AtomicInteger(0)
63+
64+
val futures = blogs.map { blog ->
65+
CompletableFuture.supplyAsync({ fetchSingleBlog(blog, failures) }, fetchExecutor)
66+
.whenComplete { _, ex ->
67+
log.info("Future done blogId={}", blog.id)
68+
}
69+
}
70+
71+
CompletableFuture.allOf(*futures.toTypedArray()).join()
72+
73+
val results = futures.map { it.join() }
74+
log.info(
75+
"RSS collect finished. total=$total, success=${total - failures.get()}, failed=${failures.get()}",
76+
)
77+
78+
return results
79+
}
80+
81+
private fun fetchSingleBlog(
82+
blog: Blog,
83+
failures: AtomicInteger,
84+
): FetchResult {
85+
return try {
86+
FetchResult(
87+
blogId = blog.id,
88+
blogName = blog.name,
89+
articles = rssClient.fetch(blog),
90+
error = null,
91+
)
92+
} catch (ex: Throwable) {
93+
failures.incrementAndGet()
94+
log.warn(
95+
"Failed to collect RSS. blogId=${blog.id}, blogName=${blog.name}",
96+
ex,
97+
)
98+
FetchResult(
99+
blogId = blog.id,
100+
blogName = blog.name,
101+
articles = emptyList(),
102+
error = ex,
103+
)
104+
}
105+
}
35106
}

0 commit comments

Comments
 (0)