CombineExt

CombineExt provides a collection of operators, publishers and utilities for Combine, that are not provided by Apple themselves, but are common in other Reactive Frameworks and standards.

The original inspiration for many of these additions came from my journey investigating Combine after years of RxSwift and ReactiveX usage.

All operators, utilities and helpers respect Combine's publisher contract, including backpressure.

Operators

• withLatestFrom
• flatMapLatest
• assign
• materialize
• values
• failures
• dematerialize
• zip(with:) and Collection.zip
• combineLatest(with:) and Collection.combineLatest
• mapMany(_:)
• setOutputType(to:)

Publishers

• AnyPublisher.create
• CurrentValueRelay
• PassthroughRelay

Note: This is still a relatively early version of CombineExt, with much more to be desired. I gladly accept PRs, ideas, opinions, or improvements. Thank you! :)

Installation

CocoaPods

Add the following line to your Podfile:

pod 'CombineExt'

Swift Package Manager

Add the following dependency to your Package.swift file:

.package(url: "https://github.com/CombineCommunity/CombineExt.git", from: "1.0.0")

Carthage

Carthage support is offered as a prebuilt binary.

Add the following to your Cartfile:

github "CombineCommunity/CombineExt"

Operators

This section outlines some of the custom operators CombineExt provides.

withLatestFrom

Merges two publishers into a single publisher by combining each value from self with the latest value from the second publisher, if any.

let taps = PassthroughSubject<Void, Never>()
let values = CurrentValueSubject<String, Never>("Hello")

taps
  .withLatestFrom(values)
  .sink(receiveValue: { print("withLatestFrom: \($0)") })

taps.send()
taps.send()
values.send("World!")
taps.send()

Output:

withLatestFrom: Hello
withLatestFrom: Hello
withLatestFrom: World!

---

flatMapLatest

Transforms an output value into a new publisher, and flattens the stream of events from these multiple upstream publishers to appear as if they were coming from a single stream of events.

Mapping to a new publisher will cancel the subscription to the previous one, keeping only a single subscription active along with its event emissions.

Note: flatMapLatest is a combination of map and switchToLatest.

let trigger = PassthroughSubject<Void, Never>()
trigger
    .flatMapLatest { performNetworkRequest() }

trigger.send()
trigger.send() // cancels previous request
trigger.send() // cancels previous request

---

assign

CombineExt provides custom overloads of assign(to:on:) that let you bind a publisher to multiple keypath targets simultaneously.

var label1: UILabel
var label2: UILabel
var text: UITextField

["hey", "there", "friend"]
    .publisher
    .assign(to: \.text, on: label1,
            and: \.text, on: label2,
            and: \.text, on: text)

---

materialize

Convert any publisher to a publisher of its events. Given a Publisher<Output, MyError>, this operator will return a Publisher<Event<Output, MyError>, Never>, which means your failure will actually be a regular value, which makes error handling much simpler in many use cases.

let values = PassthroughSubject<String, MyError>()
enum MyError: Swift.Error {
  case ohNo
}

values
  .materialize()
  .sink(receiveCompletion: { print("materialized: completed with \($0)") },
        receiveValue: { print("materialized: \($0)") })

values.send("Hello")
values.send("World")
values.send("What's up?")
values.send(completion: .failure(.ohNo))

Output:

materialize: .value("Hello")
materialize: .value("World")
materialize: .value("What's up?")
materialize: .failure(.ohNo)
materialize: completed with .finished

---

values

Given a materialized publisher, publish only the emitted upstream values, omitting failures. Given a Publisher<Event<String, MyError>, Never>, this operator will return a Publisher<String, Never>.

Note: This operator only works on publishers that were materialized with the materialize() operator.

let values = PassthroughSubject<String, MyError>()
enum MyError: Swift.Error {
  case ohNo
}

values
  .materialize()
  .values()
  .sink(receiveValue: { print("values: \($0)") })

values.send("Hello")
values.send("World")
values.send("What's up?")
values.send(completion: .failure(.ohNo))

Output:

values: "Hello"
values: "World"
values: "What's up?"

---

failures

Given a materialized publisher, publish only the emitted upstream failure, omitting values. Given a Publisher<Event<String, MyError>, Never>, this operator will return a Publisher<MyError, Never>.

Note: This operator only works on publishers that were materialized with the materialize() operator.

let values = PassthroughSubject<String, MyError>()
enum MyError: Swift.Error {
  case ohNo
}

values
  .materialize()
  .failures()
  .sink(receiveValue: { print("failures: \($0)") })

values.send("Hello")
values.send("World")
values.send("What's up?")
values.send(completion: .failure(.ohNo))

Output:

failure: MyError.ohNo

---

dematerialize

Converts a previously-materialized publisher into its original form. Given a Publisher<Event<String, MyError>, Never>, this operator will return a Publisher<String, MyError>

