Redux-saga로 웹 소켓 통신 구현

Redux-Saga

redux-saga는 redux의 비동기 작업을 구현할 수 있도록 도와주는 library이다.

redux 공식문서에서는 다음과 같이 소개하고 있다.

redux-saga is a library that aims to make application side effects easier to manage, more efficient
to execute, easy to test, better at handling failures.

redux-saga는 side effect(실행중인 현재 함수의 범위를 벗어난 것에 영향을 미치는 모든 것, server와의 
통신과 같은 비동기 작업, 인증 서비스 호출, 브라우저 캐시 액세스 등의 작업)를 보다 쉽게
관리/실행/테스트를 할 수 있도록 도와주는 라이브러리 입니다.

 

redux-saga는 기본적으로 ES6의 Generators를 사용하여 구현한다.

Generator function은 간단하게 '*'키워드를 이용하여 만들 수 있고 "yield", "next" 키워드를 이용하여 function을 사용자가 원하는 시점에 일시 정지, 시작 과 같이 사용할 수 있다.

 

saga에서 사용하는 자주 사용하는 함수가 있다.

1. call 

 - 함수를 실행할때 사용한다.

 - api를 호출할때 사용할 수 있다.

 - ex) call(함수명, 함수에 넘겨줄 인자값1, 함수에 넘겨줄 인자값2,... 함수에 넘겨줄 인자값n)

2. fork

 - call과 같이 함수를 실행할 때 사용한다.

 - call은 동기실행, fork는 비동기 실행

3. put

 - 특정 action을 dispatch할 때 사용

 - ex) put({type: 'ACTION1', payload: action.payload})

4. take

 - 특정 action에  대한 작업을 처리해줌

 - ex) take('ACTION')

5. takeEvery

 - 들어오는 모든 action에 대한 작업을 처리해줌

 - ex) takeEvery('ACTION1', 함수)

6. takeLatest

 - 들어오는 action에서 같은 action이 여러번 요청된다면 가장 마지막 action에 대해서만 동작을 실행

7. delay

 - 설정한 시간(ms) 만큼 delay를 시킨다.

 - delay(1000)

8. all

 - Generator함수를 전부 실행시켜준다.

 - Generator함수를 배열 형태의 인자로 넣어준다.

 - ex) all([generatorFunction1(), generatorFunction1()])

9. select

 - redux의 state를 가져온다

 - ex) const state = select()

 

Redux-saga를 이용한 websocket 구현

이전에 포스팅 한 글에서 업비트의 open api를 이용하여 웹소켓을 구현한적이 있다.

https://kmj24.tistory.com/153

Web Socket, 웹 소켓

업비트와 웹소켓을 연결해놓으니 끊임없이 데이터가 들어온다.(초당 수십~수백개가 들어오는듯...)

이렇게 모든 데이터들을 받아와서 상태를 갱신한다면 렌더링 병목현상이 발생할 수 있다.

이를 방지하기 위해 데이터를 받아오는 방식을 바꿀 필요가 있었다.

일정 시간 딜레이를 두고 데이터를 받아오며, 받아온 데이터의 중복된 내용을 제거할 필요가 있었다.

기본적으로 웹 소켓을 연결했을 때 Push 형태로 작동 된다.

Push방식을 그대로 두면 upbit open api에서 Push한 모든 데이터를 그대로 가져와서 컴포넌트에 뿌려주게 된다.

이를 Pull 형태로 바꾸어야 데이터를 경량화 해야 된다. 중간에 buffer를 만들어 두고 buffer에 데이터가 쌓이면 중복된 데이터를 제거한 후 경량화 된 데이터를 Pull형태로 가져와서 컴포넌트에 뿌려주도록 한다.

redux-saga에서는 이러한 형태를 구현할 수 있도록 도와주는 Channel api를 제공한다.

 

actionChannel

 - 특정 action을 버퍼링 할 수 있다.

 - 만약 여러개의 action을 처리할 때 순차적으로 처리하고 싶을 경우 actionChannel을 사용할 수 있다.

import { take, actionChannel, call, ... } from 'redux-saga/effects'

function* watchRequests() {
  const requestChan = yield actionChannel('ACTION')
  while (true) {
    const { payload } = yield take(requestChan);
    yield call(handleRequest, payload);
  }
}

