Subject

[simoniful]

Ch.3 Subjects

• 실시간으로 Observable에 새로운 값을 수동으로 추가하고 subscriber에게 방출
• Observable이자 Observer인 녀석

A. 시작하기

let subject = PublishSubject<String>()

subject.onNext("Is anyone listening?")

let subscriptionOne = subject
    .subscribe(onNext: { (string) in
        print(string)
    })

subject.on(.next("1"))
//Print: 1

subject.onNext("2")
//Print: 2

예제

• PublishSubject는 받은 정보를 가능하면 먼저 수정한 다음에 subscriber에게 배포
• PublishSubject는 현재(current)의 subscriber에만 이벤트를 방출한다. 따라서 어떤 정보가 추가되었을 때 구독하지 않았다면 그 값을 얻을 수 없다.
• 따라서 구독 이후에 값을 변경하는 onNext() 메서드를 실행하면 subscribe 내 블록이 실행된다.

B. Subject의 종류

Subject는 .next 이벤트를 받고, 이런 이벤트를 수신할 때마다 subscriber에 방출한다.

• PublishSubject: 빈 상태로 시작하여 새로운 값만을 subscriber에 방출한다.
• BehaviorSubject: 하나의 초기값을 가진 상태로 시작하여, 새로운 subscriber에게 초기값 또는 최신값을 방출한다.
• ReplaySubject: 버퍼를 두고 초기화하며, 버퍼 사이즈 만큼의 값들을 유지하면서 새로운 subscriber에게 방출한다
• Variable → BehaviorRelay: BehaviorSubject를 래핑하고, 현재의 값을 상태로 보존한다. 가장 최신/초기 값만을 새로운 subscriber에게 방출한다
• AsyncSubject: source Obsevable이 complete된 이후 가장 마지막으로 내보내진 item을 subscriber에게 방출한다

C. PublishSubjects로 작업하기

1. 개념

• PublishSubject는 구독된 순간 새로운 이벤트 수신을 알리고 싶을 때 용이
• 구독을 멈추거나, .completed, .error 이벤트를 통해 Subject가 완전 종료될 때까지 지속
• 아래로 향하는 화살표들은 이벤트의 방출, 위로 향하는 화살표들은 구독을 선언

let disposeBag = DisposeBag()
let subject = PublishSubject<String>()

subject.onNext("1")

let subscriptionOne = subject
    .subscribe(onNext: { (string) in
        print("stream 1)", string)
    })

subject.onNext("2")
// stream 1) 2

let subscriptionTwo = subject
    .subscribe({ (event) in
        print("stream 2)", event.element ?? event)
    })

subject.onNext("3")
// stream 1) 3
// stream 2) 3

subscriptionOne.dispose()
// stream 1 구독 취소, 이벤트 방출 정지

subject.onNext("4")
// stream 2) 4

subject.onCompleted()
// stream 2) completed

subject.onNext("5")
// 완전종료 이 후, 이벤트 방출 X

subscriptionTwo.dispose()
// stream 2 구독 취소, 이벤트 방출 정지

let subscriptionThree = subject
    .subscribe {
        print("stream 3)", $0.element ?? $0)
    }
    .disposed(by: disposeBag)

subject.onNext("?")
// stream 3) completed

• subject 자체가 .completed 또는 .error 같은 완전종료 이벤트들을 받으면, 새로운 subscriber에게 더이상 .next이벤트를 방출하지 않을 것으로 예상할 수 있다.
• 하지만 subject는 이러한 종료 이벤트들을 이후 새 subscriber들에게 재방출한다.
• subject가 완전종료된 후 새 subscriber가 생긴다고 다시 subject가 작동하진 않는다. (.next 이벤트는 방출하지 않는다)
• 다만, .completed 이벤트만 방출한다.

2. 사용 예시

• 시간에 민감한 데이터: 경매, 특정 타이머 만료
• 10:01am에 들어온 유저에게, 9:59am에 기존의 유저에게 날렸던 알람 "서두르세요. 경매가 1분 남았습니다." 을 계속 보내는 것은 아주 무의미

D. BehaviorSubjects로 작업하기

1. 개념

• BehaviorSubject는 마지막 .next 이벤트를 새로운 구독자에게 반복한다는 점만 빼면 PublishSubject와 유사
• 하나의 초기값을 가진 상태로 시작
• PublishSubject와는 다르게 직전의 값을 받는다

enum MyError: Error {
    case anError
}

func print<T: CustomStringConvertible>(label: String, event: Event<T>) {
    print(label, event.element ?? event.error ?? event)
}

let subject = BehaviorSubject(value: "Initial value")
let disposeBag = DisposeBag()

subject.onNext("X")

subject
    .subscribe{
        print(label: "1)", event: $0)
    }
    .disposed(by: disposeBag)
