프로젝트 구조변경

common/build.gradle

@@ -1,17 +1,27 @@
 plugins {
     id 'java'
+    id 'java-library'
+    id 'org.springframework.boot' version '2.6.4'
+    id 'io.spring.dependency-management' version '1.0.11.RELEASE'
 }
 
 group 'com.github.deogicorgi.reactive'
 version '0.0.1-SNAPSHOT'
 
+bootJar {
+    enabled = false
+}
+
+jar {
+    enabled = true
+}
+
 repositories {
     mavenCentral()
 }
 
 dependencies {
-    testImplementation 'org.junit.jupiter:junit-jupiter-api:5.7.0'
-    testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine:5.7.0'
+    implementation 'com.fasterxml.jackson.core:jackson-annotations'
 }
 
 test {

common/src/main/java/com/github/deogicorgi/reactive/common/message/AbstractKafkaProduceMessage.java

@@ -0,0 +1,38 @@
+package com.github.deogicorgi.reactive.common.message;
+
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import com.github.deogicorgi.reactive.common.value.ProduceMessageType;
+import lombok.Getter;
+import lombok.Setter;
+
+import java.time.LocalDateTime;
+
+/**
+ * 카프카 메시지 베이스
+ * 프로듀서 내 에러 발생시 처리를 쉽게하기 위해 URI 형태와 Message 형태로 나눔
+ */
+@Getter
+@Setter
+@JsonTypeInfo(
+        use = JsonTypeInfo.Id.NAME,
+        property = "type",
+        defaultImpl = KafkaUriProduceMessage.class)
+@JsonSubTypes({
+        @JsonSubTypes.Type(value = KafkaUriProduceMessage.class, names = {"uri", "Uri", "URI"}),
+        @JsonSubTypes.Type(value = KafkaBodyProduceMessage.class, names = {"message", "Message", "MESSAGE"})
+})
+public abstract class AbstractKafkaProduceMessage {
+
+    // 요청 토픽
+    protected String topic;
+
+    // 메시지 타입 (uri , message)
+    protected ProduceMessageType type;
+
+    // 요청 시간
+    protected LocalDateTime requestedAt;
+
+    public abstract String getRequestedMessage();
+
+}

common/src/main/java/com/github/deogicorgi/reactive/common/message/KafkaBodyProduceMessage.java

@@ -0,0 +1,25 @@
+package com.github.deogicorgi.reactive.common.message;
+
+import com.github.deogicorgi.reactive.common.value.ProduceMessageType;
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * JSON String message 카프카 메시지
+ */
+@Getter
+@Setter
+public class KafkaBodyProduceMessage extends AbstractKafkaProduceMessage {
+
+    // 요청 메시지
+    private String message;
+
+    public KafkaBodyProduceMessage() {
+        super.type = ProduceMessageType.Message;
+    }
+
+    @Override
+    public String getRequestedMessage() {
+        return this.message;
+    }
+}

common/src/main/java/com/github/deogicorgi/reactive/common/message/KafkaUriProduceMessage.java

@@ -0,0 +1,25 @@
+package com.github.deogicorgi.reactive.common.message;
+
+import com.github.deogicorgi.reactive.common.value.ProduceMessageType;
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * URI 카프카 메시지
+ */
+@Getter
+@Setter
+public class KafkaUriProduceMessage extends AbstractKafkaProduceMessage {
+
+    // 요청 URI
+    private String uri;
+
+    public KafkaUriProduceMessage() {
+        super.type = ProduceMessageType.URI;
+    }
+
+    @Override
+    public String getRequestedMessage() {
+        return this.uri;
+    }
+}

common/src/main/java/com/github/deogicorgi/reactive/common/model/KafkaProduceResult.java

@@ -0,0 +1,65 @@
+package com.github.deogicorgi.reactive.common.model;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.github.deogicorgi.reactive.common.message.AbstractKafkaProduceMessage;
+import com.github.deogicorgi.reactive.common.value.ProduceMessageType;
+import lombok.Getter;
+
+import java.time.LocalDateTime;
+
+/**
+ * 카프카 메시지 클래스
+ */
+@Getter
+public class KafkaProduceResult {
+
+    // 메시지 전송 상태 - true : 전송완료, false : 전송실패
+    private Boolean status = true;
+
+    // 메시지 전송 토픽
+    private String topic;
+
+    // 요청받은 메시지 타입 (uri, message)
+    private ProduceMessageType messageType;
+
+    // 요청받은 메시지 - URI 또는 JSON String
+    private String requestedMessage;
+
+    // 에러 - 전송과정 중 발생된 에러, 전송완료 일 경우 null
+    @JsonIgnore
+    private Throwable error = null;
+
+    // 에러 메시지 - 전송과정 중 발생된 에러, 전송완료 일 경우 null
+    private String errorMessage = null;
+
+    // 메시지를 요청받은 시간
+    private LocalDateTime requestedAt;
+
+    // 메시지를 처리한 시간
+    private LocalDateTime producedAt;
+
+    public KafkaProduceResult(AbstractKafkaProduceMessage message) {
+        this.setRequestedMessage(message);
+    }
+
+    public KafkaProduceResult(AbstractKafkaProduceMessage message, Throwable e) {
+        this.setRequestedMessage(message);
+        this.status = false;
+        this.error = e;
+        this.errorMessage = e.getMessage();
+        this.producedAt = null;
+    }
+
+    public Boolean hasError() {
+        return error != null;
+    }
+
+    private void setRequestedMessage(AbstractKafkaProduceMessage requestedMessage) {
+        this.topic = requestedMessage.getTopic();
+        this.messageType = requestedMessage.getType();
+        this.requestedMessage = requestedMessage.getRequestedMessage();
+        this.producedAt = LocalDateTime.now();
+        this.requestedAt = requestedMessage.getRequestedAt();
+    }
+
+}

common/src/main/java/com/github/deogicorgi/reactive/common/value/ProduceMessageType.java

@@ -0,0 +1,6 @@
+package com.github.deogicorgi.reactive.common.value;
+
+public enum ProduceMessageType {
+    URI,
+    Message;
+}