Combine: publish elements of a sequence with some delay

Question

I'm new to Combine and I'd like to get a seemingly simple thing. Let's say I have a collection of integer, such as:

let myCollection = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

I'd like to publish each element with a delay of, for example, 0.5 seconds.

print 0
wait for 0.5secs
print 1
wait for 0.5secs
and so forth

I can easily get the sequence publisher and print the elements like this:

let publisherCanc = myCollection.publisher.sink { value in
    print(value)
}

But in this case all the values are printed immediately. How can I print the values with a delay? In Combine there's a .delay modififer, but it's not for what I need (indeed, .delay delays the entire stream and not the single elements). If I try:

let publisherCanc = myCollection.publisher.delay(for: .seconds(0.5), scheduler: RunLoop.main).sink { value in
    print(value)
}

All I get it's just an "initial" delay, then the elements are printed immediately.

Thanks for your help.

Answer 1

Using the idea from the answer linked by Alexander in comments, you can create a publisher that emits a value every 0.5 seconds using Timer.publish(every:on:in:), then zip that together with your Array.publisher to make your downstream publisher emit a value every time both of your publishers have emitted a new value.

Publishers.Zip takes the n-th element of its of its upstream publishers and only emits when both of its upstreams have reached n emitted values - hence by zipping together a publisher that only emits its values at 0.5 second intervals with your original publisher that emits all of its values immediately, you delay each value by 0.5 seconds.

let delayPublisher = Timer.publish(every: 0.5, on: .main, in: .default).autoconnect()
let delayedValuesPublisher = Publishers.Zip(myCollection.publisher, delayPublisher)
let subscription = delayedValuesPublisher.sink { print($0.0) }

Answer 2

Try to use flatMap(maxPublishers:) with delay(for:scheduler:) operators.

import Foundation
import Combine

var tokens: Set<AnyCancellable> = []

let valuesToPublish = [1, 2, 3, 4, 5, 6, 7, 8, 9]
valuesToPublish.publisher
    .flatMap(maxPublishers: .max(1)) { Just($0).delay(for: 1, scheduler: RunLoop.main) }
    .sink { completion in
        print("--- completion \(completion) ---")
    } receiveValue: { value in
        print("--- value \(value) ---")
    }
    .store(in: &tokens)

Setting maxPublishers property you can specify the maximum number of concurrent publisher subscriptions. Apple

Answer 3

Based on the examples provided in other answers, I came up with a solution with generics:

import Combine
import SwiftUI

struct TimedSequence<T: Any>  {
    typealias TimedJointPublisher = (Publishers.Zip<Publishers.Sequence<[T], Never>, Publishers.Autoconnect<Timer.TimerPublisher>>)
    
    var sink: AnyCancellable?

    init(array: [T], interval: TimeInterval, closure: @escaping (T) -> Void) {
        let delayPublisher = Timer.publish(every: interval, on: .main, in: .default).autoconnect()
        let timedJointPublisher = Publishers.Zip(array.publisher, delayPublisher)
        self.sink = timedJointPublisher.sink(receiveValue: {r in
            closure(r.0)
        })
    }
}

Usage:

1 - Basic types:

let m = TimedSequence(array: [1, 2, 3], interval: 2, closure: {
    element in
    let textReceived = String(element)        //assigns 1 ...2 seconds...then 2...2 seconds...then 3 to textReceived
})

let m = TimedSequence(array: ["Hello", "World"], interval: 2, closure: {
    element in
    let textReceived = element.upperCased()   //assigns HELLO ...2 seconds... then WORLD to textReceived

2 - Custom types:

class MyClass {
    var desc: String
    init(desc: String) {
        self.desc = desc
    }
}


let m = TimedSequence(array: [MyClass(desc: "t"), MyClass(desc: "s")], interval: 2, closure: {
    str in
    let textReceived = str.desc.uppercased()
})

Same as 1, except that str.desc (S and T respectively) gets assigned to textReceived at 2 second interval.