gem-hunter

ReactiveKit is a lightweight Swift framework for reactive and functional reactive programming that enables you to get into the reactive world today.

The framework is compatible with all Apple platforms and Linux. If you are developing an iOS or macOS app, make sure to also check out Bond framework that provides UIKit and AppKit bindings, reactive delegates and data sources.

ReactiveKit is currently in a process of API alignment with Apple's Combine framework. Types and functions are being renamed, where applicable, to match those of Combine. It's important to note that ReactiveKit will not become a drop-in replacement for Combine. The goal is to make interoperability and transition smooth. All work is being done in a backward compatible way and will be done gradually over a number of releases. Check out release notes to follow the process.

This document will introduce the framework by going through its implementation. By the end you should be equipped with a pretty good understanding of how is the framework implemented and what are the best ways to use it.

To get started quickly, clone the project and explore available tutorials in the playgrounds of the workspace!

• Introduction
• Signals
• Wrapping asynchronous calls into signals
• Disposing signals
• Transforming signals
• More about errors
• Creating simple signals
• Disposing in a bag
• Threading
• Bindings
  • Binding targets
  • Binding to a property
• Sharing sequences of events
• Subjects
• Connectable signals
  • Implementing shareReplay operator
• Handling signal errors
  • Generalized error handling
• Tracking signal state
  • Single signal state tracking
• Property
• Loading signals
  • Consuming loading state
  • Transforming loading signals
  • Loading property
• Other common patterns
  • Performing an action on .next event
  • Combining multiple signals
• Debugging
• Requirements
• Installation
  • Carthage
  • CocoaPods
  • Swift Package Manager
• Communication
• Additional Documentation
• License

Consider how text of a text field changes as a user enters his name. Each entered letter gives us a new state.

---[J]---[Ji]---[Jim]--->

We can think of these state changes as a sequence of events. It is quite similar to an array or a list, but with the difference that events are generated over time as opposed to having them all in memory at once.

The idea behind reactive programming is that everything can be represented as a sequence. Let us consider another example - a network request.

---[Response]--->

The outcome of a network request is a response. Although we have only one response, we can still think of it as a sequence. An array of one element is still an array.

Arrays are finite so they have a property that we call size. It is a measure of how much memory the array occupies. When we talk about sequences over time, we do not know how many events they will generate during their lifetime. We do not know how many letters the user will enter. However, we would still like to know when the sequence is done generating the events.

To get that information, we can introduce a special kind of event - a completion event. It is an event that marks the end of a sequence. No event shall follow the completion event.

We will denote completion event visually with a vertical bar.

---[J]---[Ji]---[Jim]---|--->

The completion event is important because it tells us that whatever was going on is now over. We can finalize the work at that point and dispose any resources that might have been used in processing the sequence.

Unfortunately, the universe is not governed by order, rather by chaos. Unexpected things happen and we have to anticipate that. For example, a network request can fail so instead of a response, we can receive an error.

---!Error!--->

In order to represent errors in our sequences, we will introduce yet another kind of event. We will call it a failure event. The failure event will be generated when something unexpected happens. Just like the completion event, the failure event will also represent the end of a sequence. No event shall follow the failure event.

Let us see how the event is defined in ReactiveKit.

extension Signal {

    /// An event of a sequence.
    public enum Event {

        /// An event that carries next element.
        case next(Element)

        /// An event that represents failure. Carries an error.
        case failed(Error)

        /// An event that marks the completion of a sequence.
        case completed
    }
}

It is just an enumeration of the three kinds of events we have. Sequences will usually have zero or more .next events followed by either a .completed or a .failed event.

What about sequences? In ReactiveKit they are called signals. Here is the protocol that defines them.

/// Represents a sequence of events.
public protocol SignalProtocol {

  /// The type of elements generated by the signal.
  associatedtype Element

  /// The type of error that can terminate the signal.
  associatedtype Error: Swift.Error

