[AFP-53] ✨ Added endpoint testing for kafka update [@svalencia]

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

@@ -66,7 +66,7 @@ public class KafkaConfig {
     }
 
     @Bean
-    public KafkaReceiver<String, String> kafkaEventReceiver(@Value("${kafka.topic.input}") String posLogTopic) {
+    public KafkaReceiver<String, String> kafkaEventReceiver(@Value("${kafka.topic.output}") String posLogTopic) {
         ReceiverOptions<String, String> receiverOptions = ReceiverOptions.create(consumerFactory());
         receiverOptions.maxCommitAttempts(3);
         return KafkaReceiver.create(receiverOptions.addAssignListener(Collection::iterator).subscription(Collections.singleton(posLogTopic)));

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

@@ -42,6 +42,15 @@ public class OrderController {
     }
 
 
+    @PostMapping("/orderStatus")
+    @ResponseStatus(HttpStatus.CREATED)
+    public Mono<Order> getUpdatedOrderFromWarehouseTopic(@RequestBody Order order) {
+        sender.sendUpdatedOrderToWarehouseTopic(order);
+        return orderService.updateOrderByOrderId(order.getId(), order);
+    }
+
+    
+
     @GetMapping("/orders")
     @CrossOrigin
     public Flux<Order> getAllOrders(){

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

@@ -41,9 +41,16 @@ public class Receiver {
     @EventListener(ApplicationStartedEvent.class)
     public void consumeOrderStatus() {
         kafkaReceiver.receive()
-//                .doOnNext(record -> System.out.println(record))
-//                .doOnNext(record -> log.info("record.value(): {} ", record.value()))
-//                .doOnNext(record -> updateOrderStatus(record.value()))
+                .doOnNext(record -> System.out.println(record))
+                .doOnNext(record -> log.info("record.value(): {} ", record.value()))
+//                .doOnNext(record -> {
+//                    if (record.key() != null) {
+//                        updateOrderStatus(record.value());
+//                    } else {
+//                        log.error("{} is :::::: ", record.value());
+//                    }
+//                })
+                .doOnNext(record -> updateOrderStatus(record.value()))
                 .doOnError(throwable -> System.out.println(throwable.getMessage()))
                 .subscribe();
     }
@@ -53,20 +60,25 @@ public class Receiver {
 
     private void updateOrderStatus(String orderStr) {
         try {
-            ObjectMapper objectMapper = new ObjectMapper();
-            Order order = objectMapper.readValue(orderStr, Order.class);
-            log.info("ORDER objectMapper {}", order);
-            String orderId = order.getId();
-
-            System.out.println("About to try sending an email.");
-            emailService.emailCreator(order);
-
-            Mono<Order> updated = orderService.updateOrderByOrderId(orderId, order);
-            updated.block();
+//            if (orderStr.isEmpty()) {
+//                log.info("ORDERSTR IS EMPTYYYYYY");
+//                return;
+//            } else {
+                ObjectMapper objectMapper = new ObjectMapper();
+                Order order = objectMapper.readValue(orderStr, Order.class);
+                log.info("ORDER objectMapper {}", order);
+                String orderId = order.getId();
+
+                System.out.println("About to try sending an email.");
+                emailService.emailCreator(order);
+
+                Mono<Order> updated = orderService.updateOrderByOrderId(orderId, order);
+                updated.subscribe();
+//            }
 
         } catch (Exception e) {
             log.error("Caught error on UpdateOrderStatus method", e);
-            e.printStackTrace();
+//            e.printStackTrace();
         }
     }
 

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

@@ -25,6 +25,8 @@ public class Sender {
 
     private static final String ORDER_TOPIC = "orders";
 
+    private static final String WAREHOUSE_TOPIC = "warehouseManagement";
+
 
     public void sendOrderToWarehouse(Order orderObject) {
         log.info(String.format("##### -> Sender sending message: %s ", orderObject));
@@ -42,4 +44,17 @@ public class Sender {
     }
 
 
+    public void sendUpdatedOrderToWarehouseTopic(Order order) {
+        ProducerRecord<String, Order> record = new ProducerRecord<>(WAREHOUSE_TOPIC, order);
+        System.out.println("In sendUpdatedOrderToWarehouseTopic");
+        Flux<SenderResult<Order>> updated = kafkaEventProducer.send(Mono.just(SenderRecord.create(record, order)))
+                .doOnError(throwable -> System.out.println(throwable))
+                .doOnNext(uuidSenderResult -> {
+                    if (null != uuidSenderResult.exception()) {
+                        System.out.println("send order update");
+                    }
+                });
+        updated.doOnError(throwable -> log.error("error on sendUpdatedOrderToWarehouse method")).subscribe();
+    }
+
 }

src/main/resources/application.properties

@@ -4,6 +4,7 @@ kafka.producer.acks=all
 kafka.consumer.bootstrap-servers=localhost:9092
 kafka.consumer.group-id=group_id
 kafka.topic.input=orders
+kafka.topic.output=warehouseManagement
 
 # Config for MailTrap SMTP Mail testing service
 spring.mail.protocol=smtp