구독 공유(Sharing Subscription)

import UIKit
import RxSwift

let bag = DisposeBag()

let source = Observable<String>.create { observer in
    let url = URL(string: "https://kxcoding-study.azurewebsites.net/api/string")!
    let task = URLSession.shared.dataTask(with: url) { (data, response, error) in
        if let data = data, let html = String(data: data, encoding: .utf8) {
            observer.onNext(html)
        }
        
        observer.onCompleted()
    }
    task.resume()
    
    return Disposables.create {
        task.cancel()
    }
}
.debug()

source.subscribe().disposed(by: bag)
source.subscribe().disposed(by: bag)
source.subscribe().disposed(by: bag)

===============================================================================
결과

2024-06-05 17:18:19.170: overveiw.playground:48 (__lldb_expr_3) -> subscribed
2024-06-05 17:18:19.363: overveiw.playground:48 (__lldb_expr_3) -> subscribed
2024-06-05 17:18:19.365: overveiw.playground:48 (__lldb_expr_3) -> subscribed
2024-06-05 17:18:23.727: overveiw.playground:48 (__lldb_expr_3) -> Event next(Hello)
2024-06-05 17:18:23.730: overveiw.playground:48 (__lldb_expr_3) -> Event completed
2024-06-05 17:18:23.730: overveiw.playground:48 (__lldb_expr_3) -> isDisposed
2024-06-05 17:18:23.732: overveiw.playground:48 (__lldb_expr_3) -> Event next(Hello)
2024-06-05 17:18:23.732: overveiw.playground:48 (__lldb_expr_3) -> Event completed
2024-06-05 17:18:23.732: overveiw.playground:48 (__lldb_expr_3) -> isDisposed
2024-06-05 17:18:23.733: overveiw.playground:48 (__lldb_expr_3) -> Event next(Hello)
2024-06-05 17:18:23.733: overveiw.playground:48 (__lldb_expr_3) -> Event completed
2024-06-05 17:18:23.733: overveiw.playground:48 (__lldb_expr_3) -> isDisposed

위 코드를 보면 API를 호출하여 문자를 하나 받아오고 있다. 또한 아래 세개의 구독자가 있다.

RxSwift에서는 새로운 구독자가 추가되는 경우 첫번째 구독자가 전달받은 결과를 나머지 구독자가 공유할 수 없다. 새로운 구독자가 추가되면 항상 새로운 시퀀스가 시작된다. 따라서 위 코드를 실행하면 네트워크 요청이 세번 실행되어 리소스가 낭비된다.

이런 문제를 해결하기 위해선 모든 구독자가 하나의 구독을 공유하도록 구현해야 한다.
우선 위 코드에 debug() 뒤에 share 연산자를 추가한다.

import UIKit
import RxSwift

let bag = DisposeBag()

let source = Observable<String>.create { observer in
    let url = URL(string: "https://kxcoding-study.azurewebsites.net/api/string")!
    let task = URLSession.shared.dataTask(with: url) { (data, response, error) in
        if let data = data, let html = String(data: data, encoding: .utf8) {
            observer.onNext(html)
        }
        
        observer.onCompleted()
    }
    task.resume()
    
    return Disposables.create {
        task.cancel()
    }
}
.debug()
.share()

source.subscribe().disposed(by: bag)
source.subscribe().disposed(by: bag)
source.subscribe().disposed(by: bag)

===============================================================================
결과

2024-06-05 17:25:53.627: overveiw.playground:48 (__lldb_expr_7) -> subscribed
2024-06-05 17:25:53.701: overveiw.playground:48 (__lldb_expr_7) -> Event next(Hello)
2024-06-05 17:25:53.704: overveiw.playground:48 (__lldb_expr_7) -> Event completed
2024-06-05 17:25:53.704: overveiw.playground:48 (__lldb_expr_7) -> isDisposed

share연산자를 추가하고 결과를 보면 옵저버블에서 구현한 코드는 한번만 실행된다. 즉 새로운 구독자가 추가되어도 하나의 구독을 공유하게 된다.

