[LIME-159] 레디스 메시지 브로커 설정

lime-api/src/main/java/com/programmers/lime/domains/chat/application/ChatService.java

@@ -5,7 +5,7 @@
 import java.time.ZoneId;
 import java.util.List;
 
-import org.springframework.messaging.simp.SimpMessagingTemplate;
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.stereotype.Service;
 
 import com.programmers.lime.domains.chat.application.dto.response.ChatGetServiceResponse;
@@ -19,6 +19,7 @@
 import com.programmers.lime.error.BusinessException;
 import com.programmers.lime.error.ErrorCode;
 import com.programmers.lime.global.config.security.SecurityUtils;
+import com.programmers.lime.global.event.chat.ChatSendMessageEvent;
 import com.programmers.lime.redis.chat.ChatSessionRedisManager;
 import com.programmers.lime.redis.chat.model.ChatSessionInfo;
 
@@ -28,8 +29,6 @@
 @RequiredArgsConstructor
 public class ChatService {
 
-	private final SimpMessagingTemplate simpMessagingTemplate;
-
 	private final MemberReader memberReader;
 
 	private final ChatSessionRedisManager chatSessionRedisManager;
@@ -40,6 +39,8 @@ public class ChatService {
 
 	private final ChatRoomMemberReader chatRoomMemberReader;
 
+	private final ApplicationEventPublisher applicationEventPublisher;
+
 	public void sendMessage(final Long memberId, final String sessionId, final String message, final String timeSeq) {
 		Member member = memberReader.read(memberId);
 
@@ -62,7 +63,9 @@ public void sendMessage(final Long memberId, final String sessionId, final Strin
 
 		chatAppender.appendChat(chatInfoWithMember.toChatInfo());
 
-		simpMessagingTemplate.convertAndSend("/subscribe/rooms/" + chatRoomId, chatInfoWithMember);
+		applicationEventPublisher.publishEvent(
+			new ChatSendMessageEvent("/subscribe/rooms/" + chatRoomId, chatInfoWithMember)
+		);
 	}
 
 	public void joinChatRoom(final Long chatRoomId) {
@@ -84,7 +87,9 @@ public void joinChatRoom(final Long chatRoomId) {
 
 		chatAppender.appendChat(chatInfoWithMember.toChatInfo());
 
-		simpMessagingTemplate.convertAndSend("/subscribe/rooms/join/" + chatRoomId, chatInfoWithMember);
+		applicationEventPublisher.publishEvent(
+			new ChatSendMessageEvent("/subscribe/rooms/" + chatRoomId, chatInfoWithMember)
+		);
 	}
 
 	public void sendExitMessageToChatRoom(final Long chatRoomId) {
@@ -107,7 +112,9 @@ public void sendExitMessageToChatRoom(final Long chatRoomId) {
 
 		chatAppender.appendChat(chatInfoWithMember.toChatInfo());
 
-		simpMessagingTemplate.convertAndSend("/subscribe/rooms/exit/" + chatRoomId, chatInfoWithMember);
+		applicationEventPublisher.publishEvent(
+			new ChatSendMessageEvent("/subscribe/rooms/" + chatRoomId, chatInfoWithMember)
+		);
 	}
 
 	public ChatGetServiceResponse getChatWithMemberList(final Long chatRoomId) {

lime-api/src/main/java/com/programmers/lime/global/config/chat/message/ChatMessageBrokerConfig.java

@@ -0,0 +1,32 @@
+package com.programmers.lime.global.config.chat.message;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
+
+import com.programmers.lime.redis.chat.listener.IChatListener;
+import com.programmers.lime.redis.chat.publisher.IChatPublisher;
+import com.programmers.lime.redis.chat.publisher.RedisChatPublisher;
+
+import lombok.RequiredArgsConstructor;
+
+@Configuration
+@RequiredArgsConstructor
+public class ChatMessageBrokerConfig {
+
+	private final SimpleChatListener simpleChatListener;
+
+	private final RedisChatPublisher redisChatPublisher;
+
+	@Bean
+	@Primary
+	public IChatListener configListener() {
+		return simpleChatListener;
+	}
+
+	@Bean
+	@Primary
+	public IChatPublisher configPublisher() {
+		return redisChatPublisher;
+	}
+}

lime-api/src/main/java/com/programmers/lime/global/config/chat/message/SimpleChatListener.java

@@ -0,0 +1,48 @@
+package com.programmers.lime.global.config.chat.message;
+
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+
+import org.springframework.messaging.simp.SimpMessagingTemplate;
+import org.springframework.stereotype.Component;
+
+import com.programmers.lime.domains.chat.model.ChatInfoWithMember;
+import com.programmers.lime.redis.chat.listener.IChatListener;
+import com.programmers.lime.redis.chat.model.ChatInfoWithMemberCache;
+
+import lombok.RequiredArgsConstructor;
+
+@Component
+@RequiredArgsConstructor
+public class SimpleChatListener implements IChatListener {
+
+	private final SimpMessagingTemplate simpMessagingTemplate;
+
+	@Override
+	public void sendMessage(final ChatInfoWithMemberCache chatInfoWithMemberCache) {
+
+		if (chatInfoWithMemberCache == null) {
+			throw new IllegalStateException("Deserialized message is null. Message deserialization failed.");
+		}
+
+		simpMessagingTemplate.convertAndSend(chatInfoWithMemberCache.destination(),
+			toChatInfoWithMember(chatInfoWithMemberCache));
+	}
+
+	private ChatInfoWithMember toChatInfoWithMember(final ChatInfoWithMemberCache chatInfoWithMemberCache) {
+		return ChatInfoWithMember.builder()
+			.chatId(chatInfoWithMemberCache.chatId())
+			.chatRoomId(chatInfoWithMemberCache.chatRoomId())
+			.memberId(chatInfoWithMemberCache.memberId())
+			.nickname(chatInfoWithMemberCache.nickname())
+			.profileImage(chatInfoWithMemberCache.profileImage())
+			.message(chatInfoWithMemberCache.message())
+			.sendAt(toLocalDateTime(chatInfoWithMemberCache.sendAt()))
+			.build();
+	}
+
+	private LocalDateTime toLocalDateTime(final String sendAt) {
+		DateTimeFormatter formatter = DateTimeFormatter.ISO_LOCAL_DATE_TIME;
+		return LocalDateTime.parse(sendAt, formatter);
+	}
+}

lime-api/src/main/java/com/programmers/lime/global/event/chat/ChatSendMessageEvent.java

@@ -0,0 +1,9 @@
+package com.programmers.lime.global.event.chat;
+
+import com.programmers.lime.domains.chat.model.ChatInfoWithMember;
+
+public record ChatSendMessageEvent(
+	String destination,
+	ChatInfoWithMember chatInfoWithMember
+) {
+}

lime-api/src/main/java/com/programmers/lime/global/event/chat/ChatSendMessageEventListener.java

@@ -0,0 +1,45 @@
+package com.programmers.lime.global.event.chat;
+
+import org.springframework.context.event.EventListener;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Component;
+
+import com.programmers.lime.domains.chat.model.ChatInfoWithMember;
+import com.programmers.lime.redis.chat.publisher.IChatPublisher;
+import com.programmers.lime.redis.chat.model.ChatInfoWithMemberCache;
+
+import lombok.RequiredArgsConstructor;
+
+@Component
+@RequiredArgsConstructor
+public class ChatSendMessageEventListener {
+
+	private final IChatPublisher chatPublisher;
+
+	@Async
+	@EventListener
+	public void sendMessage(final ChatSendMessageEvent event) {
+
+		ChatInfoWithMemberCache chatInfoWithMemberCache = chatInfoWithMember(
+			event.chatInfoWithMember(),
+			event.destination()
+		);
+
+		chatPublisher.sendMessage("sub-chat", chatInfoWithMemberCache);
+	}
+
+	public ChatInfoWithMemberCache chatInfoWithMember(final ChatInfoWithMember chatInfoWithMember,
+		final String destination) {
+		return ChatInfoWithMemberCache.builder()
+			.chatId(chatInfoWithMember.chatId())
+			.chatRoomId(chatInfoWithMember.chatRoomId())
+			.memberId(chatInfoWithMember.memberId())
+			.nickname(chatInfoWithMember.nickname())
+			.profileImage(chatInfoWithMember.profileImage())
+			.message(chatInfoWithMember.message())
+			.sendAt(chatInfoWithMember.sendAt().toString())
+			.chatType(chatInfoWithMember.chatType().name())
+			.destination(destination)
+			.build();
+	}
+}

lime-infrastructure/src/main/java/com/programmers/lime/redis/chat/ChatRedisMessageBrokerConfig.java

@@ -0,0 +1,36 @@
+package com.programmers.lime.redis.chat;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.listener.ChannelTopic;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
+import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
+
+import com.programmers.lime.redis.chat.listener.ChatMessageListenerImpl;
+
+@Configuration
+public class ChatRedisMessageBrokerConfig {
+
+	@Bean
+	public RedisMessageListenerContainer redisMessageListenerContainer(
+		final RedisConnectionFactory connectionFactory,
+		final MessageListenerAdapter listenerAdapter,
+		final ChannelTopic channelTopic
+	) {
+		RedisMessageListenerContainer container = new RedisMessageListenerContainer();
+		container.setConnectionFactory(connectionFactory);
+		container.addMessageListener(listenerAdapter, channelTopic);
+		return container;
+	}
+
+	@Bean
+	public MessageListenerAdapter listenerAdapter(final ChatMessageListenerImpl listener) {
+		return new MessageListenerAdapter(listener, "onMessage");
+	}
+
+	@Bean
+	public ChannelTopic channelTopic() {
+		return new ChannelTopic("sub-chat");
+	}
+}

lime-infrastructure/src/main/java/com/programmers/lime/redis/chat/listener/ChatMessageListenerImpl.java

@@ -0,0 +1,36 @@
+package com.programmers.lime.redis.chat.listener;
+
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Component;
+
+import com.programmers.lime.redis.chat.model.ChatInfoWithMemberCache;
+
+import lombok.RequiredArgsConstructor;
+
+@Component
+@RequiredArgsConstructor
+public class ChatMessageListenerImpl implements MessageListener {
+
+	private final RedisTemplate<String, Object> redisTemplate;
+
+	private final IChatListener chatListener;
+
+	@Override
+	public void onMessage(final Message message, final byte[] pattern) {
+
+		try {
+			ChatInfoWithMemberCache chatInfoWithMemberCache = (ChatInfoWithMemberCache)redisTemplate.getValueSerializer()
+				.deserialize(message.getBody());
+
+			if (chatInfoWithMemberCache == null) {
+				throw new IllegalStateException("Deserialized message is null. Message deserialization failed.");
+			}
+
+			chatListener.sendMessage(chatInfoWithMemberCache);
+		} catch (Exception e) {
+			throw new RuntimeException("Failed to process redis message", e);
+		}
+	}
+}

lime-infrastructure/src/main/java/com/programmers/lime/redis/chat/listener/IChatListener.java

@@ -0,0 +1,7 @@
+package com.programmers.lime.redis.chat.listener;
+
+import com.programmers.lime.redis.chat.model.ChatInfoWithMemberCache;
+
+public interface IChatListener {
+	void sendMessage(final ChatInfoWithMemberCache chatInfoWithMemberCache);
+}

lime-infrastructure/src/main/java/com/programmers/lime/redis/chat/model/ChatInfoWithMemberCache.java

@@ -0,0 +1,17 @@
+package com.programmers.lime.redis.chat.model;
+
+import lombok.Builder;
+
+@Builder
+public record ChatInfoWithMemberCache(
+	Long chatId,
+	Long chatRoomId,
+	Long memberId,
+	String nickname,
+	String profileImage,
+	String message,
+	String sendAt,
+	String chatType,
+	String destination
+) {
+}

lime-infrastructure/src/main/java/com/programmers/lime/redis/chat/publisher/IChatPublisher.java

@@ -0,0 +1,7 @@
+package com.programmers.lime.redis.chat.publisher;
+
+import com.programmers.lime.redis.chat.model.ChatInfoWithMemberCache;
+
+public interface IChatPublisher {
+	void sendMessage(String channel, ChatInfoWithMemberCache chatInfoWithMemberCache);
+}

lime-infrastructure/src/main/java/com/programmers/lime/redis/chat/publisher/RedisChatPublisher.java

@@ -0,0 +1,20 @@
+package com.programmers.lime.redis.chat.publisher;
+
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Component;
+
+import com.programmers.lime.redis.chat.model.ChatInfoWithMemberCache;
+
+import lombok.RequiredArgsConstructor;
+
+@Component
+@RequiredArgsConstructor
+public class RedisChatPublisher implements IChatPublisher {
+
+	private final RedisTemplate<String, Object> redisTemplate;
+
+	@Override
+	public void sendMessage(final String channel, final ChatInfoWithMemberCache chatInfoWithMemberCache) {
+		redisTemplate.convertAndSend(channel, chatInfoWithMemberCache);
+	}
+}