Conbini provides convenience Publishers, operators, and Subscribers to squeeze the most out of Apple's Combine framework.

Usage

To use this library, you need to:

Add Conbini to your project through SPM.

// swift-tools-version:5.2
import PackageDescription

let package = Package(
    /* Your package name, supported platforms, and generated products go here */
    dependencies: [
        .package(url: "https://github.com/dehesa/package-conbini.git", from: "0.6.2")
    ],
    targets: [
        .target(name: /* Your target name here */, dependencies: ["Conbini"])
    ]
)

If you want to use Conbini's testing extension, you need to define the CONBINI_FOR_TESTING flag on your SPM targets or testing targets. Conbini testing extensions require XCTest, which is not available in runtime on some platforms (such as watchOS), or you may not want to link to such dynamic library (e.g. when building command-line tools).

targets: [
    .testTarget(name: /* Your target name here */, dependencies: ["Conbini"], swiftSettings: [.define("CONBINI_FOR_TESTING")])
]

Import Conbini in the file that needs it.

import Conbini

The testing conveniences depend on XCTest, which is not available on regular execution. That is why Conbini is offered in two flavors:

import Conbini includes all code excepts the testing conveniences.
import ConbiniForTesting includes the testing functionality only.

Operators

Publisher Operators:

handleEnd(_:)

Executes (only once) the provided closure when the publisher completes (whether successfully or with a failure) or when the publisher gets cancelled.

It performs the same operation that the standard handleEvents(receiveSubscription:receiveOutput:receiveCompletion:receiveCancel:receiveRequest:) would perform if you add similar closures to receiveCompletion and receiveCancel.

let publisher = upstream.handleEnd { (completion) in
    switch completion {
    case .none: // The publisher got cancelled.
    case .finished: // The publisher finished successfully.
    case .failure(let error): // The publisher generated an error.
    }
}

retry(on:intervals:)

Attempts to recreate a failed subscription with the upstream publisher a given amount of times waiting the specified number of seconds between failed attempts.

let apiCallPublisher.retry(on: queue, intervals: [0.5, 2, 5])
// Same functionality to retry(3), but waiting between attemps 0.5, 2, and 5 seconds after each failed attempt.

This operator accept any scheduler conforming to Scheduler (e.g. DispatchQueue, RunLoop, etc). You can also optionally tweak the tolerance and scheduler operations.

then(maxDemand:_:)

Ignores all values and executes the provided publisher once a successful completion is received. If a failed completion is emitted, it is forwarded downstream.

let publisher = setConfigurationOnServer.then {
    subscribeToWebsocket.publisher
}

This operator optionally lets you control backpressure with its maxDemand parameter. The parameter behaves like flatMap's maxPublishers, which specifies the maximum demand requested to the upstream at any given time.

Subscriber Operators:

assign(to:on:) variants.

Combine's assign(to:on:) operation creates memory cycles when the "on" object also holds the publisher's cancellable. A common situation happens when assigning a value to self.

class CustomObject {
    var value: Int = 0
    var cancellable: AnyCancellable? = nil

    func performOperation() {
        cancellable = numberPublisher.assign(to: \.value, on: self)
    }
}

Conbini's assign(to:onWeak:) operator points to the given object weakly with the added benefit of cancelling the pipeline when the object is deinitialized.

Conbini also introduces the assign(to:onUnowned:) operator which also avoids memory cycles, but uses unowned instead.

await

Wait synchronously for the response of the receiving publisher.

let publisher = Just("Hello")
    .delay(for: 2, scheduler: DispatchQueue.global())

let greeting = publisher.await

The synchronous wait is performed through DispatchGroups. Please, consider where are you using await, since the executing queue stops and waits for an answer:

Never call this property from DispatchQueue.main or any other queue who is performing any background tasks.
Awaiting publishers should never process events in the same queue as the executing queue (or the queue will become stalled).

invoke(_:on:) variants.

This operator calls the specified function on the given value/reference passing the upstream value.

struct Custom {
    func performOperation(_ value: Int) { /* do something */ }
}

let instance = Custom()
let cancellable = [1, 2, 3].publisher.invoke(Custom.performOperation, on: instance)

Conbini also offers the variants invoke(_:onWeak:) and invoke(_:onUnowned:), which avoid memory cycles on reference types.

result(onEmpty:_:)

It subscribes to the receiving publisher and executes the provided closure when a value is received. In case of failure, the handler is executed with such failure.

let cancellable = serverRequest.result { (result) in
    switch result {
    case .success(let value): ...
    case .failure(let error): ...
    }
}

The operator lets you optionally generate an error (which will be consumed by your handler) for cases where upstream completes without a value.

sink(fixedDemand:)

It subscribes upstream and request exactly fixedDemand values (after which the subscriber completes). The subscriber may receive zero to fixedDemand of values before completing, but never more than that.

let cancellable = upstream.sink(fixedDemand: 5, receiveCompletion: { (completion) in ... }) { (value) in ... }

sink(maxDemand:)

It subscribes upstream requesting maxDemand values and always keeping the same backpressure.

let cancellable = upstream.sink(maxDemand: 3) { (value) in ... }

Publishers

Deferred variants.

These publishers accept a closure that is executed once a greater-than-zero demand is requested. There are several flavors:

DeferredValue emits a single value and then completes.

The value is not provided/cached, but instead a closure will generate it. The closure is executed once a positive subscription is received.

