Committed Changes to Configure Producer and Integration with Order Generator

omd-checkout-service/pom.xml

@@ -28,6 +28,16 @@
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-simple</artifactId>
+            <version>1.7.30</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-devtools</artifactId>
@@ -43,7 +53,7 @@
             <groupId>com.github.javafaker</groupId>
             <artifactId>javafaker</artifactId>
             <version>1.0.2</version>
-</dependency>
+        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-test</artifactId>

omd-checkout-service/src/main/java/com/nisum/omd/OrderGenerator.java

@@ -2,6 +2,7 @@ package com.nisum.omd;
 
 import java.time.LocalDateTime;
 
+import com.nisum.omd.service.OmdProducerService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.CommandLineRunner;
 import org.springframework.stereotype.Component;
@@ -19,6 +20,10 @@ public class OrderGenerator implements CommandLineRunner{
 	RandomDataGeneratorImpl rdg;
 	@Autowired
 	Order order;
+
+	@Autowired
+	OmdProducerService omdProducerService;
+
 	Boolean orderGenerateFlag =  true;
 	
 	public Boolean getOrderGenerateFlag() {
@@ -33,6 +38,7 @@ public class OrderGenerator implements CommandLineRunner{
 
 		  while(orderGenerateFlag){
 		  order=getObjectData(order);
+		  omdProducerService.sendMessage(order);
 		  ObjectMapper objectMapper = new ObjectMapper();
 		  objectMapper.registerModule(new JavaTimeModule());
 		  objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);

omd-checkout-service/src/main/java/com/nisum/omd/config/OmdProducerConfig.java

+package com.nisum.omd.config;
+
+import com.nisum.omd.domain.Order;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.support.serializer.JsonSerializer;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class OmdProducerConfig {
+
+    @Value("${bootstrap-server}")
+    private String bootstrapServer;
+
+    @Bean
+    public ProducerFactory<String, Order> producerFactory(){
+
+        Map<String,Object> config = new HashMap<>();
+
+        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
+        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
+
+        return new DefaultKafkaProducerFactory<>(config);
+    }
+
+    @Bean
+    public KafkaTemplate<String, Order> kafkaTemplate(){
+        return new KafkaTemplate<>(producerFactory());
+    }
+}

omd-checkout-service/src/main/java/com/nisum/omd/service/OmdProducerService.java

+package com.nisum.omd.service;
+
+import com.nisum.omd.domain.Order;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
+import org.springframework.stereotype.Service;
+import org.springframework.util.concurrent.ListenableFuture;
+import org.springframework.util.concurrent.ListenableFutureCallback;
+
+@Service
+public class OmdProducerService {
+
+    @Autowired
+    private KafkaTemplate<String, Order> kafkaTemplate;
+
+    @Value("${topic-name}")
+    private String topic;
+
+    Logger logger = LoggerFactory.getLogger(OmdProducerService.class);
+
+    public void sendMessage(Order order) {
+        ListenableFuture<SendResult<String, Order>> future =
+                kafkaTemplate.send(topic, order);
+        future.addCallback(new ListenableFutureCallback<SendResult<String, Order>>() {
+            @Override
+            public void onSuccess(SendResult<String, Order> result) {
+                logger.info(" Published Order Data with Order Number :"
+                        + order.getOrderNum() + " In Topic :"+ result.getRecordMetadata().topic());
+            }
+            @Override
+            public void onFailure(Throwable ex) {
+                logger.error("Unable to Publish the Order Data, Due to: ",ex.getMessage());
+            }
+        });
+    }
+
+}

omd-checkout-service/src/main/resources/application.properties

@@ -67,3 +67,8 @@ app.paymentMethod.paymentStatus=AUTH
 app.paymentMethod.paymentDetail.TransactionId=021361475199
 app.paymentMethod.paymentDetail.ProcessedAmount=1000.00
 app.paymentMethod.paymentDetail.RequestAmount=1000.00
+
+
+#Kafka Properties
+bootstrap-server = localhost:9092
+topic-name = TOPIC_OMD_ORDER_DATA
\ No newline at end of file