Note: This operator only works on publishers that were materialized with the materialize() operator.

---

partition

Partition a publisher's values into two separate publishers of values that match, and don't match, the provided predicate.

let source = PassthroughSubject<Int, Never>()

let (even, odd) = source.partition { $0 % 2 == 0 }

even.sink(receiveValue: { print("even: \($0)") })
odd.sink(receiveValue: { print("odd: \($0)") })

source.send(1)
source.send(2)
source.send(3)
source.send(4)
source.send(5)

Output:

odd: 1
even: 2
odd: 3
even: 4
odd: 5

---

ZipMany

This repo includes two overloads on Combine’s Publisher.zip methods (which, at the time of writing only go up to arity three).

This lets you arbitrarily zip many publishers and receive either an array of inner publisher outputs back.

let first = PassthroughSubject<Int, Never>()
let second = PassthroughSubject<Int, Never>()
let third = PassthroughSubject<Int, Never>()
let fourth = PassthroughSubject<Int, Never>()

subscription = first
  .zip(with: second, third, fourth)
  .map { $0.reduce(0, +) }
  .sink(receiveValue: { print("zipped: \($0)") })

first.send(1)
second.send(2)
third.send(3)
fourth.send(4)

You may also use .zip() directly on a collection of publishers with the same output and failure types, e.g.

[first, second, third, fourth]
  .zip()
  .map { $0.reduce(0, +) }
  .sink(receiveValue: { print("zipped: \($0)") })

Output:

zipped: 10

CombineLatestMany

This repo includes two overloads on Combine’s Publisher.combineLatest methods (which, at the time of writing only go up to arity three) and an Collection.combineLatest constrained extension.

This lets you arbitrarily combine many publishers and receive an array of inner publisher outputs back.

let first = PassthroughSubject<Bool, Never>()
let second = PassthroughSubject<Bool, Never>()
let third = PassthroughSubject<Bool, Never>()
let fourth = PassthroughSubject<Bool, Never>()

subscription = [first, second, third, fourth]
  .combineLatest()
  .sink(receiveValue: { print("combineLatest: \($0)") })

first.send(true)
second.send(true)
third.send(true)
fourth.send(true)

first.send(false)

Output:

combineLatest: [true, true, true, true]
combineLatest: [false, true, true, true]

MapMany

Projects each element of a publisher collection into a new publisher collection form.

let intArrayPublisher = PassthroughSubject<[Int], Never>()
    
intArrayPublisher
  .mapMany(String.init)
  .sink(receiveValue: { print($0) })
    
intArrayPublisher.send([10, 2, 2, 4, 3, 8])

Output:

["10", "2", "2", "4", "3", "8"]

setOutputType

Publisher.setOutputType(to:) is an analog to .setFailureType(to:) for when Output is constrained to Never. This is especially helpful when chaining operators after an .ignoreOutput() call.

Publishers

This section outlines some of the custom Combine publishers CombineExt provides

AnyPublisher.create

A publisher which accepts a factory closure to which you can dynamically push value or completion events.

This lets you easily create custom publishers to wrap any non-publisher asynchronous work, while still respecting the downstream consumer's backpressure demand.

AnyPublisher<String, MyError>.create { subscriber in
  // Values
  subscriber(.value("Hello"))
  subscriber(.value("World!"))
  
  // Complete with error
  subscriber(.failure(MyError.someError))
  
  // Or, complete successfully
  subscriber(.finished)
}

You can also use an AnyPublisher initializer with the same signature:

AnyPublisher<String, MyError> { subscriber in 
    /// ...
}

---

CurrentValueRelay

A CurrentValueRelay is identical to a CurrentValueSubject with two main differences:

• It only accepts values, but not completion events, which means it cannot fail.
• It only publishes a .finished event upon deallocation.

let relay = CurrentValueRelay<String>("well...")

relay.sink(receiveValue: { print($0) }) // replays current value, e.g. "well..."

relay.accept("values")
relay.accept("only")
relay.accept("provide")
relay.accept("great")
relay.accept("guarantees")

Output:

well...
values
only
provide
great
guarantees

---

PassthroughRelay

A PassthroughRelay is identical to a PassthroughSubject with two main differences:

• It only accepts values, but not completion events, which means it cannot fail.
• It only publishes a .finished event upon deallocation.

let relay = PassthroughRelay<String>()
relay.accept("well...")

relay.sink(receiveValue: { print($0) }) // does not replay past value(s)

relay.accept("values")
relay.accept("only")
relay.accept("provide")
relay.accept("great")
relay.accept("guarantees")

Output:

values
only
provide
great
guarantees

License

MIT, of course ;-) See the LICENSE file.

The Apple logo and the Combine framework are property of Apple Inc.