Code and Crescendo raccoon

WebSocket이란?

https://harmony-raccoon.tistory.com/145

STOMP이란?

https://harmony-raccoon.tistory.com/144

Vue.js로 WebSocket 구현하기

https://harmony-raccoon.tistory.com/152

Spring을 이용해서 WebSocket 구현하기

WebSocket 서버를 열기 위해서는 WebSocket Config 파일이 존재해야 합니다.

WebSocketConfig

@Configuration
@EnableWebSocketMessageBroker
@RequiredArgsConstructor
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    private final JwtHandShakeInterceptor jwtHandShakeInterceptor;
    private final AuthChannelInterceptor authChannelInterceptor;

    @Value("${websocket.allowed-origin}")
    private String frontServer; // front URI

    @Bean(name = "customMessageBrokerScheduler")
    public TaskScheduler messageBrokerTaskScheduler() {
        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
        scheduler.setPoolSize(10);
        scheduler.setThreadNamePrefix("ws-heartbeat-");
        scheduler.setWaitForTasksToCompleteOnShutdown(true);
        scheduler.initialize();
        return scheduler;
    } // WebSocket을 처리할 Thread 설정

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.enableSimpleBroker("/topic", "/queue") // Topic, Queue Prefix 설정
                .setTaskScheduler(messageBrokerTaskScheduler())
                .setHeartbeatValue(new long[]{10000, 10000});
        // 하트비트 연결 및 10초 주기로 연결 확인
        registry.setApplicationDestinationPrefixes("/app"); // Client가 데이터를 보낼 Prefix 설정
        registry.setUserDestinationPrefix("/user");
        // 사용자별 메세지 라우팅
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/chat") // WebSocket HandShake를 할 URI
                .addInterceptors(jwtHandShakeInterceptor) // jwt를 통해서 인가 담당 기능
                .setAllowedOriginPatterns(frontServer) // CORS 해제 설정
                .withSockJS(); // SockJS를 통한 WebSocket 설정
    }

    @Override
    public void configureClientInboundChannel(ChannelRegistration registration) {
        registration.interceptors(authChannelInterceptor); // WebSocket에서 인증 정보를 처리할 수 있도록 저장하는 기능을 수행하는 객체
    }

    @Override
    public void configureWebSocketTransport(WebSocketTransportRegistration registration) {
        registration.setMessageSizeLimit(128 * 1024); // 128KB
        registration.setSendBufferSizeLimit(512 * 1024); // 512KB
        registration.setSendTimeLimit(1000); // 전송시간 제한 1초
    }
}

여기서 추가적인 객체가 2개가 존재합니다.

JwtHandShakeInterceptor, AuthChannelInterceptor인데, 각각 하는 역할이 다릅니다.

JwtHandShakeInterceptor : HTTP HandShake가 일어날 때, 인가 검증

AuthChannelInterceptor : STOMP에서 연결을 하기 위한 CONNECT Frame 통신이 올 때, 인가 검증

특히, AuthChannelInterceptor는 인가를 확인한 후에 사용자 데이터를 기능에 적용할 수 있도록 제공해주는 중요한 객체입니다.

JwtHandShakeInterceptor

