[Observables] 개념

RxSwift를 살펴보는 중인데 너무 헷갈려서 정리를 하면서 알아가보려고 합니다. 이번 정리는 문서의 글을 가져오고 또 이해한 바를 풀어쓰다보니 굉장히 줄글이 될 거 같습니다.

 

Observable과 Observer 개념

Observer와 Observable의 관계를 먼저 알아야해서 ReactiveX 문서를 읽어봅니다. 근데 ReactiveX 문서에 가보시면 정말 여러번 쓰여진 문장이 있어요. 

 

ReactiveX에서 Observer는 Observable을 구독한다. Observable이 배출하는 하나 또는 연속된 항목에 Observer는 반응한다.

 

뜻이 잘 이해가 안가네요.. 그래서 다 읽어보고 제일 이해하기 쉽게 쓰여진 문장을 아래 적어보았어요.

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

 

대체 Observer와 Observable이 무엇일까요? 비동기 프로그래밍을 하기 위해서는 먼저 코드 블럭이 실행 결과를 리턴할 때까지 기다릴 필요 없이 계속해서 다음 코드 블럭을 실행해야 하고 이는 한꺼번에 여러 코드를 실행시킬 수 있게 만들어 줘요. 그럼 실행 결과를 리턴해야 할 때까지 기다려주고 리턴이 되면 리턴값을 써서 작업을 해주어야겠죠? 이 때 리턴값을 써서 작업을 해주는 것이 바로 Observer입니다! 그리고 실행결과를 리턴해주는 그 작업이 바로 Observable이에요. 정확하게는 아니지만 그래도 이렇게 비유해주면 이해가 조금 되네요. 이 설명도 문서에 적혀 있는 글을 비슷하게 가져온 말이니 꼭꼭 읽어보시길 추천합니다!!!

 

🔥🔥🔥 ReactiveX 문서 🔥🔥🔥

 

음.. 다시 한 번 생각해서 써보면, Observable은 데이터를 나중에 보내주기 위해 만들어준 객체이고, Observer는 나중에 보내준 데이터를 받고 연산해주는 역할을 하는 거네요. 그래서 비동기적으로 작동하는 코드는 Observable을 만들어서 그 안에 데이터를 처리해주는 매커니즘을 만들고, Observable이 이벤트를 발생시키면 그 때 Observer가 연산을 하는 방식이네요. 그럼 위 글에서 "구독"이라는 표현은 대체 무엇일까요? Observable을 단순히 만든다고 해서 Observable 안에 만들어준 매커니즘이 동작하지 않아요. 이를 "구독"을 해주어야 매커니즘이 작동하고, 데이터를 옵저버에 보내줍니다. 여기서 문서에 나온 예제를 한 번 살펴보겠습니다.

 

기본적인 코드 동작방식은 아래의 흐름과 같아요.

1. 메서드를 호출한다.

2. 메서드가 리턴한 값을 변수에 저장한다.

3. 결과 값을 가진 변수를 통해 필요한 연산을 처리한다.

// 메서드를 호출하고, 리턴 값을 `returnVal`에 할당한다
returnVal = someMethod(itsParameters);
// returnVal을 통해 필요한 작업을 진행한다
titleLabel.text = returnVal // 예시

Reactive 에선 병력적으로 코드를 배치하기 때문에 아래와 같아요.

1. 비동기 메소드 호출로 결과를 리턴받고 필요한 동작을 처리하는 메서드를 정의한다: 이 메서드는 옵저버의 일부가 된다.

2. Observable로 비동기 호출을 정의한다.

3. 구독을 통해 옵저버를 Observable 객체에 연결 시킨다(또한, 동시에 Observable의 동작을 초기화 한다).

4. 필요한 코드를 계속 구현한다; 메서드 호출로 결과가 리턴될 때마다, 옵저버의 메서드는 리턴 값 또는 (Observable이 배출하는)항목들을 사용해서 연산을 시작한다.

// 옵저버의 onNext 핸들러를 정의한다, 하지만 실행하지는 않는다
// (이 예제에서는, 단순히 옵저버에 onNext 핸들러만 구현한다)
def myOnNext = { it -> /* 필요한 연산을 처리한다 */ };
// Observable을 정의하지만, 역시 실행하지는 않는다
def myObservable = someObservable(itsParameters);
// 옵저버가 Observable을 구독한다. 그리고 Observable을 실행한다
myObservable.subscribe(myOnNext);
// 필요한 코드를 구현한다

흐름은 어느정도 이해가 되었어요!!

근데 문제는 옵저버가 Observable을 구독을 해서 연결을 해주었지만 이 연결이 어떻게 되어있는 것인가!! 어떻게 데이터를 넘겨준다는거지?

분명 위의 예제에서는 확인 불가했어요....

 

조금 더 읽어봐야해요. 위의 예제에서 onNext 비스므레한게 있죠? 바로 그것을 이용하네요.

subscribe 메소드를 통해서 옵저버와 Observable을 연결하였고 Observable은 onNext, onCompleted, onError 이 세 가지 메소드를 호출하여 데이터를 넘겨준다고 합니다!! 그리고 옵저버도 이 세 가지 메소드를 이용해서 데이터를 받고 처리해주네요!

 

onNext : Observable은 새로운 항목들을 배출할 때마다 이 메소드를 호출해서 항목을 파라미터로 전달합니다.

