RxSwift: Observables

In my own understanding, the main idea behind Rx (Reactive Extension) Programming is that you have observables and subscribers. These two always go hand-in-hand. Observables could be a variable or a function (or a method, or however you prefer to call it), and the subscribers (subscribes to an observable) will be notified if there are any changes in your observables.

Observables

Can asynchronously send events (values) to its subscribers over time until it gets completed, or gets an error. It allows one or more subscribers to listen to any events in real time. Observables mean nothing if there are no subscribers. It can't do anything or start any process if there are no subscribers. Nothing is being observed, nothing gets emitted.

Observables can emit only three types of events:

• next event

  • The latest data value to be sent to the subscribers. We don't know how many next events an Observable may emit, but as long as it doesn't encounter a terminating event, it will continue to emit next events.

• completed event

  • A terminating event

  • It means the observable has finished (successfully) its lifecycle and will no longer emit new values to its subscribers.

  • Gets disposed after.

• error event

  • A terminating event

  • The Observable got an error and will emit this error to its subscribers and will no longer emit new values.

Vocabulary:

• Emit is what we'll call when the observables send events (it could be a next, completed or error event) and the subscribers receive these events.

• Operator In Rx, we call a method an "operator." They are methods / protocols to which Observable<Element> confirms. It allows you to manipulate the date emitted by the Observables. And to be able to take full advantage of Rx, you have to know what each operator does.

Basic Operators

I mentioned that you need to know what each operator does to take full advantage of Rx, because previously, before I started studying RxSwift (using RayWenderlich's RxSwift book - you should go buy it if you're interested), I misunderstood the basic operator and in the end, I couldn't make it work according to how I imagine it should work. I will explain later how I misunderstood some of the basic operators.

Just Operator

Observable<Element>.just(variable)

• Only accepts one element, one type

• Only emits one value, then dies. I mean, it gets terminated.

Example, integer:

var one = 1
 
 let observable = Observable<Int>.just(one)
 _ = observable.subscribe(onNext: { value in
 debugPrint("value: ", value)
 })
 
 one = 100 //Changed value

Console:

"value: " 1

I purposely made one a var (not a let) because I want to show you that the observable will only emit 1, any changes to variable one will no longer be emitted.

We can also do it this way so we can see that the Observable, indeed, got completed and terminated.

var one = 1
 
 let observable = Observable<Int>.just(one)
 _ = observable.subscribe { (value) in
 debugPrint("value: ", value)
 } onError: { (error) in
 debugPrint("error")
 } onCompleted: {
 debugPrint("completed")
 } onDisposed: {
 debugPrint("disposed")
 }
 
 one = 100 //Changed value

Console:

"value: " 1
"completed"
"disposed"

Example, array of integer:

var array = [1, 2, 4, 5]
 
 let observable = Observable<[Int]>.just(array)
 _ = observable.subscribe { (value) in
 debugPrint("value: ", value)
 } onError: { (error) in
 debugPrint("error")
 } onCompleted: {
 debugPrint("completed")
 } onDisposed: {
 debugPrint("disposed")
 }
 
 array.insert(3, at: 2)

Console:

"value: " [1, 2, 4, 5]
"completed"
"disposed"

Did you guess it right? It will only emit [1, 2, 4, 5], even though the array updated and inserted a new integer, it was no longer emitted.

Example, class:

class Person {
 var name: String
 var age: Int
 var sex: String
 
 init(name: String, age: Int, sex: String) {
 self.name = name
 self.age = age
 self.sex = sex
 }
}

let person = Person(name: "Jen", age: 29, sex: "F")
 
let observable = Observable<Person>.just(person)
 _ = observable.subscribe { (value) in
 debugPrint("value: ", value.name, value.age, value.sex)
 } onError: { (error) in
 debugPrint("error")
 } onCompleted: {
 debugPrint("completed")
 } onDisposed: {
 debugPrint("disposed")
 }
 
 person.age = 30 // Update person

Console:

"value: " "Jen" 29 "F"
"completed"
"disposed"

Example, array of class:

 let person1 = Person(name: "Jen", age: 29, sex: "F")
 let person2 = Person(name: "John", age: 30, sex: "M")
 let person3 = Person(name: "Brandon", age: 2, sex: "M")
 let persons = [person1, person2, person3]

 let observable = Observable<[Person]>.just(persons)
 _ = observable.subscribe { (value) in
     debugPrint("value: ", value)
 
     // Value is an array
     for i in 0..<value.count {
         let p = value[i]
         debugPrint("[\(i)]: ", p.name, p.age, p.sex)
     }
 } onError: { (error) in
 debugPrint("error")
 } onCompleted: {
 debugPrint("completed")
 } onDisposed: {
 debugPrint("disposed")
 }
 
 person1.age = 30 // Update person1
 person2.name = "John Kenneth" // Update person2

Console:

"value: " [JenRxPractice.Person, JenRxPractice.Person, JenRxPractice.Person]
"[0]: " "Jen" 29 "F"
"[1]: " "John" 30 "M"
"[2]: " "Brandon" 2 "M"
"completed"
"disposed"

You get it, right? After it emits a value, it simply terminates and any changes after will no longer be emitted

of Operator

Observable.of(one or more variable of same data type)

• No need to explicitly declare the type, if you remember the just operator needs to explicitly declare the type: Observable<Int>.just(var)

Example, integer:

let one = 1
let two = 2
let three = 3
 
