RxSwift - Observable

Observable

이벤트 발행의 주체인 Observable은 이벤트를 예의주시하고 있다가 이벤트가 발생하면 해당 이벤트를 Observer에게 전달합니다.

Observable이 이벤트를 확인하면 Observer에게 바로 전달해줘야하기 때문에 Observable과 Observer는 연락을 하고 있는 상태여야 합니다. 이 상태를 “subscribe”, “구독했다” 라고 합니다.

유튜브에서 좋아하는 유튜버를 구독하면 해당 유튜버가 새로운 영상을 올렸을 시에 알림을 받게 됩니다. 이와 비슷하게 Observable을 구독한 Observer는 발생한 이벤트를 받게 됩니다.

Observable 자체는 하나의 사건이 아니고, 사건들의 순서를 가지고 있는 이벤트 발생의 주체입니다. 즉, Observable은 Observable sequence, stream입니다.

Observable이 발행하는 이벤트의 형태는 총 3가지 입니다.

next : 새로운 이벤트가 발생하면 구독자에게 전달

error : 에러 발생

completed : 정상적으로 종료

onNext

Observable을 구독하고 있는 Observer에게 데이터를 전달합니다.

onNext는 Observable의 Stream이 끊기지 않았기 때문에 이벤트를 계속 방출시킬 수 있습니다.

onCompleted

Observable을 구독하고 있는 Observer에게 해당 Observable의 Stream이 성공적으로 끝났음을 알립니다.

onCompleted는 Stream을 끊고 이벤트를 종료시킬 수 있습니다.

onError

Observable을 구독하고 있는 Observer에게 오류가 발생했음을 알리고 Stream을 종료시킵니다.

onError는 이벤트에 오류가 있음을 알고 중간에 Stream을 종료시킵니다.

그럼, 이벤트 발행의 주체인 Observable은 어떻게 생성되는걸까요?

Create Observable

Observable을 생성하는 Operator를 만나봅시다.

just

하나의 항목을 방출하는 Observable을 생성

Observable.just([1, 2, 3])
    .subscribe { event in print(event) }
    .disposed(by: disposedBag)

/*
next([1, 2, 3])
completed
*/
Swift
복사

from

배열에 포함된 요소를 하나씩 순서대로 방출하는 Observable을 생성

Observable.from([1, 2, 3])
    .subscribe{ element in print(element) }
    .disposed(by: disposedBag)
    
/*
next(1)
next(2)
next(3)
completed
*/
Swift
복사

create

직접적인 코드 구현을 통해 Observer 메서드를 호출하여 Observable을 생성

Observable<String>.create { observer in 
    observer.onNext("A")
    observer.onCompleted()
    return Disposables.create()
}.subscribe(
    onNext: { print($0) },
    onError: { print($0) }, 
    onCompleted: { print("Completed") }, 
    onDisposed: { print("Disposed") }
).disposed(by: disposedBag)

/* 
A
Completed 
Disposed
*/
Swift
복사

deferred

Observer가 구독하기 전까지 Observable 생성을 지연했다가, 구독이 시작되면 Observer 별로 새로운 Observable을 생성

.subscribe() 하는 순간 .deferred()가 실행됩니다.

let a = ["a", "a"]
let b = ["b", "b"]
var flag = true 

let observable: Observable<String> = Observable.deferred {
    flag = !flag
    if flag {
        return Observable.from(a)
    } else {
        return Observable.from(b)
    }
}

observable
    .subscribe { print($0) }
    .disposed(by: disposedBag)
/*
a
a
completed
*/

observable
    .subscribe { print($0) }
    .disposed(by: disposedBag)
/*
b
b
completed
*/
Swift
복사

range

연속된 범위의 정수를 발행하는 Observable을 생성

start부터 count 크기 만큼의 값을 가지는 Observable을 생성합니다.

Observable.range(start: 1, count: 10)
    .subscribe { print($0) }
    .disposed(by: disposedBag)

/*
next(1)
next(2)
..
next(10)
*/
Swift
복사

generate

설정한 조건에 맞는 정수를 발행하는 Observable을 생성

•

initialState : 시작 값 (가장 먼저 방출되는 값)

•

condition : 조건이 true일 경우에만 요소가 방출됨 (false일 경우 completed)

•

iterate : 값을 증가시키거나 감소시킬 경우에 사용

Observable.generate(initialState: 0, condition: { $0 <= 10 }, iterate: { $0 + 2 })
    .subscribe { print($0) }
    .disposed(by: disposedBag) 
    
