[#239] RabbitMQ 적용

.docker/Dockerfile

@@ -17,6 +17,9 @@ REDIS_HOST=${REDIS_HOST} \
 REDIS_PORT=${REDIS_PORT} \
 ACTIVE_PROFILE=${ACTIVE_PROFILE} \
 SENTRY_DSN=${SENTRY_DSN} \
-SLACK_WEBHOOK=${SLACK_WEBHOOK}
+RABBITMQ_USERNAME=${RABBITMQ_USERNAME} \
+RABBITMQ_PASSWORD=${RABBITMQ_PASSWORD} \
+RABBITMQ_HOST=${RABBITMQ_HOST} \
+RABBITMQ_PORT=${RABBITMQ_PORT}
 COPY ${JAR_FILE} app.jar
 ENTRYPOINT ["java", "-Dspring.profiles.active=${ACTIVE_PROFILE}", "-jar", "/app.jar"]

.docker/docker-compose.yml

@@ -22,6 +22,21 @@ services:
     volumes:
       - ./data/redis:/data
 
+  rabbitmq:
+    image: pcloud/rabbitmq-stomp
+    container_name: kkini-rabbitmq
+    volumes:
+      - ./rabbitmq/etc/:/etc/rabbitmq/
+      - ./rabbitmq/data/:/var/lib/rabbitmq/
+      - ./rabbitmq/logs/:/var/log/rabbitmq/
+    ports:
+      - "5672:5672"
+      - "15672:15672"
+      - "61613:61613"
+    environment:
+      RABBITMQ_DEFAULT_USER: ${RABBITMQ_USERNAME}
+      RABBITMQ_DEFAULT_PASS: ${RABBITMQ_PASSWORD}
+
   green:
     container_name: green
     image: kiseo/kkini

build.gradle

@@ -101,18 +101,26 @@ dependencies {
     //https://mvnrepository.com/artifact/org.locationtech.jts/jts-core
     implementation 'org.locationtech.jts:jts-core:1.19.0'
     // logback - cloudwatch
-    implementation group: 'ca.pjer', name: 'logback-awslogs-appender', version: '1.6.0'
+    implementation 'ca.pjer:logback-awslogs-appender:1.6.0'
     // AWS
     implementation 'org.springframework.cloud:spring-cloud-starter-aws:2.2.6.RELEASE'
     // Slack Notification
-    implementation "net.gpedro.integrations.slack:slack-webhook:1.4.0"
+    implementation 'net.gpedro.integrations.slack:slack-webhook:1.4.0'
     // Websocket
     implementation 'org.springframework.boot:spring-boot-starter-websocket'
+    // RabbitMQ
+    implementation 'org.springframework.boot:spring-boot-starter-amqp'
+    testImplementation 'org.springframework.amqp:spring-rabbit-test'
+    // 외부 브로커를 사용하기 위해
+    implementation 'org.springframework.boot:spring-boot-starter-reactor-netty:2.4.6'
+    //jackson2json에서 LocalDateTime을 handling 하기 위해
+    implementation 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.12.4'
+
     // Sentry
     implementation 'io.sentry:sentry-spring-boot-starter:6.17.0'
     implementation 'io.sentry:sentry-logback:6.17.0'
     // WireMock
-    implementation "org.springframework.cloud:spring-cloud-starter-contract-stub-runner"
+    implementation 'org.springframework.cloud:spring-cloud-starter-contract-stub-runner'
     // Embedded Redis
     testImplementation 'it.ozimov:embedded-redis:0.7.2'
     // Cache

src/main/java/com/prgrms/mukvengers/MukvengersApplication.java

@@ -16,7 +16,7 @@ public static void main(String[] args) {
 	}
 
 	@PostConstruct
-	void started() {
+	void setTimeAndLocale() {
 		TimeZone.setDefault(TimeZone.getTimeZone("Asia/Seoul"));
 		Locale.setDefault(Locale.KOREA);
 	}

src/main/java/com/prgrms/mukvengers/domain/chat/api/ChatController.java

@@ -5,14 +5,15 @@
 import java.util.List;
 import java.util.Map;
 
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.web.PageableDefault;
 import org.springframework.http.ResponseEntity;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.handler.annotation.DestinationVariable;
 import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.messaging.handler.annotation.MessageMapping;
 import org.springframework.messaging.handler.annotation.Payload;
-import org.springframework.messaging.handler.annotation.SendTo;
 import org.springframework.stereotype.Controller;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PathVariable;
@@ -34,6 +35,7 @@
 public class ChatController {
 
 	private final ChatService chatService;
+	private final RabbitTemplate template;
 
 	@ResponseBody
 	@GetMapping(value = "/api/v2/crews/{crewId}/chats", produces = APPLICATION_JSON_VALUE)
@@ -58,12 +60,24 @@ public ResponseEntity<ApiResponse<ChatResponses>> getChattingList(
 		return ResponseEntity.ok(new ApiResponse<>(chatResponses));
 	}
 
-	@MessageMapping("/chat.sendMessage/{crewId}")
-	@SendTo("/topic/public/{crewId}")
-	public ChatResponse sendMessage(@DestinationVariable Long crewId,
+	// @MessageMapping("/chat.sendMsg/{crewId}")
+	// @SendTo("/topic/public/{crewId}")
+	// public ChatResponse sendMessage(@DestinationVariable Long crewId,
+	// 	@Header("simpSessionAttributes") Map<String, Object> simpSessionAttributes,
+	// 	@Payload ChatRequest chatRequest
+	// ) {
+	// 	return chatService.save(chatRequest, crewId, simpSessionAttributes);
+	// }
+
+	@MessageMapping("/chat.sendMsg/{crewId}")
+	public void sendMessageByRabbitMQ(@DestinationVariable Long crewId,
+		MessageHeaders headers,
 		@Header("simpSessionAttributes") Map<String, Object> simpSessionAttributes,
 		@Payload ChatRequest chatRequest
 	) {
-		return chatService.save(chatRequest, crewId, simpSessionAttributes);
+		log.info("sendMessageByRabbitMQ");
+		log.error("headers : {}", headers);
+		ChatResponse chatResponse = chatService.save(chatRequest, crewId, simpSessionAttributes);
+		template.convertAndSend("chat.exchange", "crews." + crewId, chatResponse);
 	}
 }

src/main/java/com/prgrms/mukvengers/domain/chat/handler/StompHandler.java

@@ -27,7 +27,7 @@
 @RequiredArgsConstructor
 public class StompHandler implements ChannelInterceptor {
 
-	public static final String DEFAULT_PATH = "/topic/public/";
+	public static final String DEFAULT_PATH = "exchange/chat.exchange/crews.";
 
 	private final JwtTokenProvider jwtTokenProvider;
 	private final UserRepository userRepository;
@@ -41,26 +41,36 @@ public Message<?> preSend(Message<?> message, MessageChannel channel) {
 		StompCommand command = accessor.getCommand();
 
 		if (StompCommand.CONNECT.equals(command)) { // websocket 연결요청 -> JWT 인증
+			log.error("CONNECT");
 
+			log.error("accessor : {}", accessor);
 			// JWT 인증
-			User user = getUserByAuthorizationHeader(
-				accessor.getFirstNativeHeader("Authorization"));
+			// User user = getUserByAuthorizationHeader(
+			// 	accessor.getFirstNativeHeader("Authorization"));
+
+			User user = userRepository.findById(92L).get();
+
+			log.error("CONNECTED userId : {}", user.getId());
+
 			// 인증 후 데이터를 헤더에 추가
 			setValue(accessor, "userId", user.getId());
 			setValue(accessor, "username", user.getNickname());
 			setValue(accessor, "profileImgUrl", user.getProfileImgUrl());
 
+			log.error("CONNECTED userId : {}", user.getId());
+
 		} else if (StompCommand.SUBSCRIBE.equals(command)) { // 채팅룸 구독요청(진입) -> CrewMember인지 검증
+			log.error("SUBSCRIBE");
 
 			Long userId = (Long)getValue(accessor, "userId");
 			Long crewId = parseCrewIdFromPath(accessor);
-			log.debug("userId : " + userId + "crewId : " + crewId);
+			log.error("userId : " + userId + " crewId : " + crewId);
 			setValue(accessor, "crewId", crewId);
-			validateUserInCrew(userId, crewId);
+			// validateUserInCrew(userId, crewId);
 
 		} else if (StompCommand.DISCONNECT == command) { // Websocket 연결 종료
-			Long userId = (Long)getValue(accessor, "userId");
-			log.info("DISCONNECTED userId : {}", userId);
+			// Long userId = (Long)getValue(accessor, "userId");
+			// log.info("DISCONNECTED userId : {}", userId);
 		}
 
 		log.info("header : " + message.getHeaders());
@@ -100,7 +110,7 @@ private Long parseCrewIdFromPath(StompHeaderAccessor accessor) {
 	private void validateUserInCrew(Long userId, Long crewId) {
 		crewMemberRepository.findCrewMemberByCrewIdAndUserId(crewId, userId)
 			.orElseThrow(() -> new WebSocketException(
-				String.format("crew Id : {} userId : {} 로 조회된 결과가 없습니다.", crewId, userId)));
+				String.format("crew Id : {0} userId : {0} 로 조회된 결과가 없습니다.", crewId, userId)));
 	}
 
 	private Object getValue(StompHeaderAccessor accessor, String key) {

src/main/java/com/prgrms/mukvengers/domain/chat/handler/WebSocketEventListener.java

@@ -1,24 +1,17 @@
 package com.prgrms.mukvengers.domain.chat.handler;
 
-import java.util.Map;
-import java.util.Objects;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.context.event.EventListener;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.messaging.simp.SimpMessageSendingOperations;
-import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
 import org.springframework.stereotype.Component;
-import org.springframework.web.socket.messaging.SessionConnectedEvent;
-import org.springframework.web.socket.messaging.SessionDisconnectEvent;
-import org.springframework.web.socket.messaging.SessionSubscribeEvent;
 
-import com.prgrms.mukvengers.domain.chat.dto.request.ChatRequest;
-import com.prgrms.mukvengers.domain.chat.dto.request.MessageType;
-import com.prgrms.mukvengers.domain.chat.exception.WebSocketException;
+import com.prgrms.mukvengers.domain.chat.dto.response.ChatResponse;
 
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 
+@Slf4j
 @Component
 @RequiredArgsConstructor
 public class WebSocketEventListener {
@@ -27,63 +20,69 @@ public class WebSocketEventListener {
 
 	private final SimpMessageSendingOperations messagingTemplate;
 
-	// 연결 요청
-	@EventListener
-	public void handleWebSocketConnectListener(SessionConnectedEvent event) {
-		logger.info("Received a new web socket connection");
-	}
-
-	// 구독 요청(입장)
-	@EventListener
-	public void handleWebSocketSubscribeListener(SessionSubscribeEvent event) {
-		logger.info("Received a new web socket subscribe");
-		StompHeaderAccessor accessor = StompHeaderAccessor.wrap(event.getMessage());
-
-		String username = (String)getValue(accessor, "username");
-		Long userId = (Long)getValue(accessor, "userId");
-		Long crewId = (Long)getValue(accessor, "crewId");
-
-		logger.info("User: {} {} Subscribe Crew : {}", userId, username, crewId);
-		ChatRequest chatRequest = new ChatRequest(MessageType.JOIN, userId,
-			username + " 님이 입장했습니다.");
-		messagingTemplate.convertAndSend("/topic/public/" + crewId, chatRequest);
-
-	}
-
-	// 연결 해제
-	@EventListener
-	public void handleWebSocketDisconnectListener(SessionDisconnectEvent event) {
-		StompHeaderAccessor accessor = StompHeaderAccessor.wrap(event.getMessage());
-
-		String username = (String)getValue(accessor, "username");
-		Long userId = (Long)getValue(accessor, "userId");
-		Long crewId = (Long)getValue(accessor, "crewId");
-
-		logger.info("User: {} {} Disconnected Crew : {}", userId, username, crewId);
-
-		ChatRequest chatRequest = new ChatRequest(
-			MessageType.LEAVE, userId, username + " 님이 떠났습니다.");
-
-		messagingTemplate.convertAndSend("/topic/public/" + crewId, chatRequest);
-	}
-
-	private Object getValue(StompHeaderAccessor accessor, String key) {
-		Map<String, Object> sessionAttributes = getSessionAttributes(accessor);
-		Object value = sessionAttributes.get(key);
-
-		if (Objects.isNull(value)) {
-			throw new WebSocketException(key + " 에 해당하는 값이 없습니다.");
-		}
-
-		return value;
+	@RabbitListener(queues = "chat.queue")
+	public void receive(ChatResponse chat) {
+		log.error("received : " + chat.content());
+		System.out.println("received : " + chat.content());
 	}
 
-	private Map<String, Object> getSessionAttributes(StompHeaderAccessor accessor) {
-		Map<String, Object> sessionAttributes = accessor.getSessionAttributes();
-
-		if (Objects.isNull(sessionAttributes)) {
-			throw new WebSocketException("SessionAttributes가 null입니다.");
-		}
-		return sessionAttributes;
-	}
+	// 연결 요청
+	// @EventListener
+	// public void handleWebSocketConnectListener(SessionConnectedEvent event) {
+	// 	logger.info("Received a new web socket connection");
+	// }
+	//
+	// // 구독 요청(입장)
+	// @EventListener
+	// public void handleWebSocketSubscribeListener(SessionSubscribeEvent event) {
+	// 	logger.info("Received a new web socket subscribe");
+	// 	StompHeaderAccessor accessor = StompHeaderAccessor.wrap(event.getMessage());
+	//
+	// 	String username = (String)getValue(accessor, "username");
+	// 	Long userId = (Long)getValue(accessor, "userId");
+	// 	Long crewId = (Long)getValue(accessor, "crewId");
+	//
+	// 	logger.info("User: {} {} Subscribe Crew : {}", userId, username, crewId);
+	// 	ChatRequest chatRequest = new ChatRequest(MessageType.JOIN, userId,
+	// 		username + " 님이 입장했습니다.");
+	// 	messagingTemplate.convertAndSend("/topic/public/" + crewId, chatRequest);
+	//
+	// }
+	//
+	// // 연결 해제
+	// @EventListener
+	// public void handleWebSocketDisconnectListener(SessionDisconnectEvent event) {
+	// 	StompHeaderAccessor accessor = StompHeaderAccessor.wrap(event.getMessage());
+	//
+	// 	String username = (String)getValue(accessor, "username");
+	// 	Long userId = (Long)getValue(accessor, "userId");
+	// 	Long crewId = (Long)getValue(accessor, "crewId");
+	//
+	// 	logger.info("User: {} {} Disconnected Crew : {}", userId, username, crewId);
+	//
+	// 	ChatRequest chatRequest = new ChatRequest(
+	// 		MessageType.LEAVE, userId, username + " 님이 떠났습니다.");
+	//
+	// 	messagingTemplate.convertAndSend("/topic/public/" + crewId, chatRequest);
+	// }
+	//
+	// private Object getValue(StompHeaderAccessor accessor, String key) {
+	// 	Map<String, Object> sessionAttributes = getSessionAttributes(accessor);
+	// 	Object value = sessionAttributes.get(key);
+	//
+	// 	if (Objects.isNull(value)) {
+	// 		throw new WebSocketException(key + " 에 해당하는 값이 없습니다.");
+	// 	}
+	//
+	// 	return value;
+	// }
+	//
+	// private Map<String, Object> getSessionAttributes(StompHeaderAccessor accessor) {
+	// 	Map<String, Object> sessionAttributes = accessor.getSessionAttributes();
+	//
+	// 	if (Objects.isNull(sessionAttributes)) {
+	// 		throw new WebSocketException("SessionAttributes가 null입니다.");
+	// 	}
+	// 	return sessionAttributes;
+	// }
 }