mindol1004/reactor-kafka
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

클래스명 수정

Browse Source
This commit is contained in:
deogicorgi
2022-04-03 00:17:32 +09:00
parent b5f2b70c76
commit 547d1da2af
9 changed files with 38 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
common/src/main/java/com/github/deogicorgi/reactive/common/message/AbstractKafkaProduceMessage.java → common/src/main/java/com/github/deogicorgi/reactive/common/message/AbstractKafkaMessage.java
View File

@@ -17,12 +17,12 @@ import java.time.LocalDateTime;
@JsonTypeInfo(
use = JsonTypeInfo.Id.NAME,
property = "type",
defaultImpl = KafkaUriProduceMessage.class)
defaultImpl = KafkaUriMessage.class)
@JsonSubTypes({
@JsonSubTypes.Type(value = KafkaUriProduceMessage.class, names = {"uri", "Uri", "URI"}),
@JsonSubTypes.Type(value = KafkaBodyProduceMessage.class, names = {"message", "Message", "MESSAGE"})
@JsonSubTypes.Type(value = KafkaUriMessage.class, names = {"uri", "Uri", "URI"}),
@JsonSubTypes.Type(value = KafkaBodyMessage.class, names = {"message", "Message", "MESSAGE"})
})
public abstract class AbstractKafkaProduceMessage {
public abstract class AbstractKafkaMessage {
// 요청 토픽
protected String topic;

4
common/src/main/java/com/github/deogicorgi/reactive/common/message/KafkaBodyProduceMessage.java → common/src/main/java/com/github/deogicorgi/reactive/common/message/KafkaBodyMessage.java
View File

@@ -9,12 +9,12 @@ import lombok.Setter;
*/
@Getter
@Setter
public class KafkaBodyProduceMessage extends AbstractKafkaProduceMessage {
public class KafkaBodyMessage extends AbstractKafkaMessage {
// 요청 메시지
private String message;
public KafkaBodyProduceMessage() {
public KafkaBodyMessage() {
super.type = ProduceMessageType.Message;
}

4
common/src/main/java/com/github/deogicorgi/reactive/common/message/KafkaUriProduceMessage.java → common/src/main/java/com/github/deogicorgi/reactive/common/message/KafkaUriMessage.java
View File

@@ -9,12 +9,12 @@ import lombok.Setter;
*/
@Getter
@Setter
public class KafkaUriProduceMessage extends AbstractKafkaProduceMessage {
public class KafkaUriMessage extends AbstractKafkaMessage {
// 요청 URI
private String uri;
public KafkaUriProduceMessage() {
public KafkaUriMessage() {
super.type = ProduceMessageType.URI;
}

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

@@ -1,14 +1,14 @@
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.message.AbstractKafkaMessage;
import com.github.deogicorgi.reactive.common.value.ProduceMessageType;
import lombok.Getter;
import java.time.LocalDateTime;
/**
* 카프카 메시지 클래스
* 카프카 메시지 전송결과 클래스
*/
@Getter
public class KafkaProduceResult {
@@ -38,11 +38,11 @@ public class KafkaProduceResult {
// 메시지를 처리한 시간
private LocalDateTime producedAt;
public KafkaProduceResult(AbstractKafkaProduceMessage message) {
public KafkaProduceResult(AbstractKafkaMessage message) {
this.setRequestedMessage(message);
}
public KafkaProduceResult(AbstractKafkaProduceMessage message, Throwable e) {
public KafkaProduceResult(AbstractKafkaMessage message, Throwable e) {
this.setRequestedMessage(message);
this.status = false;
this.error = e;
@@ -54,7 +54,7 @@ public class KafkaProduceResult {
return error != null;
}
private void setRequestedMessage(AbstractKafkaProduceMessage requestedMessage) {
private void setRequestedMessage(AbstractKafkaMessage requestedMessage) {
this.topic = requestedMessage.getTopic();
this.messageType = requestedMessage.getType();
this.requestedMessage = requestedMessage.getRequestedMessage();

2
producer/src/main/java/com/github/deogicorgi/reactive/producer/config/KafkaConfig.java
View File

@@ -27,6 +27,7 @@ public class KafkaConfig {
************************ Producer Options ************************
******************************************************************/
// 기본 설정들로 구성
@Bean("kafkaSender")
public KafkaSender<String, Object> kafkaSender() {
SenderOptions<String, Object> senderOptions = SenderOptions.create(getProducerProps());
@@ -35,6 +36,7 @@ public class KafkaConfig {
return KafkaSender.create(senderOptions);
}
// 프로듀서 옵션
private Map<String, Object> getProducerProps() {
return new HashMap<>() {{
put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getHosts());

7
producer/src/main/java/com/github/deogicorgi/reactive/producer/service/KafkaService.java
View File

@@ -1,6 +1,6 @@
package com.github.deogicorgi.reactive.producer.service;
import com.github.deogicorgi.reactive.common.message.AbstractKafkaProduceMessage;
import com.github.deogicorgi.reactive.common.message.AbstractKafkaMessage;
import com.github.deogicorgi.reactive.common.model.KafkaProduceResult;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
@@ -20,16 +20,15 @@ public class KafkaService {
private final KafkaSender<String, Object> producer;
public Mono<KafkaProduceResult> send(AbstractKafkaProduceMessage message) {
public Mono<KafkaProduceResult> send(AbstractKafkaMessage message) {
return producer.createOutbound()
// 지정된 토픽으로 메시지 전송
.send(Mono.just(new ProducerRecord<>(message.getTopic(), null, message.getRequestedMessage())))
// 전송완료 된 레코드를 Outbound로 리턴
.then()
// 에러 없이 전송이 완료 되었을 경우
.thenReturn(new KafkaProduceResult(message))
// 에러가 발생했을 경우
.onErrorResume(e -> Mono.just(new KafkaProduceResult(message, e)));
}
}

4
producer/src/main/java/com/github/deogicorgi/reactive/producer/web/controller/ProducerController.java
View File

@@ -1,6 +1,6 @@
package com.github.deogicorgi.reactive.producer.web.controller;
import com.github.deogicorgi.reactive.common.message.AbstractKafkaProduceMessage;
import com.github.deogicorgi.reactive.common.message.AbstractKafkaMessage;
import com.github.deogicorgi.reactive.common.model.KafkaProduceResult;
import com.github.deogicorgi.reactive.producer.web.service.ProduceService;
import lombok.RequiredArgsConstructor;
@@ -21,7 +21,7 @@ public class ProducerController {
private final ProduceService produceService;
@PostMapping
public Mono<KafkaProduceResult> send(@RequestBody AbstractKafkaProduceMessage message) {
public Mono<KafkaProduceResult> send(@RequestBody AbstractKafkaMessage message) {
return produceService.produceMessage(message);
}
}

14
producer/src/main/java/com/github/deogicorgi/reactive/producer/web/service/FailureMessageService.java Normal file
View File

@@ -0,0 +1,14 @@
package com.github.deogicorgi.reactive.producer.web.service;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
@Slf4j
@Service
@RequiredArgsConstructor
public class FailureMessageService {
public void produceFailure() {
}
}

6
producer/src/main/java/com/github/deogicorgi/reactive/producer/web/service/ProduceService.java
View File

@@ -1,6 +1,6 @@
package com.github.deogicorgi.reactive.producer.web.service;
import com.github.deogicorgi.reactive.common.message.AbstractKafkaProduceMessage;
import com.github.deogicorgi.reactive.common.message.AbstractKafkaMessage;
import com.github.deogicorgi.reactive.common.model.KafkaProduceResult;
import com.github.deogicorgi.reactive.producer.service.KafkaService;
import lombok.RequiredArgsConstructor;
@@ -18,10 +18,12 @@ import reactor.core.publisher.Mono;
public class ProduceService {
private final KafkaService kafkaService;
private final FailureMessageService failureMessageService;
public Mono<KafkaProduceResult> produceMessage(AbstractKafkaProduceMessage message) {
public Mono<KafkaProduceResult> produceMessage(AbstractKafkaMessage message) {
return kafkaService.send(message)
.map(produceResult -> {
failureMessageService.produceFailure();
if (produceResult.hasError()) {
// TODO 카프카 프로듀싱 실패일 경우 처리
// ex ) 처리하지못한 요청을 몽고등에 저장 후 재시도, 로깅 등등