let publisher = DeferredValue<Int,CustomError> {
    return intenseProcessing()
}

A Try variant is also offered, enabling you to throw from within the closure. It loses the concrete error type (i.e. it gets converted to Swift.Error).

DeferredResult forwards downstream a value or a failure depending on the generated Result.

let publisher = DeferredResult {
    guard someExpression else { return .failure(CustomError()) }
    return .success(someValue)
}

DeferredComplete forwards a completion event (whether success or failure).

let publisher = DeferredComplete {
    return errorOrNil
}

A Try variant is also offered, enabling you to throw from within the closure; but it loses the concrete error type (i.e. gets converted to Swift.Error).

DeferredPassthrough provides a passthrough subject in a closure to be used to send values downstream.

It is similar to wrapping a Passthrough subject on a Deferred closure, with the diferrence that the Passthrough given on the closure is already wired on the publisher chain and can start sending values right away. Also, the memory management is taken care of and every new subscriber receives a new subject (closure re-execution).

let publisher = DeferredPassthrough { (subject) in
    subject.send(something)
    subject.send(somethingElse)
    subject.send(completion: .finished)
}

There are several reason for these publishers to exist instead of using other Combine-provided closure such as Just, Future, or Deferred:

Combine's Just forwards a value immediately and each new subscriber always receive the same value.
Combine's Future executes its closure right away (upon initialization) and then cache the returned value. That value is then forwarded for any future subscription.
Deferred... publishers await for subscriptions and a greater-than-zero demand before executing. This also means, the closure will re-execute for any new subscriber.
Combine's Deferred has similar functionality to Conbini's, but it only accepts a publisher.
This becomes annoying when compounding operators.

DelayedRetry

It provides the functionality of the retry(on:intervals:) operator.

Then

It provides the functionality of the then operator.

HandleEnd

It provides the functionality of the handleEnd(_:) operator.

Extra Functionality:

Publishers.PrefetchStrategy

It has been extended with a .fatalError(message:file:line:) option to stop execution if the buffer is filled. This is useful during development and debugging and for cases when you are sure the buffer will never be filled.

publisher.buffer(size: 10, prefetch: .keepFull, whenFull: .fatalError())

Subscribers

FixedSink

It requests a fixed amount of values upon subscription and once if has received them all it completes/cancel the pipeline. The values are requested through backpressure, so no more than the allowed amount of values are generated upstream.

let subscriber = FixedSink(demand: 5) { (value) in ... }
upstream.subscribe(subscriber)

GraduatedSink

It requests a fixed amount of values upon subscription and always keep the same demand by asking one more value upon input reception. The standard Subscribers.Sink requests an .unlimited amount of values upon subscription. This might not be what we want since some times a control of in-flight values might be desirable (e.g. allowing only n in-flight* API calls at the same time).

let subscriber = GraduatedSink(maxDemand: 3) { (value) in ... }
upstream.subscribe(subscriber)

The names for these subscribers are not very good/accurate. Any suggestion is appreciated.

Testing

Conbini provides convenience subscribers to ease code testing. These subscribers make the test wait till a specific expectation is fulfilled (or making the test fail in a negative case). Furthermore, if a timeout ellapses or a expectation is not fulfilled, the affected test line will be marked in red correctly in Xcode.

expectsAll(timeout:on:)

It subscribes to a publisher making the running test wait for zero or more values and a successful completion.

let emittedValues = publisherChain.expectsAll(timeout: 0.8, on: test)

expectsAtLeast(timeout:on:)

It subscribes to a publisher making the running test wait for at least the provided amount of values. Once the provided amount of values is received, the publisher gets cancelled and the values are returned.

let emittedValues = publisherChain.expectsAtLeast(values: 5, timeout: 0.8, on: test)

This operator/subscriber accepts an optional closure to check every value received.

let emittedValues = publisherChain.expectsAtLeast(values: 5, timeout: 0.8, on: test) { (value) in
    XCTAssert...
}

expectsCompletion(timeout:on:)

It subscribes to a publisher making the running test wait for a successful completion while ignoring all emitted values.

publisherChain.expectsCompletion(timeout: 0.8, on: test)

expectsFailure(timeout:on:)

It subscribes to a publisher making the running test wait for a failed completion while ignoring all emitted values.

publisherChain.expectsFailure(timeout: 0.8, on: test)

expectsOne(timeout:on:)

It subscribes to a publisher making the running test wait for a single value and a successful completion. If more than one value are emitted or the publisher fails, the subscription gets cancelled and the test fails.

let emittedValue = publisherChain.expectsOne(timeout: 0.8, on: test)

XCTestCase has been extended to support the following functionality.

wait(seconds:)

Locks the receiving test for interval amount of seconds.

final class CustomTests: XCTestCase {
    func testSomething() {
        let subject = PassthroughSubject<Int,Never>()
        let cancellable = subject.sink { print($0) }
        
        let queue = DispatchQueue.main
        queue.asyncAfter(.now() + 1) { subject.send(1) }
        queue.asyncAfter(.now() + 2) { subject.send(2) }

        self.wait(seconds: 3)
        cancellable.cancel()
    }
}

References

Apple's Combine documentation.
The Combine book is an excellent Ray Wenderlich book about the Combine framework.
Cocoa with love has a great series of articles about the inner workings of Combine: 1, 2, 3.
OpenCombine is an open source implementation of Apple's Combine framework.
CombineX is an open source implementation of Apple's Combine framework.