Limiting Concurrent Access to a Service Class with Rxswift

Limiting concurrent access to a service class with RxSwift

You need a serial Scheduler that is dedicated to that service. Here is an example that can be pasted to a playground:

/// playground

import RxSwift

class Service {

    func handleJobA(input: String) -> Observable<String> {

        return Observable.create { observer in
            print("start job a")
            sleep(3)
            observer.onNext(input)
            print("complete job a")
            observer.onCompleted()
            return Disposables.create()
        }.subscribeOn(scheduler)
    }

    func handleJobB(input: String) -> Observable<String> {
        return Observable.create { observer in
            print("start job b")
            sleep(3)
            observer.onNext(input)
            print("complete job b")
            observer.onCompleted()
            return Disposables.create()
            return Disposables.create()
        }.subscribeOn(scheduler)
    }

    let scheduler = SerialDispatchQueueScheduler(internalSerialQueueName: "Service")
}

let service = Service()

_ = Observable.from(["hello","world","swift"])
    .flatMap { service.handleJobA(input: $0) }
    .subscribe(onNext:{
        print("result " + $0)
    })

_ = Observable.from(["hello","world","swift"])
    .flatMap { service.handleJobB(input: $0) }
    .subscribe(onNext:{
        print("result " + $0)
    })

import PlaygroundSupport

PlaygroundPage.current.needsIndefiniteExecution = true

concatMap / flatMap should run immediately, on the same scheduler

I can't explain schedulers behaviour... But I can make a small proposal

...once a function has started processing a task, no one else get's
access unless it is done...

You can pass all your handleJob calls through concatMap to get the behaviour you require:

Observable
    .from([1,2,3,4,5,6])
    .flatMap({ (value) -> Observable<String> in
        switch value % 2 == 0 {
        case true:
            return service.handleJobA(input: "\(value)")
        case false:
            return service.handleJobB(input: "\(value)")
        }
    })
    .subscribe(onNext:{
        print($0 + " √")
    })

Service class example:

private class Service {

    private lazy var result = PublishSubject<(index: Int, result: String)>()
    private lazy var publish = PublishSubject<(index: Int, input: String, transformation: (String) -> String)>()
    private lazy var index: Int = 0
    private lazy var disposeBag = DisposeBag()

    init() {
        publish
            .asObservable()
            .concatMap({ (index, input, transformation) -> Observable<(index: Int, result: String)> in
                let dueTime = RxTimeInterval(arc4random_uniform(3) + 1)
                return Observable
                    .just((index: index, result: transformation(input)))
                    .delay(dueTime, scheduler: self.scheduler)
            })
            .bind(to: result)
            .disposed(by: disposeBag)
    }

    func handleJobA(input: String) -> Observable<String> {
        let transformation: (String) -> String = { string in
            return "Job A: \(string)"
        }
        return handleJob(input: input, transformation: transformation)
    }

    func handleJobB(input: String) -> Observable<String> {
        let transformation: (String) -> String = { string in
            return "Job B: \(string)"
        }
        return handleJob(input: input, transformation: transformation)
    }

    func handleJob(input: String, transformation: @escaping (String) -> String) -> Observable<String> {
        index += 1
        defer {
            publish.onNext((index, input, transformation))
        }
        return result
            .filter({ [expected = index] (index, result) -> Bool in
                return expected == index
            })
            .map({ $0.result })
            .take(1)
            .shareReplayLatestWhileConnected()
    }

    let scheduler = SerialDispatchQueueScheduler(internalSerialQueueName: "Service")
}

Related Topics