Swift Combine - Wait for All Publishers

How can I wait until all Combine publishers finished their jobs in Swift?

A possible way is a view model. In this class merge the publishers and use the receiveCompletion: parameter

class ViewModel : ObservableObject {
    
    @Published var isFinished = false
    let pub1 = ["one", "two", "three", "four"].publisher
    let pub2 = ["five", "six", "seven", "eight"].publisher
    
    private var subscriptions = Set<AnyCancellable>()
    
    init() {
        pub1
            .sink { print($0) }
            .store(in: &subscriptions)
        pub2
            .sink { print($0) }
            .store(in: &subscriptions)
        
        pub1.merge(with: pub2)
            .sink(receiveCompletion: { _ in
                self.isFinished = true
            }, receiveValue: { _ in  })
            .store(in: &subscriptions)
    }
}

struct SwiftUIView: View {
    @StateObject private var model = ViewModel()
    var body: some View {
        if model.isFinished {
            Text("Hello, World!")
        } else {
            ProgressView()
        }
    }
}

Swift Combine - combining publishers without waiting for all publishers to emit first element

You can use prepend(…) to prepend values to the beginning of a publisher.

Here's a version of your code that will prepend nil to both publishers.

let timer = Timer.publish(every: 10, on: .current, in: .common).autoconnect()
let anotherPub: AnyPublisher<Int, Never> = Just(10).delay(for: 5, scheduler: RunLoop.main).eraseToAnyPublisher()

Publishers.CombineLatest(
    timer.map(Optional.init).prepend(nil),
    anotherPub.map(Optional.init).prepend(nil)
)
.filter { $0 != nil && $1 != nil } // Filter the event when both are nil values
.sink(receiveValue: { (timer, val) in
    print("Hello! \(timer) \(val)")
})

Apple Combine framework: How to execute multiple Publishers in parallel and wait for all of them to finish?

You can run some operations in parallel by creating a collection of publishers, applying the flatMap operator and then collect to wait for all of the publishers to complete before continuing. Here's an example that you can run in a playground:

import Combine
import Foundation

func delayedPublisher<Value>(_ value: Value, delay after: Double) -> AnyPublisher<Value, Never> {
  let p = PassthroughSubject<Value, Never>()
  DispatchQueue.main.asyncAfter(deadline: .now() + after) {
    p.send(value)
    p.send(completion: .finished)
  }
  return p.eraseToAnyPublisher()
}

let myPublishers = [1,2,3]
  .map{ delayedPublisher($0, delay: 1 / Double($0)).print("\($0)").eraseToAnyPublisher() }

let cancel = myPublishers
  .publisher
  .flatMap { $0 }
  .collect()
  .sink { result in
    print("result:", result)
  }

Here is the output:

1: receive subscription: (PassthroughSubject)
1: request unlimited
2: receive subscription: (PassthroughSubject)
2: request unlimited
3: receive subscription: (PassthroughSubject)
3: request unlimited
3: receive value: (3)
3: receive finished
2: receive value: (2)
2: receive finished
1: receive value: (1)
1: receive finished
result: [3, 2, 1]

Notice that the publishers are all immediately started (in their original order).

The 1 / $0 delay causes the first publisher to take the longest to complete. Notice the order of the values at the end. Since the first took the longest to complete, it is the last item.

How do you run a Swift Combine Publisher for a certain amount of time?

This operator will do the trick.

import PlaygroundSupport
import Foundation
import Combine

let page = PlaygroundPage.current
page.needsIndefiniteExecution = true

extension Publisher {
    func stopAfter<S>(_ interval: S.SchedulerTimeType.Stride, tolerance: S.SchedulerTimeType.Stride? = nil, scheduler: S, options: S.SchedulerOptions? = nil) -> AnyPublisher<Output, Failure> where S: Scheduler {
        prefix(untilOutputFrom: Just(()).delay(for: interval, tolerance: tolerance, scheduler: scheduler, options: nil))
            .eraseToAnyPublisher()
    }
}

let source = Timer.publish(every: 1, tolerance: nil, on: RunLoop.main, in: .default, options: nil)
    .autoconnect()
    .eraseToAnyPublisher()

let cancellable = source
    .stopAfter(10, scheduler: DispatchQueue.main)
    .sink(receiveValue: { print($0) })

Chaining Combine.Publisher and calling completion when finished

If the publishers depend on each other, you should chain with .flatMap. If not, you could use .append instead. Either way, if all of these publishers are one-shot publishers (they all send a completion after one value), then the chain will be torn down in good order when all have fired.

Example 1:

    Just(1)
        .flatMap { Just(($0,2)) }
        .flatMap { Just(($0.0, $0.1, 3)) }
        .sink(receiveCompletion: {print($0)},
              receiveValue: {print($0)})
        .store(in:&storage)
        // (1,2,3) as a tuple, then finished

Example 2:

    Just(1).append(Just(2).append(Just(3)))
        .sink(receiveCompletion: {print($0)},
              receiveValue: {print($0)})
        .store(in:&storage)
    // 1, then 2, then 3, then finished

How to chain together two Combine publishers in Swift and keep the cancellable object

If I understand correctly, you want to link two publishers but with the option to break that link at some point in the future.

I would try using sink on the inputPublisher, since that function gives me a cancellable, and then a PassthroughSubject, since I wasn't able to figure out how to pass the value from sink directly to outputPublisher.

It would look something like this:

static func connect(inputPublisher: Published<String>.Publisher, outputPublisher: inout Published<String>.Publisher) -> AnyCancellable {
  let passthrough = PassthroughSubject<String, Never>()
  passthrough.assign(to: &outputPublisher)
  let cancellable = inputPublisher.sink { string in
    passthrough.send(string)
  }
  return cancellable
}

Disclaimer: I wrote this on a Playground and it compiles, but I didn't actually run it.

---

Related Topics