@Component
public class JwtHandShakeInterceptor implements HandshakeInterceptor {
    @Override
    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Map<String, Object> attributes) throws Exception {
        if (request instanceof ServletServerHttpRequest servletServerHttpRequest) {
            HttpServletRequest httpReq = servletServerHttpRequest.getServletRequest();

            Cookie[] cookies = httpReq.getCookies();

            String jwt = findDabomJWT(cookies);
            exceptionHandler(jwt, attributes);
        }
        return true;
    }


    private void exceptionHandler(String jwt, Map<String, Object> attributes) {
        try {
            haveTokenLogic(jwt, attributes);
        } catch (ExpiredJwtException e) {
            // 만료 토큰 302 Redirect로 RefreshToken
        } catch (SecurityException | MalformedJwtException e) {
            // 잘못된 토큰 403 Error 내려주기
        } catch (UnsupportedJwtException e) {
            // 지원되지 않는 토큰 403 Error 내려주기
        } catch (IllegalArgumentException e) {
            // 빈 토큰 403 Error 내려주기
        }
    }

    private String findDabomJWT(Cookie[] cookies) {
        String jwt = null;
        if (cookies != null) {
            for (Cookie cookie : cookies) {
                if (Objects.equals(cookie.getName(), ACCESS_TOKEN)) {
                    jwt = cookie.getValue();
                }
            }
        }
        return jwt;
    }

    private void haveTokenLogic(String jwt, Map<String, Object> attributes) {
        if (jwt != null) {
            Claims claims = JwtUtils.getClaims(jwt);
            haveDabomTokenLogic(claims, attributes, jwt);
        }
    }

    private void haveDabomTokenLogic(Claims claims, Map<String, Object> attributes, String jwt) {
        if (claims != null) {
            MemberDetailsDto dto = createDetailsFromToken(claims, jwt);

            Authentication authentication = new UsernamePasswordAuthenticationToken(
                    dto,
                    null,
                    List.of(new SimpleGrantedAuthority(dto.getMemberRole().name()))) {
                @Override
                public String getName() {
                    return dto.getIdx().toString(); // getName대신 getIdx를 사용하기 위해서 설정
                }
            };
            SecurityContextHolder.getContext().setAuthentication(authentication);
            attributes.put("auth", authentication);
        }
    }

    private MemberDetailsDto createDetailsFromToken(Claims claims, String jwt) {
        Integer idx = Integer.parseInt(JwtUtils.getValue(claims, TOKEN_IDX));
        String name = JwtUtils.getValue(claims, TOKEN_NAME);
        String role = JwtUtils.getValue(claims, TOKEN_USER_ROLE);
        return MemberDetailsDto.createFromToken(idx, name, role);
    }

    @Override
    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Exception exception) {

    }
}

위 코드는 Spring Security에서 JwtAuthFilter와 매우 흡사한 코드입니다.

실제로 동작 과정도 매우 흡사합니다. (단, 2개의 코드가 완전히 같게 움직이게 할 수 없습니다. / 프로토콜이 미세하게 다르기 때문에)

AuthChannelInterceptor

@Component
public class AuthChannelInterceptor implements ChannelInterceptor {
    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
        if(StompCommand.CONNECT.equals(accessor.getCommand())) { // STOMP에서 CONNECT형태로 데이터가 오면
            Map<String, Object> attributes = accessor.getSessionAttributes();
            if(attributes!=null) {
                Authentication authentication = (Authentication) attributes.get("auth");
                if(authentication!=null) {
                    accessor.setUser(authentication); // 인가 정보를 뽑아서 StompHeaderAccessor에 저장한다.
                }
            }
        }
        return message;
    }
}

STOMP에서 CONNECT로 Frame의 Command가 날라오면, 처음 웹소캣에 연결하는 클라이언트이므로 인가 과정을 통해 처리해야 합니다.

이렇게 인가 정보는 다음 기능을 수행할 때 사용하게 됩니다.

ChatController

@RestController
@RequiredArgsConstructor
public class TogetherChatController {
    private final SimpMessagingTemplate messagingTemplate;
    private final MemberService memberService;
    private final TogetherService togetherService;
    private final Map<String, Set<Integer>> topicSessions = new ConcurrentHashMap<>();

    @MessageMapping("/together/{togetherIdx}")
    public void sendMessage(Principal principal,
                            @DestinationVariable Integer togetherIdx, @Payload String message) {
        MemberDetailsDto memberDetailsDto = getMemberDetailsDto((Authentication) principal);

        MemberInfoResponseDto memberInfo = memberService.getMemberInfo(memberDetailsDto);
        String destination = "/topic/together/" + togetherIdx; // 토픽 키
        int userCount = topicSessions.getOrDefault(destination, Collections.emptySet()).size();

        TogetherChatResponseDto res = TogetherChatResponseDto.toDtoBySend(memberInfo.name(), message, userCount, memberInfo.id());

        messagingTemplate.convertAndSend("/topic/together/" + togetherIdx, res);
    }

    @EventListener
    public void handleSubscribeEvent(SessionSubscribeEvent event) {
        SimpMessageHeaderAccessor headerAccessor = SimpMessageHeaderAccessor.wrap(event.getMessage());
        Principal userPrincipal = headerAccessor.getUser();
        String destination = headerAccessor.getDestination();

        if (destination != null && destination.startsWith("/topic/together/")) {
            // 환영 메시지 생성
            MemberDetailsDto memberDetailsDto = getMemberDetailsDto((Authentication) userPrincipal);
            MemberInfoResponseDto memberInfo = memberService.getMemberInfo(memberDetailsDto);

            // 토픽별 접속자 세션 관리
            topicSessions.computeIfAbsent(destination, k -> ConcurrentHashMap.newKeySet()).add(memberInfo.id());

            Integer userCount = topicSessions.get(destination).size();

            TogetherChatResponseDto welcome = TogetherChatResponseDto.toDtoByJoin(memberInfo.name(), userCount);
            messagingTemplate.convertAndSend(destination, welcome);
        }
    }

