public class FilterChannelInterceptor implements ChannelInterceptor {

   ...

   @Override
   public Message<?> preSend(Message<?> message, MessageChannel channel) {
       StompHeaderAccessor headerAccessor = StompHeaderAccessor.wrap(message);
       if (StompCommand.CONNECT.equals(headerAccessor.getCommand())) {
           if (!jwtTokenFilter.isJwtValid(Objects.requireNonNull(headerAccessor.getFirstNativeHeader("access-token")))) {
               throw new ResponseStatusException(HttpStatus.UNAUTHORIZED);
           }
       }
       return message;
   }

   @Override
   public void postSend(Message<?> message, MessageChannel channel, boolean sent) {
       StompHeaderAccessor accessor = StompHeaderAccessor.wrap(message);
       switch (accessor.getCommand()) {
           case CONNECT:
               String session_id = accessor.getSessionId();
               String user_id = Objects.requireNonNull(accessor.getFirstNativeHeader("user-id"));
               LoginSessionRequest loginSessionRequest = LoginSessionRequest.builder()
                               .type("login")
                               .session_id(session_id).user_id(user_id).build();
               tcpClientGateway.send(loginSessionRequest.toString());

               List<String> roomList = communityClient.getRoomList(Long.valueOf(user_id));
               StateRequest stateRequest = StateRequest.builder()
                       .type("connect")
                       .userId(user_id)
                       .ids(roomList).build();
               messageSender.signaling(stateTopic,stateRequest);
               break;
           case DISCONNECT:
               String sessionId = accessor.getSessionId();
               LoginSessionRequest logoutSessionRequest = LoginSessionRequest.builder()
                               .type("logout")
                               .session_id(sessionId).build();
               String id = tcpClientGateway.send(logoutSessionRequest.toString());
               
               ...
               
               break;
           default:
               break;
       }
   }
   ...
}

   public class MessageSender {

   ...

   public void sendToDirectChat(String topic, DirectMessage directChat) {
       
       ...
       // Direct Message 전송
       kafkaTemplateForDirectMessage.send(topic, save);
   }

   public void sendToChannelChat(String topic,ChannelMessage channelMessage) {
   
       ...
       // Community Message 전송
       kafkaTemplateForDirectMessage.send(topic,result);
   }
    public void sendToEtcDirectChat(String topic, DirectMessage directChat) {
       switch (directChat.getType()) {
           case "reply": {
               ...
               // Direct Message 답장 기능
               kafkaTemplateForDirectMessage.send(topic,result);
               break;
           }
           case "modify": {
               ...
               // Direct Message 답장 기능
               kafkaTemplateForDirectMessage.send(topic,result);
               break;
           }
           case "delete": {
               ...
               // Direct Message 답장 기능
               kafkaTemplateForDirectMessage.send(topic,result);
               break;
           }
       }
   }
   public void sendToEtcChannelChat(String topic, ChannelMessage channelMessage) {
       switch (channelMessage.getType()) {
           case "reply": {
               ...
               // Community Message 답장 기능
               kafkaTemplateForChannelMessage.send(topic,result);
               break;
           }
           case "modify": {
               ...
               // Community Message 수정 기능
               kafkaTemplateForChannelMessage.send(topic,result);
               break;
           }
           case "delete": {
               ...
               // Community Message 삭제 기능
               kafkaTemplateForChannelMessage.send(topic,result);
               break;
           }
       }
   }

   // 파일 업로드
   public void fileUpload(String topic, FileUploadResponse fileUploadResponse) {
       kafkaTemplateForFileUpload.send(topic,fileUploadResponse);
   }

   // 시그널링 상태
   public void signaling(String topic, StateRequest stateRequest) {
       kafkaTemplateForSignaling.send(topic,stateRequest);
   }