How to schedule a synchronous sequence of asynchronous calls in Combine?

user_2981130 0
Viewed 3017

I'd like to handle a series of network calls in my app. Each call is asynchronous and flatMap() seems like the right call. However, flatMap processes all arguments at the same time and I need the calls to be sequential -- the next network call starts only after the previous one is finished. I looked up an RxSwift answer but it requires concatMap operator that Combine does not have. Here is rough outline of what I'm trying to do, but flatMap fires all myCalls at the same time.

Publishers.Sequence(sequence: urls)
  .flatMap { url in
    Publishers.Future<Result, Error> { callback in 
        myCall { data, error in 
            if let data = data {
                callback(.success(data))
            } else if let error = error {
                callback(.failure(error))
            }
        }
    }
  }
combine swift ios
2 Answers

After experimenting for a while in a playground, I believe I found a solution, but if you have a better idea, please share. The solution is to add maxPublishers parameter to flatMap and set the value to max(1)

Publishers.Sequence(sequence: urls)
  .flatMap(maxPublishers: .max(1)) // <<<<--- here
  { url in 
    Publishers.Future<Result, Error> { callback in 
      myCall { data, error in 
        if let data = data {
          callback(.success(data))
        } else if let error = error {
          callback(.failure(error))
        }
      }
    }
  }
user_29811300

You can also use prepend(_:) method on observable which creates concatenated sequence which, I suppose is similar to Observable.concat(:) in RxSwift.

Here is a simple example that I tried to simulate your use case, where I have few different sequences which are followed by one another.

func dataTaskPublisher(_ urlString: String) -> AnyPublisher<(data: Data, response: URLResponse), Never> {
    let interceptedError = (Data(), URLResponse())
    return Publishers.Just(URL(string: urlString)!)
                        .flatMap {
                            URLSession.shared
                                        .dataTaskPublisher(for: $0)
                                        .replaceError(with: interceptedError)
                        }
                        .eraseToAnyPublisher()
}

let publisher: AnyPublisher<(data: Data, response: URLResponse), Never> = Publishers.Empty().eraseToAnyPublisher()


for urlString in [
    "http://ipv4.download.thinkbroadband.com/1MB.zip",
    "http://ipv4.download.thinkbroadband.com/50MB.zip",
    "http://ipv4.download.thinkbroadband.com/10MB.zip"
    ] {
        publisher = publisher.prepend(dataTaskPublisher(urlString)).eraseToAnyPublisher()
}

publisher.sink(receiveCompletion: { completion in
    print("Completed")
}) { response in
    print("Data: \(response)")
}

Here, prepend(_:) operator prefixes the sequence and so, prepended sequences starts first, completes and next sequence start.

If you run the code below, you should see that firstly 10 MB file is download, then 50 MB and at last 1 MB, since the last prepended starts first and so on.

There is other variant of prepend(_:) operator which takes array, but that does not seem to work sequentially.

user_6546660
Related