  /// Register the given observer.
  /// - Parameter observer: A function that will receive events.
  /// - Returns: A disposable that can be used to cancel the observation.
  public func observe(with observer: @escaping Observer<Element, Error>) -> Disposable
}

A signal represents the sequence of events. The most important thing you can do on the sequence is observe the events it generates. Events are received by the observer. An observer is nothing more than a function that accepts an event.

/// Represents a type that receives events.
public typealias Observer<Element, Error: Swift.Error> = (Signal<Element, Error>.Event) -> Void

We have seen the protocol that defines signals, but what about the implementation? Let us implement a basic signal type!

public struct Signal<Element, Error: Swift.Error>: SignalProtocol {

  private let producer: (Observer<Element, Error>) -> Void

  public init(producer: @escaping (Observer<Element, Error>) -> Void) {
    self.producer = producer
  }

  public func observe(with observer: @escaping Observer<Element, Error>) {
    producer(observer)
  }
}

We have defined our signal as a struct of one property - a producer. As you can see, producer is just a function that takes the observer as an argument. When we start observing the signal, what we do is basically execute the producer with the given observer. That is how simple signals are!

Signal in ReactiveKit is implemented almost like what we have shown here. It has few additions that give us some guarantees that we will talk about later.

Let us create an instance of the signal that first sends three positive integers to the observer and then completes.

Visually that would look like:

---[1]---[2]---[3]---|--->

While in the code, we would do:

let counter = Signal<Int, Never> { observer in

  // send first three positive integers
  observer(.next(1))
  observer(.next(2))
  observer(.next(3))

  // send completed event
  observer(.completed)
}

Since the observer is just a function that receives events, we just execute it with the event whenever we want to send a new one. We always finalize the sequence by sending either .completed or .failed event so that the receiver knows when the signal is over with event production.

ReactiveKit wraps the observer into a struct with various helper methods to make it easier to send events. Here is a protocol that defines it.

/// Represents a type that receives events.
public protocol ObserverProtocol {
    
    /// Type of elements being received.
    associatedtype Element
    
    /// Type of error that can be received.
    associatedtype Error: Swift.Error

    /// Send the event to the observer.
    func on(_ event: Signal<Element, Error>.Event)
}

Our observer we introduced earlier is basically the on(_:) method. ReactiveKit also provides this extension on the observer:

public extension ObserverProtocol {

    /// Convenience method to send `.next` event.
    public func receive(_ element: Element) {
        on(.next(element))
    }

    /// Convenience method to send `.failed` or `.completed` event.
    public func receive(completion: Subscribers.Completion<Error>) {
        switch completion {
        case .finished:
            on(.completed)
        case .failure(let error):
            on(.failed(error))
        }
    }

    /// Convenience method to send `.next` event followed by a `.completed` event.
    public func receive(lastElement element: Element) {
        receive(element)
        receive(completion: .finished)
    }
}

So with ReactiveKit we can implement the previous example like this:

let counter = Signal<Int, Never> { observer in

  // send first three positive integers
  observer.receive(1)
  observer.receive(2)
  observer.receive(3)

  // send completed event
  observer.receive(completion: .finished)
}

What happens when we observe such a signal? Remember, the observer is a function that receives events so we can just pass a closure to our observe method.

counter.observe(with: { event in
  print(event)
})

Of course, we will get our three events printed out.

next(1)
next(2)
next(3)
completed

We can easily wrap asynchronous calls into signals because of the way we implemented our Signal type. Let us say that we have an asynchronous function that fetches the user.

func getUser(completion: (Result<User, ClientError>) -> Void) -> URLSessionTask

The function communicates fetch result through a completion closure and a Result type whose instance will contain either a user or an error. To wrap this into a signal, all we need to do is call that function within our signal initializer's producer closure and send relevant events as they happen.

