RxSwift Documentation

Reactive Programming in Swift

24,645 stars Swift Swift 1 file · ~15,827 tokens #functional#ios#observer#reactive#reactivex#rxswift#swift#unidirectional
RAW Doc

1. Project Overview & Quickstart (ReactiveX/RxSwift)

File: README.md

Rx is a generic abstraction of computation expressed through Observable<Element> interface, which lets you broadcast and subscribe to values and other events from an Observable stream.

RxSwift is the Swift-specific implementation of the Reactive Extensions standard.

While this version aims to stay true to the original spirit and naming conventions of Rx, this project also aims to provide a true Swift-first API for Rx APIs.

Cross platform documentation can be found on ReactiveX.io.

Like other Rx implementations, RxSwift's intention is to enable easy composition of asynchronous operations and streams of data in the form of Observable objects and a suite of methods to transform and compose these pieces of asynchronous work.

KVO observation, async operations, UI Events and other streams of data are all unified under abstraction of sequence. This is the reason why Rx is so simple, elegant and powerful.

I came here because I want to ...

... understand

... install

... hack around

... interact

... compare

... understand the structure

RxSwift is as compositional as the asynchronous work it drives. The core unit is RxSwift itself, while other dependencies can be added for UI Work, testing, and more.

It comprises five separate components depending on each other in the following way:

none
┌──────────────┐    ┌──────────────┐
│   RxCocoa    ├────▶   RxRelay    │
└───────┬──────┘    └──────┬───────┘
        │                  │
┌───────▼──────────────────▼───────┐
│             RxSwift              │
└───────▲──────────────────▲───────┘
        │                  │
┌───────┴──────┐    ┌──────┴───────┐
│    RxTest    │    │  RxBlocking  │
└──────────────┘    └──────────────┘
  • RxSwift: The core of RxSwift, providing the Rx standard as (mostly) defined by ReactiveX. It has no other dependencies.
  • RxCocoa: Provides Cocoa-specific capabilities for general iOS/macOS/watchOS & tvOS app development, such as Shared Sequences, Traits, and much more. It depends on both RxSwift and RxRelay.
  • RxRelay: Provides PublishRelay, BehaviorRelay and ReplayRelay, three simple wrappers around Subjects. It depends on RxSwift.
  • RxTest and RxBlocking: Provides testing capabilities for Rx-based systems. It depends on RxSwift.

Usage

Here's an example In Action
Define search for GitHub repositories ...
let searchResults = searchBar.rx.text.orEmpty
text
.throttle(.milliseconds(300), scheduler: MainScheduler.instance)
.distinctUntilChanged()
.flatMapLatest { query -> Observable&lt;[Repository]&gt; in
    if query.isEmpty {
        return .just([])
    }
    return searchGitHub(query)
        .catchAndReturn([])
}
.observe(on: MainScheduler.instance) |  |

| ... then bind the results to your tableview | |
| searchResults
.bind(to: tableView.rx.items(cellIdentifier: "Cell")) {
(index, repository: Repository, cell) in
cell.textLabel?.text = repository.name
cell.detailTextLabel?.text = repository.url
}
.disposed(by: disposeBag) | |

Installation

RxSwift doesn't contain any external dependencies.

These are currently the supported installation options:

Manual

Open Rx.xcworkspace, choose RxExample and hit run. This method will build everything and run the sample app

XCFrameworks

Each release starting with RxSwift 6 includes *.xcframework framework binaries.

Simply drag the needed framework binaries to your Frameworks, Libraries, and Embedded Content section under your target's General tab.

[!TIP]
RxSwift's xcframework(s) are signed with an Apple Developer account, and you can always verify the Team Name: Shai Mishali