1. Multicast Operator

let bag = DisposeBag()
let subject = PublishSubject<Int>()

let source = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).take(5)

source
    .subscribe { print("🔵", $0) }
    .disposed(by: bag)

source
    .delaySubscription(.seconds(3), scheduler: MainScheduler.instance)
    .subscribe { print("🔴", $0) }
    .disposed(by: bag)


source.connect()


=========================================================================================
결과
🔵 next(0)
🔵 next(1)
🔵 next(2)
🔵 next(3)
🔴 next(0)
🔵 next(4)
🔵 completed
🔴 next(1)
🔴 next(2)
🔴 next(3)
🔴 next(4)
🔴 completed

위 코드에서 1초 주기로 5개의 정수를 방출하는 옵저버블이 있고 아래에는 두개의 구독자가 있는데 두번째 구독자는 구독 시점을 3초 지연시키고 있다.

결과를 보면 두 개의 시퀀스가 개별적으로 돌고 있고 서로 공유되지 않고 있다. 이는 RxSwift의 규칙이다.
위 코드에 multicast 연산자를 적용하면 어떻게될까?

우선 multicast 연산자를 보면 ConnectableObservable을 리턴하는데 이는 일반 옵저버블과 시퀀스가 시작되는 시점이 다르다. 즉 구독자가 추가되어도 시퀀스가 바로 시작되지 않고 connect()가 호출되는 시점에 시퀀스가 시작된다.

원본 옵저버블이 방출한 이벤트가 구독자에게 바로 전달되지 않고 첫번째 파라미터인 subject로 전달되고 이 subject가 등록된 모든 구독자에게 이벤트를 전달한다.

ConnectableObservableAdapter는 원본 옵저버블과 subject를 연결해주는 특별한 클래스이다.

multicast 연산자를 위 코드에 추가하면 변수 source에는 일반 옵저버블이 아닌 Connectable 옵저버블이 저장된다.

let bag = DisposeBag()
let subject = PublishSubject<Int>()

let source = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).take(5).multicast(subject)

source
    .subscribe { print("🔵", $0) }
    .disposed(by: bag)

source
    .delaySubscription(.seconds(3), scheduler: MainScheduler.instance)
    .subscribe { print("🔴", $0) }
    .disposed(by: bag)
    
source.connect()

=========================================================================================
결과
🔵 next(0)
🔵 next(1)
🔵 next(2)
🔴 next(2)
🔵 next(3)
🔴 next(3)
🔵 next(4)
🔴 next(4)
🔵 completed
🔴 completed

결과를 보면 모든 구독자가 원본 옵저버블을 공유한다. 구독을 지연한 3초동안 원본 옵저버블이 전달한 두개의 이벤트는 두번째 구독자에게 전달하지 않는다.

또한 connect() 메서드는 Disposable을 리턴하기 때문에 원하는 시점에 dispose()를 호출하여 공유 시퀀스를 중지할 수도 있다.

multicast 연산자는 subject를 직접 만들고 connect()를 직접 호출하는 번거로움이 있어 보통 다른 연산자를 사용한다.

2. Publish Operator

let bag = DisposeBag()

let source = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).take(5).publish()

source
    .subscribe { print("🔵", $0) }
    .disposed(by: bag)

source
    .delaySubscription(.seconds(3), scheduler: MainScheduler.instance)
    .subscribe { print("🔴", $0) }
    .disposed(by: bag)

source.connect()

=========================================================================================
결과
🔵 next(0)
🔵 next(1)
🔵 next(2)
🔴 next(2)
🔵 next(3)
🔴 next(3)
🔵 next(4)
🔴 next(4)
🔵 completed
🔴 completed

publish() 메소드를 보면 multicast 연산자를 호출하고 새로운 PublishSubject를 만들어 파라미터로 전달한다. 또한 multicast연산자와 같이 ConnectableObservable을 리턴한다.

