How to enforce minimum delay between events from Combine publisher

Question

How to enforce minimum interval between events emitted from Combine publisher? With assumption that I want all events from upstream to be emitted but with minimum interval between them, let's say 1s. If the interval between two events in the upstream is > 1s the events should be emitted as they are. So far I've tried something like this:

let subject = PassthroughSubject<Int, Never>()

let result = subject.flatMap(maxPublishers: .max(1)) {
    Just($0).delay(for: 1, scheduler: RunLoop.main)
}

let cancellable = result.sink {
    print("--- value \($0) ---")
}


// Emitting values
subject.send(1)

DispatchQueue.main.asyncAfter(deadline: .now() + 0.3) {
    subject.send(2)
}

DispatchQueue.main.asyncAfter(deadline: .now() + 0.6) {
    subject.send(3)
}

DispatchQueue.main.asyncAfter(deadline: .now() + 2.5) {
    subject.send(4)
}

but the result I get is:

--- value 1 ---
--- value 4 ---

Any idea how to achieve it?

Answer 1

I do not think that combine provides a tool for that, .throttle and .debounce are definitely not what you are looking for. Using flatMap(maxPublishers: .max(1)) will make 2 and 3 vanish as it will prevent the .delay from receive them (there is no storage so you have to add one).

This is a workaround I found using buffer that might do the trick: Add delay between values emitted by publisher in Combine Framework in iOS

I tested using you code and it work properly.

subject
    .buffer(size: Int.max, prefetch: .byRequest, whenFull: .dropOldest)
    .flatMap(maxPublishers: .max(1)) {
          Empty().delay(for: 1, scheduler: RunLoop.main).prepend($0)
    }.sink(receiveValue: { value in
          print(value)
    }).store(&cancellables)

EDIT: tI improved my first answer following @Daniel T comment, applying the delay after popping out the value using .prepend is way better.

Answer 2

@xfost wrote a great answer. Here's another solution:

let subject = PassthroughSubject<Int, Never>()

let result = Publishers.Zip(
    Timer.publish(every: 1, on: RunLoop.main, in: .default).autoconnect().prepend(Date()),
    subject
)
    .map { $1 }

let cancellable = result
    .sink {
        print("--- value \($0) ---")
    }


// Emitting values
subject.send(1)

DispatchQueue.main.asyncAfter(deadline: .now() + 0.3) {
    subject.send(2)
}

DispatchQueue.main.asyncAfter(deadline: .now() + 0.6) {
    subject.send(3)
}

DispatchQueue.main.asyncAfter(deadline: .now() + 2.5) {
    subject.send(4)
}

DispatchQueue.main.asyncAfter(deadline: .now() + 7) {
    subject.send(5)
}