project works with mongodb and kafka producer

docker-compose.yml

+version: "3.8"
+services:
+
+  mongo:
+    image: mongo:latest
+    platform: linux/x86_64
+    ports:
+      - "29514:27017"
+    environment:
+      - MONGO_INITDB_ROOT_USERNAME=root
+      - MONGO_INITDB_ROOT_PASSWORD=myPasswordMongo21c
+      - MONGO_INITDB_DATABASE=carrefur-database
+    networks:
+      - 'mongo'
+    restart: always
+
+  zookeeper:
+    image: 'bitnami/zookeeper:latest'
+    ports:
+      - '2181:2181'
+    environment:
+      - ALLOW_ANONYMOUS_LOGIN=yes
+    networks:
+      - 'kafka'
+
+  kafka:
+    image: 'bitnami/kafka:latest'
+    ports:
+      - "9092:9092"
+    environment:
+      - KAFKA_BROKER_ID=1
+      - KAFKA_LISTENERS=PLAINTEXT://:9092
+      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092
+      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
+      - ALLOW_PLAINTEXT_LISTENER=yes
+    depends_on:
+      - zookeeper
+    networks:
+      - 'kafka'
+
+networks:
+  kafka:
+    driver: bridge
+  mongo:
+    driver: bridge
\ No newline at end of file

pom.xml

@@ -15,6 +15,7 @@
 	<description>poc-measure-rest</description>
 	<properties>
 		<java.version>17</java.version>
+		<spring-cloud.version>2022.0.3</spring-cloud.version>
 	</properties>
 	<dependencies>
 		<dependency>
@@ -30,8 +31,8 @@
 			<artifactId>spring-boot-starter-web</artifactId>
 		</dependency>
 		<dependency>
-			<groupId>org.apache.kafka</groupId>
-			<artifactId>kafka-streams</artifactId>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-starter-stream-kafka</artifactId>
 		</dependency>
 
 		<dependency>
@@ -64,6 +65,17 @@
 			<scope>runtime</scope>
 		</dependency>
 	</dependencies>
+	<dependencyManagement>
+		<dependencies>
+			<dependency>
+				<groupId>org.springframework.cloud</groupId>
+				<artifactId>spring-cloud-dependencies</artifactId>
+				<version>${spring-cloud.version}</version>
+				<type>pom</type>
+				<scope>import</scope>
+			</dependency>
+		</dependencies>
+	</dependencyManagement>
 
 	<build>
 		<plugins>

src/main/java/com/cardif/pocmeasurerest/domain/model/Measure.java

+package com.cardif.pocmeasurerest.domain.model;
+
+import java.io.Serializable;
+import java.time.LocalDateTime;
+
+public record Measure(String deviceId, String value, LocalDateTime date) implements Serializable {
+
+}

src/main/java/com/cardif/pocmeasurerest/domain/repository/MeasureRepository.java

+package com.cardif.pocmeasurerest.domain.repository;
+
+import com.cardif.pocmeasurerest.domain.model.Measure;
+
+import java.util.Optional;
+
+public interface MeasureRepository {
+
+    Measure save(Measure measure);
+    Optional<Measure> getById(String id) throws RuntimeException;
+
+}

src/main/java/com/cardif/pocmeasurerest/domain/service/MeasureService.java

+package com.cardif.pocmeasurerest.domain.service;
+
+import com.cardif.pocmeasurerest.domain.model.Measure;
+import com.cardif.pocmeasurerest.domain.repository.MeasureRepository;
+import org.springframework.stereotype.Service;
+
+import java.util.Optional;
+
+@Service
+public class MeasureService {
+
+    private final MeasureRepository measureRepository;
+
+    public MeasureService(MeasureRepository measureRepository) {
+        this.measureRepository = measureRepository;
+    }
+
+    public Measure save(Measure measure) {
+        return measureRepository.save(measure);
+    }
+
+    public Optional<Measure> getById(String deviceId) {
+        return measureRepository.getById(deviceId);
+    }
+}

src/main/java/com/cardif/pocmeasurerest/infra/config/MeasureConsumer.java

+package com.cardif.pocmeasurerest.infra.config;
+
+import com.cardif.pocmeasurerest.domain.model.Measure;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+import java.util.function.Consumer;
+
+@Component("createMeasure")
+public class MeasureConsumer implements Consumer<Measure> {
+
+    private static final Logger log = LoggerFactory.getLogger(MeasureConsumer.class);
+    @Override
+    public void accept(Measure measure) {
+        log.info("Catch event {}", measure);
+    }
+}

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

