AFP[120] & AFP[121] ✨ Ecom orders to kafka and DB

pom.xml

@@ -26,6 +26,13 @@
 			<artifactId>spring-boot-starter-webflux</artifactId>
 		</dependency>
 		<!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka -->
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+			<version>2.7.0</version>
+		</dependency>
+		<!-- https://mvnrepository.com/artifact/io.confluent/kafka-json-serializer -->
+
 		<!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka-dist -->
 		<dependency>
 			<groupId>org.projectlombok</groupId>
@@ -44,12 +51,12 @@
 			<scope>test</scope>
 		</dependency>
 		<!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
-		<dependency>
+		<!--<dependency>
 			<groupId>org.projectlombok</groupId>
 			<artifactId>lombok</artifactId>
 			<version>1.18.16</version>
 			<scope>provided</scope>
-		</dependency>
+		</dependency>-->
 		<!-- https://mvnrepository.com/artifact/com.github.javafaker/javafaker -->
 		<dependency>
 			<groupId>com.github.javafaker</groupId>
@@ -89,6 +96,11 @@
 			<groupId>io.projectreactor.kafka</groupId>
 			<artifactId>reactor-kafka</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>com.fasterxml.jackson.core</groupId>
+			<artifactId>jackson-databind</artifactId>
+			<version>2.11.1</version>
+		</dependency>
 	</dependencies>
 	<build>
 		<plugins>

src/main/java/com/afp/ordermanagement/config/KafkaConfig.java

 package com.afp.ordermanagement.config;
 
-import com.afp.ordermanagement.reactivekafkaservice.Sender;
+import com.afp.ordermanagement.model.Order;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.StringDeserializer;
@@ -8,16 +8,14 @@ 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.config.ConcurrentKafkaListenerContainerFactory;
-import org.springframework.kafka.core.ConsumerFactory;
-import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
-import org.springframework.kafka.core.KafkaTemplate;*/
+import org.springframework.kafka.support.serializer.JsonSerializer;
 import reactor.kafka.receiver.KafkaReceiver;
 import reactor.kafka.receiver.ReceiverOptions;
 import reactor.kafka.sender.KafkaSender;
 import reactor.kafka.sender.SenderOptions;
 
 
+
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
@@ -42,14 +40,14 @@ public class KafkaConfig {
         senderConfigProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
         senderConfigProps.put(ProducerConfig.ACKS_CONFIG, acks);
         senderConfigProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
-        senderConfigProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        senderConfigProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
         return senderConfigProps;
-        //return new DefaultKafkaProducerFactory<>(configProps);
+        //return new DefaultKafkaProducerFactory<>(senderConfigProps);
     }
 
     @Bean
