Skip events from source sequence until a task finished

Question

I have a PublishSubject as source sequence. I need to do a api call on receiving events. Imagine if this is emitting events too fast, I want to skip all the events until api call finish. After API call finish, continue to receive events and so on..

I don't want to use Debounce or a Throttle as there are other operations related with this main sequence, So i don't want to delay all other operations.

I am trying to implement a solution with skipUntil() operator as below.

import UIKit
import RxSwift
import RxSwiftPlayground

let disposeBag = DisposeBag()

let subject = PublishSubject<String>()
let trigger = PublishSubject<String>()


subject
    .skipUntil(trigger)
    .subscribe(onNext: {
        print($0)
    })
    .disposed(by: disposeBag)


subject.onNext("A")

trigger.onNext("T1")

subject.onNext("B")
subject.onNext("C")

trigger.onNext("T2")

subject.onNext("D")

However it seems works only for the first trigger. Subsequent events from the trigger subject is no effect here.

Highly appreciate if someone can share if there is any other operator for this purpose or has implement similar kind of solution with RxSwift.

Answer 1

You would probably find my article on flatMap useful: RxSwift’s Many Faces of FlatMap

I say this because you have precisely described the functionality of flatMapFirst.

Occasionally, ... you would rather ignore events from the source until the sub-Observable is done. This is what flatMapFirst is for. ... it only allows one sub-Observable at a time... it ignores events from the source until the sub-Observable is done.

subject
    .flatMapFirst { _ in apiCall() }
    .subscribe(onNext: { apiCallResult in 
        print(apiCallResult)
    })
    .disposed(by: disposeBag)

EDIT

Looking at your question again. This should do what you want:

let disposeBag = DisposeBag()
let subject = PublishSubject<String>()
let trigger = PublishSubject<String>()

trigger
    .flatMapFirst { _ in subject.take(1) }
    .subscribe(onNext: {
        print($0)
    })
    .disposed(by: disposeBag)


subject.onNext("A")

trigger.onNext("T1")

subject.onNext("B")
subject.onNext("C")

trigger.onNext("T2")

subject.onNext("D")

The above prints only "B" and "D".