[AFP-49] Order manger can GET order status from warehouse service

backend/order-management/pom.xml

@@ -67,6 +67,11 @@
 			<artifactId>springfox-boot-starter</artifactId>
 			<version>3.0.0</version>
 		</dependency>
+		<dependency>
+			<groupId>org.apache.velocity</groupId>
+			<artifactId>velocity</artifactId>
+			<version>1.5</version>
+		</dependency>
 	</dependencies>
 
 	<build>

backend/order-management/src/main/java/com/afp/ordermanagement/controller/KafkaController.java

@@ -17,7 +17,7 @@ public class KafkaController {
         this.producer = producer;
     }
 
-    @PostMapping(value = "/publish")
+    @PostMapping(value = "/publish/{id}")
     public void sendMessageToKafkaTopic(@RequestParam String message) {
         producer.sendMessage(message);
     }

backend/order-management/src/main/java/com/afp/ordermanagement/controller/ManagerController.java

@@ -2,6 +2,7 @@ package com.afp.ordermanagement.controller;
 
 
 import com.afp.ordermanagement.model.Manager;
+import com.afp.ordermanagement.model.Order;
 import com.afp.ordermanagement.repository.ManagerRepository;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.GetMapping;
@@ -24,4 +25,5 @@ public class ManagerController {
     }
 
 
+
 }

backend/order-management/src/main/java/com/afp/ordermanagement/controller/OrderController.java

 package com.afp.ordermanagement.controller;
 
 import com.afp.ordermanagement.model.Order;
-import com.afp.ordermanagement.repository.OrderRepository;
 import com.afp.ordermanagement.service.OrderService;
+import com.afp.ordermanagement.service.Producer;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.*;
 import reactor.core.publisher.Flux;
@@ -17,12 +18,27 @@ public class OrderController {
     @Autowired
     private OrderService orderService;
 
+    @Autowired
+    Producer producer;
+
+
+    /**
+     * DESC  -   This route will let order manager get order status from warehouse
+     * @param orderId
+     */
+    @GetMapping("/orderStatus/{orderId}")
+    public void getOrderStatusFromWarehouse(@PathVariable String orderId) {
+        producer.sendOrderId(orderId);
+    }
+
     @GetMapping("/orders")
+    @ResponseStatus(HttpStatus.OK)
     public Flux<Order> getAllOrders(){
         return orderService.getAllOrders();
     }
 
     @GetMapping("/orders/{customerId}")
+    @ResponseStatus(HttpStatus.OK)
     public Flux<Order> getAllOrdersByCustomerId(@PathVariable("customerId") String customerId) {
         return orderService.getAllOrdersByCustomerId(customerId);
     }
@@ -30,6 +46,7 @@ public class OrderController {
     @PostMapping("/orders")
     @ResponseStatus(HttpStatus.CREATED)
     public Mono<Order> saveOrder(@RequestBody Order order){
+
         return orderService.createOrder(order);
     }
 

backend/order-management/src/main/java/com/afp/ordermanagement/model/Order.java

@@ -30,6 +30,8 @@ public class Order {
         CANCELLED
     };
 
+
+
     @Builder.Default
     private OrderStatus orderStatus = OrderStatus.RECEIVED;
 

backend/order-management/src/main/java/com/afp/ordermanagement/repository/OrderRepository.java

@@ -5,7 +5,6 @@ import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
 import org.springframework.stereotype.Repository;
 import reactor.core.publisher.Flux;
 
-import java.util.List;
 
 @Repository
 public interface OrderRepository extends ReactiveMongoRepository<Order, String> {

backend/order-management/src/main/java/com/afp/ordermanagement/service/Consumer.java

 package com.afp.ordermanagement.service;
 
+import com.afp.ordermanagement.model.Order;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.kafka.annotation.KafkaListener;
@@ -14,4 +15,9 @@ public class Consumer {
     public void consumerManager(String message){
         logger.info(String.format("#### -> Consumed message -> %s", message));
     }
+
+    @KafkaListener(topics = "orders")
+    public void getOrderStatusFromWarehouse(String status) {
+        logger.info(String.format("#### -> Consumed order Status: %s", status));
+    }
 }

backend/order-management/src/main/java/com/afp/ordermanagement/service/Producer.java

 package com.afp.ordermanagement.service;
 
+
+import org.apache.velocity.exception.ResourceNotFoundException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.config.ConfigDataResourceNotFoundException;
 import org.springframework.stereotype.Service;
 import org.springframework.kafka.core.KafkaTemplate;
 
+import java.sql.SQLOutput;
+
 @Service
 public class Producer {
 
@@ -19,9 +24,24 @@ public class Producer {
         this.kafkaTemplate = kafkaTemplate;
     }
 
-    public void sendMessage(String message) {
-        logger.info(String.format("#### -> Producing message: %s", message));
-        kafkaTemplate.send(MANAGER_TOPIC, message);
+    /**
+     * DESC -   Sending orderId as a message to ORDER_TOPIC for warehouse to consume
+     * @param id
+     * @throws ResourceNotFoundException
+     * @throws IllegalArgumentException
+     */
+    public void sendOrderId(String id) {
+        try {
+            logger.info(String.format("#### -> Order id sent to warehouse: %s", id));
+            kafkaTemplate.send(ORDER_TOPIC, id);
+        } catch (ResourceNotFoundException e) {
+            logger.error("Order with that Id does not exist, exception caught: " + e);
+        } catch (IllegalArgumentException e) {
+            logger.error("Not a valid input, exception caught: " + e);
+        }
+
+
+
     }
 
 }