Merge branch 'dev' into 'AFP-134'

# Conflicts:
#   src/main/java/com/afp/ordermanagement/controller/OrderController.java

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

@@ -3,6 +3,7 @@ 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.EmailService;
 import com.afp.ordermanagement.service.OrderService;
 import lombok.AllArgsConstructor;
 import lombok.Data;
@@ -41,8 +42,13 @@ public class OrderController {
     @PostMapping("/orders")
     @ResponseStatus(HttpStatus.CREATED)
     public Mono<Order> getOrderFromEcom(@RequestBody Order orderObject) {
-        sender.sendOrderToWarehouse(orderObject);
-        return orderService.createOrder(orderObject);
+//        Mono<Order> orderCreated = orderService.createOrder(orderObject);
+        return orderService.createOrder(orderObject)
+                .doOnSuccess(order -> {
+                    orderObject.setId(order.getId());
+                    System.out.println("sending order*******************" + orderObject.getId());
+                    sender.sendOrderToTopic(orderObject);
+                });
     }
 
     @GetMapping("/orders")
@@ -79,4 +85,12 @@ public class OrderController {
     public void deleteOrderbyId(@PathVariable(value = "orderId") String orderId) {
         orderService.deleteOrderById(orderId);
     }
+
+    @PostMapping("/orderStatus")
+    @ResponseStatus(HttpStatus.CREATED)
+    public Mono<Order> getUpdatedOrderFromWarehouseTopic(@RequestBody Order order) {
+        sender.sendUpdatedOrderToWarehouseTopic(order);
+        return orderService.updateOrderByOrderId(order.getId(), order);
+    }
+
 }

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

+package com.afp.ordermanagement.model;
+
+import lombok.Getter;
+import lombok.Setter;
+import lombok.ToString;
+import org.springframework.data.annotation.Id;
+
+import java.util.Date;
+import java.util.List;
+
+@Getter
+@Setter
+@ToString
+public class WarehouseOrder {
+    @Id
+    private String id;
+    private String orderId;
+    private String status;
+    private Date createdAt;
+    private Date modifiedAt;
+    private List<Item> orderItems;
+    private String address;
+
+
+    public WarehouseOrder() {
+
+
+
+    }
+}

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

@@ -4,7 +4,9 @@ import com.afp.ordermanagement.controller.OrderController;
 import com.afp.ordermanagement.model.Item;
 import com.afp.ordermanagement.model.Order;
 import com.afp.ordermanagement.model.OrderStatus;
+import com.afp.ordermanagement.model.WarehouseOrder;
 import com.afp.ordermanagement.repository.OrderRepository;
+import com.afp.ordermanagement.service.EmailService;
 import com.afp.ordermanagement.service.OrderService;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -34,6 +36,8 @@ public class Receiver {
     @Autowired
     private OrderRepository orderRepository;
 
+    @Autowired
+    private EmailService emailService;
 
     @EventListener(ApplicationStartedEvent.class)
     public void consumeOrderStatus() {
@@ -51,12 +55,19 @@ 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();
-
-            Mono<Order> updated = orderService.updateOrderByOrderId(orderId, order);
-            updated.block();
+            WarehouseOrder order = objectMapper.readValue(orderStr, WarehouseOrder.class);
+            log.info("order received from warehouse {}", order);
+            String orderId = order.getOrderId();
+
+            System.out.println("About to try sending an email.");
+            Mono<Order> convertedOrder = orderService.getOrderById(order.getOrderId());
+            convertedOrder.doOnSuccess(o -> {
+                log.info("Existing order in the db: {}", o);
+                o.setOrderStatus(OrderStatus.valueOf(order.getStatus()));
+                emailService.emailCreator(o);
+                orderService.updateOrderByOrderId(orderId, o).subscribe();
+                System.out.println("ayayayayayayayaya receiver" + o);
+            }).subscribe();
 
         } catch (Exception e) {
             log.error("Caught error on UpdateOrderStatus method", e);

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

 package com.afp.ordermanagement.reactivekafkaservice;
 
 import com.afp.ordermanagement.model.Order;
+import com.afp.ordermanagement.service.EmailService;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -11,24 +12,29 @@ import reactor.kafka.sender.KafkaSender;
 import reactor.kafka.sender.SenderRecord;
 import reactor.kafka.sender.SenderResult;
 
+import java.util.UUID;
+
 
 @Service
 @Slf4j
 public class Sender {
-
-
+    @Autowired
+    private EmailService emailService;
 
     @Autowired
     private KafkaSender<String, Order> kafkaEventProducer;
 
 
-    private static final String ORDER_TOPIC = "orders";
+    private static final String ORDER_TOPIC = "OMS_ORDER_UPDATE";
 
+    private static final String WAREHOUSE_TOPIC = "WMOS_ORDER_UPDATE";
 
-    public void sendOrderToWarehouse(Order orderObject) {
+    public void sendOrderToTopic(Order orderObject) {
         log.info(String.format("##### -> Sender sending message: %s ", orderObject));
         ProducerRecord<String, Order> record = new ProducerRecord<>(ORDER_TOPIC, orderObject);
-        Flux<SenderResult<Order>> working = kafkaEventProducer.send(Mono.just(SenderRecord.create(record, orderObject)))
+        System.out.println("In sendOrderToWarehouse");
+        emailService.emailCreator(orderObject);
+        Flux<SenderResult<UUID>> working = kafkaEventProducer.send(Mono.just(SenderRecord.create(record, UUID.randomUUID())))
                 .doOnError(throwable -> System.out.println(throwable))
                 .doOnNext(uuidSenderResult -> {
                     if (null != uuidSenderResult.exception()) {
@@ -38,5 +44,18 @@ public class Sender {
         working.doOnError(throwable -> log.error("some error")).subscribe();
     }
 
+    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/java/com/afp/ordermanagement/service/EmailService.java

@@ -22,7 +22,7 @@ public class EmailService {
         mailMessage.setSubject(subject);
         mailMessage.setText(message);
 
-        mailMessage.setFrom("NOREPLY@nisum.com");
+        mailMessage.setFrom("kaminskikeving@gmail.com");
 
         javaMailSender.send(mailMessage);
     }
@@ -32,16 +32,16 @@ public class EmailService {
         String message1 = "message ", message2 = "build failed.";
         switch (status) {
             case RECEIVED:
-                message1 = "Hello, you order #" + order.getId() + " has been received!";
-                message2 = "We hope to have your order fulfilled soon.";
+                message1 = "Hello, your order #" + order.getId() + " has been received!";
+                message2 = " We hope to have your order fulfilled soon.\n\n" + order.getOrderItems();
                 break;
             case CANCELLED:
                 message1 = "I'm sorry, your order #" + order.getId() + " has been canceled.";
-                message2 = "For more information, contact {NULL}.";
+                message2 = " For more information, contact {NULL}.";
                 break;
             case FULFILLED:
                 message1 = "Good news everyone! Your order #" + order.getId() + " has been fulfilled.";
-                message2 = "Your tracking number is: " + order.getOrderTrackingCode();
+                message2 = " Your tracking number is: " + order.getOrderTrackingCode();
                 break;
         }
         return message1 + message2;
@@ -60,7 +60,7 @@ public class EmailService {
                 status = "FULFILLED";
                 break;
         }
-        return "Your order #" + order.getId() + " has been" + status;
+        return "Your order #" + order.getId() + " has been " + status;
     }
 
     public String toCreator(Order order) {
@@ -73,6 +73,7 @@ public class EmailService {
         String to = toCreator(order);
 
         sendMail(to, subject, message);
+
         return "Email sent to customer!";
     }
 }

src/main/java/com/afp/ordermanagement/service/OrderService.java

@@ -38,6 +38,7 @@ public class OrderService {
         newOrder.setOrderTrackingCode(defaultOrderTrackingCode);
         newOrder.setOrderCreatedAt(serviceSystemTime);
         newOrder.setOrderUpdatedAt(serviceSystemTime);
+        //newOrder.setId("abc123");
         //System.out.println(newOrder);
         return orderRepository.save(newOrder);
     }
@@ -95,12 +96,13 @@ public class OrderService {
 
         return orderRepository.findById(orderId)
                 .flatMap(existingOrder -> {
-                    existingOrder.setCustomerAddress(newOrder.getCustomerAddress());
-                    existingOrder.setCustomerEmailAddress(newOrder.getCustomerEmailAddress());
-                    existingOrder.setOrderTrackingCode(newOrder.getOrderTrackingCode());
-                    existingOrder.setOrderItems(newOrder.getOrderItems());
+                    //existingOrder.setCustomerAddress(newOrder.getCustomerAddress());
+                    //existingOrder.setCustomerEmailAddress(newOrder.getCustomerEmailAddress());
+                    //existingOrder.setOrderTrackingCode(newOrder.getOrderTrackingCode());
+                    //existingOrder.setOrderItems(newOrder.getOrderItems());
                     existingOrder.setOrderStatus(newOrder.getOrderStatus());
-                    existingOrder.setOrderItems(newOrder.getOrderItems());
+                    //existingOrder.setOrderItems(newOrder.getOrderItems());
+                    System.out.println("serviceeee" + existingOrder);
                     return orderRepository.save(existingOrder);
                 });
     }

src/main/resources/application.properties

+spring.data.mongodb.uri=mongodb+srv://user:password2021@cluster0.g23rm.mongodb.net/myFirstDatabase?
 kafka.producer.bootstrap-servers=localhost:9092
 kafka.producer.acks=all
 kafka.consumer.bootstrap-servers=localhost:9092
 kafka.consumer.group-id=group_id
-kafka.topic.input=orders
+kafka.topic.input=WMOS_ORDER_UPDATE
+
+server.port=8084
+
 # Config for MailTrap SMTP Mail testing service
 spring.mail.protocol=smtp
 spring.mail.host=smtp.mailtrap.io
@@ -10,4 +14,13 @@ spring.mail.port=2525
 spring.mail.username=945a9d376253be
 spring.mail.password=96d65f623868cc
 spring.mail.properties.mail.smtp.auth = true
-spring.mail.properties.mail.smtp.starttls.enable = true
+spring.mail.properties.mail.smtp.starttls.enable = false
\ No newline at end of file
+
+## Config for GMail SMTP mail sending
+#spring.mail.host=smtp.gmail.com
+#spring.mail.port=587
+#spring.mail.username=gmail email address
+#spring.mail.password=gmail email password
+#spring.mail.properties.mail.smtp.auth=true
+#spring.mail.properties.mail.smtp.starttls.enable=true
+#mail.smtp.debug=true
\ No newline at end of file

src/main/resources/kafka.producer.bootstrap-servers

+kafka.producer.bootstrap-servers=localhost:9092
+kafka.producer.acks=all
+kafka.consumer.bootstrap-servers=localhost:9092
+kafka.consumer.group-id=group_id
+kafka.topic.input=orders
\ No newline at end of file