CODEWITHSUNDEEP
iosswiftxcoderx-swift
Douglas Silva
Douglas Silva

Reputation: 145

Scheduling subscriptions in RxSwift

I’m trying to schedule a series of downloads of images. I want to call scheduleImagesDownload as many times as I want, but wait to perform the code inside downloadImages only when the previous call is completed.

I’m trying to fetch and download images from camera for a specific ID, once all images are downloaded for that ID, I want to start downloading images for next ID, and so on.

I’m having a hard time because, even using a Serial Scheduler, all the subscriptions are called right away, before the previous download is completed. I was wondering if there’s a way to do this with pure Rx, without having to use semaphores, etc.

Thank you in advance!


    func scheduleImagesDownload(flightId: String) -> Disposable {
        let subscription = donwloadImages(flightId)
            .subscribeOn(SerialDispatchQueueScheduler(qos: .background))
            .subscribe(onCompleted: {
                /// Finished downloading images for flight.
            }, onError: { error in
                /// Error downloading images for flight.
            })

        return subscription
    }

    func donwloadImages(_ flightId: String) -> Completable {
        return Completable.create { completable in

            /// Simulate querying the drone for images to download async and start downloading them.
            DispatchQueue.global().async {
                sleep(5) // Sleep to simulate downloading time.
                completable(.completed) // Finished downloading.
            }
            return Disposables.create()
        }
    }

Upvotes: 1

Views: 1255

Answers (2)

Smit Sonani
Smit Sonani

Reputation: 115

Key operator for chaining each image downloading operation in queue is ConcatMap. I have written following code snippet based on your requirements. Snippet is pretty much self explanatory.

let flightIds: [String] = [] // Array holding flightIds
let disposeBag = DisposeBag()

func download() {
    Observable.from(flightIds) // Convert array of flightIds into Observable chain
        .flatMap(getImageURLs) // For each flightId, get array of image URLs to be downloaded
        .flatMap(convertToImageURLObservable) // Convert array of image URLs into Observable chain
        .concatMap(downloadImage) // Concate each url in observable chain so each image will be downloaded sequencially
        .subscribeOn(SerialDispatchQueueScheduler(qos: .background)) // Scheduled entire chain on background queue
        .subscribe()
        .disposed(by: disposeBag)
}

/// Fetches image URLs for given flightId
func getImageURLs(_ flightId: String) -> Single<[URL]> {
    return Single<[URL]>.create { single in

        /// fetch & pass URLs in below array inside .success
        single(.success([]))

        return Disposables.create()
    }
}

/// Convert array of image URLs into Observable chain
func convertToImageURLObservable(_ urls: [URL]) -> Observable<URL> {
    return Observable.from(urls)
}

/// Downloads image for given URL
func downloadImage(_ url: URL) -> Completable {
    return Completable.create { completable in

        /// fetch image
        DispatchQueue.global().async {
            sleep(5) // Sleep to simulate downloading time.
            completable(.completed) // Finished downloading.
        }

        return Disposables.create()
    }
}

Upvotes: 2

Mocha
Mocha

Reputation: 2253

I believe this is what you want. You basically create an array/set of all the ids that you retrieve from the camera. Then on each completion, you remove an element in your collection and then you restart the process until your collection is empty.

    var cameraIds: Set<String>()

    func scheduleImagesDownload(flightId: String) -> Disposable {
        let subscription = donwloadImages(flightId)
            .subscribeOn(SerialDispatchQueueScheduler(qos: .background))
            .subscribe(onCompleted: {
                if let nextFlightId = cameraIds.first {
                  cameraIds.removeFirst()
                  scheduleImagesDownload(flightId: nextFlightId)
               } else {
                 // Finished downloads for all cameraIds and images
               }
            }, onError: { error in
                /// Error downloading images for flight.
            })

        return subscription
    }

Upvotes: 0

Related Questions