Implements rabbitMQ rather than Kafka

docker-compose.yml

@@ -31,21 +31,15 @@ services:
     networks:
       - 'mongo'
 
-  kafka:
-    image: landoop/fast-data-dev:latest
-    environment:
-      ADV_HOST: '127.0.0.1'
-      RUNTESTS: 0
-      FORWARDLOGS: 0
-      SAMPLEDATA: 0
+  rabbitmq:
+    image: rabbitmq:3.12.3-management
     ports:
-      - 2181:2181                 # Zookeeper port
-      - 3030:3030                 # Landoop UI port
-      - 8081-8083:8081-8083       # REST Proxy, Schema Registry, Kafka Connect ports
-      - 9581-9585:9581-9585       # JMX Ports
-      - 9092:9092                 # Kafka Broker port
-    networks:
-      - 'kafka'
+      - 5672:5672
+      - 15672:15672   #management
+    environment:
+      RABBITMQ_DEFAULT_USER: "user"
+      RABBITMQ_DEFAULT_PASS: "p@ssword"
+
 
 networks:
   kafka:

pom.xml

@@ -30,11 +30,11 @@
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-web</artifactId>
 		</dependency>
+
 		<dependency>
-			<groupId>org.springframework.cloud</groupId>
-			<artifactId>spring-cloud-starter-stream-kafka</artifactId>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-amqp</artifactId>
 		</dependency>
-
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-configuration-processor</artifactId>

src/main/java/com/cardif/pocmeasurerest/infra/controller/MeasureController.java

@@ -2,6 +2,7 @@ package com.cardif.pocmeasurerest.infra.controller;
 
 import com.cardif.pocmeasurerest.domain.model.Measure;
 import com.cardif.pocmeasurerest.domain.service.MeasureService;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.http.MediaType;
 import org.springframework.web.bind.annotation.*;
 
@@ -10,6 +11,7 @@ import java.util.Optional;
 @RestController
 @CrossOrigin("*")
 @RequestMapping(value = "/measure")
