RabbitMQ + Spring boot로 Flutter용 채팅 만들기 1

먼저 도커로 rabbitMQ를 실행하여 준다

docker run -d --name rabbitmq -p 61613:61613 -p 5672:5672 -p 15672:15672 rabbitmq:3-management

 

 

다른 블로그에서는 5672, 15672만 여는데 stomp를 사용할 예정이라면 61613까지 열어주어야한다!

5672 : rabbitMQ AMQP 기본포트

15672 : 관리자 화면 포트 - localhost:15672로 웹클라이언트에서 접속 가능

61613 : stomp 플러그인 기본 포트 

 

그 뒤 stomp 플러그인을 활성화 시켜준다. 

docker desktop으로 이동하여 해당 rabbitMQ컨테이너에서 terminal로 들어가 

 

rabbitmq-plugins enable rabbitmq_web_stomp
rabbitmq-plugins enable rabbitmq_web_stomp_examples

을 입력해서 stomp플러그인 을 활성화 시켜준다. 

 

spring boot를 생성하고 

필요한 dependecies로는 

	implementation 'org.springframework.boot:spring-boot-starter-amqp'
	testImplementation 'org.springframework.amqp:spring-rabbit-test'
	implementation 'org.springframework.boot:spring-boot-starter-reactor-netty'
	implementation 'com.fasterxml.jackson.core:jackson-databind'
	implementation 'com.fasterxml.jackson.core:jackson-databind'

 

가 있으나 사용하면서 추가될 수 있다. (lombok) 추가

application.yml파일을 만들어준다.

server:
	port: 9081
spring:
	rabbitmq:
		port: 5672
		host: localhost
		username: guest
		password: guest

rabbitmq 기본 amqp수행 포트인 5672로 설정해주고 아이디와 비밀번호는 따로 만들어둔것이 없다면 guest로 통일하면 된다.

 

rabbitConfig를 만들어 준다.

import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.util.AntPathMatcher;
import org.springframework.web.socket.config.annotation.*;

@Configuration
@EnableWebSocket
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
@Override
public void configureMessageBroker(MessageBrokerRegistry registry)
{
//registry.enableSimpleBroker( "/topic");
registry.setPathMatcher(new AntPathMatcher("."));
registry.setApplicationDestinationPrefixes("/app")
.enableStompBrokerRelay("/topic")
.setRelayHost("3.34.216.149")
.setRelayPort(61613)
.setVirtualHost("/")
.setClientLogin("guest")
.setClientPasscode("guest");

}

@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
registry.addEndpoint("/chat").setAllowedOrigins("*");
}




}

@EnableWebSocket은 생략가능하다. 

다른 블로그를 보면 enableSimpleBroker를 사용하는데 rabbitmq를 통해 stomp로 뿌려줄려면 enableStompbrokerRelay로 넣어준다. 그리고 rabbitmq의 기본 stomp 포트인 61613을 세팅해준다.

 

사용자는 ws://localhost:9081/chat으로 웹소켓 연결이 가능하고 해당 연결에서 /topic/구독할 이름 으로 구독이 가능하다.

 

import com.fasterxml.jackson.databind.Module;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@EnableRabbit
public class RabbitConfig {



private static final String CHAT_QUEUE_NAME = "sample.queue";
private static final String CHAT_EXCHANGE_NAME = "topic.exchange";
private static final String ROUTING_KEY = "room.*";

@Bean
public Queue queue(){ return new Queue(CHAT_QUEUE_NAME, true); }

@Bean
public TopicExchange chatExchange() {
return new TopicExchange(CHAT_EXCHANGE_NAME);
}

@Bean
public Binding chatBinding(Queue chatQueue, TopicExchange chatExchange) {
return BindingBuilder.bind(chatQueue).to(chatExchange).with(ROUTING_KEY);
}

@Bean
public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
return new RabbitAdmin(connectionFactory);
}
@Bean
public Jackson2JsonMessageConverter jsonMessageConverter(){
//LocalDateTime serializable을 위해
ObjectMapper objectMapper = new ObjectMapper();
objectMapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, true);
objectMapper.registerModule(dateTimeModule());

Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter(objectMapper);

return converter;
}

@Bean
public Module dateTimeModule() {
return new JavaTimeModule();
}

@Bean
public RabbitTemplate rabbitTemplate(){
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
rabbitTemplate.setMessageConverter(jsonMessageConverter());
rabbitTemplate.setRoutingKey(CHAT_QUEUE_NAME);
return rabbitTemplate;
}
@Bean
public ConnectionFactory connectionFactory(){
CachingConnectionFactory factory = new CachingConnectionFactory();
factory.setHost("localhost");
factory.setUsername("guest");
factory.setPassword("guest");
return factory;
}


@Bean
public SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
final SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
factory.setMessageConverter(jsonMessageConverter());
return factory;
}

}

jackson을 통하여 객체를 json으로 만들수 있도록 한다.

 

rabbitAdmin은 생략하여도 된다. 넣었던 이유는 이 스프링 서버로 메시지를 받으면 rabbitmq exchange로 보내면서 해당 큐를 만들어줘야 한다고 생각 하여 admin을 통해 생성 한뒤 보내려 했는데, 알고보니 stomp플러그인을 사용하면 자동으로 만들어준다.  

import com.capstone.chatting.ChatWebSocketHandler;
import com.capstone.chatting.Service.MessageSenderService;
import lombok.RequiredArgsConstructor;
import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.messaging.handler.annotation.DestinationVariable;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.util.Map;

@Controller
@RequiredArgsConstructor
@Log4j2
public class ChatController {

    private final RabbitTemplate template;

    private final Map<String, WebSocketSession> sessions ;
    private final static String CHAT_EXCHANGE_NAME = "amq.topic";
    private final static String CHAT_QUEUE_NAME = "sample.queue";




    private final MessageSenderService messageSenderService;


    @MessageMapping("chat.enter.{chatRoomId}")
    public void sendMessage(String chatMessage, @DestinationVariable String chatRoomId){
        System.out.println("send message : " + chatMessage);

        template.convertAndSend(CHAT_EXCHANGE_NAME, "room." + chatRoomId,chatMessage);

    }

이제 메시지를 받고 처리해 본다.  
@MessageMapping으로 사용자가 /app/chat.enter.2와 같은 예시로 메시지를 보내게 되면 @Destinationvariable로 2값이  chatRoomId로 들어가게 되고 해당 destination으로 메시지를 보내게되면 template.convertAndSend로 RabbitTemplate가 해당 스트링을 json형태로 바꾸어서 amq.topic이라는 exchange로 보내게 된다. amq.topic exchange는 topic성격이어서  받은 키값으로 바인딩하는 큐를 stomp-xxxx형태로  큐를생성하고 큐에 메시지를 넣는다. 그러면 해당 큐에서 키값을 구독하고 있는, 즉, /topic/room.2를 구독하고 있는 웹소캣에 연결된 클라이언트들에게 전송된다.  

 

flutter에서 받는것은 다음 포스팅에서