feat: use rabbitmq

docker-compose.yaml

@@ -25,3 +25,11 @@ services:
     environment:
       MONGO_INITDB_ROOT_USERNAME: user
       MONGO_INITDB_ROOT_PASSWORD: pa55word
+  rabbitmq:
+    image: rabbitmq:3.12.0-management
+    ports:
+      - "5672:5672"
+      - "15672:15672"
+    environment:
+      RABBITMQ_DEFAULT_USER: "user"
+      RABBITMQ_DEFAULT_PASS: "p@ssword"
\ No newline at end of file

pom.xml

@@ -31,9 +31,18 @@
 			<groupId>io.micrometer</groupId>
 			<artifactId>micrometer-tracing-bridge-brave</artifactId>
 		</dependency>
+<!--		<dependency>-->
+<!--			<groupId>org.springframework.cloud</groupId>-->
+<!--			<artifactId>spring-cloud-starter-stream-kafka</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>com.fasterxml.jackson.datatype</groupId>
+			<artifactId>jackson-datatype-jsr310</artifactId>
+			<version>2.15.2</version> <!-- Use the latest version -->
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>

src/main/java/com/cardiff/pocmeasurekafka/config/RabbitMQConfig.java

+package com.cardiff.pocmeasurekafka.config;
+
+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 RabbitMQConfig {
+
+  @Value("${spring.rabbitmq.queue}")
+  private String queue;
+
+  @Bean
+  public Queue queue() {
+    return new Queue(queue, 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;
+  }
+}

src/main/java/com/cardiff/pocmeasurekafka/in/rest/MeasureController.java

 package com.cardiff.pocmeasurekafka.in.rest;
 
 
-import java.io.Serializable;
-import java.time.LocalDateTime;
-
+import com.cardiff.pocmeasurekafka.out.redis.RedisAdapter;
+import com.cardiff.pocmeasurekafka.out.redis.RedisAdapter.MeasureRedis;
+import lombok.RequiredArgsConstructor;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 
-import com.cardiff.pocmeasurekafka.model.Measure;
-import com.cardiff.pocmeasurekafka.out.redis.RedisAdapter;
-import com.cardiff.pocmeasurekafka.out.redis.RedisAdapter.MeasureRedis;
-
-import lombok.RequiredArgsConstructor;
-
 @RestController
 @RequestMapping("/measure")
 @RequiredArgsConstructor

src/main/java/com/cardiff/pocmeasurekafka/in/stream/MeasureConsumer.java

 package com.cardiff.pocmeasurekafka.in.stream;
 
-import java.util.function.Consumer;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.stereotype.Component;
 
 import com.cardiff.pocmeasurekafka.model.Measure;
@@ -13,13 +12,21 @@ import lombok.RequiredArgsConstructor;
 
 @Component("measureConsumer")
 @RequiredArgsConstructor
-public class MeasureConsumer implements Consumer<Measure> {
+public class MeasureConsumer {
+
   private static final Logger log = LoggerFactory.getLogger(MeasureConsumer.class);
   private final MeasureService measureService;
 
-  @Override
-  public void accept(Measure measure) {
-    measureService.saveMeasure(measure);
-    log.info("Catch event {}", measure);
+//  @Override
+//  public void accept(Measure measure) {
+//    measureService.saveMeasure(measure);
+//    log.info("Catch event {}", measure);
+//  }
+
+  @RabbitListener(queues = "${spring.rabbitmq.queue}")
+  public void receiveMessage(Measure message) {
+    measureService.saveMeasure(message);
+    log.info("Received message: [{}]", message);
+
   }
 }

src/main/java/com/cardiff/pocmeasurekafka/out/rest/RestAdapter.java

 package com.cardiff.pocmeasurekafka.out.rest;
 
+import com.cardiff.pocmeasurekafka.model.Measure;
 import java.io.Serializable;
 import java.time.Instant;
-import java.time.LocalDateTime;
 import java.time.ZoneOffset;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Value;
@@ -17,8 +16,6 @@ import org.springframework.stereotype.Component;
 import org.springframework.web.client.RestTemplate;
 import org.springframework.web.util.UriComponentsBuilder;
 
-import com.cardiff.pocmeasurekafka.model.Measure;
-
 
 @Component
 public class RestAdapter {

src/main/resources/application-local.properties

@@ -27,17 +27,24 @@ spring.jackson.visibility.setter=none
 spring.jackson.visibility.is-getter=none
 
 # Stream
-spring.cloud.stream.kafka.binder.brokers=localhost:9092
-spring.cloud.function.definition=measureConsumer
-spring.cloud.stream.kafka.binder.configuration.key.serializer=org.apache.kafka.common.serialization.StringSerializer
+#spring.cloud.stream.kafka.binder.brokers=localhost:9092
+#spring.cloud.function.definition=measureConsumer
+#spring.cloud.stream.kafka.binder.configuration.key.serializer=org.apache.kafka.common.serialization.StringSerializer
+
+# RabbitMQ
+spring.rabbitmq.host=localhost
+spring.rabbitmq.queue=measureConsumer
+spring.rabbitmq.port=5672
+spring.rabbitmq.username=user
+spring.rabbitmq.password=p@ssword
 
 #Producer
 #spring.cloud.stream.bindings.measureProducer.destination=measure
 
 
 ## Consumer
-spring.cloud.stream.bindings.measureConsumer-in-0.destination=measure
-spring.cloud.stream.bindings.measureConsumer-in-0.group=measureGroup
+#spring.cloud.stream.bindings.measureConsumer-in-0.destination=measure
+#spring.cloud.stream.bindings.measureConsumer-in-0.group=measureGroup
 
 
 # Redis

src/main/resources/application.properties

@@ -28,13 +28,13 @@ spring.jackson.visibility.is-getter=none
 spring.cloud.stream.bindings.measureProducer.destination=measure
 
 # Stream
-spring.cloud.stream.kafka.binder.brokers=${SPRING_CLOUD_STREAM_KAFKA_BINDER_BROKERS}
-spring.cloud.function.definition=measureConsumer
-spring.cloud.stream.kafka.binder.configuration.key.serializer=org.apache.kafka.common.serialization.StringSerializer
+#spring.cloud.stream.kafka.binder.brokers=${SPRING_CLOUD_STREAM_KAFKA_BINDER_BROKERS}
+#spring.cloud.function.definition=measureConsumer
+#spring.cloud.stream.kafka.binder.configuration.key.serializer=org.apache.kafka.common.serialization.StringSerializer
 
 ## Consumer
-spring.cloud.stream.bindings.measureConsumer-in-0.destination=measure
-spring.cloud.stream.bindings.measureConsumer-in-0.group=measureGroup
+#spring.cloud.stream.bindings.measureConsumer-in-0.destination=measure
+#spring.cloud.stream.bindings.measureConsumer-in-0.group=measureGroup
 
 # Redis
 #spring.data.redis.host=localhost