[RxJS] RxJS를 위한 지식들 : EventEmitter, Stream, 마블 다이어그램

https://rxjs-dev.firebaseapp.com/

 

java 9에 Reactive Stream이 추가되고 Spring 5에서도 사용되면서 Reactive한 프로그래밍이 주목을 받게 되었다.

 

rxjs가 담당하는 영역 : 비동기 처리, 데이터 전파, 데이터 처리, 그리고 이를 아우르는 "상태 자동 전파"

* rxjs의 본질은 함수형 프로그래밍이 아니다. 관련은 있긴 하지만...

* Rx는 FRP 구현체가 아니다. 명확한 정의보다는 현실의 문제를 Rx로 해결하는데에 집중해보도록하자.

* 굳이 비동기 처리를 Rx로 할 이유는 없다. js는 generator, promise, async/await 등 처리 방법을 이미 만들어 두었다.

 

rxjs는 일관된 방식으로 안전하게 데이터 흐름을 처리하는 라이브러리이다.
- 손찬욱(Naver) https://www.youtube.com/watch?v=2f09-veX4HA

 

모든 어플리케이션은 궁극적으로 상태머신이다. 코딩해보면 상세적인 로직의 차이는 있지만 다음과 같은 과정을 거친다.

입력(input) => 로직(logic) => 어떤 상태를 변경 => 결과 제출(output)

 

input에 대해서는 배열, 또 다른 함수의 값, 혹은 함수 그 자체(값처럼 다뤄지는 언어의 경우), 키보드 입력, 크론잡에 의한 트리거 등등 매우 많은 입력값들이 존재한다.

 

처리하는 logic에 대해서는, 크게 동기(Synchromous), 비동기(Asynchronous)로 나눌 수 있다. 

나는 흔히 동기 : 순차적으로, 처리가 완료될 때까지 기다리는 것 / 비동기 : 기다리지 않고, 별도의 백그라운드에서 작업을 계속하는 것 정도로 이해하고, 설명해왔다. 

 

그런데 rxjs의 관점에서는, 이러한 차이를 상쇄하고, 하나의 방식으로 처리한다. 구체적으론, input을 구별하지 않고 하나의 observable로 바라본다. 기존의 callback, promise, generator, async/await와는 관점 자체가 다른 것이다.

 

* observable : 시간을 인덱스로 둔 컬렉션. 좀 더 직관적인 설명으론 데이터의 컨테이너. 더욱 직관적으로는, 그냥 처리를 할 대상인 input

 

https://www.youtube.com/watch?v=2f09-veX4HA

 

Rxjs에서는 기존의 Observer pattern을 일부 개선해서 아래와 같은 흐름을 가져간다.

이것도 하나도 어려울 것 없다. 관찰값을 관찰하다가 적절한 처리를 하는게 다이다.

React했으면 솔직히 이해 금방 됨 ㅇㅇ

 

https://www.youtube.com/watch?v=2f09-veX4HA

 

 

Observer 패턴과 EventEmitter에 대한 이해

edykim.com/ko/post/events-eventemitter-translation-in-node.js/

Node.js의 Events `EventEmitter` 번역

위의 글에 근거하여 생각해보면,  DOM은 가장 잘 알려진 이벤트 에미터다. Dom에 eventListener를 붙이는 건 너무 당연하잖아요?

node에서 직접 EventEmitter를 만들어볼 수도 있습니다.

import EventEmitter from 'events';

class Calculator extends EventEmitter {}

const calc = new Calculator();

// 이벤트 구독 시작
calc.addListener('add', (a, b) => {
  calc.emit('result', a + b);
});

// 이벤트 emit
calc.emit('add', 2, 3);

 

node에서 EventEmitter가 어떻게 돌아가는 지 확인하고 싶다면  www.huskyhoochu.com/nodejs-eventemitter/

미니멀한 Event Emitter와 pubsub으로 mitt가 있음. 꽤나 유명하다. www.npmjs.com/package/mitt

node 기반 프론트에서 해당 패턴을 사용하기 위한 패키지. React에 붙여 써보자. github.com/Olical/EventEmitter

 

Stream, Observable, Observer 용어

stream은 데이터 전체를 다 읽거나 쓰지 않고도 중간에 어떠한 동작을 해줄 수 있게 해준다.

stream을 다룬 적이 과거에 있었는데 darrengwon.tistory.com/1193 이런 글을 남긴 적이 있다.

