[RxSwift] Observable(2) - Creating Observable

이번 포스팅에는 Observable을 생성하는 Operator 몇 가지에 대해 알아보자!

 

 

Marble Diagram 

Operator를 이해하기 앞서, 이 marble diagram에 대해 알 필요가 있다. 

실제 ReactiveX에 들어가서 여러 Operator들을 보게 되면, 

Operator들의 동작을 Marble Diagram을 통해 표현한다.

 

 

.just()

우선, marble diagram 부터 살펴보자

just 연산자는 marble diagram으로 이해해보면,

데이터가 들어와서(빨간 구슬),

Just Operator를 거치게 되면, 

데이터를(빨간 구슬)을 방출하는 Observable을 생성하고, 바로 Completed된다.

 

let observable = Observable.just(1) //Observable

observable  //Subsribe
    .subscribe(
        onNext: { element in
            print(element)
        },
        onCompleted: {
            print("completed")
        }
    )
    .disposed(by: disposBag
    
/* Prints:
 1
 completed
 */

 

Just Operator는 한 번만 데이터를 방출하고 Completed를 시키는데, 

배열을 방출시키면, 

let observable = Observable.just([1,2,3])

/* Prints:
 [1, 2, 3]
 completed
 */

배열이 그대로 나오게 된다. 

 

 

.of()

of operator는 Marble diagram을 찾을 수없었다. 

of operator는 여러 데이터를 순차적으로 방출할 수 있다. 

 

let observable = Observable.of(1, 2, 3, 4)

/* Prints:
 1
 2
 3
 4
 completed
 */

여러 타입을 동시에 방출시키고 싶은 경우에는 아래와 같이 하면 된다. 

타입을 안 쓰는 경우 타입 추론이 되기 때문에, 타입 제한이 생긴다.

let observable: Observable<Any> = Observable.of(1, 2, 3, "hi")

 

 

.from()

from연산자는 배열을 데이터로 받아 순차적으로 방출하고 completed 한다

 

let observable2 = Observable.from([1,3,4])


/* Prints:
 1
 3
 4
 completed
 */

of 연산자와 유사해 보이지만, of 연산자에 배열을 전달하게 되면, 배열 그대로 나오게 된다. 

let observable2 = Observable.of([1, 2, 3])


/* Prints:
 [1,2,3]
 completed
*/

 

 

 

.range()

range 연산자는 for문과 같이 Observable을 통해 값을 방출할 수 있다. 

range는 Int타입으로 제한이 걸려있기 때문에, 따로 타입 제한이 필요 없다.  

let observable = Observable.range(start: 0, count: 5)
let observable2 = Observable.range(0..<5)  // 이것도 같은 결과 나옴!


/* Prints:
 0
 1
 2
 3
 4
 completed
*/

 

 

.empty()

empty는 의도적으로 아무런 요소(element)를 방출하지 않는 Observable을 생성할 때 사용한다.

해당 경우 요소를 방출하지 않기 때문에, 타입 추론이 불가능하다. 

따라서, 타입을 명시적으로 써주어야 한다.

 

let observable = Observable<Void>.empty()

/* Prints:
 completed
*/

정말 쓸모없어 보이지만, 

즉시 종료할 수 있는 Observable을 리턴 하고 싶을 때, 

의도적으로 0개의 값을 가지는 Observable을 리턴 하고 싶을 때 사용한다.

 

 

.never()

해당 연산자는 아무런 요소만이 아닌, completed이벤트로 방출하지 않는다!

empty와 마찬가지로 타입 추론이 되지 않기 때문에, 명시적으로 타입을 써주어야 한다.

let observable = Observable<Void>.never()

/* Prints:

*/

 

 

.create()

create의 실제 구현을 보면, subscribe라는 escaping closure가 있다

해당 클로져는 AnyObserber를 input으로 Disposable을 리턴한다

public static func create(_ subscribe: @escaping (AnyObserver<Element>) -> Disposable) -> Observable<Element> {
    AnonymousObservable(subscribe)
}

 

그렇기 때문에 실제 구현할 때, 

let observable = Observable<Int>.create { observer in
    return Disposables.create()
}

이런 식으로 Disposable을 return 해주어야 한다.

 

여기서의 데이터 전달은 onNext()로,

error는 onError()로

completed는 onCompleted()를 통해 전달한다. 

 

let observable = Observable<Int>.create { observer in
    observer.onNext(1)
    observer.onCompleted()
    
    return Disposables.create()
}

/* Prints:
 1
 completed
*/

 

이제 RxSwift 포스팅 처음에 구현했던 비동기 코드를 Rx로 바꾸어 보자

//바꾸기 전!
func downLoadJson(_ url: String, _ completionHandler: @escaping (String?) -> (Void)) {
    DispatchQueue.global().async {
        let jsonURL = URL(string: url)!
        let data = try! Data(contentsOf: jsonURL)
        let json = String(data: data, encoding: .utf8)
        
        DispatchQueue.main.async {
            completionHandler(json)
        }
    }
}

//Rx 적용 후
func downLoadJson(_ url: String) -> Observable<String> {
    return Observable.create { observer in
        let jsonURL = URL(string: url)!
        
        let task = URLSession.shared.dataTask(with: jsonURL) { data, _, err in
            if err != nil {
                observer.onError(err!) //error 이벤트 전달
            }
            if let emitData = data {
                let jsonOutput = String(data: emitData, encoding: .utf8)!
                observer.onNext(jsonOutput) // next 이벤트 전달
            }
            observer.onCompleted() // completed 이벤트 전달
        }
        task.resume()
        
        return Disposables.create() {
            task.cancel()
        }
    }
}

 

 

 

.deffered()

deffered 연산자는 Observable이 생성되는 시점을 subscribe 될 때까지 미루는 연산자이다.

public static func deferred(_ observableFactory: @escaping () throws -> Observable<Element>)

deffered의 내부 클로져를 보면, Observable을 리턴 해주는 것을 알 수 있다. 

 

따라서, 이를 구현할 때, 내부에 Observable을 리턴 해주면 된다. 

let deferredOb = Observable<Int>.deferred {
    return Observable.just(1)
}

 

그러면 이 연산자는 어디에 사용할까?

우선 다음 예시를 보자

func someFunction() {
    print("its function!")
}

let observable = Observable.just(someFunction())


/* Prints:
 its function!
*/

실제 subscribe 하지도 않았는데, 해당 함수가 출력이 됐다! 

of, from, just 연산자는 이런 식으로 선언되는 시점에 미리 계산이 되기 때문에, 

이런 경우 deffered 연산자를 사용한다. 

 

var token = false

let observable = Observable.just(token)

token = true 

observable.subscribe(onNext: {print($0)})

/* Prints:
 false
*/

실제 subsribe 시점에서 true로 바꾸어 주었지만, 반영되지 않았다. 

(이역시 of, from, just 연산자에서 발생된다.)

 

따라서, 무언가의 작업을 미루고 싶을 때, 

혹은 subscribe시점의 최신 값을 받아오고 싶을 때, 해당 연산자를 사용한다.

 

var token = false

let observable = Observable.deferred {
    if token {
        return Observable<String>.of("it have token")
    }
    else {
        return Observable<String>.of("it dont have token")
    }
}

token = true
observable.subscribe(onNext: {print($0)})


/* Prints:
 it have token
*/

 

 

Reference

https://github.com/fimuxd/RxSwift/blob/master/Lectures/02_Observables/Ch2.%20Observables.md

https://www.notion.so/Wallaby-RxSwift-72194669a39a4557baa69c672268af38

https://reactivex.io/documentation/operators.html