[BE] FEAT: state server, produce "1 connect" to kafka

src/backend/chat-server/src/main/resources/static/index.html

@@ -46,7 +46,8 @@ <h1>Group Chat</h1>
 
 <script>
     let stompClient = null;
-    let userId = 'user' + Math.floor(Math.random() * 1000); // 랜덤 사용자 ID 생성
+    //let userId = 'user' + Math.floor(Math.random() * 1000); // 랜덤 사용자 ID 생성
+    let userId = '1';
     let roomId = '1'; // 테스트용 단일 방 ID
 
     function connect() {

src/backend/kafka/docker-compose.yml

@@ -21,7 +21,7 @@ services:
       KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
       KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
       KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
-      KAFKA_CREATE_TOPICS: "chatting:1:1, playlist:1:1, room:1:1, connection:1:1, invitation:1:1, friend-online-notify:1:1"
+      KAFKA_CREATE_TOPICS: "chatting:1:1, playlist:1:1, room:1:1, connection:1:1, invitation:1:1, friend_online_notify:1:1"
     volumes:
       - /var/run/docker.sock:/var/run/docker.sock
     networks:

src/backend/state-server/build.gradle

@@ -21,9 +21,13 @@ dependencies {
 	implementation 'org.springframework.boot:spring-boot-starter-web'
 	implementation 'org.springframework.kafka:spring-kafka'
 	implementation 'org.springframework.boot:spring-boot-starter-data-redis'
+	implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
+	implementation 'org.projectlombok:lombok'
+	annotationProcessor 'org.projectlombok:lombok'
 	testImplementation 'org.springframework.boot:spring-boot-starter-test'
 	testImplementation 'io.projectreactor:reactor-test'
 	testImplementation 'org.springframework.kafka:spring-kafka-test'
+	runtimeOnly 'com.mysql:mysql-connector-j'
 	testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
 }
 

src/backend/state-server/src/main/java/com/example/state/config/KafkaProducerConfig.java

@@ -0,0 +1,51 @@
+package com.example.state.config;
+
+import com.example.state.dto.UserStatusUpdateDto;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.support.serializer.JsonSerializer;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class KafkaProducerConfig {
+
+    @Value("${spring.kafka.bootstrap-servers}")
+    private String KAFKA_BROKER; // Kafka 브로커 주소 // Kafka broker 주소
+
+    @Bean
+    public ProducerFactory<String, String> producerFactory() {
+        Map<String, Object> configProps = new HashMap<>();
+        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_BROKER);
+        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        return new DefaultKafkaProducerFactory<>(configProps);
+    }
+
+    @Bean
+    public KafkaTemplate<String, String> kafkaTemplate() {
+        return new KafkaTemplate<>(producerFactory());
+    }
+
+    @Bean
+    public ProducerFactory<String, UserStatusUpdateDto> userStatusProducerFactory() {
+        Map<String, Object> configProps = new HashMap<>();
+        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_BROKER);
+        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class); // JSON 직렬화
+
+        return new DefaultKafkaProducerFactory<>(configProps);
+    }
+
+    @Bean
+    public KafkaTemplate<String, UserStatusUpdateDto> userStatusKafkaTemplate() {
+        return new KafkaTemplate<>(userStatusProducerFactory());
+    }
+}

src/backend/state-server/src/main/java/com/example/state/dto/UserStatusUpdateDto.java

@@ -0,0 +1,17 @@
+package com.example.state.dto;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class UserStatusUpdateDto {
+    private String userId;
+    private String status;
+    private List<String> friends;
+    private String timestamp;
+}

src/backend/state-server/src/main/java/com/example/state/entity/Friend.java

@@ -0,0 +1,27 @@
+package com.example.state.entity;
+
+import jakarta.persistence.*;
+import lombok.*;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+@Entity
+@Table(name = "friend")
+@Getter
+@NoArgsConstructor
+@AllArgsConstructor
+@IdClass(FriendId.class)
+public class Friend {
+
+    @Id
+    @Column(name = "friend_1")
+    private Long friend1;
+
+    @Id
+    @Column(name = "friend_2")
+    private Long friend2;
+
+    @Column(name = "created_at", nullable = false, updatable = false)
+    private String createdAt;
+}

src/backend/state-server/src/main/java/com/example/state/entity/FriendId.java

@@ -0,0 +1,30 @@
+package com.example.state.entity;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+public class FriendId implements Serializable {
+    private Long friend1;
+    private Long friend2;
+
+    public FriendId() {}
+
+    public FriendId(Long friend1, Long friend2) {
+        this.friend1 = friend1;
+        this.friend2 = friend2;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        FriendId friendId = (FriendId) o;
+        return Objects.equals(friend1, friendId.friend1) &&
+                Objects.equals(friend2, friendId.friend2);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(friend1, friend2);
+    }
+}

