produce and consume from kafka

pom.xml

@@ -29,6 +29,10 @@
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-webflux</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-web</artifactId>
+		</dependency>
 		<dependency>
 			<groupId>org.apache.kafka</groupId>
 			<artifactId>kafka-streams</artifactId>

src/main/java/com/ascendfinalproject/warehouse/Consumer.java

+package com.ascendfinalproject.warehouse;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Service;
+
+import java.io.IOException;
+
+@Service
+public class Consumer {
+
+  private final Logger logger = LoggerFactory.getLogger(Producer.class);
+
+  // this is placeholder
+  @KafkaListener(topics = "fulfilled", groupId = "WAREHOUSE_MANAGEMENT")
+  public void consume(String message) throws IOException {
+    logger.info(String.format("#### -> Consumed message -> %s", message));
+  }
+
+}

src/main/java/com/ascendfinalproject/warehouse/Producer.java

+package com.ascendfinalproject.warehouse;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Service;
+
+
+@Service
+public class Producer {
+
+  private static final Logger logger = LoggerFactory.getLogger(Producer.class);
+  private static final String FULFILLED = "fulfilled";
+  private static final String CANCELLED = "cancelled";
+
+  @Autowired
+//  publish messages to the topic
+  private KafkaTemplate<String, String> kafkaTemplate;
+
+  public void orderFulfilled(String message) {
+    logger.info(String.format("#### -> this order is fulfilled -> %s", message));
+    this.kafkaTemplate.send(FULFILLED, message);
+  }
+
+  public void orderCancelled(String message) {
+    logger.info(String.format("#### -> this order is cancelled -> %s", message));
+    this.kafkaTemplate.send(CANCELLED, message);
+  }
+
+}

src/main/java/com/ascendfinalproject/warehouse/WarehouseApplication.java

@@ -11,3 +11,4 @@ public class WarehouseApplication {
 	}
 
 }
+

src/main/java/com/ascendfinalproject/warehouse/controllers/KafkaController.java

+package com.ascendfinalproject.warehouse.controllers;
+
+
+import com.ascendfinalproject.warehouse.Producer;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController
+@RequestMapping(value = "/kafka")
+public class KafkaController {
+
+  private final Producer producer;
+
+  @Autowired
+  KafkaController(Producer producer) {
+    this.producer = producer;
+  }
+
+  @PostMapping(value = "/fulfilled")
+  public void sendMessageToKafkaTopic(@RequestParam("message") String message) {
+    this.producer.orderFulfilled(message);
+  }
+
+}

src/main/resources/application.properties

 spring.data.mongodb.uri=mongodb+srv://warehouse1:ascendWarehouseProject@warehouse-cluster.xopll.mongodb.net/myFirstDatabase?retryWrites=true&w=majority
-spring.data.mongodb.database=test
\ No newline at end of file
+spring.data.mongodb.database=test
+
+server: port: 9000
+spring:
+    kafka:
+        consumer:
+            bootstrap-servers: localhost:9092
+            group-id: WAREHOUSE_MANAGEMENT
+            auto-offset-reset: earliest
+            key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+            value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+        producer:
+            bootstrap-servers: localhost:9092
+            key-serializer: org.apache.kafka.common.serialization.StringSerializer
+            value-serializer: org.apache.kafka.common.serialization.StringSerializer
\ No newline at end of file