[Webflux] Chatting App Chapter1 (Vue + Rsocket)

Introduction

이전 글([WebFlux] RSocket Basic)에서 WebFlux에서 RSocket 기초 사용법을 알아보았다.

이번에는 WebFlux + RSocket를 이용한 간단한 Chatting App을 구현해보자.

기본적인 기술 스택은 Server는 WebFlux, Front는 Vue로 구현한다.

아직 RSocket과 WebFlux를 공부하고 있는 단계라 깊은 지식을 가진 상태는 아니지만 지금까지 공부한 내용을 공유하고 싶어 이번 글을 작성한다.

[Webflux] RSocket Basic

 

Server Example

Application.yaml

spring:
  rsocket:
    server:
      port: 6565
      transport: websocket
      mapping-path: /rs

RSocket을 websocket으로 transport 하고 port와 path정보를 설정한다.

즉, ws://localhost:6565/rs로 RSocket의 설정이 셋팅된다.

추가로 RSocket-java는 UDP, TCP, websocket을 지원한다.

 

Message Dto

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;

@Data  
@AllArgsConstructor
@NoArgsConstructor
@ToString
public class Message {
    String username;
    String message;
}

Front와 Sever가 데이터를 주고받기 위한 DTO객체다.

단순한 Chatting App을 위해 데이터도 단순하게 잡자.

 

Controller

import java.util.ArrayList;
import java.util.List;

import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.annotation.ConnectMapping;
import org.springframework.stereotype.Controller;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@Controller
public class RSocketController {

  private final List<RSocketRequester> CLIENTS = new ArrayList<>();

  @ConnectMapping
  public void onConnect(RSocketRequester requester) {
    requester.rsocket()
             .onClose()
             .doFirst(() -> { 
                  CLIENTS.add(requester);
              })
             .doOnError(error -> { 
              })
             .doFinally(consumer -> { 
                CLIENTS.remove(requester);
              })
             .subscribe();
  }

  @MessageMapping("message")
  Mono<Message> message(Message message) {
    this.sendMsg(message);
    return Mono.just(message);
  }

  @MessageMapping("send")
  void sendMsg(Message message) {
    Message msgDto = new Message();
    msgDto.setUsername(message.getUsername());
    msgDto.setMessage(message.getMessage());

    Flux.fromIterable(CLIENTS)
        .doOnNext(ea -> {
          ea.route("")
            .data(msgDto)
            .send()
            .subscribe();
        })
        .subscribe();
  }

}

Server단에 핵심 로직을 담고 있다.

단순하게 하기 위해 Contorller단에 비즈니스 로직을 작성했다.

 

👉 @ConnectMapping

1. Front에서 RSocket을 통해 Server로 연결을 시도하면 처음 만나게 되는 메소드다.
2. 연결 시도한 RSocketRequester의 정보를 알 수 있다.

 

