CombineExt

by CombineCommunity

CombineCommunity / CombineExt

CombineExt provides a collection of operators, publishers and utilities for Combine, that are not pr...

427 Stars 29 Forks Last release: 5 months ago (1.2.0) MIT License 61 Commits 3 Releases

Available items

No Items, yet!

The developer of this repository has not created any items for sale yet. Need a bug fixed? Help with integration? A different license? Create a request here:

CombineExt



Build Status Code Coverage for CombineExt on codecov

CombineExt supports CocoaPods CombineExt supports Swift Package Manager (SPM) CombineExt supports Carthage

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

Publishers

Subjects

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 up to four publishers into a single publisher by combining each value from

self
with the latest value from the other publishers, if any.
let taps = PassthroughSubject()
let values = CurrentValueSubject("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()
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)

CombineExt provides an additional overload —

assign(to:on​:ownership)
— which lets you specify the kind of ownersip you want for your assign operation:
strong
,
weak
or
unowned
.
// Retain `self` strongly
subscription = subject.assign(to: \.value, on: self)
subscription = subject.assign(to: \.value, on: self, ownership: .strong)

// Use a weak reference to self subscription = subject.assign(to: .value, on: self, ownership: .weak)

// Use an unowned reference to self subscription = subject.assign(to: .value, on: self, ownership: .unowned)


amb

Amb takes multiple publishers and mirrors the first one to emit an event. You can think of it as a race of publishers, where the first one to emit passes its events, while the others are ignored (there’s also a

Collection.amb
method to ease working with multiple publishers).

The name

amb
comes from the Reactive Extensions operator, also known in RxJS as
race
.
let subject1 = PassthroughSubject()
let subject2 = PassthroughSubject()

subject1 .amb(subject2) .sink(receiveCompletion: { print("amb: completed with ($0)") }, receiveValue: { print("amb: ($0)") })

subject2.send(3) // Since this subject emit first, it becomes the active publisher subject1.send(1) subject2.send(6) subject1.send(8) subject1.send(7)

subject1.send(completion: .finished) // Only when subject2 finishes, amb itself finishes as well, since it's the active publisher subject2.send(completion: .finished)

Output:

amb: 3
amb: 6
amb: completed with .finished

materialize

Convert any publisher to a publisher of its events. Given a

Publisher
, this operator will return a
Publisher, Never>
, which means your failure will actually be a regular value, which makes error handling much simpler in many use cases.
let values = PassthroughSubject()
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, Never>
, this operator will return a
Publisher
.

Note: This operator only works on publishers that were materialized with the

materialize()
operator.
let values = PassthroughSubject()
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, Never>
, this operator will return a
Publisher
.

Note: This operator only works on publishers that were materialized with the

materialize()
operator.
let values = PassthroughSubject()
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, Never>
, this operator will return a
Publisher

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()

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 an array of inner publisher outputs back.

let first = PassthroughSubject()
let second = PassthroughSubject()
let third = PassthroughSubject()
let fourth = PassthroughSubject()

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

MergeMany

This repo includes an extension for Collection that allows you to call

.merge()
directly on a collection of publishers with the same output and failure types.

This lets you arbitrarily merge many publishers and receive inner publisher outputs back from a single publisher.

let first = PassthroughSubject()
let second = PassthroughSubject()
let third = PassthroughSubject()
let fourth = PassthroughSubject()

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

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

Output:

output: 1
output: 2
output: 3
output: 4

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()
let second = PassthroughSubject()
let third = PassthroughSubject()
let fourth = PassthroughSubject()

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]

FilterMany

Filters element of a publisher collection into a new publisher collection. ```swift let intArrayPublisher = PassthroughSubject<[Int], Never>()

intArrayPublisher .filterMany { $0.isMultiple(of: 2) } .sink(receiveValue: { print($0) })

intArrayPublisher.send([10, 2, 4, 3, 8]) ```

Output:

none
[10, 2, 4, 8]

MapMany

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

let intArrayPublisher = PassthroughSubject()

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.

removeAllDuplicates

