[RxSwift] 카테고리별 RxSwift 정리

 

GitHub - ReactiveX/RxSwift: Reactive Programming in Swift

ReactiveX - Intro

ReactiveX는 관찰 가능한 시퀀스를 사용하여 비동기 및 이벤트 기반 프로그램을 작성하기 위한 라이브러리입니다.

Observable

Observable은 데이터 흐름을 관리하는 클래스다. 이게 Rx에서 가장 기본.

• 어떤 데이터가 내보낼지 정해져 있음.
• 외부 컨트롤에 의해 동적으로 데이터가 생성될 수 없음

Observable에서 데이터가 생기면 구독하고 있는 곳에 이벤트로 데이터를 넘겨주는 간단한 동작을 한다.

Observable<Int>.just(1) // 하나의 element만 방출함
    .subscribe(onNext:, onError:, onCompleted:, onDisposed:)

Observable에서 데이터를 보내고 구독을 하는 간단한 코드다. 

.subscribe 메서드를 보면 onNext, onError, onCompleted, onDisposed가 존재하는데 이것은 어떤 이벤트를 받았는지에 따라 호출될 함수를 정의해줄 수 있다. 이름만 봐도 직관적으로 어떤 이벤트가 발생할 수 있는지 알 수 있다.

생성된 후 데이터가 넘어가면서 부터 아래와 같은 이벤트가 나오는데, 4 가지 이벤트는 Observable의 생명주기로 불린다. 

 

• onNext 데이터 전달시 
• onError 에러 발생시
• onCompleted 성공시
• onDisposed 해제시 (release)

Observable은 Error, Complete 이벤트 발생 전 까지 Next이벤트로 데이터를 전달한다.  

Error, Complete 이벤트가 발생하면 종료되고, 더 이상 요소들을 전달할 수 없으며, Disposed 이벤트가 마지막으로 송출되고 Observable은 Dispose를 반환한다.

데이터를 처리하기위해 생성하고 구독을 했으니 끝나고 나서는 구독을 해지해야하는데 구독 해지에 사용되는게 바로 Dispose다. 

let disposeBag = DisposeBag()
Observable<Int>.just(1) // 하나의 element만 방출함
    .subscribe(onNext:{print($0)})
//    .dispose()
    .disposed(by: disposeBag)

deinit() { //disposable Release
	disposeBag = DisposeBag()
}

Disposable을 전체적으로 관리하는 방법도 있다. DisposeBag 을 사용하면 생성되는 Disposable을 Dispose 가방에 넣어놓고 관리하는 것이다. 또는  .take(2)을 사용하여 해제도 가능하다. 데이터를 몇번 까지만 받을지 설정하고 다 받으면 Dispose된다.

RxSwift 문서에는 dispose()를 사용하는 것 보다, DisposeBag 또는 takeuntil을 사용하여 관리하는것을 권장하고 있다.

Dispose를 빼먹으면 메모리 누수가 일어난다. 그러니까 절때 잊지말기.

 

Observable 생성

Observable.create { observer in
    observer.onNext(1)
    //observer.on(.next(1))
    
    observer.onCompleted()
    //observer.on(.completed)
    
    return Disposables.create {}
}

기본적인 Observable 생성은 create를 사용해 만들수 있다. 

create외 간단한게 만들어 방출할 수 있는 방법들이 있다.

//----Just----
Observable<Int>
    .just(1) // 하나의 element만 방출함
    .subscribe(onNext: {value in
        print(value)
    })

//----Of1----
Observable<Int>
    .of(1, 2, 3, 4) // elements를 순차적으로 방출
    .subscribe(onNext: {value in
        print(value)
    })

//----Of2----
Observable
    .of([1, 2, 3, 4]) // array 하나만 방출하게 됨.
    .subscribe(onNext: {value in
        print(value)
    })

//----From----
Observable
    .from([1,2,3,4]) // from은 array만 받음. array에 있는 element를 방출
    .subscribe(onNext: {value in
        print(value)
    })
//----range----
Observable.range(start: 1, count: 9) // start~count까지 늘려가면서 방출함.
    .subscribe(onNext: {
        print("2 * \($0) = \(2 * $0)")
    })

Operators

Observable 생성에서 사용했던 create, just ... 등이 Observable의 연산자로 불린다.

연산자의 종류는 엄청나게 많다. 카테고리로는 생성, 결합, 변환, 필터링, 오류처리, 유틸리티, 조건 및 불린, 수학 및 집계, 역압, 연결 등이 있다.

