Stomp, Redis 로 현재 접속중인 사용자 조회하기

현재 진행중인 도서 공유 서비스 프로젝트를 진행하며 새로운 기능을 구현한 내용입니다.

 

---

 

요구사항

• 지역을 기준으로 도서 검색 시 해당 지역에 접속중인 사용자 목록을 조회하는 기능이 추가되었으면 한다는 피드백이 들어왔음

접근

• 채팅 기능으로 STOMP를 사용, 메일 인증 기능으로 Redis를 사용중이었음.
• 이 둘을 사용하면 현재 접속중인 유저의 목록을 조회하는 것이 가능하지 않을까? 생각하여 둘을 활용하는 방향으로 접근하였음.

구현방법

• 소켓 연결 시, 현재 연결을 시도하는 사용자의 지역코드로 Redis의 Set에 사용자의 정보를 저장하기로 하였음.
  • Set을 사용한 이유는 Redis의 set은 Java와 비슷하게 중복된 값을 허용하지 않으므로 데이터에 신뢰성을 높이기 위함이었음. 같은 유저의 정보가 두명이 있다면 해당 데이터의 신뢰성은 떨어질 것이라고 판단
  • StompHeader의 Command 는 현재 들어온 요청이 어떤 타입의 요청인지를 나타내고 있음.
  • 따라서 StompCommand.CONNECT, 즉 소켓에 연결하는 요청이라면 JWT 토큰을 파싱하여 현재 사용자의 정보를 Redis에 저장
  • JWT 토큰을 파싱하여 나온 유저에서 식별자를 소켓의 sessionAttribute에 저장
    • 연결을 끊는 시점, 소켓을 DISCONNECT 할 때 사용자를 식별하고 Redis에서 유저 정보를 삭제해야하기 위함
  • StompHeader의 Command 가 DISCONNECT 라면 sessionAttribute에 저장했던 유저 식별자를 가져와, 유저 정보를 Redis에서 제거
• 도서 검색을 할 때, 지역 코드로 구독 요청을 보내게 프론트에서 구현
• 스프링에선 @Scheduled 를 적용하여 5초마다 현재 지역에 접속중인 사용자 목록을 Redis 에서 조회하여 메시지를 발행하게 하였음

테스트코드

• WebSocketStompClient 를 사용하여 테스트코드 작성.

API 문서의 일부분

위는 API 문서의 설명. SockJs를 사용하더라도 따로 설정을 추가할 필요가 없다.

• 소켓을 연결하는 과정에서의 JWT 검증에 대한 테스트를 진행
• 소켓이 연결되었다면 Redis에 유저정보가 저장되었는지 검증
• 소켓 연결을 해제하면 Redis에 유저정보가 삭제되었는지 검증

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
public class SocketTest {

    @LocalServerPort
    private int port;
    @Autowired
    private JwtProvider jwtProvider;
    @Autowired
    private UserRepository userRepository;
    @Autowired
    private PasswordEncoder passwordEncoder;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    private WebSocketStompClient stompClient;

    private String jwt;
    private String areaCode = "12345678";
    private User user;

    @BeforeEach
    void setup() throws IOException {
        // redis 초기화
        redisTemplate.getConnectionFactory().getConnection().flushAll();
        // Stomp Client 설정
        stompClient = new WebSocketStompClient(new StandardWebSocketClient());
        MappingJackson2MessageConverter messageConverter = new MappingJackson2MessageConverter();
        ObjectMapper objectMapper = messageConverter.getObjectMapper();
        objectMapper.registerModules(new JavaTimeModule(), new ParameterNamesModule());
        stompClient.setMessageConverter(messageConverter);

        user = this.userRepository.save(User.builder()
                .email("test@test.com")
                .password(passwordEncoder.encode("12345678"))
                .areaCode(areaCode)
                .nickname("testAccount")
                .build());
        UsernamePasswordAuthenticationToken token = new UsernamePasswordAuthenticationToken(user.getId(), null);
        jwt = jwtProvider.createToken(token);
    }

