What do collect, prefix, dropFirst, drop(while:), scan, reduce, removeDuplicates, compactMap, output(in:), prepend and append emit, and which of them wait for completion?

11.7.1 · L2 · Interview

Predict first

What does each operator emit for 1 through 6 (a pipe is completion)?

import Combine

var bag = Set<AnyCancellable>()
func run<P: Publisher>(_ name: String, _ p: P) where P.Failure == Never {
    var out: [String] = []
    p.sink(receiveCompletion: { _ in out.append("|") }, receiveValue: { out.append("\($0)") }).store(in: &bag)
    print(name, out.joined(separator: " "))
}
let nums = (1...6).publisher
run("collect(2):", nums.collect(2))
run("prefix(3):", nums.prefix(3))
run("dropFirst(2):", nums.dropFirst(2))
run("drop(while < 3):", nums.drop(while: { $0 < 3 }))
run("scan:", nums.scan(0, +))
run("reduce:", nums.reduce(0, +))
run("removeDuplicates:", [1, 1, 2, 2, 1].publisher.removeDuplicates())
run("compactMap:", ["1", "x", "3"].publisher.compactMap { Int($0) })
run("output(in:):", nums.output(in: 1..<3))
run("prepend/append:", nums.prefix(2).prepend(0).append(9))
  • collect(2): [1, 2] [3, 4] [5, 6] | / prefix(3): 1 2 3 | / scan: 1 3 6 10 15 21 | / reduce: 21 | ...
  • scan: 21 | / reduce: 1 3 6 10 15 21 |
  • prefix(3): 1 2 3 (no completion)

Mental model. scan is a running tally on a scoreboard; reduce is the final score announced only after the whistle. If the match never ends, you never get a final score.

Most operators are best remembered by what they do to the stream’s length and timing. prefix(3) passes the first three values then completes the stream early; dropFirst(2) and drop(while:) skip values at the start (the latter only while the predicate holds, then everything); output(in: 1..<3) selects by position. collect(2) groups into arrays of two and, at completion, emits the leftover group, so it has to wait for a full group and delays values. compactMap drops nil results; removeDuplicates drops a value only when it equals the previous one, so 1 1 2 2 1 becomes 1 2 1, not a set. prepend and append add values before and after.

The pair people mix up is scan versus reduce: scan emits every running total (1 3 6 10 15 21), so it is a good fit for state over time, while reduce emits one final value and only when the upstream completes. On a stream that never completes (a subject, a notification), reduce, collect() without a count, last, count and max never produce anything. So before reaching for an aggregating operator, ask: does this upstream finish? If not, use scan or a windowing operator (collect(.byTime)) instead.

scan emits each running total, reduce one value at the end, prefix completes early, removeDuplicates compares only neighbours.

Common traps

  • Using reduce, last or count on an endless stream. They only emit at completion, so nothing ever arrives.
  • Expecting removeDuplicates to deduplicate globally. It only compares with the previous value.

They might follow up with

  • How do I batch by time instead of count? collect(.byTime(scheduler, interval)) or .byTimeOrCount.
  • What does prefix(1) do to the upstream? It cancels it after the first value, so it is a clean “take one”.

Related questions

  • 11.7.2 What does flatMap(maxPublishers:) do, and why can limiting it silently lose values?
  • 11.7.3 Where should a catch go in a search pipeline, and what happens to the stream if you put it outside the…
  • 11.1.1 What are Publisher, Subscriber, Subscription and Subscriber.Input/Failure, and when does a pipeline actually…
  • 11.2.1 How do you read a pipeline, and how do combineLatest, merge, zip and scan differ?
  • 11.2.3 What do debounce, throttle and removeDuplicates do, and what do share and multicast change?

What does flatMap(maxPublishers:) do, and why can limiting it… →