ReactiveX - Operators

 

Swift에서 고차함수로 사용하는 map, filter, reduce, flatMap, compactMap도 RxSwift에 비슷하게 기능을 제공하고 있다.

 

Traits

RxSwift에서는 Single, Driver가 RxSwift의 Traits로 제공되는데, Driver 같은 경우는 RxCocoa 라이브러리에 있다.

Single, Maybe, Completable (RxSwift Traits)

Single

ReactiveX 문서에서는 Single을 아래와 같이 설명하고 있다.

Single은 Obvservable의 한 형태이지만, Observable처럼 존재하지 않는 곳에서부터 무한대까지의 임이의 연속된 값들을 배출하는 것과는 달리, 항상 한 가지 값 또는 오류 알림 둘 중 하나만 배출한다.
이런 이유 때문에, Single을 구독할 때는 Observable을 구독할 때 사용하는 세 개의 메서드(onNext, onError, onCompleted) 대신 다음의 두 메서드만 사용할 수 있다:
onSuccessSingle은 자신이 배출하는 하나의 값을 이 메서드를 통해 전달한다onErrorSingle은 항목을 배출할 수 없을 때 이 메서드를 통해 Throwable 객체를 전달한다
Single은 이 메서드 중 하나만 그리고, 한 번만 호출한다. 메서드가 호출되면 Single의 생명주기는 끝나고 구독도 종료된다.

struct Single<Element> {
    let source: Observable<Element>
}

Single은 일련의 요소를 방출하는 대신 항상 정확히 하나의 요소 또는 오류 방출이 보장되는 Observable의 변형이다.

Single을 사용하는 일반적인 사용 사례 중 하나는 응답이나 오류만 반환할 수 있는 HTTP 요청을 수행하는 것이지만, Single은 무한한 요소 스트림이 아닌 단일 요소만 신경쓰는 모든 경우를 모델링하는 데 사용할 수 있습니다.

생성과 구독은 Observable과 다르지 않다.

//----Single1---- Observable을 생성 후 Single로 캐스팅
Observable<String>.create({ observer -> Disposable in
    observer.onError(TraitsError.single)
    return Disposables.create()
})
.asSingle() //Single로 casting
.subscribe(onSuccess: { str in
    print(str)
}, onFailure: {
    print($0.localizedDescription)
}, onDisposed: {
    print("disposed")
}).disposed(by: disposeBag)

//----Single2---- 
struct somJSON: Decodable { let name: String }
enum JSONError: Error { case decodingError }
let json1 = "{"name":"park"}"
let json2 = "{"my_name":"bark"}"

func decode(json: String) -> Single<somJSON> {
    Single<somJSON>.create { observer -> Disposable in
        guard let data = json.data(using: .utf8),
              let json = try? JSONDecoder().decode(somJSON.self, from: data)
        else {
            observer(.failure(JSONError.decodingError))
            return Disposables.create()
        }
        observer(.success(json))
        return Disposables.create()
    }
}
decode(json: json1)
    .subscribe {
        switch $0 {
        case .success(let json):
            print(json.name)
        case .failure(let error):
            print(error.localizedDescription)
        }
    }.disposed(by: disposeBag)
decode(json: json2)
    .subscribe {
        switch $0 {
        case .success(let json):
            print(json.name)
        case .failure(let error):
            print(error.localizedDescription)
        }
    }.disposed(by: disposeBag)

subscribe할 때 이벤트가 다른것을 확인 할 수 있다. 정확히 하나의 요소 또는 오류만을 방출한다 했던 것 처럼 성공, 실패, 해제에 대한 이벤트만 존재한다.

또한 Single은 Observable의 한 형태이기 때문에 Observable에서 Single로 캐스팅이 가능하고, 그 반대도 가능하다.

asSingle, asObervable 과 같은 Operator를 사용하면 쉽게 캐스팅이 가능하다.

Maybe

Observable<String>.create { observer -> Disposable in
    observer.onError(TraitsError.maybe)
    return Disposables.create()
}.asMaybe()
    .subscribe(onSuccess: {
        print("OnSuccess: \($0)")
    }, onError: {
        print($0.localizedDescription)
    }, onCompleted: {
        print("completed")
    }, onDisposed: {
        print("disposed")
    }).disposed(by: disposeBag)

Maybe도 Single과 같이 Observable의 한 형태라서 서로 캐스팅이 가능하다!

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

이벤트는 Success, Error, Complete 가 존재하며 Success로 데이터가 전달되며 나머지는 동일하다.

Completable