onError: Observable은 기대하는 데이터가 생성되지 않았거나 오류가 났을 경우 알리기 위해 이 메소드를 호출합니다.

onCompleted: 에러가 발생하지 않았다면 Observable은 마지막 onNext를 호출한 후 이 메소드를 호출합니다.

 

Observable 계약이라는 조건이 있다는데 이 조건에서 onNext는 0번 이상 호출 가능하지만, onError, onCompleted는 단 한번만 둘 중에 하나만 호출된다고 합니다. 그리고 onNext는 emit된다 onError, onCompleted 호출은 alaram이라고 부른다고합니다..! 끝났거나 에러가 났으니 알람을 주고, 계속 정상적으로 실행되면서 데이터를 넘겨주니 emit이라고 표현하는 것 같은 느낌이네용

 

unsubscribe

음 끝난줄 알았는데 unsubscribing이라는 것이 남았습니다.ㅠㅠ

그래도 이 개념은 간단해요!! Observable을 생성했고 옵저버가 이를 구독하여 연결까지 시켜놓았지만, 더이상 옵저버가 구독을 원하지 않을 때! 이 메소드를 이용해서 구독해지를 할 수 있다고 합니다. 그럼 Observable은 연결된 옵저버가 더이상 없으니 새로운 항목들을 방출하지 않아요!

 

 

여기까지가 Observable이 무엇이고 어떤식으로 동작하는지 문서를 읽고 해석해 보았지만 역시나 직접 안 써보면 모르겠네영.. 그래서 create 메소드를 이용해서 간단한 Observable을 만들고 사용해보도록 할게요.

 

 

Create operator

Observable을 생성해주는 연산자입니다! 

바로 아래와 같이요!

사용은 아래와 같아요.

func showLabelText(_ url: String) -> Observable<String> {
        return Observable.create { emitter in
            emitter.onNext("Hello")
            emitter.onCompleted()
            return Disposables.create()
        }
}

먼저, create메소드를 이용해서 Observable을 생성해줘요. 그리고 emitter는 Observer로, onNext, onCompleted, onError 세가지 메소드를 가지고 있어요. 방금 만든 Observable안에서 데이터를 처리한 후, 옵저버한테 처리한 값을 넘겨주는 코드입니다. 물론 데이터 처리라기 보다는 단순히 "Hello"를 넘겨주고 있지만요 ㅎㅎ

showLabelText(MY_API)
        .subscribe { event in
            switch event {
            case .next(let t):
                DispatchQueue.main.async {
                    self.label.text = t
                }
            case .error(let e):
                print(e)
                break
            case .completed:
                break
            }
        }
        .dispose() // 이건 unsubscribe개념으로 Observable과 Observer의 연결을 끊어준다.

showLabelText() 함수는 Observable을 반환하고 이를 subscribe해줍니다. Observable이 항목을 방출할 때 에러가 없다면 next로 넘어와서 그 이후의 작업을 처리해줍니다.

이 코드에선 dispose() 코드를 통해 구독해놓은 Observer의 연결을 끊어주지만 만일 Observable안에서 처리가 길어지는데 바로 연결을 끊어주면, 코드가 동작이 되지 않겠죠?!

 

아래는 그 실험을 한번 해봤어용

    func showLabelText(_ url: String) -> Observable<String> {
        return Observable.create { emitter in
            let url = URL(string: url)!
            let task = URLSession.shared.dataTask(with: url) { (data, _, err) in
                guard err == nil else {
                    emitter.onError(err!)
                    return
                }
                if let d = data, let json = String(data: d, encoding: .utf8) {
                        emitter.onNext(json)
                }
                emitter.onCompleted()
            }
            task.resume()
            
            return Disposables.create() {
                task.cancel()
            }
        }
    }

위 코드 설명을 해보자면, 서버에서 json파일을 받아오는 코드예요. URLSession task를 만들어 네트워크 에러가 난다면 onError메소드 호출, data가 무사히 들어온다면 onNext호출, data를 무사히 넘겼다면 onCompleted. 그리고는 Dispose를 반환해요.

    private func updateUI() {
        showLabelText(MY_API)
            .subscribe(onNext: { t in
                DispatchQueue.main.async {
                    self.label.text = t
                }
                print("next")
            }, onError: { e in
                print(e)
            }, onCompleted: {
                print("done")
            }, onDisposed: {
                print("disposed")
            })
            .disposed(by: disposeBag)
    }

이제 이 Observable을 생성하고 구독하여 옵저버를 통해 값을 넘겨받을 준비가 완료되었어요. 지금은 disposeBag이라는 개념을 통해 코드가 아주 잘 작동되지만 만약 이게 아래와 같으면 어떻게 될까요?

            .dispose()

Observable과의 연결이 바로 끊겨버려서 서버에서 데이터를 받고 항목을 방출해도 받지를 못하게 됩니다!! 

 

그럼 보너스로 여기서 disposeBag의 역할도 알 수 있는데 이 VC 객체가 Stack(?)그니까 메모리에서 해제될 때 한꺼번에 연결을 끊어주는 역할을 해주는 객체입니다~~ 

 

이 글을 읽고 조금은 이해가 되었으면 좋겠네요! 이제 rxswift의 operator들을 하나씩 둘러봐야겠습니다.