-    public KafkaSender<String, String> kafkaEventProducer() {
-        SenderOptions<String, String> senderOptions = SenderOptions.create(producerFactoryString());
+    public KafkaSender<String, Order> kafkaEventProducer() {
+        SenderOptions<String, Order> senderOptions = SenderOptions.create(producerFactoryString());
         return KafkaSender.create(senderOptions);
     }
 

src/main/java/com/afp/ordermanagement/controller/OrderController.java

@@ -12,6 +12,7 @@ import reactor.core.publisher.Mono;
 
 @RestController
 @RequestMapping("/api")
+@CrossOrigin
 public class OrderController {
 
     @Autowired
@@ -22,12 +23,17 @@ public class OrderController {
 
 
     /**
-     * DESC  -   This route will let order manager get order status from warehouse
-     * @param orderId
+     * DESC -   Persisting order information in the database and sending the
+     *          order object via orders kafka topic so that warehouse can
+     *          handle that
+     * @param orderObject
+     * @return
      */
-    @GetMapping("/orderStatus/{orderId}")
-    public void getOrderStatusFromWarehouse(@PathVariable String orderId) {
-        sender.sendOrderIdToWarehouse(orderId);
+    @PostMapping("/ordersFromEcom")
+    @ResponseStatus(HttpStatus.CREATED)
+    public Mono<Order> getOrderFromEcom(@RequestBody Order orderObject) {
+        sender.sendOrderToWarehouse(orderObject);
+        return orderService.createOrder(orderObject);
     }
 
     @GetMapping("/orders")
@@ -46,14 +52,7 @@ public class OrderController {
     public Flux<Order> getAllOrdersByCustomerId(@PathVariable("customerId") String customerId) {
         return orderService.getAllOrdersByCustomerId(customerId);
     }
-
-    @PostMapping("/orders")
-    @ResponseStatus(HttpStatus.CREATED)
-    public Mono<Order> saveOrder(@RequestBody Order order){
-
-        return orderService.createOrder(order);
-    }
-
+    
     @PutMapping("/order/{orderId}")
     public Mono<ResponseEntity<Order>> updateOrder(@PathVariable(value = "orderId") String orderId, @RequestBody Order order){
         return orderService.updateOrderByOrderId(orderId, order)

src/main/java/com/afp/ordermanagement/reactivekafkaservice/Receiver.java

@@ -34,7 +34,7 @@ public class Receiver {
     @EventListener(ApplicationStartedEvent.class)
     public void consumeOrderStatus() {
         kafkaReceiver.receive()
-                .doOnNext(record -> log.info(String.format("##### -> Receiver receiving message: %s ", record.value())))
+                .doOnNext(record -> System.out.println(record))
                 .doOnNext(record -> log.info("record.value(): {} ", record.value()))
                 .doOnNext(record -> onOrderStatusReceived(record.value()))
                 .doOnError(throwable -> System.out.println(throwable.getMessage()))

src/main/java/com/afp/ordermanagement/reactivekafkaservice/Sender.java

 package com.afp.ordermanagement.reactivekafkaservice;
 
-import com.afp.ordermanagement.service.OrderService;
-import com.fasterxml.jackson.databind.ObjectMapper;
+import com.afp.ordermanagement.model.Order;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.clients.producer.RecordMetadata;
-import org.reactivestreams.Publisher;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.kafka.sender.KafkaSender;
-import reactor.kafka.sender.SenderOptions;
 import reactor.kafka.sender.SenderRecord;
 import reactor.kafka.sender.SenderResult;
 
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.CountDownLatch;
-
 
 @Service
 @Slf4j
@@ -28,17 +19,16 @@ public class Sender {
 
 
     @Autowired
-    private KafkaSender<String, String> kafkaEventProducer;
+    private KafkaSender<String, Order> kafkaEventProducer;
 
 
     private static final String ORDER_TOPIC = "orders";
 
 
-    public void sendOrderIdToWarehouse(String id) {
-        log.info(String.format("##### -> Sender sending message: %s ", id));
-//        sendMessage(ORDER_TOPIC, id, id);
-        ProducerRecord<String, String> record = new ProducerRecord<>(ORDER_TOPIC, id);
-        Flux<SenderResult<String>> working = kafkaEventProducer.send(Mono.just(SenderRecord.create(record, id)))
+    public void sendOrderToWarehouse(Order orderObject) {
+        log.info(String.format("##### -> Sender sending message: %s ", orderObject));
+        ProducerRecord<String, Order> record = new ProducerRecord<>(ORDER_TOPIC, orderObject);
+        Flux<SenderResult<Order>> working = kafkaEventProducer.send(Mono.just(SenderRecord.create(record, orderObject)))
                 .doOnError(throwable -> System.out.println(throwable))
                 .doOnNext(uuidSenderResult -> {
                     if (null != uuidSenderResult.exception()) {
@@ -49,11 +39,11 @@ public class Sender {
     }
 
 
-    public void sendUpdatedStatus(String id, String status) {
-        log.info(String.format("Sender sending updated status for ordernumber: %s", id));
-        ProducerRecord<String, String> stat = new ProducerRecord<>(ORDER_TOPIC, status);
-
-    }
+//    public void sendUpdatedStatus(String id, String status) {
+//        log.info(String.format("Sender sending updated status for ordernumber: %s", id));
+//        ProducerRecord<String, String> stat = new ProducerRecord<>(ORDER_TOPIC, status);
+//
+//    }
 
 
 }

src/main/java/com/afp/ordermanagement/service/OrderService.java

@@ -15,6 +15,7 @@ public class OrderService {
     OrderRepository orderRepository;
 
     public Mono<Order> createOrder(Order newOrder){
+        System.out.println("here");
         String defaultOrderTrackingCode = "N/A";
         OrderStatus defaultOrderStatus = OrderStatus.RECEIVED;
         long serviceSystemTime = System.currentTimeMillis();
@@ -22,6 +23,7 @@ public class OrderService {
         newOrder.setOrderTrackingCode(defaultOrderTrackingCode);
         newOrder.setOrderCreatedAt(serviceSystemTime);
         newOrder.setOrderUpdatedAt(serviceSystemTime);
+        //System.out.println(newOrder);
         return orderRepository.save(newOrder);
     }