add return endpoint

Author: rberrelleza

frontend/src/App.jsx

@@ -242,6 +242,9 @@ const Item = ({ item, onRent, backdrop }) => {
               <div className="Item__button Item__button--rented button">
                 Watch Now
               </div>
+              <div className="Item__button button" onClick={() => handleReturn(item)}>
+                Return
+              </div>
             </>
           }
         </div>
@@ -250,6 +253,32 @@ const Item = ({ item, onRent, backdrop }) => {
   );
 }
 
+function handleReturn(item) {
+  fetch('/rent/return', {
+    method: 'POST',
+    headers: {
+      'Content-Type': 'application/json'
+    },
+    body: JSON.stringify({
+      catalog_id: item.id
+    })
+  })
+  .then(response => {
+    if (!response.ok) {
+      throw new Error('Network response was not ok');
+    }
+    return response.json();
+  })
+  .then(data => {
+    console.log(`Returned movie with ID: ${item.id}`);
+    // Refresh data to update the UI
+    this.refreshData();
+  })
+  .catch(error => {
+    console.error('There was a problem with the return request:', error);
+  });
+}
+
 const CartIcon = () => {
   return (
     <svg xmlns="http://www.w3.org/2000/svg" fill="none" viewBox="0 0 21 22" width="16">

rent/src/main/java/com/okteto/rent/controller/RentController.java

@@ -18,7 +18,8 @@
 
 @RestController
 public class RentController {
-    private static final String KAFKA_TOPIC = "rentals";
+    private static final String KAFKA_TOPIC_RENTALS = "rentals";
+    private static final String KAFKA_TOPIC_RETURNS = "returns";
 
     private final Logger logger = LoggerFactory.getLogger(RentController.class);
 
@@ -37,7 +38,7 @@ List<String> rent(@RequestBody Rent rentInput) {
 
         logger.info("Rent [{},{}] received", catalogID, price);
 
-        kafkaTemplate.send(KAFKA_TOPIC, catalogID, price)
+        kafkaTemplate.send(KAFKA_TOPIC_RENTALS, catalogID, price)
         .thenAccept(result -> logger.info("Message [{}] delivered with offset {}",
                         catalogID,
                         result.getRecordMetadata().offset()))
@@ -50,6 +51,24 @@ List<String> rent(@RequestBody Rent rentInput) {
         return new LinkedList<>();
     }
 
+    @PostMapping(path= "/rent/return", consumes = "application/json", produces = "application/json")
+    public Map<String, String> returnMovie(@RequestBody ReturnRequest returnRequest) {
+        String catalogID = returnRequest.getMovieID();
+
+        logger.info("Return [{}] received", catalogID);
+
+        kafkaTemplate.send(KAFKA_TOPIC_RETURNS, catalogID, catalogID)
+        .thenAccept(result -> logger.info("Return message [{}] delivered with offset {}",
+                        catalogID,
+                        result.getRecordMetadata().offset()))
+        .exceptionally(ex -> {
+            logger.warn("Unable to deliver return message [{}]. {}", catalogID, ex.getMessage());
+            return null;
+        });
+
+        return Collections.singletonMap("status", "return processed");
+    }
+
     public static class Rent {
         @JsonProperty("catalog_id")
         private String movieID;
@@ -72,4 +91,17 @@ public String getPrice() {
             return price;
         }
     }
+
+    public static class ReturnRequest {
+        @JsonProperty("catalog_id")
+        private String movieID;
+
+        public void setMovieID(String movieID) {
+            this.movieID = movieID;
+        }
+
+        public String getMovieID() {
+            return movieID;
+        }
+    }
 }

worker/cmd/worker/main.go

@@ -41,28 +41,42 @@ func main() {
 	master := kafka.GetMaster()
 	defer master.Close()
 
-	consumer, err := master.ConsumePartition(*topic, 0, sarama.OffsetOldest)
+	// Consumer for "rentals" topic
+	consumerRentals, err := master.ConsumePartition("rentals", 0, sarama.OffsetOldest)
+	if err != nil {
+		log.Panic(err)
+	}
+
+	// Consumer for "returns" topic
+	consumerReturns, err := master.ConsumePartition("returns", 0, sarama.OffsetOldest)
 	if err != nil {
 		log.Panic(err)
 	}
 
 	signals := make(chan os.Signal, 1)
 	signal.Notify(signals, os.Interrupt)
 	doneCh := make(chan struct{})
+
 	go func() {
 		for {
 			select {
-			case err := <-consumer.Errors():
+			case err := <-consumerRentals.Errors():
 				fmt.Println(err)
-			case msg := <-consumer.Messages():
+			case msg := <-consumerRentals.Messages():
 				*messageCountStart++
 				fmt.Printf("Received message: movies %s price %s\n", string(msg.Key), string(msg.Value))
 				price, _ := strconv.ParseFloat(string(msg.Value), 64)
-				// price *= 0.5
 				insertDynStmt := `insert into "rentals"("id", "price") values($1, $2) on conflict(id) do update set price = $2`
 				if _, err := db.Exec(insertDynStmt, string(msg.Key), fmt.Sprintf("%f", price)); err != nil {
 					log.Panic(err)
 				}
+			case msg := <-consumerReturns.Messages():
+				catalogID := string(msg.Value)
+				fmt.Printf("Received return message: catalogID %s\n", catalogID)
+				deleteStmt := `DELETE FROM rentals WHERE id = $1`
+				if _, err := db.Exec(deleteStmt, catalogID); err != nil {
+					log.Panic(err)
+				}
 			case <-signals:
 				fmt.Println("Interrupt is detected")
 				doneCh <- struct{}{}