Observable

Observable

 

 

 

1️⃣ Observable

 

• Observable은 observer가 구독(subscribe)하는 대상이다.
• Observable이 아이템을 방출하면 observer가 반응(react)한다.
• 이 패턴은 동시성 연산(concurrent operation)을 편하게 해준다.
          ┌ 블록 상태에서 Observable이 아이템을 방출하길 기다릴 필요가 없다.
          └ 대신, observer가 보초처럼 서서 적절하게 반응(react)할 준비를 하고 있다.

 

 

2️⃣ 배경

 

• ReactiveX에서는 Observer에 의해 명령어들이 병렬(parallel)로 실행되고, 결과값이 임의 순서로 나중에 반환된다.
• 데이터를 어떻게 가져오고 변환할지에 대한 매커니즘을 Observable 형태로 정의를 한다. 그 다음, 이 Observable을 observer가 구독함으로써 이 매커니즘이 동작을 시작하고, Observable이 아이템을 방출할 때마다 observer는 정의된 매커니즘대로 동작한다.
• 이러한 방식의 장점은, 서로 의존성이 없는 작업들을 처리해야 할 때, 작업들이 순서대로 하나하나 끝나길 기다릴 필요 없이, 한 번에 여러 작업들을 동시에 시작할 수 있다는 점이다.

 

 

 

3️⃣ Method Call vs Asynchronous Model

 

• Method Call

 

	// make the call, assign its return value to `returnVal`
	returnVal = someMethod(itsParameters);
	// do something useful with returnVal

view raw MethodCall hosted with ❤ by GitHub

① 메서드를 호출한다.

② 변수에 메서드 반환 값을 저장한다.

③ 변수 값으로 다른 유용한 일을 한다.

 

 

• Asynchronous Model

 

	// defines, but does not invoke, the Subscriber's onNext handler
	// (in this example, the observer is very simple and has only an onNext handler)
	def myOnNext = { it -> do something useful with it };
	// defines, but does not invoke, the Observable
	def myObservable = someObservable(itsParameters);
	// subscribes the Subscriber to the Observable, and invokes the Observable
	myObservable.subscribe(myOnNext);
	// go on about my business

view raw AsynchronousModel hosted with ❤ by GitHub

① 비동기 호출에서 반환되는 값으로 할 일을 메서드로 정의한다. (이 메서드는 observer의 일부이다.)

② 비동기 호출을 Observable 형태로 정의한다.

③ Observable에 observer를 구독시킨다. (이는 Observable 동작을 시작시킨다.)

④ 다른 할 일을 한다. (Observable이 아이템을 방출할 때마다 observer의 메서드가 아이템을 알아서 처리한다.)

 

 

 

4️⃣ onNext, onComplete, onError

 

• observer라면 이 세가지 메서드를 구현한다.
• Observable은 특정 상황에 observer의 메서드를 호출한다.
• onNext는 방출(emission)이다.
          ① onNext : Observable이 아이템을 방출했을 때 호출한다. 메서드 파라미터로 방출된 아이템을 전달한다.
• onError, onComplete는 알림(notification)이다.
          ② onError : Observable이 예상된 데이터를 생성하는데에 실패했거나 다른 에러를 만났을 때 호출한다.
                               메서드 파라미터로 어떤 에러가 발생했는지를 전달한다.
                               이후에는 onNext나 onComplete는 실행되지 않는다.
          ③ onComplete : Observable이 에러를 발생시키지 않고 onNext를 마지막으로 실행했을 때 호출한다.

 

 

 

5️⃣ 구독 취소 (Unsubscribing)

 

• ReactiveX의 특화된 observer interface인 Subscriber에는 unsubscribe 메서드를 구현한다.
• Observable에 구독하는 observer가 없다면, Observable은 새로운 아이템 방출을 멈출지 선택할 수 있다.
• 구독 취소는 연산자(operator)에 전파(cascade back)되면서 아이템 방출을 막는다.
          ┌ 이는 동시에 발생된다고 보장되진 않는다.
          └ 또한, Observable에 구독하는 observer가 없더라도 잠깐동안은 아이템을 방출할 수도 있다.

 

 

 

6️⃣ "Hot" & "Cold" Observables

 

• Hot : Observable이 생성되자마자 방출을 시작한다.
          └ observer는 언제나 중간부터 아이템을 받을 수 있다.
