Spring 웹 소켓을 이용한 실시간 채팅 시스템 구현 공부

이번 포스팅은 실제로 채팅 서버 구축에 들어가기 이전에, 프론트, 백엔드 각자 공부를 진행하고 아이디어를 공유하기 위해 공부 및 혼자서 구현을 해본 과정을 적어보고자 합니다. 이후 회의 등에서 더 좋은 아이디어가 있을 경우 추후 포스팅으로 실제 구현기를 작성할 예정입니다

 

들어가기에 앞서

마트/배달 및 커뮤니티 기능과 유저 기능이 존재함에 따라서 각 게시글 별로 별도의 채팅창을 생성하여 유저들끼리 소통을 할 수 있는 창구를만들고자 채팅 시스템을 구현하게 되었다.

채팅 시스템은 모두가 사용하는 카카오톡 등과 같이 실시간 기능이 중요하기 때문에 이를 어떻게 처리하는지가 가장 중요한 부분이었다.

 

먼저 우리가 생각한 채팅 기능은 다음 기능을 하게 된다

• 게시글이 생성되고 요청이 들어온다면 게시글 별 채팅방이 생성된다.(게시글 생성 시점에서 채팅방을 만들 경우 불필요하게 많은 채팅방이 생성될 것 같음)
• 게시글 작성자는 방장의 권한을 가지며 유저를 강퇴할 수 있다.
• 강퇴된 유저는 Blacklist로 별도로 관리되고, Blacklist된 유저는 채팅방을 구독 시 강퇴되기 이전의 메시지만 확인이 가능하다
• 카카오톡과 같이 실시간으로 메시지 옆에 안읽은 유저 수를 표시해준다

와 같다. 그 이외로 세부적인 사항은 지금 결정하기보다는 각자 공부를 하고 가장 효율적이라고 생각되는 부분을 적용하기로 하였다.

 

WebSocket과 STOMP

검색창에 "실시간 채팅 기능 구현" 등과 같은 검색어를 입력하면 영혼의 동반자와 같이 따라오는 두 단어가 있다. 바로 WebSocket과 STOMP이다. 일단 이 친구들에 대해서 정확히 알지 못하기 때문에 공부를 하였다.

 

채팅 시스템은, 단순히 서버에서 클라이언트로 일방적으로 데이터를 보내는 것이 아닌, 클라이언트와 서버가 데이터를 양방향으로 주고받아야 한다. 즉, SSE의 경우에는 최초 연결 이후 서버 -> 클라이언트 단방향 통신이기 때문에 채팅 시스템에 어울리지 않는다.

 

기본적으로 우리는 HTTP 통신 프로토콜을 사용해서 클라이언트가 요청을 하면 서버가 응답을 하게 된다.

이 경우에는 사용자는 새로운 메시지를 확인하기 위해서 계속해서 "메시지확인" 등과 같이 별도의 요청을 계속해서 보내주어야 하는데 이렇게 되면 결국 채팅이라는 느낌을 가질 수 없다. 이러한 문제를 해결하기 위해서 다음과 같은 방법들이 등장하게 되었다.

 

Polling

Polling 방법은, 사용자의 별도 조작없이 클라이언트 서버 자체에서 일정 시간마다 주기적으로 서버에 요청을 보내는 것이다.

즉 사용자는 채팅 화면에 가만히 있더라도, 클라이언트측에서 계속해서 요청을 보내고 확인하기 때문에 새로운 메시지가 존재할 시 응답을 받고 새로운 메시지를 확인할 수 있다. 하지만, 실시간과 같은 느낌을 가지기 위해서는 polling 하는 주기가 짧아져야 하며, 주기가 짧을 경우 과하게 많은 요청이 가서 트래픽 많이 발생할 수 있는 단점이 있다. 반대로 polling하는 주기가 길다면 사용자는 실시간과 같은 느낌을 받지 못할 수 있다.

 

Long Polling

Polling 방법에서 조금 더 발전한 방법이다. 먼저 클라이언트는 서버 측으로 요청을 보낸다. 그 후, 서버는 새로운 데이터가 발생할 시에만 응답을 전송하게 되며, 응답을 받은 클라이언트는 다시 바로 요청을 보내게 된다. 이 방법은 새로운 데이터가 존재하지 않을 경우에는 응답을 하지 않아 네트워크 트래픽의 부담을 줄일 수 있다.

어떻게 보면 합리적인 방법일 수 있지만, 결국 채팅 시스템과 같이 새로운 데이터가 발생하는 주기가 짧다면 polling 방식과 같이 과하게 많은 트래픽이 많이 발생하게 된다.

 

HTTP 프로토콜에서의 실시간성은 결국 요청과 응답쌍이 이루어지며 많은 비용이 발생할 수 있는 단점이 존재하기 때문에, 이러한 단점을 극복하고자 웹 소켓 프로토콜이 등장하게 된다.

 

WebSocket