    @EventListener
    public void handleDisconnectEvent(SessionDisconnectEvent event) {
        SimpMessageHeaderAccessor headerAccessor = SimpMessageHeaderAccessor.wrap(event.getMessage());
        Principal userPrincipal = headerAccessor.getUser();

        if (userPrincipal == null) {
            return; // 인증정보 없으면 종료
        }

        MemberDetailsDto memberDetailsDto = getMemberDetailsDto((Authentication) userPrincipal);
        Integer memberId = memberDetailsDto.getIdx();

        // 연결된 모든 토픽에 대해 해당 멤버 아이디 제거
        topicSessions.forEach((topic, memberSet) -> {
            boolean removed = memberSet.remove(memberId);
            if (removed) {
                messagingTemplate.convertAndSend(topic, "");
            }
        });
    }

    private MemberDetailsDto getMemberDetailsDto(Authentication principal) {
        Authentication authentication = principal;
        MemberDetailsDto memberDetailsDto = (MemberDetailsDto)authentication.getPrincipal();
        return memberDetailsDto;
    }
}

위 코드는 실제 채팅방 기능을 개발한 코드에서 발췌한 내용입니다.

제가 저 코드를 작성할 때, 인가 정보를 가지고 채팅방의 인원을 확인하고, 없다면 새로 추가하여 인원방을 정확하게 처리하기 위해 인가 정보를 활용했었습니다.

이렇게 AuthChannelInterceptor를 활용해서 추가적인 기능 개발을 진행할 수 있습니다.

WebSocket 이란?

https://harmony-raccoon.tistory.com/145

STOMP이란?

https://harmony-raccoon.tistory.com/144

Vue.js를 이용하여 WebSocket 구현하기

Sock.js를 사용해서 WebSocket 구현

WebSocket을 Stomp를 사용해서 구현할려면 npm에서 라이브러리를 다운받아야 합니다.

npm install stompjs
npm install sockjs-client

이를 통해서 WebSocket에 연결하는 방법을 다음 코드로 구현하면 됩니다.

JavaScript로 WebSocket 연결

다음 코드는 WebSocket을 연결하는 것까지만 구현되어있습니다.

const connectWebSocket = async () => {
  const ws = new SockJS('https://[WebSocket 서버 URI 주소]', null,
      {
        transportOptions: {
          xhr: { withCredentials: true }, // Cookie를 같이 보내기 위한 코드
          xhrStreaming: { withCredentials: true } // Cookie를 같이 보내기 위한 코드
        }
      })
  const client = Stomp.over(ws)
  socket.value = client
  socket.value.connect(
      {},
    (frame) => {
      subscribed.value = true
      subscribeMasterEvent()
    },
    (error) => {
    }
  )
}

JavaScript로 WebSocket 구독

다음 코드는 연결된 WebSocket에서 STOMP로 구독하고, 메시지를 얻는 기능을 코드로 구현한 것입니다.

const socket = reactive({
  messages: [],
  socket: null
})

const chatSubscription = () => {
  socket.socket.subscribe(`/topic/[토픽 URI]`, (mes) => {
    const data = JSON.parse(mes.body)
    socket.messages.push(data)
    if(data.users !== socket.joinMember) {
      socket.joinMember = data.users
    }
  });
}

이 기능은 개발자가 지정한 Topic으로 구독을 진행하고, 그 구독으로 메시지가 전파되면 그 메시지를 캐치해서 메시지 리스트에 데이터를 저장하는 코드입니다.

JavaScript로 Message 전송

writeMessage에 담긴 데이터를 웹소캣을 통해서 서버에 전송합니다.

const writeMessage = ref("")

const sendMessage = () => {
  if(writeMessage.value === ""){
    return;
  }
  socket.socket.send(`/app/[topic의 URI]`, {}, writeMessage.value);
  writeMessage.value = ""
}