결과를 보면 multicast 연산자와 동일한 것을 볼 수 있다.

3. Replay Operator

import UIKit
import RxSwift


/*:
 # replay, replayAll
 */

let bag = DisposeBag()
let subject = PublishSubject<Int>()
let source = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).take(5).multicast(subject)

source
    .subscribe { print("🔵", $0) }
    .disposed(by: bag)

source
    .delaySubscription(.seconds(3), scheduler: MainScheduler.instance)
    .subscribe { print("🔴", $0) }
    .disposed(by: bag)

source.connect()

=========================================================================================
결과
🔵 next(0)
🔵 next(1)
🔵 next(2)
🔴 next(2)
🔵 next(3)
🔴 next(3)
🔵 next(4)
🔴 next(4)
🔵 completed
🔴 completed

위 코드는 기존에 multicast 코드이다. 첫번째 구독자는 지연없이 구독을 시작하기에 모든 이벤트를 받지만 두번째 구독자는 3초 뒤에 구독을 시작한다. 그래서 구독 전에 subject가 전달한 이벤트는 받지 못한다.

만약 두번째 구독자에게 이전에 전달되었다 이벤트도 함께 전달하려면 어떻게 해야 될까?...
Publish Subject는 별도의 버퍼를 가지고 있지 않아서 불가능하다..

하지만 이를 Replay Subject로 바꾼다면 가능하다. 아래 코드를 보면

let bag = DisposeBag()
let subject = ReplaySubject<Int>.create(bufferSize: 5)
let source = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).take(5).multicast(subject)

source
    .subscribe { print("🔵", $0) }
    .disposed(by: bag)

source
    .delaySubscription(.seconds(3), scheduler: MainScheduler.instance)
    .subscribe { print("🔴", $0) }
    .disposed(by: bag)

source.connect()

=========================================================================================
결과
🔵 next(0)
🔵 next(1)
🔴 next(0)
🔴 next(1)
🔵 next(2)
🔴 next(2)
🔵 next(3)
🔴 next(3)
🔵 next(4)
🔴 next(4)
🔵 completed
🔴 completed

기존에 두번째 구독자가 받지 못했던 0과 1 이벤트도 같이 받는 것을 볼 수 있다.
아래는 이를 좀더 간결하게 사용할 수 있는 replay() 메소드이다.

replay 메소드를 활용한 코드로 다시 고쳐 보면 아래와 같다.

let bag = DisposeBag()

let source = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).take(5).replay(5)

source
    .subscribe { print("🔵", $0) }
    .disposed(by: bag)

source
    .delaySubscription(.seconds(3), scheduler: MainScheduler.instance)
    .subscribe { print("🔴", $0) }
    .disposed(by: bag)

source.connect()

=========================================================================================
결과
🔵 next(0)
🔵 next(1)
🔴 next(0)
🔴 next(1)
🔵 next(2)
🔴 next(2)
🔵 next(3)
🔴 next(3)
🔵 next(4)
🔴 next(4)
🔵 completed
🔴 completed

결과는 같지만 좀더 단순하게 사용할 수 있다. replay 연산자를 사용할 때는 주의해야 할 것이 하나 있다.
바로 버퍼 크기를 신중히 지정해야 하는 것이다. 메모리와 연관되기에 가능한 버퍼 크기를 적게 사용하는 것이 좋다!

4. RefCount Operator

RefCount 연산자는 위에서 보았던 다른 연산자들(ObservableType)과 달리 ConnectableObservableType에 구현되어 있다.

일반 Observable에선 사용할 수 없고 ConnectableObservable에서만 사용할 수 있다.

let bag = DisposeBag()
let source = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).debug().publish()

let observer1 = source
    .subscribe { print("🔵", $0) }

source.connect()

DispatchQueue.main.asyncAfter(deadline: .now() + 3) {
    observer1.dispose()
}