Completable.create { observer -> Disposable in
	observer(.completed)
    // OR
    observer(.error(TraitsError.completable))
    return Disposables.create()
}
.subscribe(onCompleted: {
    print("complete")
}, onError: {
    print("error: \($0)")
}, onDisposed: {
    print("disposed")
}).disposed(by: disposeBag)

Completable은 complete, error 두 가지의 이벤트만 존재하며, 데이터 전달이 아닌 성공 실패만 나타낸다.

Driver, Signal, Binder Relay... (RxCocoa Traits)

UI를 처리할 때 사용하는 타입들, 이글은 RxSwift만 정리하구 RxCocoa도 따로 정리하기.

RxCocoa에는 Driver, Signal, Relay, Binder, ControlProperty, ControlEvent 가 있다.

Subject

4종류의 Subject가 있고, 특정 상황에 맞게 설계되어 있어 동작이 하나씩 다른것이 특징이다.  

• 데이터를 동적으로 생성할 수 있다.
• 양방향성을 가짐. (구독가능, 데이터 주입 가능)

데이터를 동적으로 생성할 수 있는 특징 덕분에 구독 후에 계속해서 이벤트를 보내기 받기가 가능하다.

AysncSubject

AsyncSubject는 소스 Observable로부터 배출된 마지막 값(만)을 배출하고 소스 Observalbe의 동작이 완료된 후에야 동작한다. 만약 오류로 인해 종료될 경우 AsyncSubject는 아무 항목도 배출하지 않고 발생한 오류를 그대로 전달한다.

 

BehaviorSubject

 

옵저버가 BehaviorSubject를 구독하기 시작하면, 옵저버는 소스 Observable이 가장 최근에 방출한 값을 받기 시작하며 그 이후 소스 Observable(들)에 의해 발행된 항목들을 계속 발행한다. (아무 값도 방출되지 않았다면, 처음 값이나 기본 값)오류가 방출되면 구독자에게 아무항목도 배출하지않고 오류만 보내준다.

let behaviorSubject = BehaviorSubject<String>(value: "init value")
behaviorSubject.onNext("first value")
behaviorSubject.subscribe {
    print("sub1:", $0.element ?? $0) //element가 없으면 어떤 이벤트 받았는지!
}.disposed(by: disposeBag)

//behaviorSubject.onError(SubjectError.error1)

behaviorSubject.subscribe {
    print("sub2:", $0.element ?? $0)
}.disposed(by: disposeBag)

let value = try? behaviorSubject.value()
print(value)

/* 출력
sub1: first value
sub2: first value
Optional("first value")
*/

/* 에러 발생시
sub1: first value
sub1: error(error1)
sub2: error(error1)
nil
*/

 

PublishSubject

PublishSubject는 구독 이후에 방출된 항목들만 받아볼 수 있다. 이런 특성 때문에 이미 방출된 항목들은 다시 받을 수 없다는 단점이 있다.

 

ReplaySubject

ReplaySubject는 구독 시점과 관계 없이 방출된 모든 항목들을 구독자가 받을 수 있다.

let replaySubject = ReplaySubject<String>.create(bufferSize: 2)
replaySubject.onNext("1. hello")
replaySubject.onNext("2. world")
replaySubject.onNext("3. !!!!!")

replaySubject.subscribe {
    print("sub1:", $0.element ?? $0)
}.disposed(by: disposeBag)

/*출력
  sub1: 2. world
  sub1: 3. !!!!!
 */

ReplaySubject는 몇 개의 생성자 오버로드를 제공하는데 이를 통해, 재생 버퍼의 크기가 특정 이상으로 증가할 경우, 또는 처음 배출 이후 지정한 시간이 경과할 경우 오래된 항목들을 제거한다.

 

Scheduler

Observable 연산자 체인에 멀티스레딩을 적용하고 싶다면, 특정 스케줄러를 사용해서 연산자(또는 특정 Observable)를 실행하면 된다.
ReactiveX의 일부 Observable 연산자는 사용할 스케줄러를 파라미터로 전달 받기도 하는데, 이 연산자들은 자신이 처리할 연산의 일부 또는 전체를 전달된 스케줄러 내에서 실행한다.

아래 그림처럼 observeOn(특정스케줄러)를 하면 그 아래 observer 부터는 observeOn에 설정된 스케줄러에서 동작하게 되고,

subscribeOn을 사용하면 어디서 호출되었던 가장 상위에 시작하는 observable 가 동작하는 스케줄러를 설정한 스케줄러로 변경한다.