Merge pull request 'feat/concurrency' (#63) from feat/concurrency into main

Reviewed-on: #63

src/main/java/com/magamochi/catalog/queue/consumer/MangaIngestConsumer.java

@@ -13,7 +13,9 @@ import org.springframework.stereotype.Service;
 public class MangaIngestConsumer {
   private final MangaIngestService mangaIngestService;
 
-  @RabbitListener(queues = "${queues.manga-ingest}")
+  @RabbitListener(
+      queues = "${queues.manga-ingest}",
+      concurrency = "${queues.manga-ingest-concurrency:3-5}")
   public void receiveMangaIngestCommand(MangaIngestCommand command) {
     log.info("Received manga ingest command: {}", command);
     mangaIngestService.ingestManga(command.providerId(), command.mangaTitle(), command.url());

src/main/java/com/magamochi/catalog/service/MangaIngestService.java

@@ -36,6 +36,15 @@ public class MangaIngestService {
       return;
     }
 
+    if (mangaIngestReviewRepository.existsByMangaTitleIgnoreCaseAndContentProvider_Id(
+        mangaTitle, contentProviderId)) {
+      log.info(
+          "Manga ingest review already exists for manga with mangaTitle '{}' and provider {}, skipping ingest",
+          mangaTitle,
+          contentProviderId);
+      return;
+    }
+
     var manga = mangaResolutionService.findOrCreateManga(mangaTitle);
     if (isNull(manga)) {
       createMangaIngestReview(mangaTitle, url, contentProviderId);
@@ -72,15 +81,6 @@ public class MangaIngestService {
         mangaTitle,
         contentProviderId);
 
-    if (mangaIngestReviewRepository.existsByMangaTitleIgnoreCaseAndContentProvider_Id(
-        mangaTitle, contentProviderId)) {
-      log.info(
-          "Manga ingest review already exists for manga with mangaTitle '{}' and provider {}, skipping review creation",
-          mangaTitle,
-          contentProviderId);
-      return;
-    }
-
     try {
       var contentProvider = contentProviderService.find(contentProviderId);
 

src/main/java/com/magamochi/ingestion/queue/consumer/ProviderPageIngestConsumer.java

@@ -13,7 +13,9 @@ import org.springframework.stereotype.Service;
 public class ProviderPageIngestConsumer {
   private final IngestionService ingestionService;
 
-  @RabbitListener(queues = "${queues.provider-page-ingest}")
+  @RabbitListener(
+      queues = "${queues.provider-page-ingest}",
+      concurrency = "${queues.provider-page-ingest-concurrency:2-4}")
   public void receiveProviderPageIngestCommand(ProviderPageIngestCommand command) {
     log.info("Received provider page ingest command: {}", command);
     ingestionService.fetchProviderPageMangas(command.providerId(), command.page());

src/main/resources/application.yml

@@ -20,6 +20,11 @@ spring:
       max-request-size: 4GB
   cloud:
     openfeign:
+      client:
+        config:
+          default:
+            connect-timeout: 10000
+            read-timeout: 30000
   rabbitmq:
     host: ${RABBITMQ_HOST}
     port: ${RABBITMQ_PORT}
@@ -28,6 +33,7 @@ spring:
     listener:
       simple:
         default-requeue-rejected: false
+        prefetch: 10
   data:
     redis:
       host: ${REDIS_HOST:localhost}
@@ -65,12 +71,14 @@ topics:
 
 queues:
   manga-ingest: ${MANGA_INGEST_QUEUE:mangamochi.manga.ingest}
+  manga-ingest-concurrency: 3-5
   manga-update: ${MANGA_UPDATE_QUEUE:mangamochi.manga.update}
   manga-content-ingest: ${MANGA_CONTENT_INGEST_QUEUE:mangamochi.manga.content.ingest}
   manga-content-image-ingest: ${MANGA_CONTENT_IMAGE_INGEST_QUEUE:mangamochi.manga.content.image.ingest}
   manga-content-image-update: ${MANGA_CONTENT_IMAGE_UPDATE_QUEUE:mangamochi.manga.content.image.update}
   manga-content-image-delete: ${MANGA_CONTENT_IMAGE_DELETE_QUEUE:mangamochi.manga.content.image.delete}
   provider-page-ingest: ${PROVIDER_PAGE_INGEST_QUEUE:mangamochi.provider.page.ingest}
+  provider-page-ingest-concurrency: 2-4
   image-fetch: ${IMAGE_FETCH_QUEUE:mangamochi.image.fetch}
   manga-cover-update: ${MANGA_COVER_UDPATE_QUEUE:mangamochi.manga.cover.update}
   file-import: ${FILE_IMPORT_QUEUE:mangamochi.file.import}