DispatchQueue.main.asyncAfter(deadline: .now() + 7) {
    let observer2 = source.subscribe { print("🔴", $0) }
    
    DispatchQueue.main.asyncAfter(deadline: .now() + 3) {
        observer2.dispose()
    }
}

=========================================================================================
결과
2024-06-08 14:10:34.907: refCount.playground:32 (__lldb_expr_23) -> subscribed
2024-06-08 14:10:35.943: refCount.playground:32 (__lldb_expr_23) -> Event next(0)
🔵 next(0)
2024-06-08 14:10:36.941: refCount.playground:32 (__lldb_expr_23) -> Event next(1)
🔵 next(1)
2024-06-08 14:10:37.941: refCount.playground:32 (__lldb_expr_23) -> Event next(2)
🔵 next(2)
2024-06-08 14:10:38.941: refCount.playground:32 (__lldb_expr_23) -> Event next(3)
2024-06-08 14:10:39.941: refCount.playground:32 (__lldb_expr_23) -> Event next(4)
2024-06-08 14:10:40.941: refCount.playground:32 (__lldb_expr_23) -> Event next(5)
2024-06-08 14:10:41.941: refCount.playground:32 (__lldb_expr_23) -> Event next(6)
2024-06-08 14:10:42.941: refCount.playground:32 (__lldb_expr_23) -> Event next(7)
🔴 next(7)
2024-06-08 14:10:43.941: refCount.playground:32 (__lldb_expr_23) -> Event next(8)
🔴 next(8)
2024-06-08 14:10:44.940: refCount.playground:32 (__lldb_expr_23) -> Event next(9)
🔴 next(9)
2024-06-08 14:10:45.941: refCount.playground:32 (__lldb_expr_23) -> Event next(10)
2024-06-08 14:10:46.940: refCount.playground:32 (__lldb_expr_23) -> Event next(11)
2024-06-08 14:10:47.941: refCount.playground:32 (__lldb_expr_23) -> Event next(12)
2024-06-08 14:10:48.941: refCount.playground:32 (__lldb_expr_23) -> Event next(13)
.
.
.
.계속 이어짐

위 코드에선 구독이 시작되면 첫번째 구독자의 구독이 3초 뒤에 중지되고 7초 뒤에 두번째 구독자의 구독이 시작되었다가 다시 3초 뒤에 구독이 중지된다. 하지만 Connectable Observable은 계속해서 정수를 방출한다.

아래는 refCount() 연산자를 추가한 코드이다.

let bag = DisposeBag()
let source = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).debug().publish().refCount()

let observer1 = source
    .subscribe { print("🔵", $0) }

DispatchQueue.main.asyncAfter(deadline: .now() + 3) {
    observer1.dispose()
}

DispatchQueue.main.asyncAfter(deadline: .now() + 7) {
    let observer2 = source.subscribe { print("🔴", $0) }
    
    DispatchQueue.main.asyncAfter(deadline: .now() + 3) {
        observer2.dispose()
    }
}

=========================================================================================
결과
2024-06-08 14:15:52.546: refCount.playground:32 (__lldb_expr_27) -> subscribed
2024-06-08 14:15:53.580: refCount.playground:32 (__lldb_expr_27) -> Event next(0)
🔵 next(0)
2024-06-08 14:15:54.578: refCount.playground:32 (__lldb_expr_27) -> Event next(1)
🔵 next(1)
2024-06-08 14:15:55.578: refCount.playground:32 (__lldb_expr_27) -> Event next(2)
🔵 next(2)
2024-06-08 14:15:55.754: refCount.playground:32 (__lldb_expr_27) -> isDisposed
2024-06-08 14:15:59.951: refCount.playground:32 (__lldb_expr_27) -> subscribed
2024-06-08 14:16:00.952: refCount.playground:32 (__lldb_expr_27) -> Event next(0)
🔴 next(0)
2024-06-08 14:16:01.952: refCount.playground:32 (__lldb_expr_27) -> Event next(1)
🔴 next(1)
2024-06-08 14:16:02.953: refCount.playground:32 (__lldb_expr_27) -> Event next(2)
🔴 next(2)
2024-06-08 14:16:03.107: refCount.playground:32 (__lldb_expr_27) -> isDisposed

