Implemented reply channel

Author: foolingcoder

src/main/java/mu/integration/consumer/rabbitmq/Pojo.java

@@ -6,6 +6,19 @@
  */
 public class Pojo {
 
+    public Pojo() {
+        
+    }
+
+    public Pojo(String csvLine) {
+
+        String[] array = csvLine.split(",");
+        this.setCol1(array[0]);
+        this.setCol2(array[1]);
+        this.setCol3(array[2]);
+
+    }
+
     private String col1;
 
     private String col2;

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

@@ -1,12 +1,9 @@
 package mu.integration.consumer.rabbitmq;
 
-import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
-import org.springframework.context.annotation.Bean;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
-import org.springframework.messaging.handler.annotation.SendTo;
 import org.springframework.stereotype.Component;
 
 /**
@@ -16,34 +13,23 @@
 @Component
 public class RabbitmqListener {
 
+    private static final String AMQ_CORRELATION_ID = "amqp_correlationId";
 
     @RabbitListener(queues = "myQueue")
-    @SendTo("foo_response")
-    public Message<String> receive(Message<String> message) {
-
-        System.out.println("\n\n*********************");
-
-        System.out.println(message.getHeaders());
 
-        Pojo record = new Pojo();
+    public Message<String> receive(Message<String> message) {
 
-        String[] array = message.getPayload().split(",");
-        record.setCol1(array[0]);
-        record.setCol2(array[1]);
-        record.setCol3(array[2]);
+        System.out.println("\n\n*********** Message Received ***********");
 
+        //Call a service class to validate pojo
+        Pojo record = new Pojo(message.getPayload());
         System.out.println(record.toString());
 
-        return MessageBuilder.withPayload("Hello")
+        return MessageBuilder.withPayload("PROCESSED: " + message.getHeaders().get(AMQ_CORRELATION_ID))
                 .setHeader(MessageHeaders.CONTENT_TYPE, "application/json")
                 .build();
 
     }
 
 
-    @Bean
-    public Queue queue() {
-        return new Queue("foo_response", false);
-    }
-
 }