[RxSwift] Operator(4) - Share

Observable은 구독되기 전까지는 동작하지 않다가, 

구독이 이루어지면 한 번에 데이터를 방출한다.

이걸 Cold Observable이라 한다.

 

Cold Observable은 UnitCast방식으로 동작하여, 

Subscriber마다 하나의 Stream이 생성이 된다. 

 

반면 Subject와 같은 Hot Observable은

Multicast방식으로 동작하며, 

하나의 Stream을 통해 데이터가 방출된다. 

이 내용이 이해되지 않는 다면, 해당 포스팅을 보길 바란다.

 

하나의 Stream을 통해 데이터가 방출되는 Hot Observable과 달리 

Subscriber마다 Stream이 생성되는 Cold Observable에서는 Stream이 여러개 생길 수 있다.  

 

다음과 같은 함수가 있다고 가정해 보자. 

func fetch() -> Observable<Int> {
	return Observable.create { emitter in
		emitter.onNext(1)
		emitter.onNext(2)
		
		return Disposables.create()
	}.debug("Observable")
}

fetch().subscribe(onNext: { element in
		print(element)
}).disposed(by: disposeBag)

fetch().subscribe(onNext: { element in
		print(element)
}).disposed(by: disposeBag)

해당 함수를 다음과 같이 구독하게 되면 2개의 시퀀스가 생기는 것을 볼 수 있다.

/*
Observable -> subscribed
Observable -> Event next(1)
1
Observable -> Event next(2)
2
Observable -> subscribed
Observable -> Event next(1)
1
Observable -> Event next(2)
2
Observable -> isDisposed
Observable -> isDisposed
*/

 

 

Share Operator

Share Operator는 Cold Observable을 Hot Observable처럼 여러 구독자가 시퀀스를 공유할 수 있게 된다. 

let shareOb = fetch().share()

shareOb.subscribe(onNext: { element in
		print(element)
}).disposed(by: disposeBag)

shareOb.subscribe(onNext: { element in
		print(element)
}).disposed(by: disposeBag)

/*
 Observable -> subscribed
 Observable -> Event next(1)
 1
 Observable -> Event next(2)
 2
 Observable -> isDisposed
 */

다음과 같이 시퀀스가 하나만 생성된걸 볼 수 있다. 

 

 

share(replay: , scope:)

public func share(replay: Int = 0, scope: SubjectLifetimeScope = .whileConnected)

 

replay

replay는 버퍼의 크기를 의미하며, defualt 값은 0이다.

let shareOb = fetch().share(replay: 1)
		
		shareOb.subscribe(onNext: { element in
				print(element)
		}).disposed(by: disposeBag)

		shareOb.subscribe(onNext: { element in
				print(element)
		}).disposed(by: disposeBag)

다음과 같이 두번째 subscribe에서는 마지막 이벤트인 2가 한번 더 방출된다. 

/*
 Observable -> subscribed
 Observable -> Event next(1)
 1
 Observable -> Event next(2)
 2
 2
 Observable -> isDisposed
 */

 

scope

scope에는 2가지의 종류가 있다.

• .whileConnected
• .forever 

default값은 .whileConnected이다.

 

.whileConnected

whileConnected는 해당 스트림을 구독하고 있는 Subscriber의 수가 1개 이상일 경우에만 스트림이 유지가 된다. 

반면, Subscriber가 0으로 떨어지게 되면, 스트림은 유지되지 않으며, 

새로운 Subscriber는 replay값이 정해져 있더라도 버퍼의 값을 받을 수 없다. 

즉, 버퍼 역시 유지되지 않는다.

 

다음과 같이 onCompleted이벤트를 방출하여 스트림을 끊게 되면, 

func fetch() -> Observable<Int> {
	return Observable.create { emitter in
		emitter.onNext(1)
		emitter.onNext(2)
		emitter.onCompleted()
		
		return Disposables.create()
	}.debug("Observable")
}

let shareOb = fetch().share(replay: 1)

shareOb.subscribe(onNext: { element in
		print(element)
}).disposed(by: disposeBag)

shareOb.subscribe(onNext: { element in
		print(element)
}).disposed(by: disposeBag)

 

다음과 같이 스트림은 유지되지 않고 두 번 생기게 된다. 

또한 replay를 통해 버퍼의 크기를 1로 정해주더라도 

버퍼 역시 유지되지 않는 것을 볼 수 있다.

/*
 Observable -> subscribed
 Observable -> Event next(1)
 1
 Observable -> Event next(2)
 2
 Observable -> Event completed
 Observable -> isDisposed
 
 Observable -> subscribed
 Observable -> Event next(1)
 1
 Observable -> Event next(2)
 2
 Observable -> Event completed
 Observable -> isDisposed
 */

 

.forever

forever은 whileConnected와 다르게

스트림에 Subscriber가 0개더라도 해당 스트림은 유지되며, 버퍼 역시 유지된다.

let shareOb = fetch().share(replay: 1, scope: .forever)

shareOb.subscribe(onNext: { element in
		print(element)
}).disposed(by: disposeBag)

shareOb.subscribe(onNext: { element in
		print(element)
}).disposed(by: disposeBag)

 

다음과 같이 스트림과 버퍼가 유지되는 것을 볼 수 있다. 

/*
 Observable -> subscribed
 Observable -> Event next(1)
 1
 Observable -> Event next(2)
 2
 Observable -> Event completed
 Observable -> isDisposed
 2
 */

 

 

Reference

https://www.notion.so/Wallaby-RxSwift-72194669a39a4557baa69c672268af38