Using Back-Pressure By flatMap in Combine framework

In the Combine framework, when using flatMap for irreparable processing requests using Web API, it is better to use the maxPublishers argument and use .flatMap(maxPublishers: .max(1)).

For example, by using .flatMap (maxPublishers: .max(1)) as shown below, the Web API POST will not process input events until the result is subscribed.

let cancelable = inputEvents
    .flatMap(maxPublishers: .max(1)) {
        // Example: irreparable processing requests
        createUser(name: $0.email, $0.password)
    }
    .sink {
        ... 
    }

Verification code

Verify the operation with two source codes.

• Example without maxPublishers argument for flatMap
  • .unlimited by default
• Example using .flatMap (maxPublishers: .max(1))

To conclude, .flatMap (maxPublishers: .max(1)) uses a flatMap to fix the number of Publishers created, even if there are new inputs until the result is subscribed. Do not execute.

Example without maxPublishers argument in flatMap

• Spec
  • Input events are (1...3).publisher
  • flatMap
    • In the first event, asynchronously delay the number after 1 second and stream the number as a string
    • After the first one, asynchronously stream numbers as strings

import Foundation
import Combine
import PlaygroundSupport

let cancelable = (1...3).publisher
    .flatMap { value -> Future<String, Never> in
        print("🍏flatMap: \(value)")
        if value == 1 {
            return Future<String, Never> { promise in
                let v = value
                // delay the event of the stream created in the first shot.
                DispatchQueue.main.asyncAfter(deadline: .now() + 1.0) {
                    print("🐢after: \(v)")
                    promise(.success("\(v)"))
                }
            }
        } else {
            return Future<String, Never> { promise in
                let v = value
                DispatchQueue.main.async {
                    print("🐰after : \(v)")
                    promise(.success("\(v)"))
                }
            }
        }
    }
    .sink(receiveCompletion: { completion in
        switch completion {
        case .finished:
            print("🍎sink finished:", completion)
        case .failure(let error):
            print("🍎sink failure:", error)
        }
    }) { value in
        print("🍎sink received: \(String(describing: value))")
    }


let page = PlaygroundPage.current
page.needsIndefiniteExecution = true

DispatchQueue.main.asyncAfter(deadline: .now() + 5.0) {
    // Instructs Xcode that the playground page has finished execution.
    print("🕘finished")
    page.finishExecution()
}

This code has the following output:

🍏flatMap: 1
🍏flatMap: 2
🍏flatMap: 3
🐰after : 2
🍎sink received: 2
🐰after : 3
🍎sink received: 3
🐢after: 1
🍎sink received: 1
🍎sink finished: finished
🕘finished

Use asncAfter(deadline: .now() + 1.0) to delay the event of the first stream created. However, events flow one after another and switch to a new stream with flatMap.

Example of using .flatMap (maxPublishers: .max(1))

Change the following from the previous code

.flatMap { value -> Future<String, Error> in

↓

.flatMap(maxPublishers: .max(1)) { value -> Future<String, Error> in

This code has the following output:

🍏flatMap: 1
🐢after: 1
🍎sink received: 1
🍏flatMap: 2
🐰after : 2
🍎sink received: 2
🍏flatMap: 3
🐰after : 3
🍎sink received: 3
🍎sink finished: finished
🕘finished

As before, use asncAfter(deadline: .now() + 1.0) to delay the event of the first stream created. It seems that the next event occurs after the delayed event is sink received: 1.

Conclusion

• If you do not use the maxPublishers argument in flatMap, event creation does not wait for the end.
  • It does not use Back-Pressure.
• If you specify max(1) for maxPublishers, event creation waits for completion.
  • It is requesting 1 as Back-Pressure.
    • The request repeats with each subscription.
  • Even if you specify max(1), the subscription does not end when you create a one-time event.