let source = Observable.interval(.seconds(1), scheduler: MainScheduler.instance).debug().publish()에 refCount() 연산자를 붙여 RefCountObservable로 만든다.

우선 RefCountObservable은 내부적으로 connect()메서드를 호출하므로 source.connect()는 삭제하였다.
또한 결과가 아에 다른 것을 볼 수 있는데

첫 번째 구독자가 추가되면 RefCountObservable이 connect()를 호출하고 그러면 Connectable Observable은 Subject를 통홰 모든 구독자에게 이벤트를 전달한다.

3초 뒤에 첫 번째 구독자가 구독을 중지하면 그 시점엔 다른 구독중인 구독자가 없으므로 Connectable 옵저버블은 중지된다. (2024-06-08 14:15:55.754: refCount.playground:32 (__lldb_expr_27) -> isDisposed) 이를 문서에선 disconnect라고 표현한다.

다시 7초 뒤 새로운 구독자가 추가되면 connect 되고 ConnectableObsercvable에서 새로운 시퀀스가 시작된다. 그래서 처음으로 받는 이벤트가 7이 아니라 0이 되는 것이다. (🔴 next(0))

3초 뒤에 중지하면 다시 Connectable Observable이 중지되고 (2024-06-08 14:16:03.107: refCount.playground:32 (__lldb_expr_27) -> isDisposed) 이 로그를 볼 수 있는 것이다.

이전 연산자들은 Connectable Observable을 직접 관리하여야 하지만 connect() 직접 호출과 필요한 시점에 disposed()를 호출하여 리소스를 관리하여야 하지만 refCount() 연산자는 이러한 부분이 자동으로 처리되어 코드가 좀 더 단순해진다.

5. Share Operator

위 코드는 share 연산자 구현 코드이다.

우선 share() 연산자는 두개의 파라미터를 받는다. 첫번째는 replay버퍼의 크기이고 기본값은 0이다. 0을 전달하면 multicast를 호출할 때 PublishSubject를 전달하고 0보다 큰 값이면 ReplaySubject를 전달한다.

multicast연산자를 호출하므로 하나의 subject를 통해 시퀀스를 공유한다.

두 번째 파라미터는 이 Subject의 수명을 결정한다. 기본 값은 whileConnected이다. 아래는 이에 대한 주석이다.

whileConnected의 경우 새로운 구독자가 추가되면 새로운 Subject가 생성된다. Connection이 종료되면 Subject는 사라진다.
Connection마다 새로운 Subject가 생성되기 때문에 Connection은 다른 Connection과 격리된다.

만약 forever가 전달된다면 모든 Connection이 하나의 Subject를 공유한다.

아래 코드를 보자!

let bag = DisposeBag()
let source = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).debug().share()

let observer1 = source
    .subscribe { print("🔵", $0) }

let observer2 = source
    .delaySubscription(.seconds(3), scheduler: MainScheduler.instance)
    .subscribe { print("🔴", $0) }

DispatchQueue.main.asyncAfter(deadline: .now() + 5) {
    observer1.dispose()
    observer2.dispose()
}

DispatchQueue.main.asyncAfter(deadline: .now() + 7) {
    let observer3 = source.subscribe { print("⚫️", $0) }
    
    DispatchQueue.main.asyncAfter(deadline: .now() + 3) {
        observer3.dispose()
    }
}
  