function* handleRequest(payload) { ... }

1. actionChannel을 생성

2. take함수에 actionChannel을 넣어준다. (take함수에 action을 넣은것 처럼 channel을 넣을 수 있다.)

3. redux-saga는 call이 반환될 때 까지 기다린다. 만약 그때 다른 action이 dispatch된다면 그 action은 channel의 buffer에 저장된다.

4. call이 반환된다면 take는 대기열에 저장된 메시지를 resolve한다.

5. 기본적으로 actionChannel의 버퍼링은 제한이 없지만 argument로 제한할 수 있다

 - ex) const requestChan = yield actionChannel('REQUEST', buffers.sliding(5))

 

eventChannel

 - eventChannel은 redux store가 아닌 외부 이벤트트에 대한 channel을 생성한다.

 - eventChannel의 첫번째 argument는 subscriber 함수이다.

 - subscriber함수는 외부의 이벤트 소스를 초기화 하고 emitter를 실행하여 소스에서 channel로 들어오는 모든 event를 routing한다.

 - eventChannel을 통해 null또는 undefined를 전달하지 않도록 해야 된다.

 - emitter(END)를 호출하면 더이상 다른 메시지가 이 채널로 들어올 수 없다는 것을 의미한다.

import { take, put, call } from 'redux-saga/effects'
import { eventChannel, END } from 'redux-saga'

function countdown(secs) {
  return eventChannel(emitter => {
      const iv = setInterval(() => {
        secs -= 1
        if (secs > 0) {
          emitter(secs)
        } else {
          emitter(END)
        }
      }, 1000);
      return () => {
        clearInterval(iv)
      }
    })
}

export function* saga() {
  const chan = yield call(countdown, value)
  try {    
    while (true) {
      let seconds = yield take(chan)
      console.log(`countdown: ${seconds}`)
    }
  } finally {
    console.log('countdown terminated')
  }
}

위 코드에서 saga는 take(chan)로 message가 channel에 들어가기 전 까지(emitter(secs)가 호출되기 전 까지) 기다린다.

try / finally로 감싼 이유는 eventChannel에서 emiter(END)가 실행되면 try를 빠져나오게 된다. 즉 while문의 무한루프를 빠져나오게 된다. 그리고 finally를 실행한다. 웹 소켓 연결을 예로 들자면 finally 구문에 웹 소켓의 연결을 종료시켜주는 로직을 넣으면 될 것이다.

 기본적으로 eventChannel은 buffering되지 않으며 buffering기능을 구현하려면 eventChannel에 buffer를 argument로 넣어주어야 한다.

 - buffer.none() : 버퍼링 없음, pending된 taker가 없을 경우 새로운 message는 손실된다.

 - buffer.fixed(limit) : 버퍼 크기가 limit까지 고정된다. 오버플로우될 경우 오류를 발생시킨다. limit의 기본값은 10이다.

 - buffer.expanding(initialSize) : initialSize로 버퍼의 크기가 고정되지만, 오버플로우가 발생할 경우 버퍼의 크기는 동적 으로 확장된다.

 - buffer.droping(limit) : 버퍼 크기가 limit까지 고정되며, 오버플로우의 message는 자동으로 삭제된다.

 - buffer.sliding(limit): 버퍼의 크기가 limit까지 고정되며,  Overflow는 새로운 message의 끝에 삽입하고, 가장 오래된 message는 버퍼에서 삭제된다.

 

Channel

어떠한 소스에도 연결되지 않은 채널을 직접 만들 수 있다.

import { channel } from 'redux-saga'
import { take, fork, ... } from 'redux-saga/effects'

function* watchRequests() {
  const chan = yield call(channel)
  
  for (var i = 0; i < 3; i++) {
    yield fork(handleRequest, chan)
  }

  while (true) {
    const {payload} = yield take('REQUEST')
    yield put(chan, payload)
  }
}

function* handleRequest(chan) {
  while (true) {
    const payload = yield take(chan)
  }
}

1. channel을 만든다.

2. 기본적으로 모든 메시지를 버퍼링 하는 채널이 만들어진다.