func getUser() -> Signal<User, ClientError> {
  return Signal { observer in
    getUser(completion: { result in
      switch result {
      case .success(let user):
        observer.receive(user)
        observer.receive(completion: .finished)
      case .failure(let error):
        observer.receive(completion: .failure(error))
    })
    // return disposable, continue reading
  }
}

If we now observe this signal, we will get either a user and a completion event

---[User]---|--->

or an error

---!ClientError!--->

In code, getting the user would look like:

let user = getUser()

user.observe { event in
  print(event) // prints ".next(user), .completed" in case of successful response
}

Let me ask you one important question here. When is the request to get the user executed, i.e. when is the asynchronous function getUser(completion:) called? Think about it.

We call getUser(completion:) within our producer closure that we pass to the signal initializer. That closure however is not executed when the signal is created. That means that the code let user = getUser() does not trigger the request. It merely creates a signal that knows how to execute the request.

Request is made when we call the observe(with:) method because that is the point when our producer closure gets executed. It also means that if we call the observe(with:) method more than once, we will call the producer more than once, so we will execute the request more than once. This is a very powerful aspect of signals and we will get back to it later when we will talk about sharing sequences of events. For now just remember that each call to observe(with:) means that events get produced all over again.

Our example function getUser(completion:) returns a URLSessionTask object. We often do not think about it, but HTTP requests can be cancelled. When the screen gets dismissed, we should probably cancel any ongoing requests. A way to do that is to call cancel() on URLSessionTask that we used to make the request. How do we handle that with signals?

If you have been reading the code examples carefully, you have probably noticed that we did not correctly conform our Signal to SignalProtocol. The protocol specifies that the observe(with:) method returns something called Disposable. A disposable is an object that can cancel the signal observation and any underlying tasks.

Let me give you the definition of a disposable from ReactiveKit.

public protocol Disposable {

  /// Cancel the signal observation and any underlying tasks.
  func dispose()

  /// Returns `true` if already disposed.
  var isDisposed: Bool { get }
}

It has a method to cancel the observation and a property that can tell us if it has been disposed or not. Cancelling the observation is also referred to as disposing the signal.

There are various implementations of Disposable, but let us focus on the one that is most commonly used in signal creation. When the signal gets disposed, we often want to perform some action to clean up the resources or stop underlying tasks. What a better way to do that then to execute a closure when the the signal gets disposed. Let us implement a disposable that executes a given closure when it gets disposed. We will call it BlockDisposable.

public final class BlockDisposable: Disposable {

  private var handler: (() -> Void)?

  public var isDisposed: Bool {
    return handler == nil
  }

  public init(_ handler: @escaping () -> Void) {
    self.handler = handler
  }

  public func dispose() {
    handler?()
    handler = nil
  }
}

Simple enough. It just executes the given closure when the dispose() method is called. How do we use such a disposable? Well, we will need to improve our signal implementation.

Who should create the disposable? Since the disposable represents a way to communicate the signal cancellation, it is obviously the one who created the signal that should also provide a disposable that can cancel the signal. To do that we will refactor the signal producer to return a disposable. Additionally, we will return that disposable from the observe(with:) method so that whoever will be observing the signal can cancel the observation.

public struct Signal<Element, Error: Swift.Error>: SignalProtocol {

  private let producer: (Observer<Element, Error>) -> Disposable

  public init(producer: @escaping (Observer<Element, Error>) -> Disposable) {
    self.producer = producer
  }

  public func observe(with observer: @escaping Observer<Element, Error>) -> Disposable {
    return producer(observer)
  }
}

This means that when we are creating a signal, we also have to provide a disposable. Let us refactor our asynchronous function wrapper signal to provide a disposable.

func getUser() -> Signal<User, ClientError> {
  return Signal { observer in
    let task = getUser(completion: { result in
      switch result {
      case .success(let user):
        observer.receive(user)
        observer.receive(completion: .finished)
      case .failure(let error):
        observer.receive(completion: .failure(error))
    })

    return BlockDisposable {
      task.cancel()
    }
  }
}

We just return an instance of BlockDisposable that cancels the task when it gets disposed. We can then get that disposable when observing the signal.

let disposable = getUser().observe { event in
  print(event)
}

When we are no longer interested in signal events, we can just dispose the disposable. It will cancel the observation and cancel network task.

disposable.dispose()

For the actual implementation of Signal in ReactiveKit there are additional mechanisms that prevent events from being sent when the signal is disposed so there is a guarantee that no events will be received after the signal is disposed. Any events sent from the producer after the signal is disposed are ignored.

In ReactiveKit, signals are automatically disposed when they terminate with either a .completed or .failed event.

This is all so good, but why should we do it? What are the benefits? Here comes the most interesting aspect of reactive programming - signal operators.

Operators are functions (i.e. methods) that transform one or more signals into other signals. One of the basic operations on signals is filtering. Say that we have a signal of city names, but we want only the names starting with letter P.

filter(
---[Berlin]---[Paris]---[London]---[Porto]---|--->
)

--------------[Paris]--------------[Porto]---|--->

How could we implement such an operator? Very easily.

extension SignalProtocol {

  /// Emit only elements that pass `isIncluded` test.
  public func filter(_ isIncluded: @escaping (Element) -> Bool) -> Signal<Element, Error> {
    return Signal { observer in
      return self.observe { event in
        switch event {
        case .next(let element):
          if isIncluded(element) {
            observer.receive(element)
          }
        default:
          observer(event)
        }
      }
    }
  }
}

We have written an extension method on the SignalProtocol in which we create a signal. In the created signal's producer we observe self - the signal we are filtering - and propagate .next events that pass the test. We also propagate completion and failure events in the default case.

We use the operator by calling it on a signal.

cities.filter { $0.hasPrefix("P") }.observe { event in
  print(event) // prints .next("Paris"), .next("Porto"), .completed
}

There are many operators on signals. ReactiveKit is basically a collection of signal operators. Let us see another common one.

When observing signals we often do not care about terminal events, all we care about is the elements in .next events. We could write an operator that gives us just that.

extension SignalProtocol {

  /// Register an observer that will receive elements from `.next` events of the signal.
  public func observeNext(with observer: @escaping (Element) -> Void) -> Disposable {
    return observe { event in
      if case .next(let element) = event {
        observer(element)
      }
    }
  }
}

It should be pretty straightforward - just propagate the elements from .next event and ignore everything else. Now we can do:

cities.filter { $0.hasPrefix("P") }.observeNext { name in
  print(name) // prints "Paris", "Porto"
}

ReactiveKit also provides observeFailed and observeCompleted operators when you are interested only in those events.

Writing operators on signals is as simple as writing an extension method. When you need something that is not provided by the framework, just write it by yourself! ReactiveKit is written to be simple to understand. Whenever you are stuck, just look into the implementation.

We have seen that a signal can terminate with an error. In our getUser example, when the network request fails we send the .failed event. For that reason, our Signal type is generic both over the elements it sends and the errors it can fail with. There are, however, situations when signals are guaranteed not to fail, i.e. when they can never send an error. How do we define that?

ReactiveKit provides the following type:

/// An error type that cannot be instantiated. Used to make signals non-failable.
public enum Never: Error {
}

An enum with no cases that conforms to the Swift.Error protocol. Since it has no cases, we can never make an instance of it. We will use this trick to get the compile-time guarantee that a signal will not fail.

For example, if we try

let signal = Signal<Int, Never> { observer in
  ...
  observer.failed(/* What do I send here? */)
  ...
}

we will hit the wall because we cannot create an instance of Never so we cannot send the .failed event. This is a very powerful and important feature because whenever you see a signal whose errors are specialized to the Never type you can safely assume that that signal will not fail - because it cannot.

Bindings only work with safe (non-failable) signals.

You will often need a signal that emits just one element and then completes. To make it, use the static method just.

let signal = Signal<Int, Never>.just(5)

That will give you following signal:

---5-|--->

If you need a signal that fires multiple elements and then completes, you can convert any Sequence to a signal with the static method sequence.

let signal = Signal<Int, Never>.sequence([1, 2, 3])

---1-2-3-|--->

To create a signal that just completes without sending any elements, do

let signal = Signal<Int, Never>.completed()

---|--->

To create a signal that just fails, do

let signal = Signal<Int, MyError>.failed(MyError.someError)

---!someError!--->

You can also create a signal that never sends any events (i.e. a signal that never terminates).

let signal = Signal<Int, Never>.never()

------>

Sometimes you will need a signal that sends a specific element after a certain amount of time passes:

let signal = Signal<Int, Never>(just: 5, after: 60)

---/60 seconds/---5-|-->

Finally, when you need a signal that sends an integer every interval seconds, do

let signal = Signal<Int, Never>(sequence: 0..., interval: 5)

---0---1---2---3---...>

Handling disposables can be cumbersome when doing multiple observations. To simplify it, ReactiveKit provides a type called DisposeBag. It is a container into which you can put your disposables. The bag will dispose all disposables that were put into it when it gets deallocated.

class Example {

  let bag = DisposeBag()

  init() {
    ...
    someSignal
      .observe { ... }
      .dispose(in: bag)

    anotherSignal
      .observe { ... }
      .dispose(in: bag)
    ...
  }
}

In the example, instead of handling the disposables, we just put them into a bag by calling dispose(in:) method on the disposable. Disposables will then get disposed automatically when the bag gets deallocated. Note that you can also call dispose() on the bag to dispose its contents at will.

ReactiveKit provides a bag on NSObject and its subclasses out of the box. If you are doing iOS or macOS development you will get a free bag on your view controllers and other UIKit objects since all of them are NSObject subclasses.

extension NSObject {
  public var bag: DisposeBag { get }
}

If you are like me and do not want to worry about disposing, check out bindings.

By default observers receive events on the thread or the queue where the event is sent from.

For example, if we have a signal that is created like

let someImage = Signal<UIImage, Never> { observer in
  ...
  DispatchQueue.global(qos: .background).async {
    observer.receive(someImage)
  }
  ...
}

and if we use it to update the image view

someImage
  .observeNext { image in
    imageView.image = image // called on background queue
  }
  .dispose(in: bag)

we will end up with a weird behaviour. We will be setting the image from the background queue on an instance of UIImageView that is not thread safe - just like the rest of UIKit.

We could set the image in another async dispatch to the main queue, but there is a better way. Just use the operator receive(on:) with the queue you want the observer to be called on.

someImage
  .receive(on: ExecutionContext.main)
  .observeNext { image in
    imageView.image = image // called on main queue
  }
  .dispose(in: bag)

There is also another side to this. You might have a signal that does some slow synchronous work on whatever thread or queue it is observed on.

let someData = Signal<Data, Never> { observer in
  ...
  let data = // synchronously load large file
  observer.receive(data)
  ...
}

We, however, do not want observing that signal to block the UI.

someData
  .observeNext { data in // blocks current thread
    display(data)
  }
  .dispose(in: bag)

We would like to do the loading on another queue. We could dispatch async the loading, but what if we cannot change the signal producer closure because it is defined in a framework or there is another reason we cannot change it? That is when the operator subscribe(on:) saves the day.

someData
  .subscribe(on: ExecutionContext.global(qos: .background))
  .receive(on: ExecutionContext.main)
  .observeNext { data in // does not block current thread
    display(data)
  }
  .dispose(in: bag)

By applying subscribe(on:) we define where the signal producer gets executed. We usually use it in a combination with receive(on:) to define where the observer receives events.

Note that these operators work with execution contexts. Execution context is a simple abstraction over a thread or a queue. You can see how it is implemented here.

Bindings are observations with perks. Most of the time you should be able to replace an observation with a binding. Consider the following example. Say we have a signal of users

let presentUserProfile: Signal<User, Never> = ...

and we would like to present a profile screen when a user is sent on the signal. Usually we would do something like:

presentUserProfile.receive(on: ExecutionContext.main).observeNext { [weak self] user in
  let profileViewController = ProfileViewController(user: user)
  self?.present(profileViewController, animated: true)
}.dispose(in: bag)

But that is ugly! We have to dispatch everything to the main queue, be cautious not to create a retain cycle and ensure that the disposable we get from the observation is handled.

Thankfully there is a better way. We can create an inline binding instead of the observation. Just do the following

presentUserProfile.bind(to: self) { me, user in
  let profileViewController = ProfileViewController(user: user)
  me.present(profileViewController, animated: true)
}

and stop worrying about threading, retain cycles and disposing because bindings take care of all that automatically. Just bind the signal to the target responsible for performing side effects (in our example, to the view controller responsible for presenting a profile view controller). The closure you provide will be called whenever the signal emits an element with both the target and the sent element as arguments.

You can bind to targets that conform to both the Deallocatable and the BindingExecutionContextProvider protocols.

You can actually bind to targets that conform only to the Deallocatable protocol, but then you have to pass the execution context in which to update the target by calling bind(to:context:setter).

Objects that conform to Deallocatable provide a signal that can tell us when the object gets deallocated.

public protocol Deallocatable: class {

  /// A signal that fires `completed` event when the receiver is deallocated.
  var deallocated: Signal<Void, Never> { get }
}

ReactiveKit provides conformance to this protocol for NSObject and its subclasses out of the box.

How do you conform to Deallocatable? The simplest way is conforming to DisposeBagProvider instead.

/// A type that provides a dispose bag.
/// `DisposeBagProvider` conforms to `Deallocatable` out of the box.
public protocol DisposeBagProvider: Deallocatable {

  /// A `DisposeBag` that can be used to dispose observations and bindings.
  var bag: DisposeBag { get }
}

extension DisposeBagProvider {

  public var deallocated: Signal<Void, Never> {
    return bag.deallocated
  }
}

As you can see, DisposeBagProvider inherits Deallocatable and implements it by taking the deallocated signal from the bag. So all that you need to do is provide a bag property on your type.

The BindingExecutionContextProvider protocol provides the execution context in which the object should be updated. Execution context is just a wrapper over a dispatch queue or a thread. You can see how it is implemented here.

public protocol BindingExecutionContextProvider {

  /// An execution context used to deliver binding events.
  var bindingExecutionContext: ExecutionContext { get }
}

The Bond framework provides BindingExecutionContextProvider conformance to various UIKit objects so they can be seamlessly bound to while ensuring the main thread.

You can conform to this protocol by providing execution context.

extension MyViewModel: BindingExecutionContextProvider {

  public var bindingExecutionContext: ExecutionContext {
    return .immediateOnMain
  }
}

ExecutionContext.immediateOnMain executes synchronously if the current thread is main, otherwise it makes an asynchronous dispatch to the main queue. If you want to bind on the background queue, you can return .global(qos: .background) instead.

Note that updating UIKit or AppKit objects must always happen from the main thread or queue.

Now we can peek into the binding implementation.

extension SignalProtocol where Error == Never {

  @discardableResult
  public func bind<Target: Deallocatable>(to target: Target, setter: @escaping (Target, Element) -> Void) -> Disposable
  where Target: BindingExecutionContextProvider
  {
    return take(until: target.deallocated)
      .observeIn(target.bindingExecutionContext)
      .observeNext { [weak target] element in
        if let target = target {
          setter(target, element)
        }
      }
  }
}

First of all, notice the @discardableResult annotation. It is there because we can safely ignore the returned disposable. The binding will automatically be disposed when the target gets deallocated. That is ensured by the take(until:) operator. It propagates events from self until the given signal completes - in our case until the target.deallocated signal completes. We then just observe in the right context and on each next element update the target using the provided setter closure.

Note also that bindings are implemented only on non-failable signals.

Given a string signal name, we know that we can bind it to a label by doing

name.bind(to: label) { label, name in
  label.text = name
}

but would it not be great if we could make it a one-liner? With Swift 4 key paths we can! Just do

name.bind(to: label, keyPath: \.text)

where the target is the same target as in the previous example and keyPath is a key path to the property that should be updated with each new element sent on the signal!

If you opt-in for the Bond f