👉 .doFirst(() -> { ... }

1. 연결을 맺은 requester를 CLIENTS List에 담아둔다.

 

👉 .doOnError(error -> { ... }

1. 연결 시 발생 되는 에러를 핸들링 한다.

 

👉 .doFinally(consumer -> { ... }

1. 연결 종료 시 List에서 해당 requester를 삭제 한다.

 

👉 @MessageMapping("message")

1. Front에서 Server로 메세지 전송할 EndPoint다.
2. 즉, Front에서 ws://localhost:6565/rs/message로 보낸는 것 과 같다.
3. RSocket의 FireAndForget방식을 사용해 Front에서 Server메세지를 전송만 할 뿐 Server에서 응답값을 리턴해주지 않게 할려고 했으나 메세지가 정상적으로 받았는지 여부는 알 필요가 있다고 생각해서 RequestResponse방식을 택했다.

 

👉 this.sendMsg(message)

1. 이번 예제에는 Front에서 메세지를 받고 연결된 requester들에게 바로 메세지를 전송해보자.
2. Redis, Kafka를 활용한 Chatting App을 만든다면 훨씬 상용에 가까운 App이 될 것 이다.

 

👉 void sendMsg(Message message) { ... }

1. 전송받은 메세지를 연결된 requester들에게 전송하는 로직이 담긴 메소드다.
2. 만약, 채팅방(Topic)기능이 있다면 requester를 핸들링하여 알맞는 requester에게만 전송하면 될 것 같다.

 

👉 Flux.fromIterable(CLIENTS)

1. CLIENTS는 RSocketRequester를 담고 있는 단순 List 타입이다.
2. 해당 타입을 Flux의 List로 바꾸기 위해 fromIterable()를 사용한다.

 

👉 .doOnNext(ea -> {ea.route("").data(msgDto).send().subscribe();})

1. 실제로 requester에게 메세지를 전송하는 로직이다.

 

Front Example

App.vue

<template>
  <div>
    <h1>Chatting</h1>
    <input type="text" v-model="message">
    <button @click="send">전송</button>
    <ul>
      <li v-for="item in messages" :key="item.id">
        {{ item }}
      </li>
    </ul>
  </div>
</template>

<script>
import {
  RSocketClient
  , JsonSerializer
  , IdentitySerializer
} from "rsocket-core";
import RSocketWebSocketClient from "rsocket-websocket-client";
import { EchoResponder } from "./responder";

export default {
  data() {
    return {
      message: "",
      socket: null,
      messages: [],
      responder: new EchoResponder(this.messageReceiver)
    }
  },
  created() {
    this.connect();
  },
  methods: {
    messageReceiver(payload) {
      this.messages.push(payload.data);
    },
    send() {
      // requestResponse
      this.socket.requestResponse({
        data: {
          username: "Superpil",
          message: this.message
        },
        metadata: String.fromCharCode('message'.length) + 'message',
      }).subscribe({
        onComplete: (com) => {
          console.log('com : ', com);
        },
        onError: error => {
          console.log(error);
        },
        onNext: payload => {
          console.log(payload.data);
        },
        onSubscribe: subscription => {
          console.log("subscription", subscription)
        },
      });
    },
    connect() {
      let client = new RSocketClient({
        serializers: {
          data: JsonSerializer,
          metadata: IdentitySerializer
        },
        setup: {
          // ms btw sending keepalive to server
          keepAlive: 60000,
          // ms timeout if no keepalive response
          lifetime: 180000,
          // format of `data`
          dataMimeType: 'application/json',
          // format of `metadata`
          metadataMimeType: 'message/x.rsocket.routing.v0',
        },
        responder: this.responder,
        transport: new RSocketWebSocketClient({ 
          url: 'ws://localhost:6565/rs' 
        }),
      });
      client.connect().subscribe({
        onComplete: socket => {
          this.socket = socket;
        },
        onError: error => {
          console.log(error);
        },
        onSubscribe: cancel => {
          console.log(cancel)
        }
      });

    },
  },
};
</script>

👉 this.connect()

1. App.vue파일이 로딩되는 시점에 Server와 RSocket 연결을 맺기 위해 vue의 create 라이플사이클을 이용하자.
2. connect의 자세한 로직은 아래에서 설명한다.

 

👉 new RSocketClient({ ... })

1. Front는 RSocketClient객체를 초기화 시켜 client변수에 할당한다.
2. RSocket은 js도 지원하고 있다.

 

👉 responder: this.responder

1. server로 부터 전달되는 메세지 데이터를 핸들링한다.
2. 즉, responder는 누군가 보낸 메세지를 server로 부터 받는 역할을 한다.
3. server로 부터 전달 받기만 하면되니깐 RSocket의 FireAndForget방식으로 responder를 만든다.

 

👉 transport: ...

1. transport는 RSocket을 websocket으로 사용하기 위한 설정이다.
2. 또한 url설정을 server에서 설정한 경로로 바라보게 셋팅한다.

 

👉 client.connect().subscribe({ ... })

1. 초기화를 시킨 RSocketClient을 client에게 할당되고 client의 connect() 사용해서 server와 연결을 시도 한다.

 

👉 send() { ... }

1. 전송 버튼을 클릭하면 socket에 할당된 RSocketClient객체를 가져와 서버에게 requestResponse방식으로 메세지를 전송한다.
2. 메세지 전송이 정상인지를 확인하기 위해 requestResponse방식으로 선택 했다.

 

👉 data: { ... }

1. server로 전달할 메세지 데이터를 작성한다.

 

👉 metadata: String.fromCharCode('message'.length) + 'message',

1. server의 endpoint를 등록한다.

 

Responder

export class EchoResponder {
  constructor(callback) {
    this.callback = callback;
  }
  fireAndForget(payload) {
    this.callback(payload);
  }  
}

RSocketClient를 초기화 할때 responder에 할당되는 responder class다.

 

Conclusion

RSocket, WebFlux를 활용한 간단한 Chatting App를 만들어 봤다.

비록 RSocket으로 데이터를 주고받는 심플한 코드지만 해당 예제 기반으로 Redis, Kafka를 활용하고 DB도 추가하면 쓸만한 Chatting App으로 발전 시킬 수 있다고 생각한다.

 

Example Git Code

Example Code Web

GitHub - pygmalion0909/blog

 

Example Code API

GitHub - pygmalion0909/blog

 

Reference

1. Spring WebFlux WebSocket with Vue.js - Spring Reactive Chatting Example Code
2. Lettuce Reference Guide - redis lettuce 공식문서
3. Spring Data Redis - spring redis 공식문서
4. Building a Scalable Live Stream Chat Service with Spring WebFlux, Redis PubSub, RSocket and Auth0 - spring webflux + rsocket + redis chatting server example
5. GitHub - hantsy/angular-spring-rsocket-sample: Demo for using Angular and Spring RSocket together - spring webflux + rsocket chatting server example git
6. How to handle message sent from server to client with RSocket? - Vue Rsocket Client Server에서 메세지 응답 받기
7. Getting Started With RSocket: Servers Calling Clients - RSocket Client
8. rsocket-js routing fireAndForget to Spring Boot @MessageMapping - RSocket Client JS fireAndForget