Added kafka producer to storage service and consumer to ratings service

rating-service/pom.xml

@@ -47,6 +47,11 @@
 			<artifactId>lombok</artifactId>
 			<version>1.16.16</version>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+			<version>1.2.2.RELEASE</version>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-test</artifactId>

rating-service/src/main/java/com/lohika/jclub/rating/service/ApartmentRecordListener.java

@@ -0,0 +1,35 @@
+package com.lohika.jclub.rating.service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.annotation.EnableKafka;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+
+/**
+ * @author Andriy Levchenko
+ */
+@Slf4j
+@Component
+@EnableKafka
+public class ApartmentRecordListener {
+
+    @Autowired
+    private RatingService ratingService;
+
+    @Autowired
+    private ObjectMapper objectMapper;
+
+    @KafkaListener(topics = "${kafka.topic.boot}")
+    public void receive(ConsumerRecord<String, String> record) throws IOException {
+        log.info("Received message: " + record);
+        Apartment apartment = objectMapper.readValue(record.value(), Apartment.class);
+        BigDecimal rating = ratingService.calculateRating(apartment);
+        log.info("Rating : {}", rating);
+    }
+}

rating-service/src/main/resources/application.yml

@@ -0,0 +1,9 @@
+spring:
+  kafka:
+    consumer:
+      auto-offset-reset: earliest
+      group-id: boot
+
+kafka:
+  topic:
+    boot: apartments

storage-service/pom.xml

@@ -55,6 +55,11 @@
 			<artifactId>spring-boot-starter-test</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+			<version>1.2.2.RELEASE</version>
+		</dependency>
 	</dependencies>
 
 	<dependencyManagement>

storage-service/src/main/java/com/lohika/jclub/storage/service/ApartmentRepositoryEventListener.java

@@ -0,0 +1,36 @@
+package com.lohika.jclub.storage.service;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.rest.core.event.AbstractRepositoryEventListener;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author Andriy Levchenko
+ */
+@Slf4j
+@Component
+public class ApartmentRepositoryEventListener extends AbstractRepositoryEventListener<ApartmentRecord> {
+
+    @Autowired
+    private KafkaTemplate<String, String> kafkaTemplate;
+
+    @Autowired
+    private ObjectMapper objectMapper;
+
+    @Override
+    protected void onAfterCreate(ApartmentRecord apartmentRecord) {
+        log.info("Creating message for apartmentRecord " + apartmentRecord.toString());
+        String json;
+        try {
+            json = objectMapper.writeValueAsString(apartmentRecord);
+        } catch (JsonProcessingException e) {
+            throw new RuntimeException("Error", e);
+        }
+        kafkaTemplate.send("apartments", json);
+        log.info("Message sent for apartmentRecord " + apartmentRecord.toString());
+    }
+}