해당 서버는 받은 데이터를 다시 같은 Topic을 구독하고 있는 전체 클라이언트들에게 전송합니다.

이때, 내가 보낸 데이터를 다시 내가 받기 때문에 내가 보낸 데이터인지 확인하고 처리를 하는 기능을 꼭 구현해야 합니다.

WebSocket이란?

https://harmony-raccoon.tistory.com/145

STOMP WebSocket이란?

STOMP(Simple Text Oriented Messaging Protocol)은 TCP 혹은 WebSocket과 같은 양방향 네트워크 프로토콜 기반으로 동작하는 프로토콜입니다.

STOMP는 텍스트 지향 프로토콜이지만, Message Payload에는 Text 혹은 Binary Data도 들어갈 수 있습니다.

STOMP는 TCP 위에서 동작하는 Frame 기반의 프로토콜이며, Frame은 Command, header, Body로 구성되어 있습니다.

STOMP는 메시지 브로커를 통해서 구독과 메시지 처리를 진행한다.

Command

frame의 시작으로, 주요 command는 CONNECT, SEND, SUBSCRIBE, UNSUBSCRIBE, ACK, NACK, BEGIN, COMMIT, ABORT, DISCONNECT등이 존재합니다.

Headers

각 프레임에는 0개 이상의 헤더가 포함됩니다.

key-value형태의 쌍으로 구성되어 메시지의 메타 데이터를 전달합니다.

Body

실제 메시지 데이터를 담고 있습니다.(Text, Binary Data)

연결 설정(CONNECT, CONNTECTED)

클라이언트는 CONNECT 프레임을 전송해서 서버에 연결을 요청합니다.

서버는 CONNECTED 프레임으로 응답하여 연결이 성공적으로 설정되었음을 알립니다.

메시지 전송(SEND)

클라이언트는 SEND 프레임을 사용하여 특정 주제(destination)에 메시지를 발행합니다.

destination 헤더는 메시지가 전송될 주제를 지정합니다.

구독 및 구독 취소(SUBSCRIBE, UNSUBSCRIBE)

클라이언트는 SUBSCRIBE 프레임을 사용하여 특정 주제를 구독합니다.

서버는 해당 주제에 발행된 메시지를 해당 주제를 구독한 클라이언트에게 전송합니다.

또한, UNSUBSCRIBE 프레임을 사용하여 구독을 취소할 수 있습니다.

메시지 수신(MESSAGE)

서버는 클라이언트가 구독한 주제에 메시지가 발행되면 MESSAGE 프레임을 클라이언트에게 전송합니다.

메시지 프레임에는 destination, message-id, subscription 등의 헤더가 포함됩니다.

연결 종료(DISCONNECTED)

클라이언트가 DISCONNECTED 프레임을 전송하면 서버는 클라이언트와 웹 소캣을 종료합니다.

STOMP의 Preifx

prefix는 주로 메시지 브로커의 설정과 관련됩니다.

메시지 브로커는 특정 경로를 사용해서 메시지를 라우팅하거나 특정 동작을 수행하는데, STOMP 프로토콜에서는 이러한 경로를 정의하기 위해 prefix를 사용합니다.

대표적으로 /app, /topic, /queue가 존재합니다.

/topic

브로드캐스트의 메시징, 즉 다수의 클라이언트가 같은 주제를 구독하는 pub/sub 모델에서 사용합니다.

여러 클라이언트가 같은 주제를 구독(SUBSCRIBE)하고, 특정 주제에 메시지가 발행(publish)될 때, 해당 주제를 구독한 모든 클라이언트가 메시지를 받습니다.

/app

클라이언트가 서버로 메시지를 전송할 때 사용됩니다.

서버 측에서 메시지를 처리하고, 특정 목적지로 메시지를 라우팅합니다.

즉, topic은 구독 처리하기 위한 경로이고 app은 메시지를 주고 받는데 사용하는 경로라고 생각하면 됩니다.

/queue

point to point 모델에서 사용됩니다.

특정 클라이언트 또는 서버가 Queue를 생성하고, Queue에 전송된 메시지는 단일 수신자가 처리합니다.

즉, 간단하게 일대일 채팅창 같은 것을 구현한다면 이 라우팅을 이용하면 됩니다.

참고 자료

https://hwanheejung.tistory.com/42

https://hdbstn3055.tistory.com/295