3. watchRequests함수는 3개의 channel이 만들어진다.

4. 각 REQUEST action에서 saga는 준비될 때 까지 channel에 의해 대기된다.

 

 

Upbit Open API의 Websocket을 Redux-saga의 eventChannel로 처리하기

토이프로젝트로 upbit data를 가져오는 것을 구현하던 중 eventChannel로 처리하였다.

function upbitWebSocketChannel({ws, marketList, reqType} : ReqUpbitSocketParam){
    return eventChannel<Ticker | Trade | Orderbook>(emit => {
        ws.onopen = () => {
            const sendData = JSON.stringify([
                { ticket:"test" },
                { type: reqType, codes:marketList }
            ]);
            ws.send(sendData);
        }
        ws.onmessage = (e: any) => {
            const encode =  new TextDecoder("utf-8");
            const data: Ticker | Trade | Orderbook = JSON.parse(encode.decode(e.data));
            emit(data);
        }
        ws.onerror = (e: any) => {
            ws.close();
            emit(e);
            emit(END);
        }

        const unsubscribe = () => {
            ws.close();
        }
        return unsubscribe;
    }, buffers.expanding(200) || buffers.none());
}

function socketDataFilter(socketData: Ticker | Trade | Orderbook){
  const val: Ticker[] | Trade[] | Orderbook[] = Object.values(socketData);
  const filterData: any = {};

  for(let i = 0; i < val.length; i++){
    if(filterData[val[i].code]){
      filterData[val[i].code] 
        = (filterData[val[i].code].timestamp > val[i].timestamp) ?
          filterData[val[i].code] : val[i];
    }else{
      filterData[val[i].code] = val[i];     
    }
  }
  return filterData;
}

function* getCoinDataSaga(action: ReturnType<typeof getCoinDataAsync.request>){
  try{
    const coinData: EventChannel<Ticker | Trade | Orderbook> = yield call(upbitWebSocketChannel, action.payload);
    while(1){
      const socketData: Ticker | Trade | Orderbook = yield flush(coinData);
      const filterData: Ticker | Trade | Orderbook = yield socketDataFilter(socketData);
      if(Object.keys(filterData).length){
        yield put(getCoinDataAsync.success(filterData));
      }
      yield delay(1000);
    }
  }catch(e: any){
    yield put(getCoinDataAsync.failure(e));
  }finally{
    action.payload.ws.close();
  }
}

 

 

upbitWebSocketChannel함수를 통해 eventChannel을 만든다.

buffer옵션에 buffer.none()과 buffer.expending(200) 옵션을 설정했다. 들어오는 코인 데이터는 200개 조금 안되는 것으로 보여 200으로 설정해두었다.

eventChannel을 실행한다.

flush함수를 통해 eventChannel의 buffer데이터를 가져온다. (flush함수는 buffer의 데이터를 가져오는 saga에서 지원하는 함수)

가져온 데이터를  socketDataFilter함수를 실행하여 중복데이터를 제거한다.

중복된 데이터를 제거하는 방법은 Ticker, Trade, Orderbook에 모두 중복으로 존재하는 timestamp와 code를 이용하여 중복제거를 한다. timestamp를 비교하여 가장 최신값을 남기도록 한다.

중복을 제거한 데이터를 dispatch한다.

위의 동작을 반복하도록 하며 딜레이는 1초로 설정했다.

데이터를 아주 잘 가져온다.

 

참고 : https://redux-saga.js.org/

Redux-Saga - An intuitive Redux side effect manager. | Redux-Saga

https://meetup.toast.com/posts/114

Redux-Saga에서의 WebSocket(socket.io) 이벤트 처리 : NHN Cloud Meetup

https://velog.io/@seongkyun/React-%EC%B5%9C%EC%A0%81%ED%99%94-buffer%EB%A5%BC-%ED%99%9C%EC%9A%A9%ED%95%98%EC%97%AC-%EC%83%81%ED%83%9C-%EA%B0%B1%EC%8B%A0-%EC%A4%84%EC%9D%B4%EA%B8%B0

React 최적화 - buffer를 활용하여 상태 갱신 줄이기

https://uzihoon.com/post/af9b4d60-7d39-11ea-8fbc-1767c42620cf

UZILOG