feat: configured kafka and implemented producer and consumer

Author: AdityaByte

pom.xml

@@ -50,6 +50,10 @@
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-data-redis</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+		</dependency>
 
 		<dependency>
 			<groupId>org.springframework.boot</groupId>

src/main/java/com/connect/config/KafkaProducerConfig.java

@@ -0,0 +1,15 @@
+package com.connect.config;
+
+import org.apache.kafka.clients.admin.NewTopic;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class KafkaProducerConfig {
+
+    @Bean
+    public NewTopic newTopic() {
+        return new NewTopic("chat", 1, (short) 1);
+    }
+
+}

src/main/java/com/connect/controller/ChatUserController.java

@@ -2,6 +2,7 @@
 
 import com.connect.dto.MessageDTO;
 import com.connect.dto.UserDTO;
+import com.connect.kafka.KafkaPublisherService;
 import com.connect.model.Message;
 import com.connect.model.User;
 import com.connect.service.ChatUserService;
@@ -44,9 +45,8 @@ public void handleMessage(@Payload MessageDTO message, SimpMessageHeaderAccessor
         // Fetching the RoomID from the header.
         String roomID = headerAccessor.getFirstNativeHeader("roomId");
         log.info("Sending the message to the room: {}", roomID);
-        Optional<Message> dbMessage = chatUserService.addMessagetoRoom(message, roomID);
-        dbMessage.ifPresent(msg ->
-                messagingTemplate.convertAndSend("/topic/chat/" + roomID, msg));
+        chatUserService.publishMessageToKafka(message, roomID);
+        messagingTemplate.convertAndSend("/topic/chat/" + roomID, message);
     }
 
     // Route for handling the history.

src/main/java/com/connect/kafka/KafkaListenerService.java

@@ -0,0 +1,18 @@
+package com.connect.kafka;
+
+import com.connect.model.Message;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Service;
+
+@Service
+@Slf4j
+public class KafkaListenerService {
+
+    @KafkaListener(topics = "chat", groupId = "chat-group")
+    public void consumeEvent(Message message) {
+        // logic message is being consumed.
+        log.info("Consumed message: {}", message.toString());
+    }
+
+}

src/main/java/com/connect/kafka/KafkaPublisherService.java

@@ -0,0 +1,30 @@
+package com.connect.kafka;
+
+import com.connect.model.Message;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
+import org.springframework.stereotype.Service;
+
+import java.util.concurrent.CompletableFuture;
+
+@Service
+@Slf4j
+public class KafkaPublisherService {
+
+    @Autowired
+    private KafkaTemplate<String, Object> kafkaTemplate;
+
+    public void sendEvent(Message message) {
+        CompletableFuture<SendResult<String, Object>> future = kafkaTemplate.send("chat", message);
+        future.whenComplete((result, exception) -> {
+            if (exception == null) {
+                log.info("Message sent successfully to {}", result.getRecordMetadata().offset());
+            } else {
+                log.error("Unable to send message: {}", message.toString());
+            }
+        });
+    }
+
+}

src/main/java/com/connect/model/Message.java

@@ -1,9 +1,6 @@
 package com.connect.model;
 
-import lombok.AllArgsConstructor;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.Setter;
+import lombok.*;
 import org.bson.types.ObjectId;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.mongodb.core.mapping.Document;
@@ -14,6 +11,8 @@
 @Setter
 @NoArgsConstructor
 @AllArgsConstructor
+@Builder
+@ToString
 @Document(collection = "messages")
 public class Message {
     @Id

src/main/java/com/connect/service/ChatUserService.java

@@ -2,6 +2,7 @@
 
 import com.connect.dto.MessageDTO;
 import com.connect.enums.UserStatus;
+import com.connect.kafka.KafkaPublisherService;
 import com.connect.model.Message;
 import com.connect.model.Room;
 import com.connect.model.User;
@@ -33,6 +34,9 @@ public class ChatUserService {
     @Autowired
     private UserRepository userRepository;
 
+    @Autowired
+    private KafkaPublisherService kafkaPublisherService;
+
     private Map<String, User> users = new ConcurrentHashMap<>();
 
     public void greetingHandler(String token) {
@@ -69,6 +73,25 @@ public void joiningRequestHandler(String username, String roomId) {
         roomService.addUserToRoom(requiredUser,  roomID);
     }
 
+    public void publishMessageToKafka(MessageDTO messageDTO, String roomId) {
+        if (messageDTO == null) {
+            log.error("Message DTO is null");
+            return;
+        } else if (messageDTO.getMessage().isEmpty() || roomId.isEmpty()) {
+            log.error("Empty Data occurred");
+            return;
+        }
+        ObjectId roomID = new ObjectId(roomId);
+        // We are assuming that the roomID is not null so.
+        Message message = Message.builder()
+                .roomId(roomID)
+                .sender(messageDTO.getSender())
+                .message(messageDTO.getMessage())
+                .build();
+
+        kafkaPublisherService.sendEvent(message);
+    }
+
     public Optional<Message> addMessagetoRoom(MessageDTO messageDTO, String roomId) {
         if (messageDTO == null) {
             log.error("Message DTO is null");

src/main/resources/application.properties

@@ -23,4 +23,15 @@ frontend.origin = ${FRONTEND_ORIGIN}
 spring.data.redis.host = ${REDIS_HOST}
 spring.data.redis.port = ${REDIS_PORT}
 spring.data.redis.password = ${REDIS_PASSWORD}
-spring.data.redis.ssl.enabled = false
+spring.data.redis.ssl.enabled = false
+
+# Kafka Producer configuration
+spring.kafka.producer.bootstrap-servers = ${KAFKA_SERVER}
+spring.kafka.producer.keySerializer = org.apache.kafka.common.serialization.StringSerializer
+spring.kafka.producer.valueSerializer = org.springframework.kafka.support.serializer.JsonSerializer
+
+# Kafka Consumer configuration
+spring.kafka.consumer.bootstrap-servers = ${KAFKA_SERVER}
+spring.kafka.consumer.keyDeserializer = org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.consumer.valueDeserializer = org.springframework.kafka.support.serializer.JsonDeserializer
+spring.kafka.consumer.properties.spring.json.trusted.packages = com.connect.model