    @AfterEach
    void tearDown() {
        redisTemplate.getConnectionFactory().getConnection().flushAll();
    }

    @DisplayName("소켓 연결 시 JWT 토큰이 유효하다면 정상적으로 연결된다")
    @Test
    void connectWebSocketWithValidJwtToken_success() throws Exception {
        // given
        StompHeaders headers = new StompHeaders();
        headers.add("Authorization", "Bearer " + jwt);
        URI uri = new URI(String.format("ws://localhost:%d/ws", port));

        // expected
        assertDoesNotThrow(() -> {
            this.stompClient.connectAsync(uri, null, headers, new StompSessionHandlerAdapter() {
            }).get(5, TimeUnit.SECONDS );
        });
        System.out.println(jwt);
    }

    @DisplayName("소켓 연결 시 JWT 토큰이 유효하다면 연결이 실패한다")
    @Test
    void connectWebSocketWithInvalidJwtToken_fail() throws Exception {
        // given
        StompHeaders headers = new StompHeaders();
        String invalidToken = "invalidToken";
        headers.add("Authorization", invalidToken);
        URI uri = new URI(String.format("ws://localhost:%d/ws", port));

        // expected
        assertThrows(ExecutionException.class, () -> {
            this.stompClient.connectAsync(uri, null, headers, new StompSessionHandlerAdapter() {
            }).get(5, TimeUnit.SECONDS );
        });
    }

    @DisplayName("소켓 연결 시 레디스에 지역별 접속 목록에 사용자가 저장된다")
    @Test
    void addAreaCodeSetInRedisWhenConnect() throws Exception {
        // given
        StompHeaders headers = new StompHeaders();
        headers.add("Authorization", "Bearer " + jwt);
        URI uri = new URI(String.format("ws://localhost:%d/ws", port));

        // when
        this.stompClient.connectAsync(uri, null, headers, new StompSessionHandlerAdapter() {
                }).get(5, TimeUnit.SECONDS);

        // then
        Set<Object> members = redisTemplate.opsForSet().members("area:" + areaCode);
        assertTrue(members.contains(UserOnResponse.toDto(user)));
    }

    @DisplayName("소켓 연결 해제 시 레디스에 사용자의 정보를 제거한다")
    @Test
    void removeAreaCodeSetInRedisWhenDisconnect() throws Exception {
        // given
        StompHeaders headers = new StompHeaders();
        headers.add("Authorization", "Bearer " + jwt);
        URI uri = new URI(String.format("ws://localhost:%d/ws", port));

        StompSession stompSession = this.stompClient.connectAsync(uri, null, headers, new StompSessionHandlerAdapter() {
        }).get(5, TimeUnit.SECONDS);

        Set<Object> members = redisTemplate.opsForSet().members("area:" + areaCode);
        assertTrue(members.contains(UserOnResponse.toDto(user)));

        // when
        stompSession.disconnect();

        // then
        Thread.sleep(100);
        members = redisTemplate.opsForSet().members("area:" + areaCode);
        assertFalse(members.contains(UserOnResponse.toDto(user)));
    }
}

해결하지 못한 문제

• 스케줄러를 통해 5초마다 지역코드를 구독하고 있는 사용자들에게 Redis의 유저 목록을 브로드캐스팅 하는 기능을 테스트 코드로 작성하고 테스트 하였으나 어째서인지, 로그를 찍어보면 스케줄러가 정상 동작하지만 Future타입으로 정의해둔 메시지로 메시지가 도착하지 않았다.
• stompSession.send() 로 지역코드 경로로 메시지를 발행하면 메시지가 정상적으로 들어오는 것을 확인할 수 있었다.
• 둘의 차이가 무엇인지 왜 안되었는지에 대해서는 프로젝트가 마무리되면 알아보아야겠다.