// 1) X

subject.onError(MyError.anError)
// 1) Optional(__lldb_expr_47.MyError.anError)

subject
    .subscribe {
        print(label: "2)", event: $0)
    }
    .disposed(by: disposeBag)
// 2) Optional(__lldb_expr_47.MyError.anError)

• 문자열 Initial value → X로 subject의 값이 바뀌고 subscribe 할 때 최신의 값으로 이벤트를 받는다
• onError로 커스텀 에러를 방출하면서 완전 종료 되고, 두 번째 subscribe 에서 해당 MyError.anError 이벤트를 받게 된다.

2. 사용 예시

• BehaviorSubject는 뷰를 가장 최신의 데이터로 미리 채우기에 용이
• 유저 프로필 화면의 컨트롤을 BehaviorSubject에 바인드 할 수 있다. 이렇게 하면 앱이 새로운 데이터를 가져오는 동안 최신 값을 사용하여 화면을 미리 채워놓을 수 있다.

E. ReplaySubjects로 작업하기

• ReplaySubject는 생성시 선택한 특정 크기까지, 방출하는 최신 요소를 일시적으로 캐시하거나 버퍼한다. 그런 다음에 해당 버퍼를 새 구독자에게 방출
• 이러한 버퍼들은 메모리를 가지기에 이미지나 array 같이 메모리를 크게 차지하는 값들을 큰 사이즈의 버퍼로 가지는 것은 메모리에 엄청난 부하를 준다는 것을 유념

enum MyError: Error {
    case anError
}

func print<T: CustomStringConvertible>(label: String, event: Event<T>) {
    print(label, event.element ?? event.error ?? event)
}

let subject = ReplaySubject<String>.create(bufferSize: 2)
let disposeBag = DisposeBag()

subject.onNext("1")
subject.onNext("2")
subject.onNext("3")

subject
    .subscribe {
        print(label: "1)", event: $0)
    }
    .disposed(by: disposeBag)
// 1) 2
// 1) 3

subject
    .subscribe {
        print(label: "2)", event: $0)
    }
    .disposed(by: disposeBag)
// 2) 2
// 2) 3

subject.onNext("4")
// 1) 4
// 2) 4

subject
    .subscribe {
        print(label: "3)", event: $0)
    }
    .disposed(by: disposeBag)
// 3) 3
// 3) 4

subject.onError(MyError.anError)
// 1) Optional(__lldb_expr_54.MyError.anError)
// 2) Optional(__lldb_expr_54.MyError.anError)

subject
    .subscribe {
        print(label: "3)", event: $0)
    }
    .disposed(by: disposeBag)
// 3) 3
// 3) 4
// 3) Optional(__lldb_expr_54.MyError.anError)

• 버퍼사이즈가 2기에 최근 두개의 요소2,3은 각각의 구독자에게 보여진다. 값1은 방출되지 않는다.
• 새로운 next로 값이 추가되면 기존의 subscribe는 새로운 값을 받고, 새로 subscribe 한 부분에선 버퍼 사이즈 만큼 값을 받는다
• error를 통해 완전 종료되었음에도 불구하고 세 번째 subscribe에선 버퍼의 값을 받고 onError에 대한 이벤트를 받은 후 구독이 종료 된다

2. 사용 예시

• 최근 5개의 검색어

F. BehaviorRelay

• Variable이 Depreated 되고 새롭게 등장
• 현재의 값을 상태로 보존한다. 가장 최신/초기 값만을 새로운 subscriber에게 방출
• 완전종료 이벤트를 발생하는 것이 불가능하다, 즉, 끝나지 않게끔 만들수 있다.
• .value를 사용해 현재의 값을 꺼낼 수 있다(읽기 전용)
• value 프로퍼티를 변경하기 위해서 .accept()를 사용하여 추가 해야한다. 즉 onNext(_:)를 쓸 수 없다.
• 메인스레드에서 동작한다는 보장이 없기에 .asDriver()를 활용하여 스케쥴러 관리가 적절, 이 후 UI와 bind

func print<T: CustomStringConvertible>(label: String, event: Event<T>) {
    print(label, event.element ?? event.error ?? event)
}

let relay = BehaviorRelay(value: "Initial value")
let disposeBag = DisposeBag()

relay.asObservable()
    .subscribe {
        print(label: "1)", event: $0)
    }
    .disposed(by: disposeBag)
// 1) Initial value

relay.accept("1")
// 1) 1

relay.asObservable()
     .subscribe {
         print(label: "2)", event: $0)
     }
     .disposed(by: disposeBag)
// 1) 1

relay.accept("2")
// 1) 2
// 2) 2

2. 사용 예시

• Observer, Observable의 이점을 다 가져가면서 UI가 안정적으로 돌아가게 할 수 있다