=========================================================================================
결과
2024-06-08 14:54:58.678: share.playground:32 (__lldb_expr_35) -> subscribed
2024-06-08 14:54:59.734: share.playground:32 (__lldb_expr_35) -> Event next(0)
🔵 next(0)
2024-06-08 14:55:00.730: share.playground:32 (__lldb_expr_35) -> Event next(1)
🔵 next(1)
2024-06-08 14:55:01.730: share.playground:32 (__lldb_expr_35) -> Event next(2)
🔵 next(2)
2024-06-08 14:55:02.730: share.playground:32 (__lldb_expr_35) -> Event next(3)
🔵 next(3)
🔴 next(3)
2024-06-08 14:55:03.730: share.playground:32 (__lldb_expr_35) -> Event next(4)
🔵 next(4)
🔴 next(4)
2024-06-08 14:55:04.017: share.playground:32 (__lldb_expr_35) -> isDisposed
2024-06-08 14:55:06.117: share.playground:32 (__lldb_expr_35) -> subscribed
2024-06-08 14:55:07.119: share.playground:32 (__lldb_expr_35) -> Event next(0)
⚫️ next(0)
2024-06-08 14:55:08.119: share.playground:32 (__lldb_expr_35) -> Event next(1)
⚫️ next(1)
2024-06-08 14:55:09.119: share.playground:32 (__lldb_expr_35) -> Event next(2)
⚫️ next(2)
2024-06-08 14:55:09.269: share.playground:32 (__lldb_expr_35) -> isDisposed

위 코드에선 첫 번재 파라미터 값의 기본값이 0이다. 그래서 3초뒤 구독한 두 번째 구독자는 이전에 전달한 3개의 이벤트는 받을 수 없다. -> 🔴 next(3)

그리고 아래 코드를 통해 5초 뒤에 모든 구독이 중지되면 내부에 있는 Connectable Observable 역시 중지된다.
DispatchQueue.main.asyncAfter(deadline: .now() + 5) {
observer1.dispose()
observer2.dispose()
}

그리고 7초 뒤에 새로운 구독자가 추가되면 Connectable Observable에서 새로운 시퀀스가 추가된다. 그래서 처음 받는 이벤트에는 0이 저장된다. -> ⚫️ next(0)

이번엔 replay 파라미터에 5를 넣은 경우 결과는 달라지는걸 볼 수 있다.

let bag = DisposeBag()
let source = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).debug().share(replay: 5)

let observer1 = source
    .subscribe { print("🔵", $0) }

let observer2 = source
    .delaySubscription(.seconds(3), scheduler: MainScheduler.instance)
    .subscribe { print("🔴", $0) }

DispatchQueue.main.asyncAfter(deadline: .now() + 5) {
    observer1.dispose()
    observer2.dispose()
}

DispatchQueue.main.asyncAfter(deadline: .now() + 7) {
    let observer3 = source.subscribe { print("⚫️", $0) }
    
    DispatchQueue.main.asyncAfter(deadline: .now() + 3) {
        observer3.dispose()
    }
}

=========================================================================================
결과
2024-06-08 17:39:58.765: share.playground:32 (__lldb_expr_3) -> subscribed
2024-06-08 17:39:59.847: share.playground:32 (__lldb_expr_3) -> Event next(0)
🔵 next(0)
2024-06-08 17:40:00.843: share.playground:32 (__lldb_expr_3) -> Event next(1)
🔵 next(1)
2024-06-08 17:40:01.844: share.playground:32 (__lldb_expr_3) -> Event next(2)
🔵 next(2)
🔴 next(0)
🔴 next(1)
🔴 next(2)
2024-06-08 17:40:02.850: share.playground:32 (__lldb_expr_3) -> Event next(3)
🔵 next(3)
🔴 next(3)
2024-06-08 17:40:03.844: share.playground:32 (__lldb_expr_3) -> Event next(4)
🔵 next(4)
🔴 next(4)
2024-06-08 17:40:04.113: share.playground:32 (__lldb_expr_3) -> isDisposed
2024-06-08 17:40:06.215: share.playground:32 (__lldb_expr_3) -> subscribed
2024-06-08 17:40:07.217: share.playground:32 (__lldb_expr_3) -> Event next(0)
⚫️ next(0)
2024-06-08 17:40:08.216: share.playground:32 (__lldb_expr_3) -> Event next(1)
⚫️ next(1)
2024-06-08 17:40:09.217: share.playground:32 (__lldb_expr_3) -> Event next(2)
⚫️ next(2)
2024-06-08 17:40:09.371: share.playground:32 (__lldb_expr_3) -> isDisposed