/*
next(0)
next(2)
next(4)
next(6)
next(8)
next(10)
completed
*/
Swift
복사

repeatElement

특정 요소를 반복적으로 방출하는 Observable을 생성

Observable.repeatElement("A")
    //.take(7) // take Operater를 넣으면 7개만 방출하고 종료 
    .subscribe { print($0) }
    .disposed(by: disposedBag) 

/*
next(A)
next(A)
next(A)
...
*/
Swift
복사

interval

지정된 시간에 한 번씩 이벤트를 방출

Observable<Int>.interval(.seconds(3), scheduler: MainScheduler.instance)
    .take(10)
    .subscribe { print($0) }
    .disposed(by: disposedBag)

// 3초마다 0부터 숫자가 증가하게 된다.
/* 
next(0)
next(1)
...
next(9)
completed
*/
Swift
복사

empty

항목을 가지지 않는 Observable을 생성

empty는 .completed 이벤트만 방출합니다. 항목은 없지만 정상적으로 종료됩니다.

Observable<Void>.empty()
    .subscribe { print($0) }
    .disposed(by: disposedBag)
    
/* 
completed
*/
Swift
복사

never

항목을 방출하지 않고 종료하지 않는 Observable을 생성

Observable<Void>.never()
    .subscribe { print($0) }
    .disposed(by: disposedBag)

// never는 complete 호출되지 않음.
Swift
복사

error

항목을 방출하지 않고 오류로 종료되는 Observable을 생성

Observable<Void>.error(NSError(domain: "400", code: 400))
    .subscribe { print($0) }
    .disposed(by: disposedBag)

/*
error(Error Domain=400 Code=400 "(null)")
*/
Swift
복사

Subscribe Observable

Observable을 생성 연산자를 사용해서 만들어도 구독이 되어 있지 않으면 아무런 이벤트를 받을 수 없습니다.

내가 좋아하는 유튜버를 구독하지 않으면 새로운 영상에 대한 아무런 알림을 받을 수 없는 거처럼요. 하지만, 해당 유튜버를 구독하는 순간부터는 새로운 영상에 대한 정보를 받을 수 있습니다.

구독은 “subscribe”만 해주면 됩니다.

만약, 이런 Observable이 있다고 해봅시다. Observable은 배열에서 String 타입의 항목을 하나씩 빼서 전달해줍니다.

let strObservable = Observable.from(["안녕", "하세요", "반갑", "습니다"])
Swift
복사

하지만, 해당 데이터를 받고 싶어하는 Observer가 없기 때문에 아무런 데이터를 받을 수가 없습니다.

그러면, 데이터를 받을 수 있게끔 해당 데이터를 구독하는 코드를 작성해볼게요.

strObservable.subscribe { print($0) }
Swift
복사

이렇게 하면, Observable이 방출하는 데이터를 받을 Observer가 있기 때문에 데이터가 전달됩니다.

/*
next(안녕)
next(하세요)
next(반갑)
next(습니다)
completed
*/
Swift
복사

하지만, 구독하고 끝이 아닙니다.

구독하고 나서 Observable과 관련된 리소스를 그대로 남겨둔다면 나중에 메모리 문제가 발생할 수 있습니다.

그러면, 메모리 문제를 어떻게 해결할 수 있나요?

subscribe 에는 next, error, completed말고도 disposed라는 옵셔널 타입 클로저 인자가 존재합니다.

disposed는 Observable과 관련된 모든 리소스가 제거된 다음에 호출되는 클로저입니다.

Observable 메모리를 해제할 때에 실행시키고 싶은 코드는 disposed 부분에 넣을 수 있습니다.

strObservable.subscribe(onNext: {
    print($0)
}, onError: {_ in
    // Observable error
}, onCompleted: {
    // Observable completed
}, onDisposed: {
    // 메모리 해제 시에 호출하고 싶은 코드
})
Swift
복사

onError나 onCompleted되어서 Stream이 끝나면 onDisposed가 불리게 됩니다.

error, completed 이후에 리소스 관리해줄 필요없이 정상적으로 메모리가 해제되지만 공식 가이드라인에 따르면 이런 경우에도 수동적으로 리소스를 해제해줄 것을 권고하고 있습니다.

수동적인 리소스 해제를 위해서 disposed를 사용해주면 됩니다.

Observable.from([1, 2, 3])
    .subscribe{ element in print(element) }
    .disposed(by: disposedBag)
Swift
복사

disposed 메서드에서 인수로 사용되는 disposedBag은 Disposable 타입을 모두 모아서 한꺼번에 해제합니다. 따라서, Observable을 생성하고 구독한 다음에 바로 dispose 시키면 편리하게 사용할 수 있습니다.