let observable = Observable.of(one, two, three)
_ = observable.subscribe { (value) in
debugPrint("value: ", value)
} onError: { (error) in
debugPrint("error")
} onCompleted: {
debugPrint("completed")
} onDisposed: {
debugPrint("disposed")
}

Console:

"value: " 1
"value: " 2
"value: " 3
"completed"
"disposed"

In this example, it emitted thrice, that's why the value: was printed thrice.

This is one of the operators that I misunderstood. I thought that if I update these variables (after the observable was created), the new values will be emitted, too. But apparently, not.

var one = 1
var two = 2
var three = 3
 
let observable = Observable.of(one, two, three)
_ = observable.subscribe { (value) in
 debugPrint("value: ", value)
} onError: { (error) in
 debugPrint("error")
} onCompleted: {
 debugPrint("completed")
} onDisposed: {
 debugPrint("disposed")
}
 
 // Update variables
 one = 100
 two = 200
 three = 300

Console:

"value: " 1
"value: " 2
"value: " 3
"completed"
"disposed"

The updated or the new values are not emitted. Only the last value of these integers were emitted. In this case, the 'last' value and the initial value is the same, so it only emitted the initial values.

What happens if we update the integer values before we created the observable?

var one = 1
var two = 2
var three = 3
 
 // Update variables before creating the observable, these make it its 'last' value
 one = 100
 two = 200
 three = 300
 
 let observable = Observable.of(one, two, three)
 _ = observable.subscribe { (value) in
 debugPrint("value: ", value)
 } onError: { (error) in
 debugPrint("error")
 } onCompleted: {
 debugPrint("completed")
 } onDisposed: {
 debugPrint("disposed")
 }

Console:

"value: " 100
"value: " 200
"value: " 300
"completed"
"disposed"

Example, array of integer:

let one = 1
let two = 2
let three = 3
let array = [one, two, three]
 
let observable = Observable.of(array)
_ = observable.subscribe { (value) in
 debugPrint("value: ", value)
} onError: { (error) in
 debugPrint("error")
} onCompleted: {
 debugPrint("completed")
} onDisposed: {
 debugPrint("disposed")
}

The array here is considered one element, so it will only emit once.

Console:

"value: " [1, 2, 3]
"completed"
"disposed"

And even if you change the value of the elements inside the array, it will no longer be emitted.

var one = 1
var two = 2
var three = 3
let array = [one, two, three]
 
let observable = Observable.of(array)
_ = observable.subscribe { (value) in
 debugPrint("value: ", value)
} onError: { (error) in
 debugPrint("error")
} onCompleted: {
 debugPrint("completed")
} onDisposed: {
 debugPrint("disposed")
}
 
 // Update variables in array
 one = 100
 two = 200
 three = 300

Console:

"value: " [1, 2, 3]
"completed"
"disposed"

from Operator

Observable.from(array)

• Only takes an array

• Creates an observable of individual elements from an array.

var person1 = Person(name: "Jen", age: 29, sex: "F")
var person2 = Person(name: "John", age: 30, sex: "M")
var person3 = Person(name: "Brandon", age: 2, sex: "M")
var persons = [person1, person2, person3]
 
let observable = Observable.from(persons)
_ = observable.subscribe { (value) in
 debugPrint("value: ", value.name, value.age, value.sex)
} onError: { (error) in
 debugPrint("error")
} onCompleted: {
 debugPrint("completed")
} onDisposed: {
 debugPrint("disposed")
}

 // These changes never gets emitted
 person1.age = 30 // Update age of person1
 person2.name = "John Kenneth" // Update name of person2
 persons.removeLast() // Update array
 person1 = Person(name: "Mary", age: 0, sex: "F") // Change person1

Console:

"value: " "Jen" 29 "F"
"value: " "John" 30 "M"
"value: " "Brandon" 2 "M"
"completed"
"disposed"

I am honestly confused on this part as well, because the book did say it creates an observable of each element in the array. So I expected that any changes I made to each element (first two changes) will be emitted, but it didn't. I don't know if that's the new Observable.from and that the book just wasn't updated on that change, let me know if you experienced the same thing.

One of my mistakes in the past is that if I update the array (append, insert, delete), it will be emitted, but it didn't.

And lastly, I tried updating the whole person1 variable to a different Person, and it still didn't get emitted.

That will be the end for the basic operators. But before I end this post, I want to discuss how we can manually terminate our subscribers.

Disposing and Terminating Subscribers

If you noticed in my codes, I have this:

_ = observable.subscribe { (value) in ...

This means the subscriber will return something. It will actually return a "disposable." That way, we can manually cancel our subscription anytime we want.

let subscription = observable.subscribe { (value) in ...

and just call:

subscription.dispose()

Once we call this, the observable will stop emitting events to this subscriber. One way we can use this is if we want to cancel our subscriptions when we go out of the screen, or if we deallocate the screen.

But sometimes, we use too many subscriptions in our class, managing them one by one, and making sure we dispose of it can be a headache.

Introducing DisposeBag

It holds all the disposables and will call .dispose() on each one of them when the bag gets deallocated. You can do this by declaring the disposeBag in the class:

let bag = DisposeBag()

And on each subscription, you add .disposed(by: ):

observable.subscribe(onNext: { value in
 debugPrint("value received from observable2: ", value)
 })
 .disposed(by: bag)

When the class instance is to be deallocated, the disposeBag will be deallocated as well, and will dispose/cancel all subscriptions inside that class. So make sure your class instance is deallocated when it should be so that all these subscriptions inside the bag will be cancelled, too.