Refactored code

Author: foolingcoder

src/main/java/mu/integration/consumer/rabbitmq/entity/CsvLineInformation.java renamed to src/main/java/mu/integration/consumer/rabbitmq/dto/CsvLineInformation.java

@@ -1,4 +1,4 @@
-package mu.integration.consumer.rabbitmq.entity;
+package mu.integration.consumer.rabbitmq.dto;
 
 import lombok.Data;
 

src/main/java/mu/integration/consumer/rabbitmq/dto/CsvLineStatusDto.java

@@ -0,0 +1,50 @@
+package mu.integration.consumer.rabbitmq.listener;
+
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.cloud.stream.messaging.Processor;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.messaging.handler.annotation.Payload;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.stereotype.Component;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import mu.integration.consumer.rabbitmq.dto.CsvLineInformation;
+import mu.integration.consumer.rabbitmq.service.CsvLineInformationSender;
+import mu.integration.consumer.rabbitmq.service.CsvLineValidationService;
+
+/**
+ *
+ * @author priteela
+ */
+@RequiredArgsConstructor
+@Slf4j
+@Component
+public class CsvLineInformationConsumer {
+
+    private final CsvLineValidationService csvLineValidationService;
+    private final CsvLineInformationSender csvLineInformationSender;
+    private final ObjectMapper mapper;
+
+    @StreamListener(Processor.INPUT)
+    public void receiveOrder(@Header Message<String> header, @Payload CsvLineInformation csvLineInformation)
+            throws JsonProcessingException {
+
+        log.debug("\n\n CsvLine received: {}", mapper.writeValueAsString(csvLineInformation));
+        log.debug("\n\n Header received: {}", header);
+
+        //update csv line information
+        csvLineInformation = csvLineValidationService.validate(csvLineInformation);
+
+        Message reply = MessageBuilder.withPayload(csvLineInformation)
+                .copyHeaders(header.getHeaders())
+                .build();
+
+        csvLineInformationSender.send(reply);
+
+    }
+}

src/main/java/mu/integration/consumer/rabbitmq/entity/CsvLine.java

@@ -3,11 +3,9 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.cloud.stream.messaging.Source;
 import org.springframework.messaging.Message;
-import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.stereotype.Service;
 
 import lombok.extern.slf4j.Slf4j;
-import mu.integration.consumer.rabbitmq.entity.CsvLineInformation;
 
 @Slf4j
 @Service
@@ -16,11 +14,8 @@ public class CsvLineInformationSender {
     @Autowired
     private Source source;
 
-    public void send(CsvLineInformation csvLineInformation) {
-        Message message = MessageBuilder.withPayload(csvLineInformation).build();
-        log.info("\n\nCsvLine sent: {}", message);
-        this.source.output().send(MessageBuilder.withPayload(csvLineInformation).build());
+    public void send(Message message) {
+        log.debug("\n\n CsvLine sent: {}", message);
+        this.source.output().send(message);
     }
-
-
 }

src/main/java/mu/integration/consumer/rabbitmq/listener/CsvLineInformationConsumer.java

@@ -1,6 +1,6 @@
 package mu.integration.consumer.rabbitmq.service;
 
-import mu.integration.consumer.rabbitmq.entity.CsvLineInformation;
+import mu.integration.consumer.rabbitmq.dto.CsvLineInformation;
 
 /**
  *

src/main/java/mu/integration/consumer/rabbitmq/listener/RabbitmqListener.java

@@ -7,7 +7,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 
 import lombok.extern.slf4j.Slf4j;
-import mu.integration.consumer.rabbitmq.entity.CsvLineInformation;
+import mu.integration.consumer.rabbitmq.dto.CsvLineInformation;
 
 /**
  *