위에서 설명한 HTTP 통신 프로토콜의 단점을 극복하기 위해 양방향 통신 프로토콜인 WebSocket 프로토콜이 등장하게 된다.

WebSocket은 기본적으로 초기 연결시에는 HTTP 통신 프로토콜을 통해 핸드쉐이크를 진행하고, 연결이 된 후에는 TCP 연결을 통해 양방향으로 지속적으로 유지된다. 또한, HTTP 통신은 헤더의 데이터가 크기 때문에 요청이 잦을 수록 헤더에 대한 비용이 많이 발생하게 되는데 WebSocket은 이러한 부분도 해결을 할 수 있다.

HTTP 통신 프로토콜이 HTTP(80포트)/HTTPS(8080포트) 를 사용하듯이 WebSocket 또한 ws(80)/wss(8080)포트를 사용하여 통신을 하게 된다.

즉, 초기 핸드쉐이크 시에는 (http://~~:80)을 이용하고, 연결이 된 후에는 (ws://~~:80)을 이용하여 통신을 하는 것이다.

TCP Socket vs WebSocket

TCP 소켓은 OSI 7 계층 중 전송 계층(4계층)에서 동작하지만 웹 소켓의 경우에는 애플리케이션 계층(7계층)에서 동작을 한다.
애플리케이션 계층의 경우 전송 계층의 상위 계층이기 때문에 TCP를 기반으로 동작하며 웹 소켓이 이를 이용하게 된다.
즉 웹 소켓의 동작에서 애플리케이션 계층에서 메시지의 포맷, 전송 방식, 핸드쉐이크 절차 등을 정의 한 후 실질적인 데이터 전송은 전송 계층에서 진행하게 된다.

둘은 완전 다른 개념이라기보다는 웹 소켓은 TCP를 기반으로 동작하며, TCP 소켓과 달리 웹 환경에 특화된 애플리케이션 계층 프로토콜이다라고 보는 것이 맞다.

웹 소켓은 기존 TCP 소켓과 달리 HTTP 요청을 통해서 Handshake가 이루어진다. 따라서 HTTP 통신과 같은 포트를 사용해 양방향 통신이 가능하며 SSL/TLS 등의 보안적인 기능을 가져갈 수 있다.

 

WebSocket 초기 핸드쉐이크

GET http://localhost:8080/chat/654/bzadodjw/websocket HTTP/1.1
Host: localhost:8080
Connection: Upgrade
Pragma: no-cache
Cache-Control: no-cache
User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36
Upgrade: websocket
Origin: http://localhost:3000
Sec-WebSocket-Version: 13
Accept-Encoding: gzip, deflate, br, zstd
Accept-Language: ko,en-US;q=0.9,en;q=0.8
Sec-WebSocket-Key: vdMUyzWIfw4fWe6MO63UzA==
Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bits

초기 웹소켓 연결 수립을 위한 핸드쉐이크 시에는 이러한 형태로 http://~~ 로 HTTP 요청을 보내게 된다.

Sec-Websocket-Key는 클라이언트에서 무작위로 생성한 16바이트 Base64인코딩 키라고 한다. 이를 통해 클라이언트와 서버가 유효한 핸드쉐이크가 이루어졌음을 보장해서 공격을 막을 수 있다고 한다.

 

응답은 다음과 같은 형태로 오게 된다.

HTTP/1.1 101
Vary: Origin
Vary: Access-Control-Request-Method
Vary: Access-Control-Request-Headers
Access-Control-Allow-Origin: http://localhost:3000
Access-Control-Expose-Headers: Access-Token
Access-Control-Allow-Credentials: true
Upgrade: websocket
Connection: upgrade
Sec-WebSocket-Accept: xCcY0xCPpvMiyS059Fk4jGAhBBw=
Sec-WebSocket-Extensions: permessage-deflate;client_max_window_bits=15
X-Content-Type-Options: nosniff
X-XSS-Protection: 0
Cache-Control: no-cache, no-store, max-age=0, must-revalidate
Pragma: no-cache
Expires: 0
X-Frame-Options: DENY
Date: Wed, 27 Nov 2024 07:52:19 GMT

정상적으로 연결이 이루어졌다면, 우리가 흔히 아는 200 상태코드가 아닌 101 상태코드를 통해서 알려준다.

https://www.iana.org/assignments/http-status-codes/http-status-codes.xhtml

Hypertext Transfer Protocol (HTTP) Status Code Registry

101 상태코드는 switching protocol 상태코드로서, HTTP 프로토콜에서 웹 소켓 프로토콜로 전환을 의미한다.

WebSocket 프레임

웹 소켓은 프레임이라는 단위로 데이터를 주고받게 된다.

위 사진은 RFC 문서에서 가져온 것으로(https://datatracker.ietf.org/doc/html/rfc6455#section-5) 웹 소켓 프레임의 구조에 대해서설명해주고 있다.

• FIN(1비트): 마지막 프레임 여부를 나타낸다. 1이면 마지막 프레임, 0이면 추가적인 프레임이 존재한다. 메시지가 여러 프레임으로 나뉘는 경우 FIN 비트를 통해 마지막 여부를 파악할 수 있다.
• RSV1, RSV2, RSV3(1비트): 예약 비트로서, 기본적으로 0으로 설정되며 확장이 필요할 경우 사용된다.
• opcode(4비트): 프레임의 유형을 나타낸다. (Text, Binary, Ping/Pong 등)
• MASK(1비트), Masking-key: 클라이언트에서 서버로 데이터 전송 시 데이터 변조 방지를 위한 마스킹 여부를 나타내며 클라이언트에서 전송 시 1, 서버에서 전송 시 0으로 설정된다.
• Payload len, Extended payload length(7비트, 7+16비트, 7+64비트) : 전송할 데이터의 길이를 나타낸다. 길이에 따라 7, 16, 64비트로 확장 가능하다.
  • 데이터의 크기가 125바이트 이하인 경우 Payload len은 7비트로 표현된다. 이 경우에는 Payload len 필드 자체가 데이터의 크기를 나타낸다.
  • 데이터가 126~65,535 바이트인 경우 Payload len은 126으로 설정된다. 이후의 16비트가 실제 데이터의 크기를 나타낸다.
  • 데이터의 크기가 65,536 바이트 이상인 경우 Payload len은 127로 설정된다. 이후 64비트가 실제 데이터의 크기를 나타낸다.
• Payload Data : 실제 메시지 데이터가 담기게 된다.

STOMP

이제 웹 소켓에 대해서 알아보았는데, 그렇다면 STOMP란 무엇일까?

STOMP는 Simple Text Oriented Messaging Protocol의 약자로서 웹 소켓 프로토콜에서 데이터를 주고 받을 때, 표준화된 방법을 정의한 프로토콜이다.

웹 소켓 프로토콜을 통해서 우리는 메시지를 마음대로 주고 받을 수 있지만, 어떠한 형태로 메시지를 주고받을지는 프로젝트를 진행하면서 계속해서 정의하여야 한다. STOMP는 이러한 부분을 해결하기 위해 표준화된 방법을 사용한다.

https://stomp.github.io/stomp-specification-1.2.html

https://stomp.github.io/stomp-specification-1.2.html

 

STOMP 메시지 구조

구조는 다음과 같이 되어있다.

명령어
헤더

데이터

예시를 들면 다음과 같다.

SEND                     // 명령어
destination: /chat       // 헤더1
content-type: text/plain // 헤더2
...                      // 추가적인 헤더들

Hello, World!            // 데이터

명령어를 조금 더 자세하게 확인하면 다음과 같다

• 클라이언트 명령어
  • SEND : 서버로 메시지를 전송한다. destination 헤더를 통해 어디로 메시지를 전달할 지 표현한다.
  • SUBSCRIBE : 메시지를 받을 대상 경로을 구독한다.
  • UNSUBSCRIBE : 구독을 해제한다.
  • BEGIN : 트랜잭션을 시작한다.
  • COMMIT : 트랜잭션을 커밋하여 메시지를 적용한다.
  • ABORT : 트랜잭션을 취소하여 변경사항을 롤백한다.
  • ACK : 메시지 처리가 성공했음을 서버에 알린다.
  • NACK : 메시지 처리가 실패했음을 서버에 알린다.
  • CONNECT : 서버와의 연결을 시작한다.
  • DISCONNECT : 서버와의 연결을 종료한다.
• 서버 명령어
  • CONNECTED : 클라이언트의 연결 요청을 수락하고, 연결이 성공적으로 이루어졌음을 알린다.
  • MESSAGE : 서버가 클라이언트에게 메시지를 보낼 때 사용한다. destination헤더를 통해 메시지가 전달될 경로를 표현한다.
  • RECEIPT : 클라이언트가 보낸 명령이 정상적으로 처리되었음을 확인하는 응답이다.
  • ERROR : 서버 처리 오류, 클라이언트 명령이 잘못되었을 때 발생하는 오류 메시지이다.

각 명령어들은 요구하는 헤더가 별도로 존재한다. 이는 다음과 같다.

내가 지금 만들려고 하는 실시간 채팅 시스템에서는 SEND, SUBSCRIBE, MESSAGE가 사용될 것으로 생각된다.

 

STOMP pub/sub

STOMP 프로토콜의 주요 특징 중 하나는 pub/sub(발행/구독) 패턴을 지원하는 것이다.

STOMP 메시지는 메시지 브로커를 통해 전송된다. 즉 내가 publish(발행)한 메시지는 메시지 브로커에게 전달되며, 메시지 브로커는 이를 통해 destination 등을 파악하여 subscribe(구독)을 하고 있는 대상들에게 메시지를 전달해준다.

따라서 우리는 설정 시에 구독 prefix와 발행 prefix를 따로 설정해주게 된다.

 

Spring Boot + STOMP을 이용한 채팅방 시스템 구현

메시지 수신/발신 부터 체크해보자

먼저 웹 소켓을 연결하고 pub/sub과 메시지 수신/발신이 정상적으로 되는지부터 확인하였다.

Spring에 WebSocketConfig 파일을 생성하여 웹 소켓 관련 설정을 해주었다.

WebSocketConfig

implementation 'org.springframework.boot:spring-boot-starter-websocket'

build.gradle에 먼저 spring-boot-starter-websocket 의존성을 추가부터 해주었다. 그 이후

위 코드와 같이 설정을 해주었다. 

• @EnableWebSocketMessageBroker : WebSocket 메시지 브로커를 활성화하고, STOMP 기반 WebSocket 통신을 가능하게 한다
• .addEndpoint() : 어떠한 경로로 WebSocket 연결을 허용할 지 설정한다
• .setAllowedOriginPatterns() : 어떠한 Origin에서 연결을 허용할 지 설정한다. 지금은 단순 테스트이기 때문에 *을 사용해 모든 연결을 허용했다.
• .withSockJS() : SockJS 폴백을 활성화한다. SockJS는 웹 소켓을 지원하지 않는 환경에서도 웹 소켓을 사용할 수 있게 해주는 기술이라고 한다.
• .enableSimpleBroker() : 메모리 기반의 메시지 브로커를 활성화하고, 내 코드에서는 "/sub"으로 시작하는 목적지를 가진 메시지를 브로커가 처리하게 된다. 클라이언트 측에서 서버로 메시지를 구독할 때는 항상 /sub/chat/~~ 과 같이 /sub으로 구독 경로 시작을 하여야 한다.
• .setApplicationDestinationPrefixed() : 메시지를 발행할 때 어떠한 접두사를 사용할 지를 설정한다.

위와 같이 설정을 한다면, 유저 A,B가 /sub/chat/1 을 구독하고 있는 경우에, 유저 A가 /pub/chat/1 로 메시지를 보내면 메시지 브로커가 이를 처리하고 /sub/chat/1 을 구독하고 있는 유저들에게 메시지를 보내줄 수 있다.

ChatRoomController

컨트롤러는 위와 같이 설정하였다.

• @MessageMapping() : 지정한 경로로 메시지가 발행될 경우 이 메소드가 호출된다. Rest API의 @RequestMapping과 비슷하다고 생각하면 된다.
• @SendTo() : 메시지 브로커와 통신하여 지정 경로를 구독중인 모든 클라이언트에게 메시지를 전달한다.
• @DestinationVariable : URL경로에서 chatRoomId 부분을 추출한다. @PathVariable과 비슷하다고 생각할 수 있다.
• ChatMessage는 단순 DTO로 나 같은 경우는 아래처럼 작성되어 있다.

 

이제 한 번 채팅 수신/발신이 되는지 체크해보자.

브라우저를 2개 켜놓고 진행한 결과 채팅이 정상적으로 되는 것을 확인하였다.

 

유저 인증 기능을 추가해보자

일단 기본적으로 pub/sub 이 이루어지는 것을 확인했으니 이제 유저를 파악하고 이에 따라서 닉네임 등을 보여주면 좋을 것이라고 생각했다.

기본적으로 유저를 파악하기 위해서는 지금 프로젝트에서는 Jwt토큰을 활용하고 있다. 따라서 HTTP 통신과 같이 Authorization 헤더를 통해서 Access Token을 받고, 이에 따라서 유저를 파악해주면 간단할 것이라고 생각했다.

또한 유저를 인증하는 부분을 어느 정도까지 포함할 것인지를 고민해보았는데, 초기 핸드쉐이크 + 웹소켓 통신이 이루어질때마다 계속해서 토큰을 검사한다면, 불필요한 로직이 계속해서 생길 것이라 생각했고 굳이 계속 검증하기보다는 초기 핸드쉐이크에서만 검증을 진행해도 될 것 같다고 생각했다.

어느 시점에서 검증을 하는 것이 좋을까?

먼저 클라이언트 코드에 조금 추가해서 socket 연결 시에? 헤더를 같이 보낼 수 있도록 했다.

헤더를 넣어주었다.

그리고, WebsocketConfig에서 registryStompEndpoints을 할 때에 Interceptor을 통해서 핸드쉐이크 이전, 이후의 설정을 진행해 줄 수가 있다. 이 Interceptor는 HandShakeInterceptor을 implements하여 설정해주면 된다.

먼저, request쪽에서 Authorization 헤더가 잘 넘어오는 지 확인하기 위해 우선 request.getHeaders()을 확인해보기로 했다.

결과는 다음과 같았다.

확인 결과 내가 추가한 Authorization 헤더가 없었다. 원인을 정확하게 파악하지는 못했지만, 클라이언트 코드의 SockJS가 커스텀 헤더를 지원하지 않는 것인지, HTTP 통신에서 WebSocket 통신으로 변경되는 과정에서 담길 수 없는 것인지는 파악하지 못했다.

 

그렇다면, 내가 여기서 선택할 수 있는 방법은 2가지이다.

1. 헤더는 오지 않지만, 쿠키는 오기 때문에 Refresh Token에서 userId값을 가져와서 검증한다.
2. 맨 처음 연결을 시도하는 CONNECT 명령문일 경우에만 Access Token을 검증한다.

두 방법 모두, 초기 연결 시에 한 번만 검증을 진행하고 이후 채팅을 보내고 받을 때는 검증을 하지 않기 때문에 적절한 방법을 선택하면 될 것 같았다.

 

HandShakeInterceptor을 이용하여 Refresh Token을 이용한 검증

핸드쉐이크 과정이 일어나기 전에 쿠키로 들어온 Refresh Token 값에서 userId를 빼내어 웹 소켓 세션 attribute에 저장을 해주었다.

userId 를 정확히 파악할 수 있게 된다.

ChannelInterceptor을 이용하여 Access Token을 이용한 검증

기존의 다른 서비스 로직들에서 Access Token을 사용하고 있고, 채팅도 똑같은 방법으로 Access Token을 사용하는 것이 조금 더 좋을 것 같다고 생각한다면, 다음과 같이 코드를 작성할 수 있다.

이에 따라 클라이언트 코드를 조금 수정하였다.

소켓을 만들고 connect 일 경우에만 Authorization 헤더를 붙여서 보내게 했다. 그러고 나서 웹 소켓 연결을 한번 확인해보면

구독을 진행하거나 메시지를 보낼 때는 헤더가 달리지 않고, CONNECT 명령문에서만 헤더가 포함된 것을 볼 수 있었다.

ChannelInterceptor을 implements 하여 accessor의 커맨드가 CONNECT일 경우에만 헤더를 가져와 userId를 빼낸다.

그 후에는 WebSocket 세션 attribute에 userId 값을 넣게 된다.

두 방법 모두 정상적으로 userId를 판단할 수 있게 된다는 것을 알 수 있다.

 

그렇다면 어떤 방법을 사용하는 것이 좋을까?

1. 핸드쉐이크 이전에 쿠키값에서 Refresh Token을 이용해 유저를 검증하는 방법
   • 이 방법을 사용하면 Token 값이 올바르지 않을 경우 핸드쉐이크 자체가 수립되지 않게 할 수 있어서 조금 더 보안적으로 안전할 것 같다
   • 하지만 Access Token 만 사용하는 프로젝트에서는 커스텀 헤더가 제한적이기 때문에 사용을 하지 못할 가능성이 있다
2. 메시지를 보내기 이전 preSend에서 Access Token을 이용해 유저를 검증하는 방법
   • 이 방법 또한 초기 CONNECT 시에 한 번만 검증을 진행하게 되며, WebSocket 헤더를 이용할 수 있다.
   • 핸드쉐이크 연결이 진행된 이후 검증을 하기 때문에 불필요한 연결이 수립될 수 있다.
3. 핸드쉐이크 이전에는 Refresh Token으로 검증하고 preSend에서 Access Token으로 한번 더 검증하기
   • 확실히 두 방법을 동시에 사용하면 더 안전할 수 있을 것이라 생각한다.
   • 이렇게까지 진행하는 것이 필요한지는 아직까지는 잘 모르겠다. 프로젝트의 성격에 따라 달라질 것 같다.

우리 프로젝트에서는 Access Token(헤더) + Refresh Token(쿠키)을 모두 사용하고 있고 아무래도 불필요한 연결이 수립되지 않는 것이 더 좋을 것 같다고 생각해서 1번 방법으로 진행하기로 했다.

 

유지 기능을 이용한 채팅방 기능 로직

이제 유저를 세션 attribute에 담을 수 있게 되었기 때문에 채팅방을 구독하는 기능을 추가할 수 있게 된다. 먼저 DB의 경우에는 다음과 같이 구성을 했다.

유저와 채팅방은 다대다(N:M) 관계이기에 일대다 2개의 연결을 통해 풀어내었으며, 채팅방과 메시지는 일대다(1:N) 관계로 형성해주었다.

• 유저 채팅방 입장(구독X) : 최초 입장이기 때문에 User_Chatroom에 데이터 추가
• 유저 채팅방 구독 : 입장 시에 채팅방의 메시지 목록을 보여줌.
  • 입장을 하게 되면 가장 최신 메시지를 읽는 것과 같기 때문에 lastMessageId를 해당 채팅방의 최신 메시지 id로 변경해줌
• 유저 채팅방 구독해제 : 재입장 시 유저가 마지막으로 읽은 지점을 파악하기 위해 해당 채팅방에서 유저가 마지막으로 읽은 메시지 id를 저장
• 유저가 채팅방에서 강퇴당했을 경우 : lastMessageId를 최신화하지 않으며, lastMessageId 이전 메시지만 보여줌
• 유저 채팅방 나가기 : User_Chatroom에서 데이터 제거

채팅방 입장(최초 입장)과 입장 시 메시지 리스트를 불러오는 것, 채팅방 나가기 기능은 Rest API를 이용하고, 채팅방 구독(채팅방 키는 것)과 구독해제(채팅방 끄는 것)는 웹 소켓의 구독/해제 기능을 이용하면 된다.

 

우선은 먼저 간단하게 ChatroomService를 만들어서, 유저가 보낸 메시지를 저장하고 다른 유저들에게 보내는 기능부터 구현해보자.

ChatroomService

이전에는 컨트롤러에서 @SendTo 어노테이션을 이용해서 전달하였다면, 이번에는 SimpMessagingTemplate의 convertAndSend 메소드를 이용해서 채팅방에 메시지를 보내게 된다.(unreadCount는 추후 설명)

• SimpMessagingTemplate을 이용하더라도 대부분 브로커를 통해서 메시지가 전달된다고 한다.
• convertAndSend 메소드는 STOMP 프로토콜 기반 메시지 브로커를 통해 클라이언트로 메시지를 전송하는 데 사용된다. 메시지의 변환 및 전송 과정을 자동화하여 편하게 메시지를 보낼 수 있도록 돕는다.

따라서 우리가 메시지를 작성하게 된다면, 새로운 Message 데이터를 만들어 테이블에 저장하고 해당 destination을 구독하는 사용자들에게 모두 메시지를 보내게 된다.

3개의 브라우저에서 3개의 다른 계정을 진행했을 때 작성자가 다른 모습을 확인할 수 있었다.

DB에 메시지 또한 잘 저장된 모습을 볼 수 있다.

 

그렇다면 이제 우리가 할 일은, 채팅방 구독 시에 각 채팅방에 저장된 메시지를 불러와서 리스트를 뿌려주어야 한다.

이 부분은 Rest API 로 작성하여 클라이언트 측에서 데이터를 불러와 렌더링하도록 하였다.

 

실시간 읽지 않은 메시지 refresh 기능을 어떻게 할 것인가?

현재 내가 구현하고 싶은 기능은, 카카오톡 입장 시에 읽지 않은 인원 숫자가 사라지는 부분을 처리하는 것을 원하고 있다.

이에 따라서, 메시지를 저장함에 있어 현재 읽지 않은 인원을 체크하는 컬럼을 추가했다(unreadCount)

이 unreadCount를 어떻게 설정해야할 지 고민 끝에, (채팅방에 속한 모든 인원 - 현재 채팅방 구독중인 인원)으로 설정한 이후,

채팅방에 속해있지만, 메시지를 읽지 않았고, 구독을 하는 인원이 접속하는 경우 숫자를 1씩 내려가면 될 것이라고 생각했다.

 

스프링에서는 구독, 구독 해제, 연결, 연결 해제 시에 원하는 처리를 진행할 수 있도록 EventListner을 제공하고 있다.

이런식으로 EventListner 클래스를 생성 후 @EventListner 어노테이션과 어떤 이벤트를 처리할 지에 대해 내용을 작성해주면 된다.

• StompHeaderAccessor : STOMP 프로토콜의 헤더 정보에 접근할 수 있게 해주는 클래스
• .wrap() : 원본 메시지에서 STOMP헤더 정보를 추출하여 accessor 객체를 생성함

따라서, 구독 시에 어떤 유저가 어떤 채팅방에 접속하였는 지를 로그로 찍게 하였다. 또한 우리는 Message 테이블의 unreadCount도 있지만, User_Chatroom 테이블에서 유저가 속한 특정 채팅방의 마지막 메시지 id를 저장하는 컬럼이 있기 때문에 이를 활용하기로 했다.

 

먼저 ChatroomService에서

메시지 엔티티를생성 시에 (chatRoom의 참여 인원 - 현재 구독중인 인원)으로 unreadCount를 만들기 위해

getSubscribedUserCount(chatRoomId) 로직을 생성했다.

SimpUserRegistry라는 클래스는 스프링 웹 소켓에서 현재 연결된 웹소켓 사용자들의 정보를 관리하고 추적하는 역할을 하는 클래스이다.

현재는 다소 비효율적으로 3중첩 for문으로 작성되어있지만 단순히 기능 구현을 위해 작성한 것으로 봐주시면 된다..(나중에 더 최적화를 해야 할 로직이라고 생각한다)

먼저 SimpUserRegistry에서 유저를 찾고, 이 유저가 속한 세션을 찾은 후, 그 세션의 destination이 목표로하는 destination과 같을 경우 count를 올린다. 따라서 현재 채팅방을 구독 중인 유저들을 찾을 수 있게 되는 것이다.

이 부분에서 문제가 발생하였는데..

문제 발생 - SimpUserRegistry에 등록된 유저가 존재하지 않음

아무리 해도 count의 값이 계속 0으로 찍히는 문제가 있었다.

분명히 구독도 하고 채팅을 쳐도 구독중인 유저 수는 0에서 변하지 않았다.

원인을 파악해보니,

 

DefaultSimpUserRegistry를 보면

SessionConnectedEvent 시에 유저를 등록하게 된다. 디버그 모드로 더 정확하게 확인을 하다 보면

user가 null이라는 것을 확인할 수 있고, user가 null이기 때문에 SimpUserRegistry에는 담기지 않는 것을 확인할 수 있다.

 

DefaultSimpUserRegistry를 다시 확인해보면

Principal user = subProtocolEvent.getUser();

구문을 볼 수 있는데, 여기서 Principal 객체는 Spring Security에서 관리하는 유저 객체이다. 하지만 우리는 연결 혹은 구독 시에 따로 Spring Security에 유저를 등록해주지 않기 때문에 user가 null값이 되는 문제를 겪게 된 것이다.

따라서 위와 같이 Principal 객체를 생성하여 accessor.setUser() 을 통해 객체를 등록해주었다.

그 이후 다시 디버그 모드로 확인해보면,

Principal 객체를 인식하는 것을 확인할 수 있었다.

 

이를 만약 preSend 구문이 아닌 beforeHandshake에서 등록하기 위해서는

SecurityContextHolder에 직접 등록을 해주면 인식을 하게 되었다.

하지만 왜인지, preSend에서 SecurityContextHolder에 등록을 하면 등록이 되지 않았는데 동작 방식의 차이인 것 같은데 정확히 파악하지는 못하였다.

 

하지만, Spring Security의 Context Holder을 사용하는 것이 맞을지.. 내가 따로 세션을 만들어서 단순 하게 userId와 destination을 묶어서 관리하는 것이 좋을지는 더 고민을 해보아야 할 사항인 것 같다.

 

다시 서비스 로직으로 돌아와서, unreadCount를 제대로 파악할 수 있게 되었고, 이제 메시지를 전송 하기 전에updateLastMessageToallSubscribers 라는 메소드를 통해 세션에 속한 유저들의 lastMessageId값을 전송될 메시지의 messageId 값으로 변경해주었다.

그럼 먼저, 정상적으로 DB에 값이 들어가는지 확인해보자

 

테스트 - unreadCount 값과 lastMessageId 값이 잘 저장되는가?

현재 1번 채팅방은 4명의 인원이 참여해 있다.

유저 1,2,3,4가 1번 채팅방에 포함되어 있으며 현재 lastMessageId 값은 null이다.

이제 유저 2명만 연결을 해서 채팅을 한번 쳐보겠다.

유저 1과 2만 로그인을 하여 채팅을 쳤고, DB데이터는 다음과 같이 되었다.

메시지는 다음과 같이 unreadCount가 2로 생성되었으며,

UserChatroom 테이블에서는 구독중인 유저 1,2 만 lastMessageId가 159로 가장 최근 메시지 값으로 update 되었다.

 

구독 시 처리를 어떻게 할 것인가?

위와 같은 상황에서 이제 유저3이 채팅방을 구독한다고 가정해보자. 그렇다면 서버에서 해야 할 일은 다음과 같다.

• 유저3은 채팅방에서 아직 아무 메시지도 읽지 않았다(lastMessageId = null). 따라서 채팅방에 존재하는 모든 메시지의 unreadCount를 1 내린다.
• 유저3의 lastMessageId 값을 159 라는 가장 최근 messaeId값으로 변경한다.

우리는 구독 이벤트 발생시에 원하는 로직을 작성할 수 있다. 따라서 다음과 같이 해주었다.

웹 소켓 연결 시에 attribute에 userId값을 담았기 떄문에 userId를 파악할 수 있고, destination으로 지정된 chatroom의 번호를 확인할 수 있다. 그리고 lastMessageId값과 채팅방과 일치하는 Message 테이블의 id값들을 비교해 이후 id들의 unreadCount를 1씩 감소시켰다. id는 1부터 시작하여 순차적으로 올라가는 번호로 생성되기 때문에 문제 없을 것이라 판단했다.

 

그럼 이 상태로 3번 유저로 접속을 진행해보았다.

정상적으로 채팅방의 메시지 목록이 불러와지고, 유저3으로 채팅을 한번 더 쳤다. 이 상태에서 다시 DB를 확인하면

기존의 메시지들의 unreadCount는 정상적으로 감소하였고, 유저3번의 채팅은 3으로 잘 작성되어있다.

lastMessageId 또한 정상적으로 업데이트 된 것을 확인할 수 있다.

유저에게 실시간으로 안읽은 메시지를 감소시키는 방법에 대해 생각해보자.

이제 unreadCount라던지, lastMessageId 등 서버에서 처리해야 할 로직에 대해서는 끝났다고 생각했다.

그렇다면 실제로 화면이 계속 켜져 있는 상태에서 유저 채팅의 안읽은 메시지 수를 실시간으로 감소시키려면 어떻게 해야 할까?

• 가장 간단한 방법은, 주기적으로 채팅방 메시지를 불러와서 덮어씌우는 것이다
  • 그런데 이러한 방법을 쓰면 결국 Polling, Long-Polling 처럼 될텐데, 웹 소켓을 쓴 의미가 없지 않을까?
• 유저가 접속했을 때, 메시지가 전송되었을 때만 채팅방 메시지를 불러와서 덮어씌운다
  • 이것도 결국 Long-Polling이 되는 것 아닐까?

계속해서 고민을 하던 도중 아이디어가 하나 떠올랐다.

만약 백엔드와 프론트가 정확한 구조를 가지고 데이터가 어긋나지 않는다는 확신만 존재한다면 프론트에서 계속 DB 데이터를 불러오는 게아니라 유저가 들어올 때만 프론트에서 자체적으로 읽지 않은 메시지 수를 감소시키면 안되는걸까?

 

와 같은 생각이 들게 되었다. 하지만 이 경우에도 문제가 한 가지 있는데

• 클라이언트측은 채팅방 메시지에 대한 모든 리스트를 가지고 있으며 유저가 접속할 때마다, 맨 처음 메시지부터 마지막 메시지까지 1씩 카운트를 계속해서 감소시켜야한다 -> 채팅방 메시지가 너무 많다면 부담이 될 수 있지 않을까?

와 같은 부분이었다. 따라서 읽지 않은 수가 0인 메시지는 굳이 처리하지 않을 수 있는 방법에 대해서도 다시 고민하던 와중, 유저 입장 시에 본인의 lastMessageId를 같이 보내는 방법이 괜찮을 것 같다는 생각을 했다.

로직은 다음과 같다.

• 초기 채팅방 구독 시에, 다른 유저가 입장했을 때 다른 유저의 lastMessageId를 받는 구독을 한개 더 한다(총 2개의 구독)
• 채팅방 구독은 메시지 전달로만 사용을 하고, 다른유저입장 구독의 경우는 읽지 않은 수를 줄이는 용도로 사용을 한다.

생각보다 괜찮을 수도 있을 것 같다는 생각을 했다.

최종적으로 구독 이벤트 발생 시의 로직은 위와 같이 작성되었다. /sub/chat/room/read/{chatroomId} 를 프론트에서 구독하기 때문에 이쪽으로 구독이 연결되는 경우에만 나의 lastMessageId를 보내준다(예시에서는 cursorId라고 표현했다)

클라이언트에서는 cursor가 null인 경우 모든 메시지의 unreadCount를 제거하면 되고(한 번도 채팅방의 메시지를 읽지 않은 것이기 때문에) cursor값이 존재한다면 그 값보다 messageId값이 이상인 메시지들의 unreadCount를 내려주면 된다.

이렇게 할 경우, 서버측 db를 다시 가져오는 것이 아니기 때문에 로직이 완벽하지 않으면 오류를 발생할 수 있지만, 확실히 서버의 부담을 줄일 수 있을 것 같다고 생각했다.

결과는 다음과 같다.

 

테스트 - 실시간으로 안읽은 개수가 잘 줄어드는가?

실제로 사용자에게 보이는 읽지 않은 회원 수는 줄어들게 되었고 DB 데이터를 확인해보면

정상적으로 업데이트 되어 구독해제 이후 다시 구독을 하더라도 문제없이 데이터가 불러와지는 것을 확인할 수 있었다.

4번 유저로 입장한 결과는 위와 같다.

이러한 방식을 이용하면, 서버로의 요청도 덜하게 되서 부담이 적어지고 클라이언트에서 unreadCount가 0인 데이터는 읽음 수를 관리하는 map에서 뺄 수 있지 않을까 생각을 해보았다.

 

글을 마치며

실시간이라는 형식이 보기에는 단순하지만 실제로 공부를 하면서 정말 구조가 복잡해지고 어렵다는 것을 느꼈다. 또한 이 방법은 프로젝트에 적용하기로 확정된 방식이 아니라, 내가 혼자 고민을 해본 방법이어서 어떤 오류가 있을 지 정확하게 체크를 해보지 못한 점도 있다.

또한 테스트를 진행함에 있어서는 단일 채팅방 구조였기 때문에 채팅방이 많아진다면 어떠한 문제가 있을지는 고려되지 않았다.

실제 구현 시에는 오류 등을 해결하기 위해 구조가 변경될 수도 있다. 또한 성능적인 부분을 고려하지 않았기 때문에 변경점이 많을 수 있을 것 같은데 이러한 점은 파악 후 포스팅을 또 해보겠다..