정리하자면, 파일 I/O에 관련된 것이라고 정도로 알고 있었다. 정확히 설명하자면 '바이트 스트림'이 맞겠다.

그런데, Rx의 세계에선 "모든 것이 stream"이다. 일반적인 stream의 의미보다 조금 더 확장한 셈이다.

 

 

생산자(Input, Observable). 어렵게 생각하지 마세요. 관찰의 대상이기 때문에 Observable이라고 부를 뿐입니다.

그리고 Observable로부터 emit된 이벤트를 처리하는 녀석을 Observer라고 부릅니다.

너무 직관적이죠? Observer는 Oberservable을 관찰(구독)하고 있다가, 이벤트에 따라 적절한 로직을 처리합니다.

 

Stream으로 생각한다는 것은 무엇인가?에 대한 답도 여기에 있다.

매우, 매우 쉽게 말하자면 이거다

"어떤 값이 바뀌면 해당 값을 참고하고 있는 값도 따라 바뀐다.(상태 자동 전파)"

마치, react에서 state가 바뀌면 해당 state를 참고하고 있는 컴포넌트가 달리 렌더링 되듯 ㅇㅇ

 

 

RxJS introduction

 

Think of RxJS as Lodash for events. (RxJS는 이벤트를 위한 lodash 정도로 생각하세요)
rxjs-dev.firebaseapp.com/guide/overview

 

일반적인 dom에 이벤트를 붙였습니다.

document.addEventListener('click', () => console.log('Clicked!'));

 

RxJS를 적용하면 다음과 같이 됩니다.

import { fromEvent } from 'rxjs';

fromEvent(document, 'click').subscribe(() => console.log('Clicked!'));

 

RxJS왜 좋느냐?는 질문에 공식 문서에는 아래와 같은 이점을 듭니다.

 

1. 순수성

 

아래 같은 경우 외부 변수를 건드렸습니다. 함수가 순수하지 못하죠.

let count = 0;
document.addEventListener('click', () => console.log(`Clicked ${++count} times`));

 

반면 아래와 같은 코드의 경우 순수하게 구성할 수 있습니다.

import { fromEvent } from 'rxjs';
import { scan } from 'rxjs/operators';

fromEvent(document, 'click')
  .pipe(scan(count => count + 1, 0))
  .subscribe(count => console.log(`Clicked ${count} times`));

 

2. 좀 더 매끄러운 흐름 제어

 

간단한 쓰로틀을 바닐라 자바스크립트로 구현한 케이스입니다.

let count = 0;
let rate = 1000;
let lastClick = Date.now() - rate;
document.addEventListener('click', () => {
  if (Date.now() - lastClick >= rate) {
    console.log(`Clicked ${++count} times`);
    lastClick = Date.now();
  }
});

 

이벤트 기반인 RxJS에서는 쓰로틀을 바로 걸어서 더욱 직관적입니다.

import { fromEvent } from 'rxjs';
import { throttleTime, scan } from 'rxjs/operators';

fromEvent(document, 'click')
  .pipe(
    throttleTime(1000),
    scan(count => count + 1, 0)
  )
  .subscribe(count => console.log(`Clicked ${count} times`));

 

3. 값의 변경

 

2번에 따라 오는 거라고 생각하긴하는데, 여튼 그렇답니다.

import { fromEvent } from 'rxjs';
import { throttleTime, map, scan } from 'rxjs/operators';

fromEvent(document, 'click')
  .pipe(
    throttleTime(1000),
    map(event => event.clientX),
    scan((count, clientX) => count + clientX, 0)
  )
  .subscribe(count => console.log(count));

 

마블(marble) 다이어그램

 

rxmarbles.com/

RxMarbles: Interactive diagrams of Rx Observables

Rx를 설명하는데 마블 다이어그램이 사용된다.

 

상단 화살표 : observable

중간 : operator

하단 화살표 : output

| 표시 : 스트림이 종료되었음

🔴 : marble. 단어 그대로 '구슬'임. 이 표에서는 'observable'이 emit하는 데이터를 의미함.

X : Error! 

 

 

 

ref)

www.youtube.com/watch?v=Yh2_xsr8mXQ 

www.youtube.com/watch?v=2f09-veX4HA

www.youtube.com/watch?v=oHF8PEkteq0

www.youtube.com/watch?v=AslncyG8whg