Deployment 58

deployment.yaml

@@ -14,7 +14,7 @@ spec:
     spec:
       containers:
         - name: afp-prods-container
-          image: nexus.mynisum.com/afp-prods:58
+          image: nexus.mynisum.com/afp-prods:59
           imagePullPolicy: Always
           ports:
             - containerPort: 8083

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.Order;
-//import com.nisum.ascend.inventory.dto.WareHouseOrder;
-//import lombok.extern.slf4j.Slf4j;
-//import org.springframework.beans.factory.annotation.Autowired;
-//import org.springframework.beans.factory.annotation.Qualifier;
-//import org.springframework.boot.context.event.ApplicationStartedEvent;
-//import org.springframework.context.event.EventListener;
-//import org.springframework.stereotype.Component;
-//import reactor.kafka.receiver.KafkaReceiver;
-//
-//import java.util.List;
-//
-//@Component
-//@Slf4j
-//public class KafkaListenerService {
-//
-//    @Autowired
-//    @Qualifier("kafkaWarehouseOrderReceiver")
-//    private KafkaReceiver<String, String> kafkaWarehouseOrderReceiver;
-//
-//    @Autowired
-//    @Qualifier("kafkaOrderReceiver")
-//    private KafkaReceiver<String, String> kafkaOrderReceiver;
-//
-//    @Autowired
-//    private ProductService productService;
-//
-//    @EventListener(ApplicationStartedEvent.class)
-//    public void consumeWarehouseOrderStatus() {
-//        kafkaWarehouseOrderReceiver.receive()
-//                .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 {
-//            ObjectMapper objectMapper = new ObjectMapper();
-//            WareHouseOrder warehouseOrder = objectMapper.readValue(warehouseOrderString, WareHouseOrder.class);
-//            List<Item> itemList = warehouseOrder.getOrderItems();
-//            String status = warehouseOrder.getStatus();
-//            log.info("Received this data: {}", warehouseOrder);
-//            log.info("recieved this list: {}", itemList);
-//            for (Item item : itemList) {
-//                productService.updateProductInventoryBySkuWareHouse(item.getItemSku(), status, item.getItemQuantity()).subscribe();
-//            }
-//        } catch (Exception e) {
-//            log.error("error", e);
-//        }
-//    }
-//
-//
-//    @EventListener(ApplicationStartedEvent.class)
-//    public void consumeOrderStatus() {
-//        kafkaOrderReceiver.receive()
-//                .doOnNext(record -> log.info("record: {}", record))
-//                .doOnNext(record -> onOrderStatusReceived(record.value()))
-//                .doOnError(throwable -> System.out.println(throwable.getMessage()))
-//                .subscribe();
-//    }
-//
-//    private void onOrderStatusReceived(String orderString) {
-//        try {
-//            ObjectMapper objectMapper = new ObjectMapper();
-//            Order order = objectMapper.readValue(orderString, Order.class);
-//            List<Item> itemList = order.getOrderItems();
-//            String status = order.getOrderStatus();
-//            log.info("Received this data: {}", order);
-//            log.info("recieved this list: {}", itemList);
-//            for (Item item : itemList) {
-//                productService.updateProductInventoryBySkuOrder(item.getItemSku(), status, item.getItemQuantity()).block();
-//            }
-//
-//        } catch (Exception e) {
-//            log.error("error", e);
-//        }
-//    }
-//}
+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.Order;
+import com.nisum.ascend.inventory.dto.WareHouseOrder;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.context.event.ApplicationStartedEvent;
+import org.springframework.context.event.EventListener;
+import org.springframework.stereotype.Component;
+import reactor.kafka.receiver.KafkaReceiver;
+
+import java.util.List;
+
+@Component
+@Slf4j
+public class KafkaListenerService {
+
+    @Autowired
+    @Qualifier("kafkaWarehouseOrderReceiver")
+    private KafkaReceiver<String, String> kafkaWarehouseOrderReceiver;
+
+    @Autowired
+    @Qualifier("kafkaOrderReceiver")
+    private KafkaReceiver<String, String> kafkaOrderReceiver;
+
+    @Autowired
+    private ProductService productService;
+
+    @EventListener(ApplicationStartedEvent.class)
+    public void consumeWarehouseOrderStatus() {
+        kafkaWarehouseOrderReceiver.receive()
+                .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 {
+            ObjectMapper objectMapper = new ObjectMapper();
+            WareHouseOrder warehouseOrder = objectMapper.readValue(warehouseOrderString, WareHouseOrder.class);
+            List<Item> itemList = warehouseOrder.getOrderItems();
+            String status = warehouseOrder.getStatus();
+            log.info("Received this data: {}", warehouseOrder);
+            log.info("recieved this list: {}", itemList);
+            for (Item item : itemList) {
+                productService.updateProductInventoryBySkuWareHouse(item.getItemSku(), status, item.getItemQuantity()).subscribe();
+            }
+        } catch (Exception e) {
+            log.error("error", e);
+        }
+    }
+
+
+    @EventListener(ApplicationStartedEvent.class)
+    public void consumeOrderStatus() {
+        kafkaOrderReceiver.receive()
+                .doOnNext(record -> log.info("record: {}", record))
+                .doOnNext(record -> onOrderStatusReceived(record.value()))
+                .doOnError(throwable -> System.out.println(throwable.getMessage()))
+                .subscribe();
+    }
+
+    private void onOrderStatusReceived(String orderString) {
+        try {
+            ObjectMapper objectMapper = new ObjectMapper();
+            Order order = objectMapper.readValue(orderString, Order.class);
+            List<Item> itemList = order.getOrderItems();
+            String status = order.getOrderStatus();
+            log.info("Received this data: {}", order);
+            log.info("recieved this list: {}", itemList);
+            for (Item item : itemList) {
+                productService.updateProductInventoryBySkuOrder(item.getItemSku(), status, item.getItemQuantity()).block();
+            }
+
+        } catch (Exception e) {
+            log.error("error", e);
+        }
+    }
+}