Implement MessageSenderService

processor/src/main/java/com/myoa/engineering/crawl/ppomppu/processor/configuration/H2ConsoleConfiguration.java

@@ -11,7 +11,7 @@ import org.springframework.context.event.ContextRefreshedEvent;
 import org.springframework.context.event.EventListener;
 
 @Slf4j
-@Profile("datasource-development")
+@Profile({"datasource-local", "datasource-development"})
 @Configuration
 public class H2ConsoleConfiguration {
 

processor/src/main/java/com/myoa/engineering/crawl/ppomppu/processor/controller/CrawlAPIController.java

@@ -48,8 +48,8 @@ public class CrawlAPIController {
         Mono<List<PpomppuArticle>> articles =
             ppomppuRSSFeedService.getArticles(boardName)
                                  .doOnNext(e -> ppomppuArticleService.filterOnlyNewArticles(boardName, e))
-                                 .doOnNext(e -> ppomppuArticleService.save(boardName, e))
-                                 .doOnNext(messageSenderService::sendMessageToSlack);
+                                 .doOnNext(e -> messageSenderService.sendMessageToSlack(e))
+                                 .doOnNext(e -> ppomppuArticleService.save(boardName, e));
 
         return articles.then(Mono.just(APIResponse.success(result.done())));
     }

processor/src/main/java/com/myoa/engineering/crawl/ppomppu/processor/infrastructure/client/MessageSenderAPIClient.java

@@ -1,24 +1,26 @@
 package com.myoa.engineering.crawl.ppomppu.processor.infrastructure.client;
 
-import org.springframework.stereotype.Component;
-import org.springframework.web.reactive.function.client.WebClient;
-import org.springframework.web.reactive.function.client.WebClientRequestException;
-
 import com.myoa.engineering.crawl.ppomppu.processor.dto.constant.WebClientPropertiesUnitName;
 import com.myoa.engineering.crawl.ppomppu.support.dto.SimpleMessageDTO;
 import com.myoa.engineering.crawl.ppomppu.support.webclient.factory.WebClientFilterFactory;
+import com.myoa.engineering.crawl.ppomppu.support.webclient.factory.WebFluxExchangeStragiesFactory;
 import com.myoa.engineering.crawl.ppomppu.support.webclient.properties.WebClientProperties;
 import com.myoa.engineering.crawl.ppomppu.support.webclient.properties.WebClientProperties.WebClientPropertiesUnit;
-
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.stereotype.Component;
+import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.web.reactive.function.client.WebClientRequestException;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
 
 /**
  * PpomppuNotifierSenderAPIClient
+ *
  * @author Shin Woo-jin (woo-jin.shin@linecorp.com)
  * @since 2021-11-17
- *
  */
 @Slf4j
 @Component
@@ -31,6 +33,8 @@ public class MessageSenderAPIClient {
             webClientProperties.find(WebClientPropertiesUnitName.PPOMPPU_NOTIFIER_SENDER_API.getUnitName());
         this.webClient = WebClient.builder()
                                   .baseUrl(webClientPropertiesUnit.getBaseUrl())
+                                  .exchangeStrategies(WebFluxExchangeStragiesFactory.ofDefault())
+                                  .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
                                   .filter(WebClientFilterFactory.logRequest())
                                   .filter(WebClientFilterFactory.logResponse())
                                   .build();
@@ -40,11 +44,12 @@ public class MessageSenderAPIClient {
         return webClient.post()
                         .uri("/api/v1/messages/sendMessage/messengers/slack")
                         .bodyValue(dto)
-                        .exchangeToMono(e -> e.bodyToMono(String.class))
+                        .exchangeToMono(e -> e.bodyToMono(new ParameterizedTypeReference<String>() {}))
                         .publishOn(Schedulers.boundedElastic())
                         .onErrorResume(WebClientRequestException.class, t -> {
                             log.info("Exception occured, ignoring. : {}", t.getClass().getSimpleName());
                             return Mono.empty();
-                        });
+                        })
+                        .doOnNext(e -> log.info("response: {} ", e));
     }
 }

processor/src/main/java/com/myoa/engineering/crawl/ppomppu/processor/service/MessageSenderService.java