+@Slf4j
 public class MeasureController {
 
     private final MeasureService measureService;
@@ -20,6 +22,7 @@ public class MeasureController {
 
     @PostMapping
     public Measure save(@RequestBody Measure measure) {
+        log.info("Create Measure by Rest, request: {}", measure);
         return measureService.save(measure);
     }
 

src/main/java/com/cardif/pocmeasurerest/infra/publisher/MeasureRabbitMQPublisher.java

+package com.cardif.pocmeasurerest.infra.publisher;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.annotation.EnableRabbit;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+@Component
+@EnableRabbit
+@Slf4j
+@RequiredArgsConstructor
+public class MeasureRabbitMQPublisher {
+
+    private final RabbitTemplate rabbitTemplate;
+    private final Queue queue;
+
+    public void sendMessage(Object message) {
+        log.info("Sending message {} to RabbitMQ...", message);
+        rabbitTemplate.convertAndSend(queue.getName(), message);
+        log.info("Message sent successfully");
+    }
+}

src/main/java/com/cardif/pocmeasurerest/infra/publisher/MeasureRabbitMQPublisherConfig.java

+package com.cardif.pocmeasurerest.infra.publisher;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import org.springframework.amqp.core.AmqpTemplate;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class MeasureRabbitMQPublisherConfig {
+
+    @Value("${spring.rabbitmq.queue}")
+    private String queueName;
+
+    @Bean
+    public Queue queue() {
+        return new Queue(queueName, false);
+    }
+
+    @Bean
+    public MessageConverter messageConverter(ObjectMapper objectMapper) {
+        return new Jackson2JsonMessageConverter(objectMapper);
+    }
+
+    @Bean
+    public ObjectMapper objectMapper() {
+        ObjectMapper objectMapper = new ObjectMapper();
+        objectMapper.registerModule(new JavaTimeModule());
+        return objectMapper;
+    }
+
+    @Bean
+    public AmqpTemplate template(ConnectionFactory connectionFactory,
+                                 MessageConverter messageConverter) {
+        RabbitTemplate template = new RabbitTemplate(connectionFactory);
+        template.setMessageConverter(messageConverter);
+        return template;
+    }
+
+}
\ No newline at end of file

src/main/java/com/cardif/pocmeasurerest/infra/repository/MeasureRepositoryImpl.java

@@ -2,22 +2,24 @@ package com.cardif.pocmeasurerest.infra.repository;
 
 import com.cardif.pocmeasurerest.domain.model.Measure;
 import com.cardif.pocmeasurerest.domain.repository.MeasureRepository;
+import com.cardif.pocmeasurerest.infra.publisher.MeasureRabbitMQPublisher;
 import com.cardif.pocmeasurerest.infra.repository.entity.MeasureEntity;
-import org.springframework.cloud.stream.function.StreamBridge;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Repository;
 
 import java.time.LocalDateTime;
 import java.util.Optional;
 
 @Repository
+@Slf4j
 public class MeasureRepositoryImpl implements MeasureRepository {
 
     private final JpaMeasureRepository jpaMeasureRepository;
-    private final StreamBridge streamBridge;
+    private final MeasureRabbitMQPublisher rabbitMQPublisher;
 
-    public MeasureRepositoryImpl(JpaMeasureRepository jpaMeasureRepository, StreamBridge streamBridge) {
+    public MeasureRepositoryImpl(JpaMeasureRepository jpaMeasureRepository, MeasureRabbitMQPublisher rabbitMQPublisher) {
         this.jpaMeasureRepository = jpaMeasureRepository;
-        this.streamBridge = streamBridge;
+        this.rabbitMQPublisher = rabbitMQPublisher;
     }
 
     @Override
@@ -27,9 +29,11 @@ public class MeasureRepositoryImpl implements MeasureRepository {
         entity.setDeviceId(measure.deviceId());
         entity.setValue(measure.value());
         MeasureEntity entitySaved = jpaMeasureRepository.save(entity);
-        streamBridge.send("measureProducer", entitySaved);
-        return new Measure(entitySaved.getDeviceId(), entitySaved.getValue(),
+        var measureSaved = new Measure(entitySaved.getDeviceId(), entitySaved.getValue(),
                 entitySaved.getDate());
+        log.info("Publish Message: {}", measureSaved);
+        rabbitMQPublisher.sendMessage(measureSaved);
+        return measureSaved;
     }
 
     @Override

src/main/java/com/cardif/pocmeasurerest/infra/repository/entity/MeasureEntity.java

 package com.cardif.pocmeasurerest.infra.repository.entity;
 
+import lombok.ToString;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.mongodb.core.mapping.Document;
 
 import java.time.LocalDateTime;
 
 @Document(collection = "measure")
+@ToString
 public class MeasureEntity {
 
     @Id

src/main/resources/application-local.properties

@@ -8,5 +8,9 @@ spring.data.mongodb.username=root
 spring.data.mongodb.password=myPasswordMongo21c
 spring.data.mongodb.authentication-database=admin
 
-#Kafka configurations stream
-spring.cloud.stream.kafka.binder.brokers=localhost:9092
\ No newline at end of file
+# RabbitMQ Configuration
+spring.rabbitmq.host=localhost
+spring.rabbitmq.queue=measureConsumer
+spring.rabbitmq.port=5672
+spring.rabbitmq.username=user
+spring.rabbitmq.password=p@ssword
\ No newline at end of file

src/main/resources/application.properties

@@ -23,8 +23,3 @@ spring.jackson.visibility.field=any
 spring.jackson.visibility.getter=none
 spring.jackson.visibility.setter=none
 spring.jackson.visibility.is-getter=none
-
-spring.cloud.function.definition=measureConsumer
-
-#Producer
-spring.cloud.stream.bindings.measureProducer.destination=measure
\ No newline at end of file