새로운 구독자는 구독이 시작되는 시점에 버퍼에 저장되어 있는 이벤트를 함께 전달 받기 때문에
🔴 next(0)
🔴 next(1)

위 이벤트도 같이 전달 받는다. 하지만 여전히 마지막 세번째 구독자는 새로운 Subject로부터 이벤트를 전달 받기 때문에
⚫️ next(0) 그대로 0을 받는다.

마지막으로 2번째 파라미터를 Defulat인 whileConnected에서 forever로 받는 경우의 예제이다.
이 경우 모든 구독자가 하나의 Subject를 공유하게 된다.

let bag = DisposeBag()
let source = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).debug().share(replay: 5, scope: .forever)

let observer1 = source
    .subscribe { print("🔵", $0) }

let observer2 = source
    .delaySubscription(.seconds(3), scheduler: MainScheduler.instance)
    .subscribe { print("🔴", $0) }

DispatchQueue.main.asyncAfter(deadline: .now() + 5) {
    observer1.dispose()
    observer2.dispose()
}

DispatchQueue.main.asyncAfter(deadline: .now() + 7) {
    let observer3 = source.subscribe { print("⚫️", $0) }
    
    DispatchQueue.main.asyncAfter(deadline: .now() + 3) {
        observer3.dispose()
    }
}

=========================================================================================
결과
2024-06-08 17:45:38.924: share.playground:32 (__lldb_expr_7) -> subscribed
2024-06-08 17:45:39.970: share.playground:32 (__lldb_expr_7) -> Event next(0)
🔵 next(0)
2024-06-08 17:45:40.965: share.playground:32 (__lldb_expr_7) -> Event next(1)
🔵 next(1)
2024-06-08 17:45:41.965: share.playground:32 (__lldb_expr_7) -> Event next(2)
🔵 next(2)
🔴 next(0)
🔴 next(1)
🔴 next(2)
2024-06-08 17:45:42.965: share.playground:32 (__lldb_expr_7) -> Event next(3)
🔵 next(3)
🔴 next(3)
2024-06-08 17:45:43.964: share.playground:32 (__lldb_expr_7) -> Event next(4)
🔵 next(4)
🔴 next(4)
2024-06-08 17:45:44.258: share.playground:32 (__lldb_expr_7) -> isDisposed
⚫️ next(0)
⚫️ next(1)
⚫️ next(2)
⚫️ next(3)
⚫️ next(4)
2024-06-08 17:45:46.358: share.playground:32 (__lldb_expr_7) -> subscribed
2024-06-08 17:45:47.360: share.playground:32 (__lldb_expr_7) -> Event next(0)
⚫️ next(0)
2024-06-08 17:45:48.360: share.playground:32 (__lldb_expr_7) -> Event next(1)
⚫️ next(1)
2024-06-08 17:45:49.360: share.playground:32 (__lldb_expr_7) -> Event next(2)
⚫️ next(2)
2024-06-08 17:45:49.512: share.playground:32 (__lldb_expr_7) -> isDisposed

위 결과를 보면 마지막 세번째 구독자가 추가되는 시점에 버퍼에 저장되어 있는 5개의 이벤트가 함께 전달된다.
⚫️ next(0)
⚫️ next(1)
⚫️ next(2)
⚫️ next(3)
⚫️ next(4)

그런데 이후를 보면 5가 아닌 다시 ⚫️ next(0) 0이 전달되는 것을 볼 수 있다. 이는 시퀀스가 중지된 다음 새로운 구독자가 추가되면 새로운 시퀀스가 시작되기 때문이다. 이는 RefCount Observable의 특징이기에 당연한 결과이다.

KXCoding 강의
Marble Diagram
ReactiveX 사이트