Publisher.removeAllDuplicates
and
.removeAllDuplicates(by:)
are stricter forms of Apple’s
Publisher.removeDuplicates
and
.removeDuplicates(by:)
—the operators de-duplicate across all previous value events, instead of pairwise.

If your

Output
doesn‘t conform to
Hashable
or
Equatable
, you may instead use the comparator-based version of this operator to decide whether two elements are equal.
subscription = [1, 1, 2, 1, 3, 3, 4].publisher
  .removeAllDuplicates()
  .sink(receiveValue: { print("removeAllDuplicates: \($0)") })
removeAllDuplicates: 1
removeAllDuplicates: 2
removeAllDuplicates: 3
removeAllDuplicates: 4

share(replay:)

Similar to

Publisher.share
,

.share(replay:)
can be used to create a publisher instance with reference semantics which replays a pre-defined amount of value events to further subscribers.
let subject = PassthroughSubject()

let replayedPublisher = subject .share(replay: 3)

subscription1 = replayedPublisher .sink(receiveValue: { print("first subscriber: ($0)") })

subject.send(1) subject.send(2) subject.send(3) subject.send(4)

subscription2 = replayedPublisher .sink(receiveValue: { print("second subscriber: ($0)") })

first subscriber: 1
first subscriber: 2
first subscriber: 3
first subscriber: 4
second subscriber: 2
second subscriber: 3
second subscriber: 4

prefix(duration:)

An overload on

Publisher.prefix
that that republishes values for a provided
duration
(in seconds), and then completes.
let subject = PassthroughSubject()

subscription = subject .prefix(duration: 0.5, on: DispatchQueue.main) .sink(receiveValue: { print($0) })

subject.send(1) subject.send(2) subject.send(3)

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

1
2
3

toggle()

Toggle each boolean element of a publisher collection.

let subject = PassthroughSubject()

subscription = subject .toggle() .sink(receiveValue: { print($0) })

subject.send(true) subject.send(false) subject.send(true)

false
true
false

nwise

Groups the elements of the source publisher into arrays of N consecutive elements.

let subject = PassthroughSubject()

subscription = subject .nwise(3) .sink(receiveValue: { print($0) })

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

[1, 2, 3]
[2, 3, 4]
[3, 4, 5]

pairwise

Groups the elements of the source publisher into tuples of the previous and current elements

let subject = PassthroughSubject()

subscription = subject .pairwise() .sink(receiveValue: { print("($0.0) -> ($0.1)") })

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

1 -> 2
2 -> 3
3 -> 4
4 -> 5

Publishers

This section outlines some of the custom Combine publishers CombineExt provides

AnyPublisher.create

A publisher which accepts a closure with a subscriber argument, to which you can dynamically send 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.

You should return a

Cancelable
-conforming object from the closure in which you can define any cleanup actions to execute when the pubilsher completes or the subscription to the publisher is canceled.
AnyPublisher.create { subscriber in
  // Values
  subscriber.send("Hello")
  subscriber.send("World!")

// Complete with error subscriber.send(completion: .failure(MyError.someError))

// Or, complete successfully subscriber.send(completion: .finished)

return AnyCancellable { // Perform cleanup } }

You can also use an

AnyPublisher
initializer with the same signature:
AnyPublisher { subscriber in 
    /// ...
    return AnyCancellable { }

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("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()
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

Subjects

ReplaySubject

A Combine analog to Rx’s

ReplaySubject
type. It’s similar to a
CurrentValueSubject
in that it buffers values, but, it takes it a step further in allowing consumers to specify the number of values to buffer and replay to future subscribers. Also, it will handle forwarding any completion events after the buffer is cleared upon subscription.

let subject = ReplaySubject(bufferSize: 3)

subject.send(1) subject.send(2) subject.send(3) subject.send(4)

subject .sink(receiveValue: { print($0) })

subject.send(5)

Output:

2
3
4
5

License

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

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

We use cookies. If you continue to browse the site, you agree to the use of cookies. For more information on our use of cookies please see our Privacy Policy.