kafka-messages has been added to Person and Salary service

Author: cevheri

person-service/README.md

@@ -34,6 +34,16 @@ services:
 We will use github public repository for our configuration:
 https://github.com/cevheri/microservices-config-server
 
+## Kafka Configuration
+
+---
+### Create Kafka Topic
+```shell
+kafka-topics.sh --create --zookeeper zookeeper:2181 --replication-factor 1 --partitions 5 --topic salary
+```
+---
+
+
 
 ---
 ## Development

person-service/pom.xml

@@ -83,7 +83,15 @@
             <artifactId>springfox-swagger-ui</artifactId>
             <version>3.0.0</version>
         </dependency>
-
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka-test</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
     <dependencyManagement>
         <dependencies>

person-service/src/main/java/com/cevher/ms/person/service/PersonService.java

@@ -20,10 +20,28 @@ public class PersonService {
 
     private final PersonRepository personRepository;
     private final RestTemplate restTemplate;
+    private final SalaryProducer salaryProducer;
+
+    private final Double PERSON_DEFAULT_SALARY = 1000D;
+
+    private void sendFirstSalary(Person person) {
+        log.info("sendFirstSalary method of PersonService for CurrentPerson");
+
+        SalaryMessage salaryMessage = SalaryMessage
+                .builder()
+                .personId(person.getId())
+                .amount(PERSON_DEFAULT_SALARY)
+                .build();
+        salaryProducer.produce(salaryMessage.toString());
+
+        log.info("Send Kafka Message: " + salaryMessage);
+    }
 
     public Person savePerson(Person person) {
         log.info("savePerson method of PersonService");
-        return personRepository.save(person);
+        Person resultPerson = personRepository.save(person);
+        sendFirstSalary(resultPerson);
+        return resultPerson;
     }
 
     public ResponseTempVM getPersonWithDepartment(Long personId) {

person-service/src/main/java/com/cevher/ms/person/service/SalaryProducer.java

@@ -0,0 +1,54 @@
+package com.cevher.ms.person.service;
+
+import lombok.*;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Service;
+
+import java.io.Serializable;
+import java.util.UUID;
+
+@Service
+@Slf4j
+@RequiredArgsConstructor
+public class SalaryProducer {
+    private static final String KAFKA_TOPIC = "salary";
+
+    private final KafkaTemplate<String, String> kafkaTemplate;
+
+    public void produce(String message) {
+        kafkaTemplate.send(KAFKA_TOPIC, message);
+    }
+}
+
+
+@AllArgsConstructor
+@NoArgsConstructor
+@ToString
+@EqualsAndHashCode
+@Builder()
+class SalaryMessage implements Serializable {
+
+    private String toService;
+    private String fromService;
+    private String uuid;
+
+    @Getter
+    @Setter
+    private Long personId;
+
+    @Getter
+    @Setter
+    private Double amount;
+
+    public String getUuid() {
+        return UUID.randomUUID().toString();
+    }
+
+    public String getFromService() {
+        return "person-service";
+    }
+    public String getToService() {
+        return "salary-service";
+    }
+}

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

@@ -3,7 +3,17 @@ spring:
     name: person-service
   zipkin:
     base-url: ${ZIPKIN_URI:http://zipkin:9411/}
-
+  kafka:
+    consumer:
+      bootstrap-servers: localhost:9092
+      group-id: salary_group
+      auto-offset-reset: earliest
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+    producer:
+      bootstrap-servers: localhost:9092
+      key-serializer: org.apache.kafka.common.serialization.StringSerializer
+      value-serializer: org.apache.kafka.common.serialization.StringSerializer
 server:
   port: 9002
 
@@ -28,4 +38,4 @@ eureka:
 #inform the Eureka service that client wants to be advertised by IP address.
 #Personally, we always set this attribute to true. Cloud-based microservices are sup-
 #posed to be ephemeral and stateless. They can be started up and shut down at will.
-#IP addresses are more appropriate for these types of services.
+#IP addresses are more appropriate for these types of services.

salary-service/src/main/java/com/cevher/ms/salary/domain/Salary.java

@@ -14,7 +14,7 @@
 @ToString
 @EqualsAndHashCode
 @Entity
-@Table(name = "salary")
+@Table(name = "salary", catalog = "public")
 public class Salary implements Serializable {
 
     @Id

salary-service/src/main/java/com/cevher/ms/salary/service/SalaryConsumer.java

@@ -0,0 +1,56 @@
+package com.cevher.ms.salary.service;
+
+import lombok.*;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Service;
+
+import java.io.Serializable;
+import java.util.UUID;
+
+@Service
+@Slf4j
+@RequiredArgsConstructor
+public class SalaryConsumer {
+
+    private static final String KAFKA_TOPIC = "salary";
+    private static final String KAFKA_CONSUMER_GROUP = "salary_group";
+
+    @KafkaListener(topics = KAFKA_TOPIC,
+            groupId = KAFKA_CONSUMER_GROUP)
+    public void consume(String message) {
+        log.info("Consumer Group : " + message.toString());
+    }
+}
+
+@AllArgsConstructor
+@NoArgsConstructor
+@ToString
+@EqualsAndHashCode
+@Builder()
+class SalaryMessage implements Serializable {
+
+    private String toService;
+    private String fromService;
+    private String uuid;
+
+    @Getter
+    @Setter
+    private Long personId;
+
+    @Getter
+    @Setter
+    private Double amount;
+
+    public String getUuid() {
+        return UUID.randomUUID().toString();
+    }
+
+    public String getFromService() {
+        return "person-service";
+    }
+
+    public String getToService() {
+        return "salary-service";
+    }
+}

…evher/ms/salary/rest/SalaryResource.java …r/ms/salary/web/rest/SalaryResource.javasalary-service/src/main/java/com/cevher/ms/salary/rest/SalaryResource.java renamed to salary-service/src/main/java/com/cevher/ms/salary/web/rest/SalaryResource.java salary-service/src/main/java/com/cevher/ms/salary/rest/SalaryResource.java renamed to salary-service/src/main/java/com/cevher/ms/salary/web/rest/SalaryResource.java

@@ -1,4 +1,4 @@
-package com.cevher.ms.salary.rest;
+package com.cevher.ms.salary.web.rest;
 
 import com.cevher.ms.salary.dto.SalaryDto;
 import com.cevher.ms.salary.service.SalaryService;

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

@@ -3,6 +3,18 @@ spring:
     name: salary-service
   zipkin:
     base-url: ${ZIPKIN_URI:http://zipkin:9411/}
+  kafka:
+    consumer:
+      bootstrap-servers: localhost:9092
+      group-id: salary_group
+      auto-offset-reset: earliest
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+    producer:
+      bootstrap-servers: localhost:9092
+      key-serializer: org.apache.kafka.common.serialization.StringSerializer
+      value-serializer: org.apache.kafka.common.serialization.StringSerializer
+
 server:
   port: 9004