RxSwift(2)-Subject

1. Subject란?

RxSwift에서 Subject는 Observable과 Observer의 특성을 모두 갖춘 하이브리드 요소입니다. Observable처럼 값을 방출할 수 있으며, 동시에 Observer처럼 다른 Observable을 구독할 수도 있습니다.

흠... 그니까 양방향이라고 생각하면된다. 

• Observer로서: onNext(_:), onError(_:), onCompleted() 메서드를 호출하여 값을 수신
• Observable로서: 구독자들에게 값을 방출하고 이벤트 전달

Subject가 필요한 이유?

1. 이벤트 브릿징: 명령형코드와 반응형 코드 사이의 연결고리(너와 나의 연결 고리)

2. 멀티캐스팅: 하나의 Observable 실행 결과를 여러 구독자에게 공유

3. 상태 관리: UI상태나 앱 상태를 반응형으로 관리

4. 이벤트 변환: 다양한 이벤트 소스를 일관된 Observable스트림으로 통합

자 그러면 이제 4가지 타입의 Subject를 보자.

2. PublishSubject

구독된 순간 새로운 이벤트 수신을 알리고 싶을 때 용이합니다.

즉, 구독 이전에 발생한 이벤트는 모두 무시, 빈상태로 시작되며 초기값 없기에 -> 실시간 이벤트 처리에 적합합니다. (버튼 탭,)

3. BehaviorSubject

초기값을 가지며 신규 구독자에게 최신값을 전달하는 특성이 있다.

- 항상 초기값으로 시작하고, 구독자는 구독즉시 가장 최근에 방출된 값을 받는다. -> 상태유지, 최신 값 제공이 필요한 경우 적합

// 초기값 "초기 상태"로 BehaviorSubject 생성
let behaviorSubject = BehaviorSubject<String>(value: "초기 상태")

// 값 변경 전 첫 번째 구독
let subscription1 = behaviorSubject
    .subscribe(onNext: { value in
        print("첫 번째 구독자: \(value)")
    })
    .disposed(by: disposeBag)
// 출력: 첫 번째 구독자: 초기 상태

// 값 변경
behaviorSubject.onNext("업데이트된 상태")
// 출력: 첫 번째 구독자: 업데이트된 상태

// 값 변경 후 두 번째 구독
let subscription2 = behaviorSubject
    .subscribe(onNext: { value in
        print("두 번째 구독자: \(value)")
    })
    .disposed(by: disposeBag)
// 출력: 두 번째 구독자: 업데이트된 상태 (가장 최근 값 즉시 수신)

// 현재 값 직접 접근
do {
    let currentValue = try behaviorSubject.value()
    print("현재 값: \(currentValue)")
} catch {
    print("에러 발생: \(error)")
}
// 출력: 현재 값: 업데이트된 상태

실무에서는 처음에 상태를 넣어두고 해당 상태를 업데이트함에따라 UI가 달라지는 경우 사용하면 좋을꺼 같다.

// 네트워크 연결 상태 관리
enum ConnectionState {
    case disconnected
    case connecting
    case connected
}

let connectionState = BehaviorSubject<ConnectionState>(value: .disconnected)

// 다른 컴포넌트에서
connectionState
    .subscribe(onNext: { state in
        switch state {
        case .disconnected:
            self.showOfflineUI()
        case .connecting:
            self.showConnectingIndicator()
        case .connected:
            self.showOnlineUI()
        }
    })
    .disposed(by: disposeBag)

// 상태 업데이트
func connectToServer() {
    connectionState.onNext(.connecting)
    
    apiClient.connect()
        .subscribe(onSuccess: { _ in
            connectionState.onNext(.connected)
        }, onError: { _ in
            connectionState.onNext(.disconnected)
        })
        .disposed(by: disposeBag)
}

value()메서드는 동기적으로 현재 값에 접근할 수 있다.

4. ReplaySubject

지정된 크기의 버퍼를 유지하며, 새 구독자에게 버퍼에 있는 최근 이벤트들!!!을 전달합니다.

생성 시 지정된 버퍼 크기만큼의 최근 이벤트들을 저장하고 새 구독자는 구독 시점에 버퍼에 저장된 이벤트들을 즉시 수신합니다.

-> 과거 이벤트의 재생이 필요한 경우(알람 히스토리??)

// 채팅 메시지 히스토리 관리
let messageHistory = ReplaySubject<Message>.create(bufferSize: 20)

// 새 메시지 수신 시
func receiveMessage(_ message: Message) {
    messageHistory.onNext(message)
}

// UI에 연결
func configureUI() {
    messageHistory
        .subscribe(onNext: { [weak self] message in
            self?.addMessageToUI(message)
        })
        .disposed(by: disposeBag)
}

// 새 화면에 진입할 때 - 자동으로 최근 20개 메시지를 받음
func viewDidLoad() {
    configureUI()
}

 

5. AsyncSubject

: completed event가 전달되기 전까지 어떠한 이벤트도 방출하지 않습니다. completed이벤트가 전달되면 가장 최근에 전달된 값을 방출합니다.

3가지와 다른게 Observer가 subject구독하는 즉시 바로 이벤트를 전달받을 수 있다면 이것은 완료시점이 중요한 subject인거같다

-> 비동기 계산 결과가 필요할때 유용( 파일 다운로드완료시 결과전달할때?)

// 파일 다운로드 완료 후 전체 데이터 처리
func downloadLargeFile() -> Observable<Data> {
    let asyncSubject = AsyncSubject<Data>()
    
    let downloadTask = URLSession.shared.dataTask(with: largeFileURL) { data, response, error in
        if let error = error {
            asyncSubject.onError(error)
            return
        }
        
        guard let data = data else {
            asyncSubject.onError(DownloadError.noData)
            return
        }
        
        // 다운로드 진행 중에는 중간 데이터를 보내지 않음
        asyncSubject.onNext(data)
        asyncSubject.onCompleted()  // 완료 시점에 전체 데이터 전달
    }
    
    downloadTask.resume()
    return asyncSubject.asObservable()
}

 

6.주의해야할점

1. 목적에 맞는 Subject 선택: 단순 이벤트위해서는 PublishSubject, 상태관리는 BehaviorSubject, 히스토리 ReplaySubject, 최종 결과만 AsyncSubject

2. Subject 접근 제한: 클래스 내부에서는 Subject를 사용하되, 외부에서는 Observable 노출하는것이 안전하다.

3. 메모리 관리: 순환 참조를 방지하기 위해 [weak self]

4. 스레드 안정성 고려: 동시성 문제 고려

개인적인 회고:

PublishSubject는 PassthorughtSubject와 유사하다고 생각한다.

BehaviorSubject는 CurrentValueSubject?

 

참고자료

https://github.com/fimuxd/RxSwift/blob/master/Lectures/03_Subjects/Ch3.%20Subjects.md