+package com.cardif.pocmeasurerest.infra.controller;
+
+import com.cardif.pocmeasurerest.domain.model.Measure;
+import com.cardif.pocmeasurerest.domain.service.MeasureService;
+import org.springframework.http.MediaType;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.Optional;
+
+@RestController
+@CrossOrigin("*")
+@RequestMapping(value = "/measure", consumes = MediaType.APPLICATION_JSON_VALUE, produces = MediaType.APPLICATION_JSON_VALUE)
+public class MeasureController {
+
+    private final MeasureService measureService;
+
+    public MeasureController(MeasureService measureService) {
+        this.measureService = measureService;
+    }
+
+    @PostMapping
+    public Measure save(@RequestBody Measure measure) {
+        return measureService.save(measure);
+    }
+
+    @GetMapping("/{deviceId}")
+    public Optional<Measure> getByDeviceId(@PathVariable String deviceId) {
+        return measureService.getById(deviceId);
+    }
+
+}

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

+package com.cardif.pocmeasurerest.infra.repository;
+
+import com.cardif.pocmeasurerest.infra.repository.entity.MeasureEntity;
+import org.springframework.data.mongodb.repository.MongoRepository;
+import org.springframework.data.repository.PagingAndSortingRepository;
+
+import java.util.Optional;
+
+public interface JpaMeasureRepository extends MongoRepository<MeasureEntity, String> {
+
+    Optional<MeasureEntity> findByDeviceId(String deviceId);
+
+}

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

+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.repository.entity.MeasureEntity;
+import org.springframework.cloud.stream.function.StreamBridge;
+import org.springframework.stereotype.Repository;
+
+import java.time.LocalDateTime;
+import java.util.Optional;
+
+@Repository
+public class MeasureRepositoryImpl implements MeasureRepository {
+
+    private final JpaMeasureRepository jpaMeasureRepository;
+    private final StreamBridge streamBridge;
+
+    public MeasureRepositoryImpl(JpaMeasureRepository jpaMeasureRepository, StreamBridge streamBridge) {
+        this.jpaMeasureRepository = jpaMeasureRepository;
+        this.streamBridge = streamBridge;
+    }
+
+    @Override
+    public Measure save(Measure measure) {
+        MeasureEntity entity = new MeasureEntity();
+        entity.setDate(LocalDateTime.now());
+        entity.setDeviceId(measure.deviceId());
+        entity.setValue(measure.value());
+        MeasureEntity entitySaved = jpaMeasureRepository.save(entity);
+        streamBridge.send("createMeasure", entitySaved);
+        return new Measure(entitySaved.getDeviceId(), entitySaved.getValue(),
+                entitySaved.getDate());
+    }
+
+    @Override
+    public Optional<Measure> getById(String id) {
+        return jpaMeasureRepository.findByDeviceId(id).map(entity -> new Measure(entity.getDeviceId(),
+                entity.getValue(), entity.getDate()));
+    }
+}

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

+package com.cardif.pocmeasurerest.infra.repository.entity;
+
+import org.springframework.data.annotation.Id;
+import org.springframework.data.mongodb.core.mapping.Document;
+
+import java.time.LocalDateTime;
+
+@Document
+public class MeasureEntity {
+
+    @Id
+    private String id;
+    private String deviceId;
+    private String value;
+    private LocalDateTime date;
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id;
+    }
+
+    public String getDeviceId() {
+        return deviceId;
+    }
+
+    public void setDeviceId(String deviceId) {
+        this.deviceId = deviceId;
+    }
+
+    public String getValue() {
+        return value;
+    }
+
+    public void setValue(String value) {
+        this.value = value;
+    }
+
+    public LocalDateTime getDate() {
+        return date;
+    }
+
+    public void setDate(LocalDateTime date) {
+        this.date = date;
+    }
+}

src/main/resources/application-local.properties

+#Mongo configurations
+spring.data.mongodb.database=carrefur-database
+spring.data.mongodb.host=localhost
+spring.data.mongodb.port=29514
+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
+spring.cloud.function.definition=createMeasure
+
+#Producer
+spring.cloud.stream.bindings.createMeasure.destination=measure
+
+# Actuator
+management.endpoints.jmx.exposure.exclude=*
+management.endpoints.web.exposure.include=health,info,metrics,prometheus
+management.info.git.enabled=true
+management.info.java.enabled=true
+management.info.os.enabled=true
+
+# If it has security preferred when-authorized
+management.endpoint.health.show-components=always
+management.endpoint.health.show-details=always
\ No newline at end of file

src/test/java/com/cardif/pocmeasurerest/PocMeasureRestApplicationTests.java

-package com.cardif.pocmeasurerest;
-
-import org.junit.jupiter.api.Test;
-import org.springframework.boot.test.context.SpringBootTest;
-
-@SpringBootTest
-class PocMeasureRestApplicationTests {
-
-	@Test
-	void contextLoads() {
-	}
-
-}