Merge pull request #41 from lotteon2/O2F-778-SNS-/feat/member-event

O2 f 778 sns /feat/member event

build.gradle

@@ -38,7 +38,7 @@ dependencies {
     implementation 'org.springframework.cloud:spring-cloud-starter-config'
     implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
     implementation 'org.springframework.boot:spring-boot-starter-data-redis'
-//    implementation 'org.springframework.kafka:spring-kafka'
+    implementation 'org.springframework.kafka:spring-kafka'
     implementation 'org.springframework.boot:spring-boot-starter-validation'
     implementation 'org.springframework.boot:spring-boot-starter-web'
     implementation 'org.springframework.cloud:spring-cloud-starter-circuitbreaker-resilience4j'
@@ -54,6 +54,8 @@ dependencies {
     // aws
     implementation 'org.springframework.cloud:spring-cloud-starter-aws:2.2.4.RELEASE'
 
+    implementation group: 'io.github.dailyon-maven', name: 'daily-on-common', version: '0.0.6'
+
     compileOnly 'org.projectlombok:lombok'
     runtimeOnly 'com.h2database:h2'
     runtimeOnly 'com.mysql:mysql-connector-j'

src/main/java/com/dailyon/snsservice/entity/Member.java

@@ -59,4 +59,13 @@ public void decreaseFollowingCount() {
   public void decreaseFollowerCount() {
     this.followerCount -= 1;
   }
+
+  public static Member createMember(String nickname, String profileImgUrl, String code) {
+    return Member.builder().nickname(nickname).profileImgUrl(profileImgUrl).code(code).build();
+  }
+
+  public void updateMember(String nickname, String profileImgUrl) {
+    this.nickname = nickname;
+    this.profileImgUrl = profileImgUrl;
+  }
 }

src/main/java/com/dailyon/snsservice/kafka/MemberEventListener.java

@@ -0,0 +1,54 @@
+package com.dailyon.snsservice.kafka;
+
+import com.dailyon.snsservice.service.member.MemberService;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import dailyon.domain.common.KafkaTopic;
+import dailyon.domain.sns.kafka.dto.MemberCreateDTO;
+import dailyon.domain.sns.kafka.dto.MemberUpdateDTO;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.Acknowledgment;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class MemberEventListener {
+
+  private final MemberService memberService;
+  private final ObjectMapper objectMapper;
+
+  @KafkaListener(topics = KafkaTopic.CREATE_MEMBER_FOR_SNS)
+  public void createMember(String message, Acknowledgment ack) {
+    MemberCreateDTO memberCreateDTO;
+    try {
+      memberCreateDTO = objectMapper.readValue(message, MemberCreateDTO.class);
+      memberService.createMember(memberCreateDTO);
+
+      ack.acknowledge();
+    } catch (JsonProcessingException e) {
+      log.error(e.getMessage());
+      e.printStackTrace();
+    } catch (Exception e) {
+      log.error(e.getMessage());
+    }
+  }
+
+  @KafkaListener(topics = KafkaTopic.UPDATE_MEMBER_FOR_SNS)
+  public void updateMember(String message, Acknowledgment ack) {
+    MemberUpdateDTO memberUpdateDTO;
+    try {
+      memberUpdateDTO = objectMapper.readValue(message, MemberUpdateDTO.class);
+      memberService.updateMember(memberUpdateDTO);
+
+      ack.acknowledge();
+    } catch (JsonProcessingException e) {
+      log.error(e.getMessage());
+      e.printStackTrace();
+    } catch (Exception e) {
+      log.error(e.getMessage());
+    }
+  }
+}

src/main/java/com/dailyon/snsservice/repository/member/MemberRepository.java

@@ -1,8 +1,11 @@
 package com.dailyon.snsservice.repository.member;
 
 import com.dailyon.snsservice.dto.response.member.OOTDMemberProfileResponse;
+import com.dailyon.snsservice.entity.Member;
 
 public interface MemberRepository {
 
   OOTDMemberProfileResponse findOOTDMemberProfile(Long memberId, Long followerId);
+
+  void save(Member member);
 }

src/main/java/com/dailyon/snsservice/repository/member/MemberRepositoryImpl.java

@@ -1,6 +1,7 @@
 package com.dailyon.snsservice.repository.member;
 
 import com.dailyon.snsservice.dto.response.member.OOTDMemberProfileResponse;
+import com.dailyon.snsservice.entity.Member;
 import com.dailyon.snsservice.exception.MemberEntityNotFoundException;
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Repository;
@@ -17,4 +18,9 @@ public OOTDMemberProfileResponse findOOTDMemberProfile(Long ootdMemberId, Long m
         .findOOTDMemberProfile(ootdMemberId, memberId)
         .orElseThrow(MemberEntityNotFoundException::new);
   }
+
+  @Override
+  public void save(Member member) {
+    memberJpaRepository.save(member);
+  }
 }

src/main/java/com/dailyon/snsservice/service/member/MemberService.java

@@ -1,7 +1,10 @@
 package com.dailyon.snsservice.service.member;
 
 import com.dailyon.snsservice.dto.response.member.OOTDMemberProfileResponse;
+import com.dailyon.snsservice.entity.Member;
 import com.dailyon.snsservice.repository.member.MemberRepository;
+import dailyon.domain.sns.kafka.dto.MemberCreateDTO;
+import dailyon.domain.sns.kafka.dto.MemberUpdateDTO;
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
@@ -12,8 +15,23 @@
 public class MemberService {
 
   private final MemberRepository memberRepository;
+  private final MemberReader memberReader;
 
   public OOTDMemberProfileResponse getOOTDMemberProfile(Long ootdMemberId, Long memberId) {
     return memberRepository.findOOTDMemberProfile(ootdMemberId, memberId);
   }
+
+  public void createMember(MemberCreateDTO memberCreateDTO) {
+    Member member =
+        Member.createMember(
+            memberCreateDTO.getNickname(),
+            memberCreateDTO.getProfileImgUrl(),
+            memberCreateDTO.getCode());
+    memberRepository.save(member);
+  }
+
+  public void updateMember(MemberUpdateDTO memberUpdateDTO) {
+    Member member = memberReader.read(memberUpdateDTO.getId());
+    member.updateMember(memberUpdateDTO.getNickname(), memberUpdateDTO.getProfileImgUrl());
+  }
 }

src/main/resources/application-dev.yml

@@ -19,6 +19,17 @@ spring:
     host: redis-cluster
     port: 6379
     password: 11111111
+  kafka:
+    consumer:
+      bootstrap-servers: kafka-service:9092
+      # 컨슈머 그룹 지정 - 컨슈머 그룹안의 하나의 컨슈머가 다운되더라도 컨슈머 그룹 안의 다른 컨슈머가 읽을 수 있도록 함 또한 Offset으로 재시작시 메시지 관리 가능
+      group-id: sns-service
+      # Kafka consumer가 다운되었을 때 가장 빠른 컨슈머 오프셋을 가지는 것 부터 읽는다.
+      auto-offset-reset: earliest
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+    listener:
+      ack-mode: manual
 eureka:
   client:
     register-with-eureka: true

src/main/resources/application-local.yml

@@ -32,6 +32,17 @@ spring:
     host: 127.0.0.1
     port: 6379
     password: 11111111
+  kafka:
+    consumer:
+      bootstrap-servers: localhost:9092
+      # 컨슈머 그룹 지정 - 컨슈머 그룹안의 하나의 컨슈머가 다운되더라도 컨슈머 그룹 안의 다른 컨슈머가 읽을 수 있도록 함 또한 Offset으로 재시작시 메시지 관리 가능
+      group-id: sns-service
+      # Kafka consumer가 다운되었을 때 가장 빠른 컨슈머 오프셋을 가지는 것 부터 읽는다.
+      auto-offset-reset: earliest
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+    listener:
+      ack-mode: manual
 
 eureka:
   client:

src/main/resources/application-prod.yml

@@ -19,6 +19,17 @@ spring:
     host: 127.0.0.1
     port: 6379
     password: 11111111
+  kafka:
+    consumer:
+      bootstrap-servers: kafka-service:9092
+      # 컨슈머 그룹 지정 - 컨슈머 그룹안의 하나의 컨슈머가 다운되더라도 컨슈머 그룹 안의 다른 컨슈머가 읽을 수 있도록 함 또한 Offset으로 재시작시 메시지 관리 가능
+      group-id: sns-service
+      # Kafka consumer가 다운되었을 때 가장 빠른 컨슈머 오프셋을 가지는 것 부터 읽는다.
+      auto-offset-reset: earliest
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+    listener:
+      ack-mode: manual
 eureka:
   client:
     register-with-eureka: true

src/test/resources/application.yml

@@ -36,6 +36,9 @@ spring:
   sql:
     init:
       data-locations: classpath:data.sql
+  kafka:
+    consumer:
+      group-id: sns-service
 
 cloud:
   aws: