[RxSwift] Observables

ReactiveX - Observable

 

Observer Pattern

이 패턴은 무한한 데이터 스트림에 매력적이다.
데이터 스트림의 끝을 알리는 기능은 없고, 에러 메커니즘도 없다. 
객체 상태 변화를 관찰하는 Observer들의 목록을 객체에 등록하여 상태 변화가 있을 때마다 메서드를 통해
객체가 직접 목록의 각 옵저버에게 통지하도록 하는 디자인 패턴이다.

상태를 가지고 있는 주체 객체와, 상태의 변경을 알아야하는 관찰 객체가 존재하며 이들의 관계는 1:1 혹은 1:N이 될 수 있는데
이 때 객체들 사이에서 다양한 처리를 할 수 있도록 해주는 패턴을 말한다. 
한 객체의 상태가 바뀌면 그 객체에 의존하는 다른 객체들한테 연락이 가고 자동으로 내용이 갱신되는 방식으로 one-to-many 의존성을 정의한다.
Iterator 로 스트림의 끝을 나타내고 Observer pattern의 비동기식 이벤트 실행을 결합하면 아래와 같다.

Observer Pattern 에서는 한 객체의 상태가 바뀌면 그 객체에 의존하는 다른 객체들에게 연락이 가고, 자동으로 내용이 갱신되는 방식으로 일대다(one - to many) 의존성을 정의한다.

• Observer
  - 데이터의 갱신을 통보 받는 인터페이스
  - Subject에서는 Observer 인터페이스의 update 메서드를 호출함으로 Subject의 데이터 변경을 Observer에게 통보한다.
• Subject
  - Observer 객체를 관리하는 요소
• ConcreteSubject
  - 객체에게 알려줘야할 상태를 저장하고, notify 해야할 함수를 만들도록 한다.
• ConcreteObserver
  - noti 를 받았을 때 행동할 로직을 작성한다.

 

장점 
- 느슨한 결합으로 시스템이 유연해지고 객체간의 의존성을 제거할 수 있다.
- pull 방식이 아닌 push 방식을 사용한다.

단점
- Thread safe를 하지 않아 구독을 신청/취소하는 동안 원하는 결과값을 얻기 힘들 수 있다.
- Observer 을 제 때 제거해주지 않으면 메모리 누수가 일어날 수 있다.
- 너무 많이 사용하게 되면 상태 관리가 힘들 수 있다.
- 비동기 방식이기 때문에 이벤트 구독을 원하는 순서대로 받지 못할 수 있다.

 

Observables

Reactivex.io - Observable

옵저버는 Observable 을 구독한다. 
Observable 이 배출하는 하나의 항목 또는 연속된 항목에 옵저버는 반응한다.
Observable은 항목들을 배출하거나 Observable의 메소드 호출을 통해 옵저버에게 알림을 보낸다.

Observable 이 객체를 배출할 때까지 기다릴 필요 없이 어떤 객체가 배출되면, 그 시점을 관찰하는 관찰자를 옵저버 안에 두고 그 관찰자를 통해 배출 알림을 받으면 된다.

ReativeX에서는 "옵저버"에 의해 임의의 순서에 따라 병렬로 실행되고 결과는 나중에 연산된다. 
메서드 호출보다는 "Observable"안에 데이터를 조회하고 변환하는 메커니즘을 정의한 후,
Observable이 이벤트를 발생시키면 옵저버의 관찰자가 그 순간을 감지하고 준비된 연산을 실행시켜 결과를 리턴하는 메커니즘 때문에
Observable을 구독한다고 표현하는 것이 올바르다.

이 접근 방법은 의존관계가 없는 많은 코드들을 실행할 때, 하나의 코드 블럭이 실행 결과를 리턴할 때까지 기다릴 필요 없이 계속해서 다음 코드 블럭을 실행할 수 있기 때문에 한 번에 여러 코드들을 실행시킬 수 있어서 - 결과적으로 전체 코드의 실행 시간은 그 중 실행시간이 긴 시간만큼 밖에 걸리지 않는 큰 장점을 제공한다.

"옵저버"라고 부르는 것이 "구독자"나 "관찰자" 또는 "리액터"로 불려지고 있지만, 통상적으로 이 모델은 리액터 패턴을 말한다.

리액터 패턴이란 하나 이상의 입력에 의해 서비스 핸들러에게 동시에 전달되는 서비스 요청을 처리하기 위한 이벤트 처리 패턴이다. 

 

 