src/backend/state-server/src/main/java/com/example/state/repository/FriendRepository.java

@@ -0,0 +1,20 @@
+package com.example.state.repository;
+
+import com.example.state.entity.Friend;
+import com.example.state.entity.FriendId;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.stereotype.Repository;
+
+import java.util.List;
+
+@Repository
+public interface FriendRepository extends JpaRepository<Friend, FriendId> {
+
+    @Query("SELECT CASE " +
+            "WHEN f.friend1 = :userId THEN f.friend2 " +
+            "WHEN f.friend2 = :userId THEN f.friend1 " +
+            "END " +
+            "FROM Friend f WHERE f.friend1 = :userId OR f.friend2 = :userId")
+    List<Long> findFriendsByUserId(Long userId);
+}

src/backend/state-server/src/main/java/com/example/state/service/FriendService.java

@@ -0,0 +1,19 @@
+package com.example.state.service;
+
+import com.example.state.repository.FriendRepository;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+@Service
+@RequiredArgsConstructor
+public class FriendService {
+    private final FriendRepository friendRepository;
+
+    public List<String> getFriends(String userId) {
+        List<Long> friendIds = friendRepository.findFriendsByUserId(Long.parseLong(userId));
+        return friendIds.stream().map(String::valueOf).collect(Collectors.toList());
+    }
+}

src/backend/state-server/src/main/java/com/example/state/service/KafkaProducerService.java

@@ -0,0 +1,18 @@
+package com.example.state.service;
+
+import com.example.state.dto.UserStatusUpdateDto;
+import lombok.RequiredArgsConstructor;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Service;
+
+@Service
+@RequiredArgsConstructor
+public class KafkaProducerService {
+    private final KafkaTemplate<String, UserStatusUpdateDto> userStatusKafkaTemplate;
+
+    private static final String TOPIC = "friend_online_notify";
+
+    public void sendUserStatus(UserStatusUpdateDto userStatusUpdateDto) {
+        userStatusKafkaTemplate.send(TOPIC, userStatusUpdateDto);
+    }
+}

src/backend/state-server/src/main/java/com/example/state/service/StateManager.java

@@ -1,18 +1,26 @@
 package com.example.state.service;
 
+import com.example.state.dto.UserStatusUpdateDto;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.springframework.stereotype.Service;
 
+import java.util.List;
+
 @Service
 public class StateManager {
 
     private final RedisService redisService;
     private final ObjectMapper objectMapper;
+    private final FriendService friendService;
+    private final KafkaProducerService kafkaProducerService;
 
-    public StateManager(RedisService redisService, ObjectMapper objectMapper) {
+    public StateManager(RedisService redisService, ObjectMapper objectMapper,
+                        FriendService friendService, KafkaProducerService kafkaProducerService) {
         this.redisService = redisService;
         this.objectMapper = objectMapper;
+        this.friendService = friendService;
+        this.kafkaProducerService = kafkaProducerService;
     }
 
     // 메시지를 수신하고 처리하는 간단한 로직
@@ -29,10 +37,18 @@ public void consumeMessage(String message) {
 
             if ("JOIN".equals(eventType)) {
                 //online 상태 저장
-                redisService.saveUserState(userId, "online", serverPort, timestamp);
+                redisService.saveUserState(userId, "online", serverPort, String.valueOf(timestamp));
+
+                List<String> friends = friendService.getFriends(userId);
+                UserStatusUpdateDto statusUpdateDto = new UserStatusUpdateDto(userId, "online", friends, timestamp);
+                kafkaProducerService.sendUserStatus(statusUpdateDto);
             } else {
                 //유저 상태 제거 ("LEAVE")
                 redisService.deleteUserState(userId);
+
+                List<String> friends = friendService.getFriends(userId);
+                UserStatusUpdateDto statusUpdateDto = new UserStatusUpdateDto(userId, "offline", friends, timestamp);
+                kafkaProducerService.sendUserStatus(statusUpdateDto);
             }
         } catch (Exception e) {
             e.printStackTrace();

src/backend/state-server/src/main/resources/application-dev.properties

@@ -3,4 +3,6 @@ spring.kafka.bootstrap-servers=localhost:19092
 spring.data.redis.database=1
 spring.data.redis.host=localhost
 spring.data.redis.port=7003
-spring.data.redis.timeout=60000
+spring.data.redis.timeout=60000
+
+spring.datasource.url=jdbc:mysql://localhost:7001/kickzo

src/backend/state-server/src/main/resources/application-docker.properties

@@ -3,4 +3,6 @@ spring.kafka.bootstrap-servers=kafka:9092
 spring.data.redis.database=1
 spring.data.redis.host=redis
 spring.data.redis.port=6379
-spring.data.redis.timeout=60000
+spring.data.redis.timeout=60000
+
+spring.datasource.url=jdbc:mysql://mysql:3306/kickzo