[RxSwift] Using Single with share

[ Single ]

 

RxSwift에서는 Single Trait 을 지원합니다. Single은 Observable의 한 형태이며, 한 가지 값 또는 에러를 발행합니다.

따라서 구독시, success와 error 두 가지의 이벤트에 처리를 할 수 있습니다.

 

대표적인 예로는 네트워크 요청 구독을 위해 request를 Single로 wrapping 하여 사용할 수 있습니다.

 

 

[ Share ]

또한 RxSwift에서는 share 를 제공합니다. share는 생성된 Observable 시퀀스를 공유하기 위해 사용되는 개념입니다.

 

처음 Rx를 시작하는 경우, 자주 실수하는 부분이 있습니다. 바로 subscribe 할 때마다, 새로운 시퀀스가 생성되는 것을 모르는 경우가 많은 것입니다.

 

 

따라서 위의 경우, 실제로 같은 시퀀스를 사용하고자 하지만, 중복적으로 네트워크 요청을 하게 됩니다. 이를 방지하기 위해, share라는 개념을 이용해 시퀀스를 공유할 수 있습니다.

 

share의 파라미터로는 replay와 scope가 있습니다. 기본적으로 replay는 0, scope는 whileConnected 옵션이 디폴트이며, whileConnected 옵션을 권장하고 있습니다. 이는 forever일 경우 버퍼가 생성된 이후, subscription의 유무와 관계없이 버퍼가 지속적으로 유지되기 때문이라고 생각합니다.

 

[ Single과 share를 같이 사용하는 경우 발생할 수 있는 문제점 ]

 

네트워크 요청 구독을 위해 주로 사용하는 Single, 네트워크 중복 요청을 방지하기 위해 주로 사용하는 share 이 둘을 함께 사용하기 위해서는 주의해야 할 점이 있습니다.

 

Single은 success 자체가 next, completed 두 가지의 성격을 지니고 있습니다. (코드1)

이는 success가 된다면 동시에 complete가 되며, 완료되었기 때문에 dispose 하는 과정을 거치게 됩니다. 따라서 연결이 끊어졌다고 판단하며, 이후 구독을 하게 될 경우 새로운 시퀀스를 생성하게 됩니다.

 

네트워크 요청의 경우, 비동기 작업을 하기 때문에 3,4 line의 경우 subscribe 하는 시점이 response 받는 시점보다 대부분의 경우 빠르다고 할 수 있습니다.

 

하지만 7 line의 경우 버튼 클릭 이벤트가 발생하는 경우 구독하기 때문에 이미 3,4 line의 구독이 종료된 이후에 구독이 진행될 것입니다. 따라서 새로운 시퀀스를 생성하며 중복적인 네트워크 요청을 다시 진행하게 됩니다.

 

그렇다고 3,4 line과 같은 경우 안정적으로 시퀀스를 공유할 수 있다고 판단하기는 어렵습니다.

 

위 예는 네트워크 요청과 유사한 상황을 만들어 테스트하는 과정입니다.

다양한 결과가 나올 수 있는데 그중 하나는 다음과 같습니다.

 

[그림 1] 테스트 코드 결과 화면

 

이를 좀 더 이해하기 쉽게, Rx의 마블과 유사하게 표현하면 다음과 같습니다.

 

[그림 2] 테스트 코드 결과 마블화

0~4번은 반복문의 인덱스를 바탕으로 이름을 부여했습니다.

 

0번과 1번을 먼저 살펴보면, 0번이 구독을 한 시점부터, shareResult는 시퀀스를 생성하게 됩니다. 따라서 0번은 이벤트를 받는다는 보장을 할 수 있습니다.

 

http://reactivex.io

An API for asynchronous programming
with observable streams

 

위 인용구는, Rx 사이트에 들어가면 바로 볼 수 있는 문구입니다. Rx는 모든 것을 비동기적인 스트림으로 간주하며, 비동기 프로그래밍에 사용하는 유용한 도구입니다.

 

이런 비동기적인 특성으로 인해 0번이 complete - dispose 되는 시점과 1번이 구독하는 시점을 동기화 할 수 없으며, 타이밍적으로 겹칠 수 있습니다. 그럴 경우 share()의 기본 replay 카운트는 0이므로 이벤트를 수신하지 못하고, 대기하다가 종료하게 됩니다.

 

나머지를 살펴보면, 0번과 1번의 shareResult에 대한 구독이 종료된 이후 구독을 시작하기 때문에, shareRequest는 새로운 시퀀스를 생성하게 됩니다. (scope가 whileConnected 이기 때문입니다.)

나머지 경우들은 새로운 이벤트를 수신하기 전에 모두 구독한 상태가 되었기 때문에, 하나의 공유된 이벤트를 수신할 수 있습니다.

 

[ 해결 방안 ]

 

위의  문제들을 해결할 수 있는 다양한 방법들이 있겠지만, 단순하게 생각하여 연결의 끊김을 본인이 결정할 수 있도록 바꾸는 것이라고 생각했습니다.

 

RxSwift에서 제공하는 Concat과 Never 연산자를 활용하여 해결방안을 만들 수 있었습니다.

 

Concat은 각 Observable을 이어서 붙여주는 연산자로, 이전 Observable이 완료될 때까지 전달한 추가 Observable을 구독합니다.

Never는 이벤트를 방출하지도, 종료하지도 않는 Observable로 만들어 주는 것입니다.

 

 

 

 

위와 같이 ignoreTerminate()를 구현하게 되면, single의 complete가 호출되더라도, Observable.never()가 종료되기 전까지 구독을 유지하게 됩니다. 하지만 이것만 추가하게 된다면,

 

`single.asObservable().ignoreTerminate().share()`

 

와 같은 방식으로 사용해야 하므로 다소 복잡하다는 생각이 듭니다. 따라서 위 개념을 확장하여, Single에 share를 추가하는 작업을 진행했습니다.

 

Single은 한 가지 값 또는 에러를 발행하므로, replay 개수를 최대 1개만 가질 수 있도록 SingleReplayBuffer enum을 추가했으며, ignoreTerminate 옵션을 추가하여 종료 옵션을 제외시킬 수 있는 여지를 만들었습니다.

 

이제 위 테스트 코드를 조금 수정하면 완성할 수 있습니다.

 

이를 통한 결과 중 하나입니다.

 

[그림 3] 변경 사항 적용한 테스트 코드 결과 화면

 

이를 좀 더 이해하기 쉽게, Rx의 마블과 유사하게 표현하면 다음과 같습니다.

[그림 4] 변경 사항 적용한 테스트 코드 결과 마블화

 

위의 onDisposed는 disposeBag을 재설정하는 경우 호출이 됩니다. 이외에 주의 깊게 봐야 하는 항목은 replay를 .one으로 적용했다는 것입니다.

0번과 1번의 경우, replay가 zero이더라도 결과를 받기 전에 둘 다 구독한 상태가 되었으므로 이벤트 결과를 받을 수 있습니다.

하지만 2번~4번의 경우 replay를 0으로 하게 된다면, 이후에 발행될 이벤트만을 기다리며 종료될 것입니다.

따라서 replay를 .one으로 설정하여 이전의 값을 공유할 수 있도록 설정할 수 있습니다.