Creating Observables

ReactiveX는 기본적으로 비동기와 병렬로 메소드를 호출하지만, 일반적인 메소드 호출은 이러하다.

// 메소드를 호출하고 리턴 값을 변수에 저장한다.
returnVal = someMethod(itsParameters);

1. 메소드를 호출한다.
2. 메소드가 리턴한 값을 변수에 저장한다.
3. 결과 값을 가진 변수를 통해 필요한 연산을 처리한다.

비동기 모델에서는 아래와 같은 흐름으로 코드가 실행된다.

// 옵저버의 onNext 핸들러를 정의하지만 실행하지 않는다.
def myOnNext = { it -> /* 필요한 연산 처리 */ };

// Observable을 정의하지만 실행하지 않는다.
def myObservable = someObservable(itsParameters);

// 옵저버가 Observable을 구독한다.
// Observable을 실행한다.
myObservable.subscribe(myOnNext);

1. 비동기 메소드 호출로 결과를 리턴받고 필요한 동작을 처리하는 메소드를 정의한다.
   이 메소드는 옵저버의 일부가 된다.
2. Observable로 비동기 호출을 정의한다.
3. 구독을 통해 옵저버를 Observable 객체에 연결시킨다. (또한 동시에 Observable의 동작을 초기화한다.)
4. 필요한 코드를 계속 구현한다.
   메소드 호출로 결과가 리턴될 때마다, 옵저버의 메소드는 리턴 값 또는 Observable이 배출하는 항목들을 사용해서 연산을 시작한다.

 

Subscribe

Subscribe 메소드를 통해 옵저버와 Observable을 연결한다. 
Subscribe를 통해 구현되는 메소드는 아래와 같다.

• onNext
  - Observable은 새로운 항목을 배출할 때마다 이 메소드를 호출한다. 
  - Observable이 배출하는 항목을 파라미터로 전달 받는다.
• onError
  - Observable은 기대하는 데이터가 생성되지 않았거나 다른 이유로 오류가 발생할 경우 오류를 알리기 위해 이 메소드를 호출한다.
  - onError가 호출되면 onNext나 onCompleted는 더 이상 호출되지 않는다.
  - onError 메소드는 오류 정보를 저장하고 있는 객체를 파라미터로 전달받는다.
• onCompleted
  - 오류가 발생하지 않았다면 Observable은 마지막 onNext를 호출한 후 이 메소드를 호출한다.

Observable 에 명시된 조건에 따라, onNext는 0번 이상 호출될 수 있다.
그 후에는 onCompleted 또는 onError 둘 중 하나를 마지막으로 호출한다.
이 둘 모두를 호출하지는 않으며 onNext 호출을 항목의 배출이라고 부르며, onCompleted 혹은 onError 호출을 알림으로 부른다.

 

구독 해지

ReactiveX 구현체 중에는 Subscriber 라는 특별한 옵저버 인터페이스가 있으며 unsubscribe 메소드를 제공한다.
현재 구독중인 Observable 중, 옵저버가 더 이상 구독을 원하지 않는 경우에는 이 메소드를 호출해서 구독을 해지할 수 있다.
만약 더 관심있는 다른 옵저버가 존재하지 않는다면 Observable들은 새로운 항목들을 배출하지 않는다.

unsubscribe 는 연산자 체인을 통해 옵저버가 구독하고 있던 Observable들이 더는 항목을 배출하지 못하도록 체인 안에 연결된 링크들을 끊어버린다.

 

Hot, Cold Observable

Observable에 따라 연속된 항목들을 배출하는 게 다르다.
뜨거운 Observable은 생성되자마자 항목들을 배출하기도 하기에 이 Observable을 구독하는 옵저버들은 어떤 경우에도
항목들이 배출되는 중간부터 Observable을 구독할 수 있다. 

반대로 차가운 Observable은 옵저버가 구독할 때까지 항목을 배출하지 않기 때문에 이 Observable을 구독하는 옵저버는 Observable이 배출하는 항목 전체를 구독할 수 있도록 보장 받는다.
ReactiveX의 구현 코드 중에는 연결 가능한(Connectable) Observable이라 불리는 Observable 객체가 존재한다.
이 Observable은 옵저버의 구독 여부와 상관없이 자신의 Connect 메소드가 호출되기 전까지 항목들을 배출하지 않는다.