@@ -9,6 +9,7 @@ import com.myoa.engineering.crawl.ppomppu.processor.dto.PpomppuArticleTransforme
 import com.myoa.engineering.crawl.ppomppu.processor.infrastructure.client.MessageSenderAPIClient;
 
 import lombok.extern.slf4j.Slf4j;
+import reactor.core.publisher.Mono;
 
 /**
  * MessageSenderService
@@ -26,12 +27,12 @@ public class MessageSenderService {
         this.messageSenderAPIClient = messageSenderAPIClient;
     }
 
-    public void sendMessageToSlack(PpomppuArticle article) {
-        messageSenderAPIClient.sendMessageToSlack(PpomppuArticleTransformer.TRANSFORM_TO_MESSAGE_DTO.apply(article));
+    public Mono<String> sendMessageToSlack(PpomppuArticle article) {
+        return messageSenderAPIClient.sendMessageToSlack(PpomppuArticleTransformer.TRANSFORM_TO_MESSAGE_DTO.apply(article));
     }
 
-    public void sendMessageToSlack(List<PpomppuArticle> articles) {
-        messageSenderAPIClient.sendMessageToSlack(PpomppuArticleTransformer.transform(articles));
+    public Mono<String> sendMessageToSlack(List<PpomppuArticle> articles) {
+        return messageSenderAPIClient.sendMessageToSlack(PpomppuArticleTransformer.transform(articles));
     }
 
 }

processor/src/main/java/com/myoa/engineering/crawl/ppomppu/processor/service/PpomppuArticleService.java

@@ -38,7 +38,7 @@ public class PpomppuArticleService {
     }
 
     @Transactional
-    public void save(PpomppuBoardName boardName, List<PpomppuArticle> articles) {
+    public List<PpomppuArticle> save(PpomppuBoardName boardName, List<PpomppuArticle> articles) {
         Long latestArticleId = articles.stream()
                                        .map(PpomppuArticle::getArticleId)
                                        .max(Long::compareTo)
@@ -54,6 +54,6 @@ public class PpomppuArticleService {
                                                                                                               latestArticleId)));
 
         // save real articles.
-        ppomppuArticleRepository.saveAll(articles);
+        return ppomppuArticleRepository.saveAll(articles);
     }
 }

processor/src/main/resources/application-local.yml

@@ -0,0 +1,12 @@
+spring:
+  config:
+    activate:
+      on-profile: local
+    import:
+      - "configserver:http://localhost:20085"
+      - classpath:/local/webclient.yml
+
+server:
+  port: 20081
+
+    # import: optional:configserver:http://localhost:11080 # can be start up even config server was not found.

processor/src/main/resources/application.yml

@@ -4,8 +4,9 @@ spring:
   main:
     allow-bean-definition-overriding: true
   profiles:
-    active: ${SPRING_ACTIVE_PROFILE:development}
+    active: ${SPRING_ACTIVE_PROFILE:local}
     group:
+      local: "local,datasource-local"
       development: "development,datasource-development"
       production: "production, datasource-production"
   freemarker:

processor/src/main/resources/development/webclient.yml

@@ -2,4 +2,4 @@ webclient:
   init: true
   units:
     - unit-name: ppn-sender-api
-      base-url: http://localhost:20081
+      base-url: http://localhost:20082

processor/src/main/resources/local/webclient.yml

@@ -0,0 +1,5 @@
+webclient:
+  init: true
+  units:
+    - unit-name: ppn-sender-api
+      base-url: http://localhost:20082

processor/src/main/resources/logback-spring.xml

@@ -1,6 +1,10 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <configuration>
   <springProperty name="DEFAULT_LEVEL_CONFIG" source="log.defaultLevel" />
+  <springProfile name="local">
+    <include resource="logback/logback-development.xml" />
+    <logger name="org.apache.kafka" level="INFO" />
+  </springProfile>
   <springProfile name="development">
     <include resource="logback/logback-development.xml" />
     <logger name="org.apache.kafka" level="INFO" />

processor/src/main/resources/production/webclient.yml

@@ -2,4 +2,4 @@ webclient:
   init: true
   units:
     - unit-name: ppn-sender-api
-      base-url: http://ppn_sender:20081
+      base-url: http://ppn_sender:20082