이벤트 소싱을 이상하게 구현하던 것을 수정함

gracefulife · gracefulife · commit ba3fd06e2331 · 2018-05-31T15:20:27.000+09:00
요청받았을 때 저장하고 -&gt; 저장된 이벤트를 발송하여, 사후처리를 하던 것을, 저장 요청 -&gt; 커맨드 이벤트 발송으로 변경함
diff --git a/profile/src/main/java/com/example/research/profile/entity/cache/Profile.java b/profile/src/main/java/com/example/research/profile/entity/cache/Profile.java
@@ -1,7 +1,7 @@
 package com.example.research.profile.entity.cache;
 
 import com.example.research.profile.entity.command.ProfileChangedCommand;
-import com.example.research.profile.entity.command.ProfileSavedCommand;
+import com.example.research.profile.entity.command.CreateProfileCommand;
 import com.example.research.profile.entity.storage.ProfileEvent;
 
 import org.springframework.data.annotation.Id;
@@ -36,7 +36,7 @@ public static Profile from(com.example.research.profile.entity.storage.Profile p
   }
 
   // 생성 이벤트에서 왔으므로, 항상 active
-  public static Profile from(ProfileSavedCommand event) {
+  public static Profile from(CreateProfileCommand event) {
     return from(event.getId(), event.getName(), event.getAge(),
         event.getSex(), true, Collections.emptySet());
   }
diff --git a/profile/src/main/java/com/example/research/profile/entity/command/CreateProfileCommand.java b/profile/src/main/java/com/example/research/profile/entity/command/CreateProfileCommand.java
@@ -1,6 +1,6 @@
 package com.example.research.profile.entity.command;
 
-import com.example.research.profile.entity.storage.Profile;
+import com.example.research.profile.v1.profile.ProfileSaveRequest;
 
 import java.time.LocalDateTime;
 
@@ -12,20 +12,20 @@
 @ToString
 @Getter
 @RequiredArgsConstructor
-public class ProfileSavedCommand extends ProfileCommand {
-  public static final String TAG = ProfileSavedCommand.class.getSimpleName();
+public class CreateProfileCommand extends ProfileCommand {
+  public static final String TAG = CreateProfileCommand.class.getSimpleName();
 
   @NonNull String id;
   @NonNull String name;
   @NonNull Integer age;
   @NonNull String sex; // man, woman
   @NonNull LocalDateTime createdAt;
 
-  public static ProfileSavedCommand from(Profile storeProfile) {
-    return new ProfileSavedCommand(
+  public static CreateProfileCommand from(ProfileSaveRequest storeProfile) {
+    return new CreateProfileCommand(
         storeProfile.getId(),
         storeProfile.getName(), storeProfile.getAge(),
-        storeProfile.getSex(), storeProfile.getCreatedAt()
+        storeProfile.getSex(), LocalDateTime.now()
     );
   }
 
diff --git a/profile/src/main/java/com/example/research/profile/entity/storage/Profile.java b/profile/src/main/java/com/example/research/profile/entity/storage/Profile.java
@@ -1,5 +1,6 @@
 package com.example.research.profile.entity.storage;
 
+import com.example.research.profile.entity.command.CreateProfileCommand;
 import com.example.research.profile.v1.profile.ProfileSaveRequest;
 
 import org.springframework.data.annotation.CreatedDate;
@@ -47,15 +48,15 @@ public class Profile {
   @LastModifiedDate
   @Column(name = "updated_at") LocalDateTime updatedAt;
 
-  public static Profile from(ProfileSaveRequest request) {
+  public static Profile from(CreateProfileCommand command) {
     Profile profile = new Profile();
-    profile.id = request.getId();
-    profile.name = request.getName();
-    profile.age = request.getAge();
-    profile.sex = request.getSex();
+    profile.id = command.getId();
+    profile.name = command.getName();
+    profile.age = command.getAge();
+    profile.sex = command.getSex();
     profile.active = true;
-    profile.createdAt = LocalDateTime.now();
-    profile.updatedAt = LocalDateTime.now();
+    profile.createdAt = command.getCreatedAt();
+    profile.updatedAt = command.getCreatedAt();
     return profile;
   }
 }
diff --git a/profile/src/main/java/com/example/research/profile/v1/profile/ProfileEventHandler.java b/profile/src/main/java/com/example/research/profile/v1/profile/ProfileEventHandler.java
@@ -1,6 +1,10 @@
 package com.example.research.profile.v1.profile;
 
+import com.example.research.profile.entity.cache.Profile;
+import com.example.research.profile.entity.cache.ProfileRepository;
+import com.example.research.profile.entity.command.CreateProfileCommand;
 import com.example.research.profile.entity.command.ProfileCommand;
+import com.example.research.profile.entity.command.UpdateProfileCommand;
 import com.example.research.profile.entity.storage.ProfileEventStoreRepository;
 import com.example.research.profile.entity.storage.ProfileStorageRepository;
 import com.fasterxml.jackson.core.JsonProcessingException;
@@ -19,23 +23,27 @@
 @Async
 @Service
 public class ProfileEventHandler {
+  @Autowired ProfileRepository profileRepository;
   @Autowired ProfileStorageRepository profileStorageRepository;
   @Autowired ProfileEventStoreRepository profileEventStoreRepository;
 
   @Autowired ObjectMapper objectMapper;
 
   @Async
   @Transactional
-  public CompletableFuture<com.example.research.profile.entity.storage.ProfileEvent> save(ProfileCommand event) {
+  public CompletableFuture<com.example.research.profile.entity.storage.ProfileEvent> save(CreateProfileCommand command) {
     return CompletableFuture.supplyAsync(() -> {
+      // TODO chaining 이 나을지, 그냥 소비시켜버릴지.
+      profileRepository.save(Profile.from(command)).subscribe();
+      profileStorageRepository.save(com.example.research.profile.entity.storage.Profile.from(command));
       Long latestVersion = profileEventStoreRepository
-          .findTopByIdentifierOrderByNoDesc(event.getId())
+          .findTopByIdentifierOrderByNoDesc(command.getId())
           .map(com.example.research.profile.entity.storage.ProfileEvent::getVersion)
           .orElse(0L);
-      String payload = generatePayload(event);
+      String payload = generatePayload(command);
 
       com.example.research.profile.entity.storage.ProfileEvent profileRawEvent
-          = com.example.research.profile.entity.storage.ProfileEvent.from(event, latestVersion + 1, payload);
+          = com.example.research.profile.entity.storage.ProfileEvent.from(command, latestVersion + 1, payload);
       return profileEventStoreRepository.save(profileRawEvent);
     });
   }
diff --git a/profile/src/main/java/com/example/research/profile/v1/profile/ProfileEventListener.java b/profile/src/main/java/com/example/research/profile/v1/profile/ProfileEventListener.java
@@ -3,7 +3,7 @@
 import com.example.research.profile.entity.cache.Profile;
 import com.example.research.profile.entity.cache.ProfileRepository;
 import com.example.research.profile.entity.command.ProfileChangedCommand;
-import com.example.research.profile.entity.command.ProfileSavedCommand;
+import com.example.research.profile.entity.command.CreateProfileCommand;
 import com.example.research.profile.entity.storage.ProfileStorageRepository;
 
 import org.springframework.beans.factory.annotation.Autowired;
@@ -23,12 +23,11 @@ public class ProfileEventListener {
   @Autowired ProfileRepository profileRepository;
   @Autowired ProfileStorageRepository profileStorageRepository;
 
-  @EventListener public void onProfileSavedEventReceived(ProfileSavedCommand event) {
-    log.info("onProfileSavedEventReceived received : {}" + event);
-    profileRepository.save(Profile.from(event))
-        .flatMap(profile -> Mono.fromCompletionStage(profileEventHandler.save(event)))
+  @EventListener public void onProfileSavedEventReceived(CreateProfileCommand command) {
+    log.info("onProfileSavedEventReceived received : {}" + command);
+    Mono.fromCompletionStage(profileEventHandler.save(command))
         .doOnError(e -> log.error("onProfileSavedEventReceived error : {}", e))
-        .onErrorResume(throwable -> profileRepository.deleteById(event.getId()).flatMap(aVoid -> Mono.empty()))
+        .onErrorResume(throwable -> profileRepository.deleteById(command.getId()).flatMap(aVoid -> Mono.empty()))
         .subscribe();
   }
 
diff --git a/profile/src/main/java/com/example/research/profile/v1/profile/ProfileHandler.java b/profile/src/main/java/com/example/research/profile/v1/profile/ProfileHandler.java
@@ -2,8 +2,8 @@
 
 import com.example.research.profile.entity.cache.Profile;
 import com.example.research.profile.entity.cache.ProfileRepository;
+import com.example.research.profile.entity.command.CreateProfileCommand;
 import com.example.research.profile.entity.command.ProfileChangedCommand;
-import com.example.research.profile.entity.command.ProfileSavedCommand;
 import com.example.research.profile.entity.storage.ProfileStorageRepository;
 
 import org.springframework.beans.factory.annotation.Autowired;
@@ -16,13 +16,15 @@
 import org.springframework.web.reactive.function.server.ServerResponse;
 
 import java.time.LocalDateTime;
+import java.util.function.Consumer;
 import java.util.function.Function;
 
 import lombok.extern.slf4j.Slf4j;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import static org.springframework.web.reactive.function.BodyInserters.fromObject;
+import static org.springframework.web.reactive.function.server.ServerResponse.noContent;
 import static org.springframework.web.reactive.function.server.ServerResponse.ok;
 
 @Slf4j
@@ -54,24 +56,14 @@ public Mono<ServerResponse> fetch(ServerRequest request) {
 
   @NonNull
   public Mono<ServerResponse> save(Mono<ProfileSaveRequest> request) {
-    Function<ProfileSaveRequest, com.example.research.profile.entity.storage.Profile> mapToProfile
-        = com.example.research.profile.entity.storage.Profile::from;
+    // TODO validation
+    Function<ProfileSaveRequest, CreateProfileCommand> requestToCommand = CreateProfileCommand::from;
+    Consumer<CreateProfileCommand> sendCommand = command -> applicationEventPublisher.publishEvent(command);
 
-    Function<com.example.research.profile.entity.storage.Profile, Mono<com.example.research.profile.entity.storage.Profile>>
-        saveProfile = profile -> {
-      log.info("profile is {}", profile);
-      com.example.research.profile.entity.storage.Profile storeProfile = profileStorageRepository.save(profile);
-      applicationEventPublisher.publishEvent(ProfileSavedCommand.from(storeProfile));
-      return Mono.just(storeProfile);
-    };
-
-    Function<com.example.research.profile.entity.storage.Profile, ProfileSaveResponse> mapToResponse =
-        profile -> new ProfileSaveResponse(profile.getId());
-
-    Mono<ProfileSaveResponse> save = request.flatMap(saveProfile.compose(mapToProfile))
-        .map(mapToResponse); // FIXME throw 처리
-
-    return ok().body(save, ProfileSaveResponse.class);
+    return request.single().flatMap(profile -> {
+      sendCommand.accept(requestToCommand.apply(profile));
+      return noContent().build();
+    });
   }
 
   @NonNull

Original file line number	Diff line number	Diff line change
`@@ -1,7 +1,7 @@`
`1`	`1`	`package com.example.research.profile.entity.cache;`
`2`	`2`
`3`	`3`	`import com.example.research.profile.entity.command.ProfileChangedCommand;`
`4`		`-import com.example.research.profile.entity.command.ProfileSavedCommand;`
	`4`	`+import com.example.research.profile.entity.command.CreateProfileCommand;`
`5`	`5`	`import com.example.research.profile.entity.storage.ProfileEvent;`
`6`	`6`
`7`	`7`	`import org.springframework.data.annotation.Id;`
`@@ -36,7 +36,7 @@ public static Profile from(com.example.research.profile.entity.storage.Profile p`
`36`	`36`	`}`
`37`	`37`
`38`	`38`	`// 생성 이벤트에서 왔으므로, 항상 active`
`39`		`- public static Profile from(ProfileSavedCommand event) {`
	`39`	`+ public static Profile from(CreateProfileCommand event) {`
`40`	`40`	`return from(event.getId(), event.getName(), event.getAge(),`
`41`	`41`	`event.getSex(), true, Collections.emptySet());`
`42`	`42`	`}`