split kafka consumer methods into order and warehouse

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

+package com.nisum.ascend.inventory.dto;
+
+import lombok.Data;
+
+@Data
+public class CustomerAddress {
+    private String street;
+    private String city;
+    private String state;
+    private String zip;
+}

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

@@ -8,12 +8,12 @@ import java.util.List;
 @Data
 public class Order {
     private String id;
-    private String orderUpdatedAt;
-    private String orderCreatedAt;
+    private long orderUpdatedAt;
+    private long orderCreatedAt;
     private String customerId;
     private String customerEmailAddress;
     private String orderStatus;
     List<Item> orderItems;
     private String orderTrackingCode;
-    private String customerAddress;
+    private CustomerAddress customerAddress;
 }

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

@@ -3,6 +3,7 @@ package com.nisum.ascend.inventory.dto;
 import com.nisum.ascend.inventory.dto.Item;
 import lombok.Data;
 
+import java.util.Date;
 import java.util.List;
 
 @Data
@@ -10,11 +11,9 @@ public class WareHouseOrder {
     private String id;
     private String orderId;
     private String status;
-    //LocalDateTime
-    private String createdAt;
-    private String modifiedAt;
+    private Date createdAt;
+    private Date modifiedAt;
     private List<Item> orderItems;
     private String address;
-
 }
 

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

@@ -47,9 +47,8 @@ public class KafkaListenerService {
             log.info("Received this data: {}", warehouseOrder);
             log.info("recieved this list: {}", itemList);
             for (Item item : itemList) {
-                productService.updateProductInventoryBySku(item.getItemSku(), status, item.getItemQuantity()).block();
+                productService.updateProductInventoryBySkuWareHouse(item.getItemSku(), status, item.getItemQuantity()).subscribe();
             }
-
         } catch (Exception e) {
             log.error("error", e);
         }
@@ -74,7 +73,7 @@ public class KafkaListenerService {
             log.info("Received this data: {}", order);
             log.info("recieved this list: {}", itemList);
             for (Item item : itemList) {
-                productService.updateProductInventoryBySku(item.getItemSku(), status, item.getItemQuantity()).block();
+                productService.updateProductInventoryBySkuOrder(item.getItemSku(), status, item.getItemQuantity()).block();
             }
 
         } catch (Exception e) {

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

@@ -38,46 +38,77 @@ public class ProductService {
     }
 
 
-    public Mono<Product> updateProductInventoryBySku(String sku, String status, int itemQuantity) {
-        int stock = itemQuantity;
-        if (status.equals("CANCELLED")) {
-            stock *= -1;
+    public Mono<Product> updateProductBySku(String sku, Product product){
+        return productRepository.findBySku(sku)
+                .flatMap(dbProduct -> {
+                    dbProduct.setUpc(product.getUpc());
+                    dbProduct.setProductName(product.getProductName());
+                    dbProduct.setProductDescription(product.getProductDescription());
+                    dbProduct.setBrand(product.getBrand());
+                    dbProduct.setCategory(product.getCategory());
+                    dbProduct.setPrice(product.getPrice());
+                    dbProduct.setAvailableStock(product.getAvailableStock());
+                    dbProduct.setProductImageUrl(product.getProductImageUrl());
+                    return productRepository.save(dbProduct);
+                });
+    }
+
+    public Mono<Product> createProduct(Product product) {
+        return productRepository.save(product);
+    }
+
+
+    public Mono<Product> updateProductInventoryBySkuWareHouse(String sku, String status, int itemQuantity) {
+        System.out.printf("sku = %s, status = %s, itemquanity = %d \n", sku, status, itemQuantity);
+
+        if (status.equals("CANCELED")) {
+             return productRepository.findBySku(sku)
+                    .flatMap(dbProduct -> {
+                        dbProduct.setAvailableStock(dbProduct.getAvailableStock() + itemQuantity);
+                        return productRepository.save(dbProduct);
+                    });
         } else if (status.equals("RECEIVED")) {
-            System.out.printf("sku = %s, status = %s, itemquanity = %d, stock = %d, finalStock = %d \n", sku, status, itemQuantity);
              return productRepository.findBySku(sku)
                     .flatMap(dbProduct -> {
                         dbProduct.setBlockedStock(dbProduct.getBlockedStock() + itemQuantity);
                         return productRepository.save(dbProduct);
                     });
-        }
-        int finalStock = stock;
-        System.out.printf("sku = %s, status = %s, itemquanity = %d, stock = %d, finalStock = %d \n", sku, status, itemQuantity, stock, finalStock);
+        } else { // if status.equals("FULFILLED")
              return productRepository.findBySku(sku)
                     .flatMap(dbProduct -> {
-                    dbProduct.setAvailableStock(dbProduct.getAvailableStock() - finalStock);
+                        dbProduct.setAvailableStock(dbProduct.getAvailableStock() - itemQuantity);
                         dbProduct.setFulfilledStock(dbProduct.getFulfilledStock() + itemQuantity);
                         dbProduct.setBlockedStock(dbProduct.getBlockedStock() - itemQuantity);
                         return productRepository.save(dbProduct);
                     });
         }
+    }
 
-    public Mono<Product> updateProductBySku(String sku, Product product){
+        public Mono<Product> updateProductInventoryBySkuOrder(String sku, String status, int itemQuantity) {
+            System.out.printf("sku = %s, status = %s, itemquanity = %d \n", sku, status, itemQuantity);
+
+            if (status.equals("CANCELED")){
                 return productRepository.findBySku(sku)
                         .flatMap(dbProduct -> {
-                    dbProduct.setUpc(product.getUpc());
-                    dbProduct.setProductName(product.getProductName());
-                    dbProduct.setProductDescription(product.getProductDescription());
-                    dbProduct.setBrand(product.getBrand());
-                    dbProduct.setCategory(product.getCategory());
-                    dbProduct.setPrice(product.getPrice());
-                    dbProduct.setAvailableStock(product.getAvailableStock());
-                    dbProduct.setProductImageUrl(product.getProductImageUrl());
+                            dbProduct.setAvailableStock(dbProduct.getAvailableStock() + itemQuantity);
+                            dbProduct.setBlockedStock(dbProduct.getBlockedStock() - itemQuantity);
                             return productRepository.save(dbProduct);
                         });
             }
-
-    public Mono<Product> createProduct(Product product) {
-        return productRepository.save(product);
+            else if (status.equals("RECEIVED")){
+                return productRepository.findBySku(sku)
+                        .flatMap(dbProduct -> {
+                            dbProduct.setBlockedStock(dbProduct.getBlockedStock() + itemQuantity);
+                            return productRepository.save(dbProduct);
+                        });
             }
+            else { // if status.equals("FULFILLED")
+                return productRepository.findBySku(sku)
+                        .flatMap(dbProduct -> {
+                            return productRepository.save(dbProduct);
+                        });
+            }
+    }
+
 
 }