RxSwift replay the last value of a completed observable

user_1625364 0
Viewed 2442

I have a cold observable that may get called multiple times. This observable does an expensive task (a network request) and then completes. I would like this observable to only ever make a single network call and if I need to call it again in the future I would like to get the last emitted value.

If an observable doesn't complete (i.e. just sends a next value without a completed event) I can use the .share(replay: 1, scope: .whileConnected) function to always get the last value. Unfortunately, this doesn't work with observables that completes at the end of the request. Bellow is an example:

let disposeBag = DisposeBag()
let refreshSubject = PublishSubject<Void>()

override func viewDidLoad() {
    super.viewDidLoad()

    let observable = Observable<String>.create { observer in
        let seconds = 2.0
        DispatchQueue.main.asyncAfter(deadline: .now() + seconds) {
            observer.onNext("Hello, World")
            observer.onCompleted() // <-- Works when commented out
        }

        return Disposables.create()
    }
    .share(replay: 1, scope: .whileConnected)

    refreshSubject
        .flatMap { _ in observable }
        .subscribe(onNext: { response in
            print("response: ", response)
        })
        .disposed(by: disposeBag)
}

@IBAction func refreshButtonHandler(_ sender: Any) {
    refreshSubject.onNext(())
}

Every time the refreshSubject is triggered it takes 2 seconds for the Hello, World to be printed. If I remove the observer.onCompleted() line however, it only takes 2 seconds the first time and subsequently returns a cached response.

Obviously this is just an example, in the real world I would not have any control if the observable completes or not but I would like to always just replay the last value regardless.

rx-swift swift
2 Answers

So you don't want the cold observable to be re-subscribed to even when the refresh is triggered. In that case, this is the solution:

Observable.combineLatest(refreshSubject.startWith(()), yourColdObservable)
    .map { $0.1 }
    .subscribe(onNext: { val in
        print("*** val: ", val)
    })

Using flatMap means that the observable is re-subscribed to every time an event enters the flatMap. By using combineLatest instead, the cold observable will only be subscribed to once. The combineLatest operator will store the result of the observable internally and emit it again every time the refresh subject emits. (No share is needed for this method.)

user_5064410 
let yourColdObservable = Observable<String>.create { observer in
    let seconds = 2.0
    DispatchQueue.main.asyncAfter(deadline: .now() + seconds) {
        observer.onNext("Hello, World")
        observer.onCompleted()
    }
    return Disposables.create()
}

let cacheObservable = refreshButton.rx.tap
    .startWith(())
    .flatMapLatest { _ in yourColdObservable }
    .share(replay: 1)

makeRequestWithCacheButton.rx.tap
    .flatMapLatest { _ in cacheObservable }
    .subscribe(onNext: { response in
        print("response: ", response)
    })
    .disposed(by: disposeBag)

StartWith

emit a specified sequence of items before beginning to emit the items from the source Observable Such a fake tap on the refreshButton when starting the sequence

FlatMapLatest

The FlatMap operator transforms an Observable by applying a function that you specify to each item emitted by the source Observable, where that function returns an Observable that itself emits items. FlatMap then merges the emissions of these resulting Observables, emitting these merged results as its own sequence. FlatMapLatest is special type of FlatMap, because it cancels previous Observable when refreshButton.rx.tap emit onNext event.

user_67568520
Related