[Carthage](https://github.com/Carthage/Carthage)

Add this to Cartfile

text
github "ReactiveX/RxSwift" "6.10.0"
bash
$ carthage update

Carthage as a Static Library

Carthage defaults to building RxSwift as a Dynamic Library.

If you wish to build RxSwift as a Static Library using Carthage you may use the script below to manually modify the framework type before building with Carthage:

bash
carthage update RxSwift --platform iOS --no-build
sed -i -e 's/MACH_O_TYPE = mh_dylib/MACH_O_TYPE = staticlib/g' Carthage/Checkouts/RxSwift/Rx.xcodeproj/project.pbxproj
carthage build RxSwift --platform iOS

[Swift Package Manager](https://github.com/swiftlang/swift-package-manager)

Note: There is a critical cross-dependency bug affecting many projects including RxSwift in Swift Package Manager. We've filed a bug (SR-12303) in early 2020 but have no answer yet. Your mileage may vary. A partial workaround can be found here.

Create a Package.swift file.

swift
// swift-tools-version:5.0

import PackageDescription

let package = Package(
  name: "RxProject",
  dependencies: [
    .package(url: "https://github.com/ReactiveX/RxSwift.git", .upToNextMajor(from: "6.0.0"))
  ],
  targets: [
    .target(name: "RxProject", dependencies: ["RxSwift", .product(name: "RxCocoa", package: "RxSwift")]),
  ]
)
bash
$ swift build

To build or test a module with RxTest dependency, set TEST=1.

bash
$ TEST=1 swift test

Manually using git submodules

  • Add RxSwift as a submodule
bash
$ git submodule add [email protected]:ReactiveX/RxSwift.git
  • Drag Rx.xcodeproj into Project Navigator
  • Go to Project > Targets > Build Phases > Link Binary With Libraries, click + and select RxSwift, RxCocoa and RxRelay targets

References

File: Documentation/ComparisonWithOtherLibraries.md

Comparison with other libraries

This section attempts to compare RxSwift with other known and popular alternatives.

The best part about reactive programming is that most of the concepts you'll learn from either of these frameworks are entirely applicable to RxSwift, and vice-versa, with only relatively minor framework-specific concepts and naming conventions.

Apple's Combine

Combine is Apple's own implementation of the Reactive Streams standard. The main differences to consider are that Combine uses typed errors, and supports backpressure. Its huge downside is that it's closed source, limited to iOS 13 and up, and does not support Linux as of today.

It also does not have its own testing framework like RxTest or RxBlocking, and doesn't have its own UIKit-supporting framework such as RxCocoa, since it is more aimed towards SwiftUI. Some of these issues have been addressed by community projects such as CombineCocoa, combine-schedulers and CombineExpectations.

It is also (as of 2020) a relatively young and not as battle-tested as RxSwift or ReactiveSwift which are around for over half a decade.

If you're interested in using Combine for your relatively modern codebase while still using your RxSwift codebase, you can leverage RxCombine, a community project which aims to provide interoperability between Combine and RxSwift.

ReactiveSwift

RxSwift is somewhat similar to ReactiveSwift since ReactiveSwift borrows a large number of concepts from Rx.

One of the main goals of this project was to create a significantly simpler interface that is more aligned with other Rx implementations, offers a richer concurrency model, offers more optimization opportunities and is more aligned with built-in Swift error handling mechanisms.

We've also decided to only rely on the Swift/llvm compiler and not introduce any external dependencies.

Probably the main difference between these projects is in their approach in building abstractions.

The main goal of RxSwift project is to provide environment-agnostic compositional computation glue abstracted in the form of observable sequences.

We then aim to improve the experience of using RxSwift on specific platforms. To do this, RxCocoa uses generic computations to build more practical abstractions and wrap Foundation/Cocoa/UIKit frameworks. That means that other libraries give context and semantics to the generic computation engine RxSwift provides such as Driver, Signal, ControlProperty, ControlEvents and more.

One of the benefits to representing all of these abstractions as a single concept - ​_observable sequences_​ - is that all computation abstractions built on top of them are also composable in the same fundamental way. They all follow the same contract and implement the same interface.
It is also easy to create flexible subscription (resource) sharing strategies or use one of the built-in ones: share, publish, multicast ...

This library also offers a fine-tunable concurrency model. If concurrent schedulers are used, observable sequence operators will preserve sequence properties. The same observable sequence operators will also know how to detect and optimally use known serial schedulers. ReactiveSwift has a more limited concurrency model and only allows serial schedulers.

Multithreaded programming is really hard and detecting non-trivial loops is even harder. That's why all operators are built in a fault tolerant way. Even if element generation occurs during element processing (recursion), operators will try to handle that situation and prevent deadlocks. This means that in the worst possible case programming error will cause stack overflow, but users won't have to manually kill the app, and you will get a crash report in error reporting systems so you can find and fix the problem.

File: Documentation/DesignRationale.md

Design Rationale

Why error type isn't generic

swift
enum Event<Element>  {
    case next(Element)      // next element of a sequence
    case error(Error)       // sequence failed with error
    case completed          // sequence terminated successfully
}

Let's discuss the pros and cons of Error being generic.

If you have a generic error type, you create additional impedance mismatch between two observables.

Let's say you have:

Observable<String, E1> and Observable<String, E2>

There isn't much you can do with them without figuring out what the resulting error type will be.

Will it be E1, E2 or some new E3 maybe? So, you would need a new set of operators just to solve that impedance mismatch.

This hurts composition properties, and Rx isn't concerned with why a sequence fails, it just usually forwards failures further down the observable chain.

There is an additional problem that, in some cases, operators might fail due to some internal error, in which case you wouldn't be able to construct a resulting error and report failure.

But OK, let's ignore that and assume we can use that to model sequences that don't error out. Could it be useful for that purpose?

Well yes, it potentially could be, but let's consider why you would want to use sequences that don't error out.

One obvious application would be for permanent streams in the UI layer that drive the entire UI. When you consider that case, it's not really sufficient to only use the compiler to prove that sequences don't error out, you also need to prove other properties. For instance, those elements are observed on MainScheduler.

What you really need is a generic way to prove traits for observable sequences. There are a lot of properties you could be interested in. For example:

  • sequence terminates in finite time (server side)
  • sequence contains only one element (if you are running some computation)
  • sequence doesn't error out, never terminates and elements are delivered on main scheduler (UI)
  • sequence doesn't error out, never terminates and elements are delivered on main scheduler, and has refcounted sharing (UI)
  • sequence doesn't error out, never terminates and elements are delivered on specific background scheduler (audio engine)

What you really want is a general compiler-enforced system of traits for observable sequences, and a set of invariant operators for those wanted properties.

A good analogy would be:

text
1, 3.14, e, 2.79, 1 + 1i      <->    Observable<E>
1m/s, 1T, 5kg, 1.3 pounds     <->    Errorless observable, UI observable, Finite observable ...

There are many ways to do such a thing in Swift by either using composition or inheritance of observables.

An additional benefit of using a unit system is that you can prove that UI code is executing on the same scheduler and thus use lockless operators for all transformations.

Since RxSwift already doesn't have locks for single sequence operations, and all of the locks that do exist are in stateful components (e.g. UI), there are practically no locks in RxSwift code, which allows for such details to be compiler enforced.

There really is no benefit to using typed Errors that couldn't be achieved in other cleaner ways while preserving Rx compositional semantics.

File: Documentation/ExampleApp.md

RxExamples

To run the example app:

  • Open Rx.xcworkspace
  • Choose one of example schemes (RxExample-iOS, RxExample-macOS) and hit Run.

File: Documentation/Examples.md

Examples

  1. Reactive values
  2. Simple UI bindings
  3. Automatic input validation
  4. more examples
  5. Playgrounds

Reactive values

First, let's start with some imperative code.
The purpose of this example is to bind the identifier c to a value calculated from a and b if some condition is satisfied.

Here is the imperative code that calculates the value of c:

swift
// this is standard imperative code
var c: String
var a = 1       // this will only assign the value `1` to `a` once
var b = 2       // this will only assign the value `2` to `b` once

if a + b >= 0 {
    c = "\(a + b) is positive" // this will only assign the value to `c` once
}

The value of c is now 3 is positive. However, if we change the value of a to 4, c will still contain the old value.

swift
a = 4           // `c` will still be equal to "3 is positive" which is not good
                // we want `c` to be equal to "6 is positive" since 4 + 2 = 6

This is not the desired behavior.

This is the improved logic using RxSwift:

swift
let a /*: Observable*/ = BehaviorRelay(value: 1)   // a = 1
let b /*: Observable*/ = BehaviorRelay(value: 2)   // b = 2

// Combines latest values of relays `a` and `b` using `+`
let c = Observable.combineLatest(a, b) { $0 + $1 }
	.filter { $0 >= 0 }               // if `a + b >= 0` is true, `a + b` is passed to the map operator
	.map { "\($0) is positive" }      // maps `a + b` to "\(a + b) is positive"

// Since the initial values are a = 1 and b = 2
// 1 + 2 = 3 which is >= 0, so `c` is initially equal to "3 is positive"

// To pull values out of the Rx `Observable` `c`, subscribe to values from `c`.
// `subscribe(onNext:)` means subscribe to the next (fresh) values of `c`.
// That also includes the initial value "3 is positive".
c.subscribe(onNext: { print($0) })          // prints: "3 is positive"

// Now, let's increase the value of `a`
a.accept(4)                                   // prints: 6 is positive
// The sum of the latest values, `4` and `2`, is now `6`.
// Since this is `>= 0`, the `map` operator produces "6 is positive"
// and that result is "assigned" to `c`.
// Since the value of `c` changed, `{ print($0) }` will get called,
// and "6 is positive" will be printed.

// Now, let's change the value of `b`
b.accept(-8)                                 // doesn't print anything
// The sum of the latest values, `4 + (-8)`, is `-4`.
// Since this is not `>= 0`, `map` doesn't get executed.
// This means that `c` still contains "6 is positive"
// Since `c` hasn't been updated, a new "next" value hasn't been produced,
// and `{ print($0) }` won't be called.

Simple UI bindings

  • Instead of binding to Relays, let's bind to UITextField values using the rx.text property.
  • Next, map the String into an Int and determine if the number is prime using an async API.
  • If the text is changed before the async call completes, a new async call will replace it via concat.
  • Bind the results to a UILabel.
swift
let subscription/*: Disposable */ = primeTextField.rx.text.orEmpty // type is Observable<String>
            .map { WolframAlphaIsPrime(Int($0) ?? 0) }             // type is Observable<Observable>
            .concat()                                              // type is Observable
            .map { "number \($0.n) is prime? \($0.isPrime)" }      // type is Observable<String>
            .bind(to: resultLabel.rx.text)                         // return Disposable that can be used to unbind everything

// This will set `resultLabel.text` to "number 43 is prime? true" after
// server call completes. You manually trigger a control event since those are
// the UIKit events RxCocoa observes internally.
primeTextField.text = "43"
primeTextField.sendActions(for: .editingDidEnd)

// ...

// to unbind everything, just call
subscription.dispose()

All of the operators used in this example are the same operators used in the first example with relays. There's nothing special about it.

Automatic input validation

If you are new to Rx, the next example will probably be a little overwhelming at first. However, it's here to demonstrate how RxSwift code looks in the real-world.

This example contains complex async UI validation logic with progress notifications.
All operations are canceled the moment disposeBag is deallocated.

Let's give it a shot.

text
/* Detailed source-code truncated for AI context efficiency. */

It doesn't get any simpler than that. There are more examples in the repository, so feel free to check them out.

They include examples on how to use Rx in the context of MVVM pattern or without it.

File: Documentation/GettingStarted.md

Getting Started

This project tries to be consistent with ReactiveX.io. The general cross platform documentation and tutorials should also be valid in case of RxSwift.

  1. Observables aka Sequences
  2. Disposing
  3. Implicit Observable guarantees
  4. Creating your first Observable (aka observable sequence)
  5. Creating an Observable that performs work
  6. Sharing subscription and share operator
  7. Operators
  8. Custom operators
  9. Infallible
  10. Playgrounds
  11. Error handling
  12. Debugging Compile Errors
  13. Debugging
  14. Enabling Debug Mode
  15. Debugging memory leaks
  16. KVO
  17. UI layer tips
  18. Making HTTP requests
  19. RxDataSources
  20. Driver
  21. Traits: Driver, Single, Maybe, Completable
  22. Examples

Observables aka Sequences

Basics

The equivalence of observer pattern (Observable<Element> sequence) and normal sequences (Sequence) is the most important thing to understand about Rx.

Every Observable sequence is just a sequence. The key advantage for an Observable vs Swift's Sequence is that it can also receive elements asynchronously. This is the kernel of RxSwift, documentation from here is about ways that we expand on that idea.

  • Observable(ObservableType) is equivalent to Sequence
  • ObservableType.subscribe method is equivalent to Sequence.makeIterator method.
  • Observer (callback) needs to be passed to ObservableType.subscribe method to receive sequence elements instead of calling next() on the returned iterator.

Sequences are a simple, familiar concept that is easy to visualize.

People are creatures with huge visual cortexes. When we can visualize a concept easily, it's a lot easier to reason about it.

We can lift a lot of the cognitive load from trying to simulate event state machines inside every Rx operator onto high level operations over sequences.

If we don't use Rx but model asynchronous systems, that probably means our code is full of state machines and transient states that we need to simulate instead of abstracting away.

Lists and sequences are probably one of the first concepts mathematicians and programmers learn.

Here is a sequence of numbers:

text
--1--2--3--4--5--6--| // terminates normally

Another sequence, with characters:

text
--a--b--a--a--a---d---X // terminates with error

Some sequences are finite while others are infinite, like a sequence of button taps:

text
---tap-tap-------tap--->

These are called marble diagrams. There are more marble diagrams at rxmarbles.com.

If we were to specify sequence grammar as a regular expression it would look like:

next* (error | completed)?

This describes the following:

  • Sequences can have 0 or more elements.
  • Once an error or completed event is received, the sequence cannot produce any other element.

Sequences in Rx are described by a push interface (aka callback).

swift
enum Event<Element>  {
    case next(Element)      // next element of a sequence
    case error(Swift.Error) // sequence failed with error
    case completed          // sequence terminated successfully
}

class Observable<Element> {
    func subscribe(_ observer: Observer<Element>) -> Disposable
}

protocol ObserverType {
    func on(_ event: Event<Element>)
}

When a sequence sends the completed or error event all internal resources that compute sequence elements will be freed.

To cancel production of sequence elements and free resources immediately, call dispose on the returned subscription.

If a sequence terminates in finite time, not calling dispose or not using disposed(by: disposeBag) won't cause any permanent resource leaks. However, those resources will be used until the sequence completes, either by finishing production of elements or returning an error.

If a sequence does not terminate on its own, such as with a series of button taps, resources will be allocated permanently unless dispose is called manually, automatically inside of a disposeBag, with the takeUntil operator, or in some other way.

Using dispose bags or takeUntil operator is a robust way of making sure resources are cleaned up. We recommend using them in production even if the sequences will terminate in finite time.

If you are curious why Swift.Error isn't generic, you can find the explanation here.

Disposing

There is one additional way an observed sequence can terminate. When we are done with a sequence and we want to release all of the resources allocated to compute the upcoming elements, we can call dispose on a subscription.

Here is an example with the interval operator.

swift
let scheduler = SerialDispatchQueueScheduler(qos: .default)
let subscription = Observable.interval(.milliseconds(300), scheduler: scheduler)
    .subscribe { event in
        print(event)
    }

Thread.sleep(forTimeInterval: 2.0)

subscription.dispose()

This will print:

text
0
1
2
3
4
5

Note that you usually do not want to manually call dispose; this is only an educational example. Calling dispose manually is usually a bad code smell. There are better ways to dispose of subscriptions such as DisposeBag, the takeUntil operator, or some other mechanism.

So can this code print something after the dispose call is executed? The answer is: it depends.

  • If the scheduler is a serial scheduler (ex. MainScheduler) and dispose is called on the same serial scheduler, the answer is no.

  • Otherwise it is yes.

You can find out more about schedulers here.

You simply have two processes happening in parallel.

  • one is producing elements
  • the other is disposing of the subscription

The question "Can something be printed after?" does not even make sense in the case that those processes are on different schedulers.

A few more examples just to be sure (observeOn is explained here).

In case we have something like:

swift
let subscription = Observable.interval(.milliseconds(300), scheduler: scheduler)
            .observe(on: MainScheduler.instance)
            .subscribe { event in
                print(event)
            }

// ....

subscription.dispose() // called from main thread

After the dispose call returns, nothing will be printed. That is guaranteed.

Also, in this case:

swift
let subscription = Observable.interval(.milliseconds(300), scheduler: scheduler)
            .observe(on: MainScheduler.instance)
            .subscribe { event in
                print(event)
            }

// ...

subscription.dispose() // executing on same `serialScheduler`

After the dispose call returns, nothing will be printed. That is guaranteed.

Dispose Bags

Dispose bags are used to return ARC like behavior to RX.

When a DisposeBag is deallocated, it will call dispose on each of the added disposables.

It does not have a dispose method and therefore does not allow calling explicit dispose on purpose. If immediate cleanup is required, we can just create a new bag.

swift
self.disposeBag = DisposeBag()

This will clear old references and cause disposal of resources.

If that explicit manual disposal is still wanted, use CompositeDisposable. It has the wanted behavior but once that dispose method is called, it will immediately dispose any newly added disposable.

Take until

Additional way to automatically dispose subscription on dealloc is to use takeUntil operator.

swift
sequence
    .take(until: self.rx.deallocated)
    .subscribe {
        print($0)
    }

Implicit `Observable` guarantees

There is also a couple of additional guarantees that all sequence producers (Observables) must honor.

It doesn't matter on which thread they produce elements, but if they generate one element and send it to the observer observer.on(.next(nextElement)), they can't send next element until observer.on method has finished execution.

Producers also cannot send terminating .completed or .error in case .next event hasn't finished.

In short, consider this example:

swift
someObservable
  .subscribe { (e: Event<Element>) in
      print("Event processing started")
      // processing
      print("Event processing ended")
  }

This will always print:

text
Event processing started
Event processing ended
Event processing started
Event processing ended
Event processing started
Event processing ended

It can never print:

text
Event processing started
Event processing started
Event processing ended
Event processing ended

Creating your own `Observable` (aka observable sequence)

There is one crucial thing to understand about observables.

When an observable is created, it doesn't perform any work simply because it has been created.

It is true that Observable can generate elements in many ways. Some of them cause side effects and some of them tap into existing running processes like tapping into mouse events, etc.

However, if you just call a method that returns an Observable, no sequence generation is performed and there are no side effects. Observable just defines how the sequence is generated and what parameters are used for element generation. Sequence generation starts when subscribe method is called.

E.g. Let's say you have a method with similar prototype:

swift
func searchWikipedia(searchTerm: String) -> Observable<Results> {}
swift
let searchForMe = searchWikipedia("me")

// no requests are performed, no work is being done, no URL requests were fired

let cancel = searchForMe
  // sequence generation starts now, URL requests are fired
  .subscribe(onNext: { results in
      print(results)
  })

There are a lot of ways to create your own Observable sequence. The easiest way is probably to use the create function.

RxSwift provides a method that creates a sequence which returns one element upon subscription. That method is called just. Let's write our own implementation of it:

This is the actual implementation

swift
func myJust<E>(_ element: E) -> Observable<E> {
    return Observable.create { observer in
        observer.on(.next(element))
        observer.on(.completed)
        return Disposables.create()
    }
}

myJust(0)
    .subscribe(onNext: { n in
      print(n)
    })

This will print:

text
0

Not bad. So what is the create function?

It's just a convenience method that enables you to easily implement subscribe method using Swift closures. Like subscribe method it takes one argument, observer, and returns disposable.

Sequence implemented this way is actually synchronous. It will generate elements and terminate before subscribe call returns disposable representing subscription. Because of that it doesn't really matter what disposable it returns, process of generating elements can't be interrupted.

When generating synchronous sequences, the usual disposable to return is singleton instance of NopDisposable.

Lets now create an observable that returns elements from an array.

This is the actual implementation

swift
func myFrom<E>(_ sequence: [E]) -> Observable<E> {
    return Observable.create { observer in
        for element in sequence {
            observer.on(.next(element))
        }

        observer.on(.completed)
        return Disposables.create()
    }
}

let stringCounter = myFrom(["first", "second"])

print("Started ----")

// first time
stringCounter
    .subscribe(onNext: { n in
        print(n)
    })

print("----")

// again
stringCounter
    .subscribe(onNext: { n in
        print(n)
    })

print("Ended ----")

This will print:

text
Started ----
first
second
----
first
second
Ended ----

Creating an `Observable` that performs work

Ok, now something more interesting. Let's create that interval operator that was used in previous examples.

This is equivalent of actual implementation for dispatch queue schedulers

swift
func myInterval(_ interval: DispatchTimeInterval) -> Observable {
    return Observable.create { observer in
        print("Subscribed")
        let timer = DispatchSource.makeTimerSource(queue: DispatchQueue.global())
        timer.schedule(deadline: DispatchTime.now() + interval, repeating: interval)

        let cancel = Disposables.create {
            print("Disposed")
            timer.cancel()
        }

        var next = 0
        timer.setEventHandler {
            if cancel.isDisposed {
                return
            }
            observer.on(.next(next))
            next += 1
        }
        timer.resume()

        return cancel
    }
}
swift
let counter = myInterval(.milliseconds(100))

print("Started ----")

let subscription = counter
    .subscribe(onNext: { n in
        print(n)
    })


Thread.sleep(forTimeInterval: 0.5)

subscription.dispose()

print("Ended ----")

This will print

text
Started ----
Subscribed
0
1
2
3
4
Disposed
Ended ----

What if you would write

swift
let counter = myInterval(.milliseconds(100))

print("Started ----")

let subscription1 = counter
    .subscribe(onNext: { n in
        print("First \(n)")
    })

let subscription2 = counter
    .subscribe(onNext: { n in
        print("Second \(n)")
    })

Thread.sleep(forTimeInterval: 0.5)

subscription1.dispose()

print("Disposed")

Thread.sleep(forTimeInterval: 0.5)

subscription2.dispose()

print("Disposed")

print("Ended ----")

This would print:

text
Started ----
Subscribed
Subscribed
First 0
Second 0
First 1
Second 1
First 2
Second 2
First 3
Second 3
First 4
Second 4
Disposed
Second 5
Second 6
Second 7
Second 8
Second 9
Disposed
Ended ----

Every subscriber upon subscription usually generates it's own separate sequence of elements. Operators are stateless by default. There are vastly more stateless operators than stateful ones.

Sharing subscription and `share` operator

But what if you want that multiple observers share events (elements) from only one subscription?

There are two things that need to be defined.

  • How to handle past elements that have been received before the new subscriber was interested in observing them (replay latest only, replay all, replay last n)
  • How to decide when to fire that shared subscription (refCount, manual or some other algorithm)

The usual choice is a combination of replay(1).refCount(), aka share(replay: 1).

swift
let counter = myInterval(.milliseconds(100))
    .share(replay: 1)

print("Started ----")

let subscription1 = counter
    .subscribe(onNext: { n in
        print("First \(n)")
    })
let subscription2 = counter
    .subscribe(onNext: { n in
        print("Second \(n)")
    })

Thread.sleep(forTimeInterval: 0.5)

subscription1.dispose()

Thread.sleep(forTimeInterval: 0.5)

subscription2.dispose()

print("Ended ----")

This will print

text
Started ----
Subscribed
First 0
Second 0
First 1
Second 1
First 2
Second 2
First 3
Second 3
First 4
Second 4
Second 5
Second 6
Second 7
Second 8
Second 9
Disposed
Ended ----

Notice how now there is only one Subscribed and Disposed event.

Behavior for URL observables is equivalent.

This is how HTTP requests are wrapped in Rx. It's pretty much the same pattern like the interval operator.

swift
extension Reactive where Base: URLSession {
    public func response(request: URLRequest) -> Observable<(response: HTTPURLResponse, data: Data)> {
        return Observable.create { observer in
            let task = self.base.dataTask(with: request) { (data, response, error) in

                guard let response = response, let data = data else {
                    observer.on(.error(error ?? RxCocoaURLError.unknown))
                    return
                }

                guard let httpResponse = response as? HTTPURLResponse else {
                    observer.on(.error(RxCocoaURLError.nonHTTPResponse(response: response)))
                    return
                }

                observer.on(.next((httpResponse, data)))
                observer.on(.completed)
            }

            task.resume()

            return Disposables.create {
                task.cancel()
            }
        }
    }
}

Operators

There are numerous operators implemented in RxSwift.

Marble diagrams for all operators can be found on ReactiveX.io

Almost all operators are demonstrated in Playgrounds.

To use playgrounds please open Rx.xcworkspace, build RxSwift-macOS scheme and then open playgrounds in Rx.xcworkspace tree view.

In case you need an operator, and don't know how to find it there is a decision tree of operators.

Custom operators

There are two ways how you can create custom operators.

Easy way

All of the internal code uses highly optimized versions of operators, so they aren't the best tutorial material. That's why it's highly encouraged to use standard operators.

Fortunately there is an easier way to create operators. Creating new operators is actually all about creating observables, and previous chapter already describes how to do that.

Lets see how an unoptimized map operator can be implemented.

swift
extension ObservableType {
    func myMap<R>(transform: @escaping (Element) -> R) -> Observable<R> {
        return Observable.create { observer in
            let subscription = self.subscribe { e in
                    switch e {
                    case .next(let value):
                        let result = transform(value)
                        observer.on(.next(result))
                    case .error(let error):
                        observer.on(.error(error))
                    case .completed:
                        observer.on(.completed)
                    }
                }

            return subscription
        }
    }
}

So now you can use your own map:

swift
let subscription = myInterval(.milliseconds(100))
    .myMap { e in
        return "This is simply \(e)"
    }
    .subscribe(onNext: { n in
        print(n)
    })

This will print:

text
Subscribed
This is simply 0
This is simply 1
This is simply 2
This is simply 3
This is simply 4
This is simply 5
This is simply 6
This is simply 7
This is simply 8
...

Infallible

Infallible is another flavor of Observable which is identical to it, but is guaranteed to never fail and thus cannot emit errors. This means that when creating your own Infallible (Using Infallible.create or one of the methods mentioned in Creating your first Observable), you will not be allowed to emit errors.

Infallible is useful when you want to statically model and guarantee a stream of values that is known to never fail, but don't want to commit to using MainScheduler and don't want to implicitly use share() to share resources and side-effects, such as the case in Driver and Signal.

Life happens

So what if it's just too hard to solve some cases with custom operators? You can exit the Rx monad, perform actions in imperative world, and then tunnel results to Rx again using Subjects.

This isn't something that should be practiced often, and is a bad code smell, but you can do it.

swift
let magicBeings: Observable<MagicBeing> = summonFromMiddleEarth()

  magicBeings
    .subscribe(onNext: { being in     // exit the Rx monad
        self.doSomeStateMagic(being)
    })
    .disposed(by: disposeBag)

  //
  //  Mess
  //
  let kitten = globalParty(   // calculate something in messy world
    being,
    UIApplication.delegate.dataSomething.attendees
  )
  kittens.on(.next(kitten))   // send result back to rx
  //
  // Another mess
  //

  let kittens = BehaviorRelay(value: firstKitten) // again back in Rx monad

  kittens.asObservable()
    .map { kitten in
      return kitten.purr()
    }
    // ....

Every time you do this, somebody will probably write this code somewhere:

swift
kittens
    .subscribe(onNext: { kitten in
      // do something with kitten
    })
    .disposed(by: disposeBag)

So please try not to do this.

Playgrounds

If you are unsure how exactly some of the operators work, playgrounds contain almost all of the operators already prepared with small examples that illustrate their behavior.

To use playgrounds please open Rx.xcworkspace, build RxSwift-macOS scheme and then open playgrounds in Rx.xcworkspace tree view.

To view the results of the examples in the playgrounds, please open the Assistant Editor. You can open Assistant Editor by clicking on View > Assistant Editor > Show Assistant Editor

Error handling

There are two error mechanisms.

Asynchronous error handling mechanism in observables

Error handling is pretty straightforward. If one sequence terminates with error, then all of the dependent sequences will terminate with error. It's usual short circuit logic.

You can recover from failure of observable by using catch operator. There are various overloads that enable you to specify recovery in great detail.

There is also retry operator that enables retries in case of errored sequence.

Hooks and Default error handling

RxSwift offers a global Hook that provides a default error handling mechanism for cases when you don't provide your own onError handler.

Set Hooks.defaultErrorHandler with your own closure to decide how to deal with unhandled errors in your system, if you need that option. For example, sending the stacktrace or untracked-error to your analytics system.

By default, Hooks.defaultErrorHandler simply prints the received error in DEBUG mode, and does nothing in RELEASE. However, you can add additional configurations to this behavior.

In order to enable detailed callstack logging, set Hooks.recordCallStackOnError flag to true.

By default, this will return the current Thread.callStackSymbols in DEBUG mode, and will track an empty stack trace in RELEASE. You may customize this behavior by overriding Hooks.customCaptureSubscriptionCallstack with your own implementation.

Debugging Compile Errors

When writing elegant RxSwift/RxCocoa code, you are probably relying heavily on compiler to deduce types of Observables. This is one of the reasons why Swift is awesome, but it can also be frustrating sometimes.

swift
images = word
    .filter { $0.containsString("important") }
    .flatMap { word in
        return self.api.loadFlickrFeed("karate")
            .catchError { error in
                return just(JSON(1))
            }
      }

If compiler reports that there is an error somewhere in this expression, I would suggest first annotating return types.

swift
images = word
    .filter { s -> Bool in s.containsString("important") }
    .flatMap { word -> Observable<JSON> in
        return self.api.loadFlickrFeed("karate")
            .catchError { error -> Observable<JSON> in
                return just(JSON(1))
            }
      }

If that doesn't work, you can continue adding more type annotations until you've localized the error.

swift
images = word
    .filter { (s: String) -> Bool in s.containsString("important") }
    .flatMap { (word: String) -> Observable<JSON> in
        return self.api.loadFlickrFeed("karate")
            .catchError { (error: Error) -> Observable<JSON> in
                return just(JSON(1))
            }
      }

I would suggest first annotating return types and arguments of closures.

Usually after you have fixed the error, you can remove the type annotations to clean up your code again.

Debugging

Using debugger alone is useful, but usually using debug operator will be more efficient. debug operator will print out all events to standard output and you can add also label those events.

debug acts like a probe. Here is an example of using it:

swift
let subscription = myInterval(.milliseconds(100))
    .debug("my probe")
    .map { e in
        return "This is simply \(e)"
    }
    .subscribe(onNext: { n in
        print(n)
    })

Thread.sleepForTimeInterval(0.5)

subscription.dispose()

will print

text
[my probe] subscribed
Subscribed
[my probe] -> Event next(Box(0))
This is simply 0
[my probe] -> Event next(Box(1))
This is simply 1
[my probe] -> Event next(Box(2))
This is simply 2
[my probe] -> Event next(Box(3))
This is simply 3
[my probe] -> Event next(Box(4))
This is simply 4
[my probe] dispose
Disposed

You can also easily create your version of the debug operator.

swift
extension ObservableType {
    public func myDebug(identifier: String) -> Observable<Self.E> {
        return Observable.create { observer in
            print("subscribed \(identifier)")
            let subscription = self.subscribe { e in
                print("event \(identifier)  \(e)")
                switch e {
                case .next(let value):
                    observer.on(.next(value))

                case .error(let error):
                    observer.on(.error(error))

                case .completed:
                    observer.on(.completed)
                }
            }
            return Disposables.create {
                   print("disposing \(identifier)")
                   subscription.dispose()
            }
        }
    }
 }

Enabling Debug Mode

In order to Debug memory leaks using RxSwift.Resources or Log all HTTP requests automatically, you have to enable Debug Mode.

In order to enable debug mode, a TRACE_RESOURCES flag must be added to the RxSwift target build settings, under Other Swift Flags.

For further discussion and instructions on how to set the TRACE_RESOURCES flag for Cocoapods & Carthage, see #378

Debugging memory leaks

In debug mode Rx tracks all allocated resources in a global variable Resources.total.

In case you want to have some resource leak detection logic, the simplest method is just printing out RxSwift.Resources.total periodically to output.

swift
/* add somewhere in
    func application(_ application: UIApplication, didFinishLaunchingWithOptions launchOptions: [UIApplicationLaunchOptionsKey : Any]? = nil)
    */
    _ = Observable.interval(.seconds(1), scheduler: MainScheduler.instance)
        .subscribe(onNext: { _ in
            print("Resource count \(RxSwift.Resources.total)")
        })

Most efficient way to test for memory leaks is:

  • navigate to your screen and use it
  • navigate back
  • observe initial resource count
  • navigate second time to your screen and use it
  • navigate back
  • observe final resource count

In case there is a difference in resource count between initial and final resource counts, there might be a memory
leak somewhere.

The reason why 2 navigations are suggested is because first navigation forces loading of lazy resources.

KVO

KVO is an Objective-C mechanism. That means that it wasn't built with type safety in mind. This project tries to solve some of the problems.

There are two built in ways this library supports KVO.

swift
// KVO
extension Reactive where Base: NSObject {
    public func observe<E>(type: E.Type, _ keyPath: String, options: KeyValueObservingOptions, retainSelf: Bool = true) -> Observable<E?> {}
}

#if !DISABLE_SWIZZLING
// KVO
extension Reactive where Base: NSObject {
    public func observeWeakly<E>(type: E.Type, _ keyPath: String, options: KeyValueObservingOptions) -> Observable<E?> {}
}
#endif

Example how to observe frame of UIView.

WARNING: UIKit isn't KVO compliant, but this will work.

swift
view
  .rx.observe(CGRect.self, "frame")
  .subscribe(onNext: { frame in
    ...
  })

or

swift
view
  .rx.observeWeakly(CGRect.self, "frame")
  .subscribe(onNext: { frame in
    ...
  })

`rx.observe`

rx.observe is more performant because it's just a simple wrapper around KVO mechanism, but it has more limited usage scenarios

  • it can be used to observe paths starting from self or from ancestors in ownership graph (retainSelf = false)
  • it can be used to observe paths starting from descendants in ownership graph (retainSelf = true)
  • the paths have to consist only of strong properties, otherwise you are risking crashing the system by not unregistering KVO observer before dealloc.

E.g.

swift
self.rx.observe(CGRect.self, "view.frame", retainSelf: false)

`rx.observeWeakly`

rx.observeWeakly is somewhat slower than rx.observe because it has to handle object deallocation in case of weak references.

It can be used in all cases where rx.observe can be used and additionally

  • because it won't retain observed target, it can be used to observe arbitrary object graph whose ownership relation is unknown
  • it can be used to observe weak properties

E.g.

swift
someSuspiciousViewController.rx.observeWeakly(Bool.self, "behavingOk")

Observing structs

KVO is an Objective-C mechanism so it relies heavily on NSValue.

RxCocoa has built in support for KVO observing of CGRect, CGSize and CGPoint structs.

When observing some other structures it is necessary to extract those structures from NSValue manually.

Here are examples how to extend KVO observing mechanism and rx.observe* methods for other structs by implementing KVORepresentable protocol.

UI layer tips

There are certain things that your Observables need to satisfy in the UI layer when binding to UIKit controls.

Threading

Observables need to send values on MainScheduler(UIThread). That's just a normal UIKit/Cocoa requirement.

It is usually a good idea for your APIs to return results on MainScheduler. In case you try to bind something to UI from background thread, in Debug build RxCocoa will usually throw an exception to inform you of that.

To fix this you need to add observeOn(MainScheduler.instance).

URLSession extensions don't return result on MainScheduler by default.

Errors

You can't bind failure to UIKit controls because that is undefined behavior.

If you don't know if Observable can fail, you can ensure it can't fail using catchErrorJustReturn(valueThatIsReturnedWhenErrorHappens), but after an error happens the underlying sequence will still complete.

If the wanted behavior is for underlying sequence to continue producing elements, some version of retry operator is needed.

Sharing subscription

You usually want to share subscription in the UI layer. You don't want to make separate HTTP calls to bind the same data to multiple UI elements.

Let's say you have something like this:

swift
let searchResults = searchText
    .throttle(.milliseconds(300), scheduler: MainScheduler.instance)
    .distinctUntilChanged()
    .flatMapLatest { query in
        API.getSearchResults(query)
            .retry(3)
            .startWith([]) // clears results on new search term
            .catchErrorJustReturn([])
    }
    .share(replay: 1)    // <- notice the `share` operator

What you usually want is to share search results once calculated. That is what share means.

It is usually a good rule of thumb in the UI layer to add share at the end of transformation chain because you really want to share calculated results. You don't want to fire separate HTTP connections when binding searchResults to multiple UI elements.

Also take a look at Driver unit. It is designed to transparently wrap those share calls, make sure elements are observed on main UI thread and that no error can be bound to UI.

Making HTTP requests

Making http requests is one of the first things people try.

You first need to build URLRequest object that represents the work that needs to be done.

Request determines is it a GET request, or a POST request, what is the request body, query parameters ...

This is how you can create a simple GET request

swift
let req = URLRequest(url: URL(string: "http://en.wikipedia.org/w/api.php?action=parse&page=Pizza&format=json"))

If you want to just execute that request outside of composition with other observables, this is what needs to be done.

swift
let responseJSON = URLSession.shared.rx.json(request: req)

// no requests will be performed up to this point
// `responseJSON` is just a description how to fetch the response


let cancelRequest = responseJSON
    // this will fire the request
    .subscribe(onNext: { json in
        print(json)
    })

Thread.sleep(forTimeInterval: 3.0)

// if you want to cancel request after 3 seconds have passed just call
cancelRequest.dispose()

URLSession extensions don't return result on MainScheduler by default.

In case you want a more low level access to response, you can use:

swift
URLSession.shared.rx.response(myURLRequest)
    .debug("my request") // this will print out information to console
    .flatMap { (data: NSData, response: URLResponse) -> Observable<String> in
        if let response = response as? HTTPURLResponse {
            if 200 ..< 300 ~= response.statusCode {
                return just(transform(data))
            }
            else {
                return Observable.error(yourNSError)
            }
        }
        else {
            rxFatalError("response = nil")
            return Observable.error(yourNSError)
        }
    }
    .subscribe { event in
        print(event) // if error happened, this will also print out error to console
    }

Logging HTTP traffic

RxCocoa will log all HTTP request info to the console by default when run in debug mode. You may overwrite the URLSession.rx.shouldLogRequest closure to define which requests should and shouldn't be logged.

swift
URLSession.rx.shouldLogRequest = { request in
    // Only log requests to reactivex.org     
    return request.url?.host == "reactivex.org" || request.url?.host == "www.reactivex.org"
}

RxDataSources

... is a set of classes that implement fully functional reactive data sources for UITableViews and UICollectionViews.

RxDataSources are bundled here.

Fully functional demonstration how to use them is included in the RxExample project.

File: Documentation/HotAndColdObservables.md

Hot and Cold Observables

IMHO, I would suggest to more think of this as property of sequences and not separate types because they are represented by the same abstraction that fits them perfectly, Observable sequence.

This is a definition from ReactiveX.io

When does an Observable begin emitting its sequence of items? It depends on the Observable. A “hot” Observable may begin emitting items as soon as it is created, and so any observer who later subscribes to that Observable may start observing the sequence somewhere in the middle. A “cold” Observable, on the other hand, waits until an observer subscribes to it before it begins to emit items, and so such an observer is guaranteed to see the whole sequence from the beginning.

Hot Observables Cold observables
... are sequences ... are sequences
Use resources ("produce heat") no matter if there is any observer subscribed. Don't use resources (don't produce heat) until observer subscribes.
Variables / properties / constants, tap coordinates, mouse coordinates, UI control values, current time Async operations, HTTP Connections, TCP connections, streams
Usually contains ~ N elements Usually contains ~ 1 element
Sequence elements are produced no matter if there is any observer subscribed. Sequence elements are produced only if there is a subscribed observer.
Sequence computation resources are usually shared between all of the subscribed observers. Sequence computation resources are usually allocated per subscribed observer.
Usually stateful Usually stateless

File: Documentation/MathBehindRx.md

Math Behind Rx

Duality between Observer and Iterator / Enumerator / Generator / Sequences

There is a duality between the observer and generator patterns. This is what enables us to transition from the async callback world to the synchronous world of sequence transformations.

In short, the enumerator and observer patterns both describe sequences. It's fairly obvious why the enumerator defines a sequence, but the observer is slightly more complicated.

There is, however, a pretty simple example that doesn't require a lot of mathematical knowledge. Assume that you are observing the position of your mouse cursor on screen at given times. Over time, these mouse positions form a sequence. This is, in essence, an observable sequence.

There are two basic ways elements of a sequence can be accessed:

  • Push interface - Observer (observed elements over time make a sequence)
  • Pull interface - Iterator / Enumerator / Generator

You can also see a more formal explanation in this video:

File: Documentation/NewFeatureRequestTemplate.md

Please copy the following template here and fill in the missing fields so we can help you as soon as possible.

text
*Short description of missing functionality*:

  E.g. I want this library to implement xxx operator.

*Short code example of how you would like to use the API*:

  code goes here

*The reason why I need this functionality*:

  E.g. I was trying to ....

*Code I have right now*:

  code snippet that demonstrates how you've attempted to solve the problem

File: Documentation/Playgrounds.md

Playgrounds

To use playgrounds:

  • Open Rx.xcworkspace
  • Build the RxSwift scheme on My Mac.
  • Open Rx playground in the Rx.xcworkspace tree view.
  • Choose View > Debug Area > Show Debug Area

2. Official Technical Reference & Guides (ReactiveX/ReactiveX.github.io)

reactivex.github.io

ReactiveX Website - Jekyll + Bootstrap.

If you want to contribute, please make sure to make your changes to the develop branch, from which the master branch is generated.

Use Docker to Build & Run (Optional)

If you have Docker installed, you can use the Dockerfile in this repository to build and run the ReactiveX website. This one-liner will perform the build and run:
docker build -t reactivex.io - < Dockerfile && docker run -p 4000:4000 -it --rm -v $PWD:/app -t reactivex.io

Install tools

Jekyll is the static site generator. Uglifier, LESS, and Rake are build tools necessary for JavaScript and CSS compilation prior to Jekyll.

text
$ gem install jekyll uglifier rake
$ npm install less

Run

Rake's default task will compile styles the run jekyll serve.

text
$ rake

Marble diagrams

The marble diagrams are created in OmniGraffle. The source file is found at the RxJava Wiki (https://github.com/Netflix/RxJava.wiki.git) or can be downloaded here: https://raw.githubusercontent.com/wiki/ReactiveX/RxJava/images/rx-operators.graffle

If you want to make changes to this file, please check it out from the wiki, make changes, and check it back in there.

As of this writing, this OmniGraffle source is used to make two separate copies of the .png marble diagrams: one set hosted at the RxJava wiki which also serves as the source for the images in the javadocs, and one set hosted at this reactivex.github.io site. You have to generate these yourself by using OmniGraffle's "export" feature. If you check in a change that you want to be visible in both places, you'll have to check in the images in both repos.

The OmniGraffle file has grown to be cumbersome, and so there are some false-canvases with names that begin and end with lots of equals signs that are used as separators/headings to make individual diagrams easier to find. When you export the .png images based on this OmniGraffle file, it will also spit out empty .png images for these false-canvases. You'll just have to delete those manually or put them in your .gitignore or something. At least they're easy to spot.

Explore Related Technologies & Topics

View All 9,111+ Repositories →

Discover similar high-velocity repositories, agent skills, and OpenAPI specifications across the ecosystem.

#functional Browse functional catalog → #ios Browse ios catalog → #observer Browse observer catalog → #reactive Browse reactive catalog → #reactivex Browse reactivex catalog → #rxswift Browse rxswift catalog → ⚡ MCP Servers Model Context Protocol → 🧠 Agent Skills Prompt & rule files → 🔌 OpenAPI Specs REST & Swagger schemas →