add kafka warehouse order status

pom.xml

@@ -25,10 +25,10 @@
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-webflux</artifactId>
 		</dependency>
-<!--		<dependency>-->
-<!--			<groupId>org.springframework.kafka</groupId>-->
-<!--			<artifactId>spring-kafka</artifactId>-->
-<!--		</dependency>-->
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+		</dependency>
 		<dependency>
 			<groupId>io.projectreactor.kafka</groupId>
 			<artifactId>reactor-kafka</artifactId>

src/main/java/com/nisum/ascend/inventory/configuration/KafkaReceiverConfig.java

 package com.nisum.ascend.inventory.configuration;
 
 
+import com.nisum.ascend.inventory.dto.WareHouseOrder;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.producer.ProducerConfig;
@@ -8,6 +9,7 @@ import org.apache.kafka.common.serialization.StringDeserializer;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
 import reactor.kafka.receiver.KafkaReceiver;
 import reactor.kafka.receiver.ReceiverOptions;
 
@@ -32,6 +34,7 @@ public class KafkaReceiverConfig {
         return KafkaReceiver.create(receiverOptions.addAssignListener(Collection::iterator)
                 .subscription(Collections.singleton(posLogTopic)));
     }
+
     private Map<String, Object> receiverConfig() {
         Map<String, Object> config = new HashMap<>();
         config.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);

src/main/java/com/nisum/ascend/inventory/dto/Item.java

+package com.nisum.ascend.inventory.dto;
+
+import lombok.Data;
+
+@Data
+public class Item {
+    private String itemId;
+    private String itemName;
+    private int itemSku;
+    private int itemQuantity;
+    private double itemPrice;
+}
+

src/main/java/com/nisum/ascend/inventory/dto/WareHouseOrder.java

+package com.nisum.ascend.inventory.dto;
+
+import com.nisum.ascend.inventory.dto.Item;
+import lombok.Data;
+
+import java.util.List;
+
+@Data
+public class WareHouseOrder {
+    private String id;
+    private String orderId;
+    private String status;
+    //LocalDateTime
+    private String createdAt;
+    private String modifiedAt;
+    private List<Item> orderItems;
+    private String address;
+
+}
+

src/main/java/com/nisum/ascend/inventory/repository/ProductRepository.java

 package com.nisum.ascend.inventory.repository;
 
-import com.nisum.ascend.inventory.dto.ProductDto;
+
 import com.nisum.ascend.inventory.model.Product;
 import org.springframework.data.mongodb.repository.Query;
 import org.springframework.data.mongodb.repository.ReactiveMongoRepository;

src/main/java/com/nisum/ascend/inventory/service/KafkaListenerService.java

 package com.nisum.ascend.inventory.service;
 
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.nisum.ascend.inventory.dto.Item;
+import com.nisum.ascend.inventory.dto.WareHouseOrder;
+import com.sun.tools.javac.jvm.Items;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.context.event.ApplicationStartedEvent;
 import org.springframework.context.event.EventListener;
 import org.springframework.stereotype.Component;
-import org.springframework.stereotype.Service;
 import reactor.kafka.receiver.KafkaReceiver;
 
+import java.util.List;
+
 @Component
 @Slf4j
 public class KafkaListenerService {
@@ -16,11 +21,31 @@ public class KafkaListenerService {
     @Autowired
     private KafkaReceiver<String, String> kafkaReceiver;
 
+    @Autowired
+    private ProductService productService;
+
     @EventListener(ApplicationStartedEvent.class)
-    public void consumeOrderMessages() {
+    public void consumeWarehouseOrderStatus() {
         kafkaReceiver.receive()
-                .doOnNext(record -> System.out.println(record))
+                .doOnNext(record -> log.info("record: {}", record))
+                .doOnNext(record -> onWarehouseOrderStatusReceived(record.value()))
                 .doOnError(throwable -> System.out.println(throwable.getMessage()))
                 .subscribe();
     }
+
+
+    private void onWarehouseOrderStatusReceived(String warehouseOrderString) {
+        try {
+            System.out.print(warehouseOrderString);
+            ObjectMapper objectMapper = new ObjectMapper();
+            WareHouseOrder warehouseOrder = objectMapper.readValue(warehouseOrderString, WareHouseOrder.class);
+            List<Item> itemList = warehouseOrder.getOrderItems();
+            for (Item item : itemList) {
+                productService.updateProductInventoryBySku(String.valueOf(item.getItemSku()),warehouseOrder.getStatus(), item.getItemQuantity());
+            }
+//            log.info("Received this data: {}", warehouseOrder);
+        } catch (Exception e) {
+            log.error("error", e);
+        }
+    }
 }

src/main/java/com/nisum/ascend/inventory/service/ProductService.java

 package com.nisum.ascend.inventory.service;
-import com.nisum.ascend.inventory.dto.ProductDto;
+
+import com.nisum.ascend.inventory.dto.Item;
 import com.nisum.ascend.inventory.exception.ResourceNotFoundException;
 import com.nisum.ascend.inventory.model.Product;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -36,6 +37,22 @@ public class ProductService {
         return productRepository.findAll();
     }
 
+
+    public Mono<Product> updateProductInventoryBySku(String sku, String status, int itemQuantity) {
+        int stock = itemQuantity;
+        if (status == "CANCELLED") {
+            stock *= -1;
+      }
+        int finalStock = stock;
+        return productRepository.findBySku(sku)
+                .flatMap(dbProduct -> {
+                    dbProduct.setAvailableStock(dbProduct.getAvailableStock() - finalStock);
+                    dbProduct.setFulfilledStock(dbProduct.getFulfilledStock() + finalStock);
+                    dbProduct.setBlockedStock(dbProduct.getBlockedStock() - itemQuantity);
+                    return productRepository.save(dbProduct);
+                });
+    }
+
     public Mono<Product> updateProductBySku(String sku, Product product){
         return productRepository.findBySku(sku)
                 .flatMap(dbProduct -> {

src/test/java/com/nisum/ascend/inventory/controller/ProductControllerTest.java

 package com.nisum.ascend.inventory.controller;
 
-import com.nisum.ascend.inventory.dto.ProductDto;
 import com.nisum.ascend.inventory.model.Product;
 import com.nisum.ascend.inventory.repository.ProductRepository;
 import com.nisum.ascend.inventory.service.ProductService;