[AFP-53] ✨ Receive updated status from warehouse and update database [@svalencia]

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

@@ -12,6 +12,7 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.KafkaTemplate;*/
+import reactor.core.publisher.Flux;
 import reactor.kafka.receiver.KafkaReceiver;
 import reactor.kafka.receiver.ReceiverOptions;
 import reactor.kafka.sender.KafkaSender;
@@ -44,7 +45,6 @@ public class KafkaConfig {
         senderConfigProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
         senderConfigProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
         return senderConfigProps;
-        //return new DefaultKafkaProducerFactory<>(configProps);
     }
 
     @Bean
@@ -53,6 +53,7 @@ public class KafkaConfig {
         return KafkaSender.create(senderOptions);
     }
 
+
 //    @Bean
 //    public KafkaTemplate<String, Object> kafkaTemplateString() {
 //        return new KafkaTemplate<>(producerFactoryString());

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

 package com.afp.ordermanagement.controller;
 
 import com.afp.ordermanagement.model.Order;
+import com.afp.ordermanagement.reactivekafkaservice.Receiver;
 import com.afp.ordermanagement.reactivekafkaservice.Sender;
 import com.afp.ordermanagement.service.OrderService;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -20,6 +21,9 @@ public class OrderController {
     @Autowired
     Sender sender;
 
+    @Autowired
+    Receiver receiver;
+
 
     /**
      * DESC  -   This route will let order manager get order status from warehouse
@@ -30,6 +34,8 @@ public class OrderController {
         sender.sendOrderIdToWarehouse(orderId);
     }
 
+
+
     @GetMapping("/orders")
     @CrossOrigin
     public Flux<Order> getAllOrders(){

src/main/java/com/afp/ordermanagement/model/Order.java

@@ -14,6 +14,7 @@ import java.util.Objects;
 @Document(collection = "orders")
 public class Order {
 
+
     @Id
     private String id;
 
@@ -32,4 +33,8 @@ public class Order {
     public Order(){
     }
 
+    public Order(OrderStatus status) {
+        this.orderStatus = status;
+    }
+
 }

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

 package com.afp.ordermanagement.reactivekafkaservice;
 
+import com.afp.ordermanagement.model.Order;
+import com.afp.ordermanagement.model.OrderStatus;
+import com.afp.ordermanagement.service.OrderService;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.AutoConfigurationPackage;
 import org.springframework.boot.context.event.ApplicationStartedEvent;
 import org.springframework.context.event.EventListener;
 import org.springframework.stereotype.Service;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 import reactor.kafka.receiver.KafkaReceiver;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
 @Service
 @Slf4j
 public class Receiver {
@@ -14,11 +26,65 @@ public class Receiver {
     @Autowired
     private KafkaReceiver<String, String> kafkaReceiver;
 
+    @Autowired
+    private OrderService orderService;
+
+
+
     @EventListener(ApplicationStartedEvent.class)
     public void consumeOrderStatus() {
         kafkaReceiver.receive()
                 .doOnNext(record -> log.info(String.format("##### -> Receiver receiving message: %s ", record.value())))
+                .doOnNext(record -> log.info("record.value(): {} ", record.value()))
+                .doOnNext(record -> onOrderStatusReceived(record.value()))
                 .doOnError(throwable -> System.out.println(throwable.getMessage()))
                 .subscribe();
     }
+
+
+
+    private void onOrderStatusReceived(String orderStatusStr) {
+        try {
+            //deserialize JSON string into java object using ObjectMapper
+            ObjectMapper objectMapper = new ObjectMapper();
+            Map<String, String> orderStatus = objectMapper.readValue(orderStatusStr, Map.class);
+            String id = orderStatus.get("id");
+            String status = orderStatus.get("orderStatus").toUpperCase(Locale.ROOT);
+            updateOrderStatus(id, status);
+            log.info("orderStatus: {}", orderStatus);
+        } catch (Exception e) {
+            log.error("Caught error", e);
+        }
+    }
+
+
+    private void updateOrderStatus(String orderId, String orderStatus) {
+        if (checkExistingOrder(orderId)) {
+            log.info("Updating {} with status {}", orderId, orderStatus);
+            Order newOrder = new Order(OrderStatus.valueOf(orderStatus));
+            Mono<Order> updateOrder = orderService.updateOrderByOrderId(orderId, newOrder);
+            updateOrder.subscribe();
+//            updateOrder.block(); //subscribe vs block?
+        }
+    }
+
+
+
+    private boolean checkExistingOrder(String orderId) {
+        Flux<Order> orders = orderService.getAllOrders();
+        List<Order> orderList = orders.collectList().block();
+        Order res = orderList.stream()
+                .filter(order -> orderId.equals(order.getId()))
+                .findAny()
+                .orElse(null);
+
+        if (res == null) {
+            log.error("Order {} not found", orderId);
+            return false;
+        }
+
+        log.info("Order exists on the database");
+        return true;
+    }
+
 }

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 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
 public class Sender {
 
 
+
     @Autowired
     private KafkaSender<String, String> 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)))
                 .doOnError(throwable -> System.out.println(throwable))
@@ -33,4 +47,13 @@ public class Sender {
                 });
         working.doOnError(throwable -> log.error("some error")).subscribe();
     }
+
+
+    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

@@ -37,7 +37,7 @@ public class OrderService {
         return orderRepository.findByCustomerId(customerId);
     }
 
-    public Mono<Order> updateOrderByOrderId(String orderId, Order newOrder){
+    public Mono<Order> updateOrderByOrderId(String orderId, Order newOrder) {
         return orderRepository.findById(orderId)
                 .flatMap(existingOrder -> {
                     existingOrder.setCustomerAddress(newOrder.getCustomerAddress());