그럼, 구독하고 나서 바로 dispose되는건 아닌가요?

아닙니다. disposeBag에 담겨있는 disposable은 disposeBag이 해제될 때 함께 해제됩니다.

Trait

Trait는 일반적인 Observable보다 좁은 범위의 Observable입니다.

Single

단일 항목이나 오류 방출

Single은 Observable의 한 형태입니다. 특이한 점이 있다면 한 가지 값이나 에러를 방출한다는 점입니다.

값이 success 또는 error 형태로 방출됩니다.

따라서, “성공이냐”, “실패냐”를 따지는 일에 적합한 Observable 입니다.

func getRepo(_ repo: String) -> Single<[String: Any]> {
    return Single<[String: Any]>.create { single in
        let task = URLSession.shared.dataTask(with: URL(string: "https://api.github.com/repos/\(repo)")!) { data, _, error in
            if let error = error {
                single(.error(error))
                return
            }

            guard let data = data,
                  let json = try? JSONSerialization.jsonObject(with: data, options: .mutableLeaves),
                  let result = json as? [String: Any] else {
                single(.error(DataError.cantParseJSON))
                return
            }

            single(.success(result))
        }

        task.resume()

        return Disposables.create { task.cancel() }
    }
}

getRepo("ReactiveX/RxSwift")
    .subscribe(onSuccess: { json in
        print("JSON: ", json)
    }, onError: { error in
        print("Error: ", error)
    })
    .disposed(by: disposeBag)
Swift
복사

Observable은 completed 이벤트를 발행할 수 있지만, Single은 completed 이벤트를 발행할 수 없습니다. 즉, Single의 이벤트인 success 자체가 next, completed 두 개의 성격을 다 포함하고 있기 때문에, completed 이벤트가 발행된 상태에서 이전에 next 이벤트가 들어오지 않으면 에러를 전달하게 됩니다.

case .completed:
	if self.seenValue {
    self.forwardOn(.completed)
	} else {
    self.forwardOn(.error(RxError.noElements)) // 에러 발생
	}
	self.dispose()
}
Swift
복사

Observable은 asSingle()이라는 메서드를 사용해서 Observable 타입을 Single 타입으로 바꿀 수 있습니다. 이 상황에서 이전과 동일하게 completed 이벤트가 발행된다면 그 이전에 next 이벤트가 들어와야 합니다. 그렇지 않으면, 에러를 발생시킵니다.

case .completed:
	if let element = self.element {
    self.forwardOn(element)
    self.forwardOn(.completed)
	}
	else {
    self.forwardOn(.error(RxError.noElements))
	}
	self.dispose()
}
Swift
복사

Single을 사용해야 한다면 반드시 Single로 시작하도록 하고, Observable로 시작한 경우에는 불필요한 에러를 발행할 수 있기 때문에 가급적 중간에 Single로 타입 변경을 하는 일은 지양하는 것이 좋습니다.

Completable

Single과 기능이 유사하지만, value값을 방출하지 않음

어느 항목도 방출하지 않고, 에러나 완료만 가능한 Observable의 변형입니다.

따라서, 일이 제대로 됐는지만 검토할 때 사용하기 좋습니다.

Completable는 Single, Maybe와 다르게 Observable을 Completable로 바꿀 수 없습니다.

func cacheLocally() -> Completable {
    return Completable.create { completable in
       guard success else {
           completable(.error(CacheError.failedCaching))
           return Disposables.create {}
       }

       completable(.completed)
       return Disposables.create {}
    }
}

cacheLocally()
    .subscribe { completable in
        switch completable {
            case .completed:
                print("Completed with no error")
            case .error(let error):
                print("Completed with an error: \(error.localizedDescription)")
        }
    }
    .disposed(by: disposeBag)
Swift
복사

Maybe

Single과 Completable을 섞어 놓은 형태

Maybe는 .success(value), .completed, .error를 방출합니다. 프로세스의 성공, 실패 여부와 더불어 출력된 값도 방출할 수 있습니다.

func generateString() -> Maybe<String> {
    return Maybe<String>.create { maybe in
        maybe(.success("RxSwift"))
        maybe(.completed)
        maybe(.error(error))

        return Disposables.create {}
    }
}

generateString()
    .subscribe(onSuccess: { element in
        print("Completed with element \(element)")
    }, onError: { error in
        print("Completed with an error \(error.localizedDescription)")
    }, onCompleted: {
        print("Completed with no element")
    })
    .disposed(by: disposeBag)
Swift
복사