Feat: 카프카 추가 및 연동(weekly -> master)

linknamu/build.gradle

@@ -47,7 +47,7 @@ dependencies {
     implementation 'org.springframework.boot:spring-boot-starter-security'
     implementation 'org.springframework.boot:spring-boot-starter-validation'
     implementation 'org.springframework.boot:spring-boot-starter-web'
-
+    implementation 'org.springframework.kafka:spring-kafka'
     implementation 'org.springdoc:springdoc-openapi-starter-webmvc-ui:2.0.2'
 
     compileOnly 'org.projectlombok:lombok'
@@ -59,6 +59,7 @@ dependencies {
 
     testImplementation 'org.springframework.boot:spring-boot-starter-test'
     testImplementation 'org.springframework.security:spring-security-test'
+    testImplementation 'org.springframework.kafka:spring-kafka-test'
 
     // testcontainers
     testImplementation "org.junit.jupiter:junit-jupiter:5.8.1"
@@ -89,7 +90,6 @@ dependencies {
 
     // AWS S3
     implementation 'org.springframework.cloud:spring-cloud-starter-aws:2.2.6.RELEASE'
-    //버전호환안되는거임
 }
 
 tasks.named('test') {

linknamu/src/main/java/com/kakao/linknamu/core/kafka/consumer/GoogleDocsConsumer.java

@@ -0,0 +1,32 @@
+package com.kakao.linknamu.core.kafka.consumer;
+
+import static com.kakao.linknamu.core.util.KafkaTopics.*;
+
+import java.util.List;
+
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Service;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.kakao.linknamu.bookmark.entity.Bookmark;
+import com.kakao.linknamu.bookmark.service.BookmarkService;
+import com.kakao.linknamu.thirdparty.googledocs.entity.GooglePage;
+import com.kakao.linknamu.thirdparty.googledocs.util.GoogleDocsProvider;
+
+import lombok.RequiredArgsConstructor;
+
+@RequiredArgsConstructor
+@Service
+public class GoogleDocsConsumer {
+	private final BookmarkService bookmarkService;
+	private final ObjectMapper om;
+	private final GoogleDocsProvider googleDocsProvider;
+
+	@KafkaListener(topics = {GOOGLE_DOCS_TOPIC}, groupId = "group-id-linknamu")
+	public void googleDocsConsumer(String message) throws JsonProcessingException {
+		GooglePage googlePage = om.readValue(message, GooglePage.class);
+		List<Bookmark> bookmarkList = googleDocsProvider.getLinks(googlePage);
+		bookmarkService.batchInsertBookmark(bookmarkList);
+	}
+}

linknamu/src/main/java/com/kakao/linknamu/core/kafka/consumer/NotionConsumer.java

@@ -0,0 +1,63 @@
+package com.kakao.linknamu.core.kafka.consumer;
+
+import static com.kakao.linknamu.core.util.KafkaTopics.*;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import org.json.simple.JSONArray;
+import org.json.simple.JSONObject;
+import org.json.simple.parser.JSONParser;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Service;
+import org.springframework.web.client.HttpClientErrorException;
+import org.springframework.web.client.RestTemplate;
+import org.yaml.snakeyaml.parser.ParserException;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.kakao.linknamu.bookmark.entity.Bookmark;
+import com.kakao.linknamu.bookmark.service.BookmarkService;
+import com.kakao.linknamu.category.entity.Category;
+import com.kakao.linknamu.core.util.JsoupResult;
+import com.kakao.linknamu.core.util.JsoupUtils;
+import com.kakao.linknamu.thirdparty.notion.dto.NotionKafkaReqeusetDto;
+import com.kakao.linknamu.thirdparty.notion.entity.NotionPage;
+import com.kakao.linknamu.thirdparty.notion.repository.NotionPageJpaRepository;
+import com.kakao.linknamu.thirdparty.notion.util.InvalidNotionApiException;
+import com.kakao.linknamu.thirdparty.notion.util.NotionApiUriBuilder;
+import com.kakao.linknamu.thirdparty.notion.util.NotionProvider;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@RequiredArgsConstructor
+@Service
+public class NotionConsumer {
+	private final BookmarkService bookmarkService;
+	private final ObjectMapper om;
+	private final NotionProvider notionProvider;
+
+
+	@KafkaListener(topics = {NOTION_TOPIC}, groupId = "group-id-linknamu")
+	public void notionConsumer(String message) throws JsonProcessingException {
+		NotionKafkaReqeusetDto requsetDto = om.readValue(message, NotionKafkaReqeusetDto.class);
+
+		List<Bookmark> bookmarkList = notionProvider.getPageLinks(
+			requsetDto.pageId(),
+			requsetDto.accessToken(),
+			requsetDto.categoryId()
+		);
+		bookmarkService.batchInsertBookmark(bookmarkList);
+	}
+}

linknamu/src/main/java/com/kakao/linknamu/core/util/KafkaTopics.java

@@ -0,0 +1,7 @@
+package com.kakao.linknamu.core.util;
+
+public class KafkaTopics {
+	public static final String GOOGLE_DOCS_TOPIC = "google_docs";
+	public static final String NOTION_TOPIC = "notion";
+	public static final String KAKAO_TOPIC = "kakao";
+}

linknamu/src/main/java/com/kakao/linknamu/thirdparty/googledocs/service/GoogleDocsApiBatchService.java

@@ -11,6 +11,7 @@
 import com.kakao.linknamu.core.config.GoogleDocsConfig;
 import com.kakao.linknamu.thirdparty.googledocs.entity.GooglePage;
 import com.kakao.linknamu.thirdparty.googledocs.repository.GooglePageJpaRepository;
+import com.kakao.linknamu.thirdparty.googledocs.util.GoogleDocsProvider;
 import com.kakao.linknamu.thirdparty.googledocs.util.InvalidGoogleDocsApiException;
 
 import com.kakao.linknamu.core.util.JsoupResult;
@@ -22,20 +23,15 @@
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 
-import java.io.IOException;
-import java.security.GeneralSecurityException;
-import java.util.HashSet;
 import java.util.List;
-import java.util.Set;
 
-import static com.kakao.linknamu.core.config.GoogleDocsConfig.getCredentials;
 
 @Slf4j
 @RequiredArgsConstructor
 @Service
 public class GoogleDocsApiBatchService {
 	private final GooglePageJpaRepository googlePageJpaRepository;
-	private final JsoupUtils jsoupUtils;
+	private final GoogleDocsProvider googleDocsProvider;
 	private final BookmarkService bookmarkService;
 
 	@Scheduled(cron = "0 0 0/1 * * *", zone = "Asia/Seoul")
@@ -46,64 +42,16 @@ public void googleDocsApiCronJob() {
 		// 활성화된 구글 독스 페이지들에 대해 배치를 실행한다.
 		activeGoogleDocsPages.forEach((GooglePage gp) -> {
 			try {
-				List<Bookmark> resultBookmarks = getLinks(gp);
+				List<Bookmark> resultBookmarks = googleDocsProvider.getLinks(gp);
 				bookmarkService.batchInsertBookmark(resultBookmarks);
 			} catch (InvalidGoogleDocsApiException e) {
 				gp.deactivate();
 				googlePageJpaRepository.save(gp);
+			} catch (Exception e) {
+				log.error(e.getMessage());
 			}
 		});
 	}
 
-	private List<Bookmark> getLinks(GooglePage googlePage) {
-		Set<Bookmark> resultBookmarks = new HashSet<>();
-		try {
-			// 서비스 생성
-			final NetHttpTransport httpTransport = GoogleNetHttpTransport.newTrustedTransport();
-			Docs service = new Docs.Builder(
-				httpTransport,
-				GoogleDocsConfig.getJSON_FACTORY(),
-				getCredentials(httpTransport))
-				.setApplicationName(GoogleDocsConfig.getAPPLICATION_NAME())
-				.build();
 
-			// google docs 객체 생성 및 get API를 사용해서 link 항목 불러오기
-			Document response = service.documents().get(googlePage.getDocumentId()).execute();
-			List<StructuralElement> contents = response.getBody().getContent();
-			for (StructuralElement e : contents) {
-				if (e.getParagraph() != null && e.getParagraph().getElements() != null) {
-					List<ParagraphElement> elements = e.getParagraph().getElements();
-					for (ParagraphElement pe : elements) {
-						if (pe.getTextRun() != null && pe.getTextRun().getTextStyle() != null
-							&& pe.getTextRun().getTextStyle().getLink() != null) {
-							String link = pe.getTextRun().getTextStyle().getLink().getUrl();
-							if (link != null) {
-								// 만약 한번 연동한 링크라면 더 이상 진행하지 않는다.
-								if (bookmarkService.existByBookmarkLinkAndCategoryId(link,
-									googlePage.getCategory().getCategoryId())) {
-									continue;
-								}
-
-								JsoupResult jsoupResult = jsoupUtils.getTitleAndImgUrl(link);
-								resultBookmarks.add(Bookmark.builder()
-									.bookmarkLink(link)
-									.bookmarkName(jsoupResult.getTitle())
-									.bookmarkThumbnail(jsoupResult.getImageUrl())
-									.category(googlePage.getCategory())
-									.build());
-							}
-						}
-					}
-				}
-			}
-		} catch (GeneralSecurityException error) {
-			log.error("구글 인증에 문제가 있습니다.");
-			throw new InvalidGoogleDocsApiException();
-		} catch (IOException error) {
-			log.error("구글Docs 연동 중 문제가 발생했습니다.");
-			throw new InvalidGoogleDocsApiException();
-		}
-
-		return resultBookmarks.stream().toList();
-	}
 }

linknamu/src/main/java/com/kakao/linknamu/thirdparty/googledocs/service/GoogleDocsApiService.java

@@ -1,5 +1,11 @@
 package com.kakao.linknamu.thirdparty.googledocs.service;
 
+import static com.kakao.linknamu.core.util.KafkaTopics.*;
+
+import java.util.concurrent.CompletableFuture;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.kakao.linknamu.category.entity.Category;
 import com.kakao.linknamu.category.service.CategoryService;
 import com.kakao.linknamu.core.exception.Exception400;
@@ -17,6 +23,9 @@
 
 import lombok.RequiredArgsConstructor;
 
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -28,8 +37,19 @@ public class GoogleDocsApiService {
 	private final CategoryService categoryService;
 	private final GoogleDocsProvider googleDocsProvider;
 	private final WorkspaceService workspaceService;
+	private final KafkaTemplate<String, String> kafkaTemplate;
+	private final ObjectMapper om;
+
 	private static final String DEFAULT_WORKSPACE_NAME = "Google Docs";
 
+	/*
+		로직
+		1. 입력 문서ID가 유효한지 판단
+		2. 기존에 등록한 적이 있는지 확인 -> 있다면 예외처리
+		3. 링크들을 연동할 워크스페이스 및 카테고리 생성
+		4. GooglePage 생성
+		5. 해당 페이지를 파싱하는 Task를 다른 쓰레드에게 위임 후 종료
+	 */
 	public void createDocsApi(RegisterGoogleDocsRequestDto dto, User user) {
 		String pageName = googleDocsProvider.getGoogleDocsTitle(dto.documentId());
 
@@ -54,6 +74,8 @@ public void createDocsApi(RegisterGoogleDocsRequestDto dto, User user) {
 			.pageName(pageName)
 			.build();
 		googlePageJpaRepository.save(googlePage);
+
+		googleDocsRequestToKafka(googlePage);
 	}
 
 	public void deleteDocsPage(User user, Long docsPageId) {
@@ -68,4 +90,17 @@ private void validUser(GooglePage googlePage, User user) {
 			throw new Exception403(GoogleDocsExceptionStatus.DOCS_FORBIDDEN);
 		}
 	}
+
+	// 초기 구글문서 연동 생성 시 데이터를 가져오는 것을 다른 쓰레드에 위임
+	private void googleDocsRequestToKafka(GooglePage googlePage) {
+		try {
+			String message = om.writeValueAsString(googlePage);
+			CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(
+				GOOGLE_DOCS_TOPIC,
+				message
+			);
+		} catch (JsonProcessingException ignored) {
+		}
+
+	}
 }

linknamu/src/main/java/com/kakao/linknamu/thirdparty/googledocs/util/GoogleDocsProvider.java

@@ -5,10 +5,17 @@
 import com.google.api.client.http.javanet.NetHttpTransport;
 import com.google.api.services.docs.v1.Docs;
 import com.google.api.services.docs.v1.model.Document;
+import com.google.api.services.docs.v1.model.ParagraphElement;
+import com.google.api.services.docs.v1.model.StructuralElement;
+import com.kakao.linknamu.bookmark.entity.Bookmark;
+import com.kakao.linknamu.bookmark.service.BookmarkService;
 import com.kakao.linknamu.core.config.GoogleDocsConfig;
 import com.kakao.linknamu.core.exception.Exception400;
 import com.kakao.linknamu.core.exception.Exception500;
+import com.kakao.linknamu.core.util.JsoupResult;
+import com.kakao.linknamu.core.util.JsoupUtils;
 import com.kakao.linknamu.thirdparty.googledocs.GoogleDocsExceptionStatus;
+import com.kakao.linknamu.thirdparty.googledocs.entity.GooglePage;
 
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -17,13 +24,18 @@
 
 import java.io.IOException;
 import java.security.GeneralSecurityException;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
 
 import static com.kakao.linknamu.core.config.GoogleDocsConfig.getCredentials;
 
 @Slf4j
 @RequiredArgsConstructor
 @Service
 public class GoogleDocsProvider {
+	private final BookmarkService bookmarkService;
+	private final JsoupUtils jsoupUtils;
 
 	public String getGoogleDocsTitle(String documentId) {
 		try {
@@ -53,4 +65,56 @@ public String getGoogleDocsTitle(String documentId) {
 			throw new Exception500(GoogleDocsExceptionStatus.GOOGLE_DOCS_LINK_ERROR);
 		}
 	}
+
+	public List<Bookmark> getLinks(GooglePage googlePage) {
+		Set<Bookmark> resultBookmarks = new HashSet<>();
+		try {
+			// 서비스 생성
+			final NetHttpTransport httpTransport = GoogleNetHttpTransport.newTrustedTransport();
+			Docs service = new Docs.Builder(
+				httpTransport,
+				GoogleDocsConfig.getJSON_FACTORY(),
+				getCredentials(httpTransport))
+				.setApplicationName(GoogleDocsConfig.getAPPLICATION_NAME())
+				.build();
+
+			// google docs 객체 생성 및 get API를 사용해서 link 항목 불러오기
+			Document response = service.documents().get(googlePage.getDocumentId()).execute();
+			List<StructuralElement> contents = response.getBody().getContent();
+			for (StructuralElement e : contents) {
+				if (e.getParagraph() != null && e.getParagraph().getElements() != null) {
+					List<ParagraphElement> elements = e.getParagraph().getElements();
+					for (ParagraphElement pe : elements) {
+						if (pe.getTextRun() != null && pe.getTextRun().getTextStyle() != null
+							&& pe.getTextRun().getTextStyle().getLink() != null) {
+							String link = pe.getTextRun().getTextStyle().getLink().getUrl();
+							if (link != null) {
+								// 만약 한번 연동한 링크라면 더 이상 진행하지 않는다.
+								if (bookmarkService.existByBookmarkLinkAndCategoryId(link,
+									googlePage.getCategory().getCategoryId())) {
+									continue;
+								}
+
+								JsoupResult jsoupResult = jsoupUtils.getTitleAndImgUrl(link);
+								resultBookmarks.add(Bookmark.builder()
+									.bookmarkLink(link)
+									.bookmarkName(jsoupResult.getTitle())
+									.bookmarkThumbnail(jsoupResult.getImageUrl())
+									.category(googlePage.getCategory())
+									.build());
+							}
+						}
+					}
+				}
+			}
+		} catch (GeneralSecurityException error) {
+			log.error("구글 인증에 문제가 있습니다.");
+			throw new InvalidGoogleDocsApiException();
+		} catch (IOException error) {
+			log.error("구글Docs 연동 중 문제가 발생했습니다.");
+			throw new InvalidGoogleDocsApiException();
+		}
+
+		return resultBookmarks.stream().toList();
+	}
 }