• Cold : Observable이 구독되었을 때 방출을 시작한다.
          └ observer는 처음부터 아이템을 받을 수 있다.
• Connectable Observable : observer 유무와 상관없이 Connet라는 메서드가 실행되기 전까지는 아이템을 방출하지 않는다.

 

 

 

7️⃣ Observable 연산을 통한 조합

 

• Observables & observers는 표준 옵저버 패턴의 약간의 확장에 불과하다.
• ReactiveX의 강력함은 연산자(operator)에서 제공한다.
• 체이닝도 가능하다.

 

 

 

---

​

 

RxJS

 

• 비동기 코드가 많아지면 그만큼 제어의 흐름이 복잡하게 얽혀 코드를 예측하기가 어려워진다.
• RxJS는 JavaScript의 비동기 프로그래밍의 문제를 해결하는데 도움을 준다.
• RxJS는 Observer 패턴을 적용한 Observable이라는 객체를 중심으로 동작한다.

 

 

 

1️⃣ RxJS Observable

 

• Observable은 특정 객체를 관찰하는 Observer에게 여러 이벤트나 값을 보내는 역할을 한다.

 

	Observable.create(
	(observer) => {
	try {
	observer.next('item');
	} catch(e) {
	observer.error(e);
	} finally {
	observer.complete();
	}
	}).subscribe(
	(x) => console.log(x),
	(err) => console.error(err),
	() => console.log('complete')
	)

view raw RxJsObservable.js hosted with ❤ by GitHub

└ 위와 마찬가지로 observer에는 3가지 메서드가 존재한다.

        ┌ ① next : Observable 구독자에게 데이터를 전달한다.

        ├ ② complete : Observable 구독자에게 완료 되었음을 알린다. next는 더 이상 데이터를 전달하지 않는다.

        └ ③ error : Observable 구독자에게 에러를 전달한다. 이후에 next 및 complete 이벤트가 발생하지 않는다.

 

 

 

2️⃣ RxJS Observable Lifecycle

 

• 생성
          ┌ Observable.create()
          └ 생성시점에는 어떠한 이벤트도 발생되지 않는다.
• 구독
          ┌ Observable.subscribe()
          └ 구독시점에 이벤트를 구독할 수 있다.
• 실행
          ┌ observer.next()
          └ 실행시점에 이벤트를 구독하고 있는 대상에게 값을 전달한다.
• 구독 해제
          ┌ observer.complete()
          ├ Observable.unsubscribe()
          └ 구독 해제 시점에 구독하고 있는 모든 대상의 구독을 종료한다.

 

 

 

3️⃣ RxJS Subject

 

• RxJS에서는 Subject를 지원한다.
• Subject는 여러 Observer에서 구독이 가능하며, Subject에서 전달하는 값을 Observer들이 전달받는다.

 

	const subject = new Subject();
	subject.subscribe(
	(item) => console.log('a', item),
	(err) => console.error(err),
	() => console.log('a complete'),
	);
	subject.subscribe(
	(item) => console.log('b', item),
	(err) => console.error(err),
	() => console.log('b complete'),
	);
	subject.next('item');
	subject.complete();
	// a item
	// b item
	// a complete
	// b complete

view raw RxJsSubject.js hosted with ❤ by GitHub

 

 

 

4️⃣ RxJS Scheduler

 

• RxJS에서는 Scheduler를 지원한다.
• 자바스크립트는 싱글스레드, 이벤트 루프로 동작하기 때문에, RxJS에서의 Scheduler는 이벤트 루프에 어떤 순서로 처리될지로 구현되었다.
• 스케줄러에는 3가지 종류가 존재한다.
          ① AsyncScheduler
              ┌ setTimeout과 비슷하다.
              └ asyncScheduler.schedule(() => console.log("async"), 2000);
          ② AsapScheduler
              ┌ asapScheduler.schedule(() => console.log("asap"));
              └ 다음 이벤트 루프에서 실행된다.
          ③ QueueScheduler
              └ 스케줄러에 전달된 state를 처리한다.

 

	queueScheduler.schedule(function (x) {
	if(state) { this.schedule(state-1) }
	console.log(state);
	}, 0, 3)
	// 3
	// 2
	// 1
	// 0

view raw RxJsScheduler.js hosted with ❤ by GitHub