RxSwift part 1

ReactiveX logo


Good day, habrovchane. In this series of articles, I would like to talk about reactive programming, namely about the
RxSwift framework . On Habré and the network were articles on RxSwift, but, in my opinion, they are too difficult for beginners. Therefore, if you start to comprehend reactive programming in iOS, then I ask for cat.


Let's start by defining what reactive programming is.


Reactive programming is a programming paradigm focused on data flows and propagation of changes.

So says the great Wikipedia .


In other words, in the case when we program in the imperative style, we write in the code a set of commands that must be executed sequentially. Reactive programming style adheres to several different concepts. With a reactive programming style, our program is a "listener" of state changes in our observed objects. It sounds complicated, but it’s not like that; it’s enough to just get into this concept and everything will become extremely easy and understandable.no bugs yet.


I will not describe how to install the framework, it is easy to do by clicking on the link . Let's get down to practice.


Observable


Let's start with a simple, but important, observable object or Observable. Observable is what will give us data, it is needed to generate the data stream.


let observable = Observable<String>.just("Первый observable")

BINGO ! We created the first observable.


So what?


Since we created an observable object, it is logical that we need to create an object that will be observed.


let observable = Observable<String>.just("Первый observable")
_ = observable.subscribe { (event) inprint(event)
}

in the log we get the following: 


next(Первый observable)  
completed

completed?


Observable sends us information about its events, there are only 3 types:


  • next
  • error
  • completed

Along with the next element comes the element that we sent and all the events sent by us, the error is sent as it is clear from the name in case of an error, and completed in the case when our observable sent all the data and completes the work.


We can create more detailed the observer subscriber and get a more convenient view for handling all events.


_ = observable.subscribe(onNext: { (event) inprint(event)
}, onError: { (error) inprint(error)
}, onCompleted: {
    print("finish")
}) {
    print("disposed") 
    //о том, что это такое и зачем это мы поговорим позже
}

Первая последовательность
finish  
disposed

In observable, you can create a sequence not only from a single line, but in general not only from lines, we can put any data type there.


let sequence = Observable<Int>.of(1, 2, 4, 5, 6)
_ = sequence.subscribe { (event) in
    print(event)
}

next(1)
next(2)
... 
completed

Observable can be created from an array of values.


let array = [1, 2, 3]
let observable = Observable<Int>.from(array)
_ = observable.subscribe { (event) inprint(event)
}

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

One observable can have as many subscriberes as it wishes . And now terminology, what is Observable?


Observable is the foundation of all Rx, which asynchronously generates a sequence of immutable data and allows others to subscribe to it.


Disposing


Now that we can create a sequence and subscribe to them, we need to deal with such a thing as disposing .


It is important to remember that Observable is a “cold” type, that is, our observable does not “emit” any events until it is subscribed to. An observable exists until it sends an error message ( error ) or a completion message ( completed ). If we want to explicitly cancel the subscription, we can do the following.


//вариант №1//создали массив значенийlet array = [1, 2, 3]
//создали observable из массива значенийlet observable = Observable<Int>.from(array)
//создали подписку на observablelet subscription = observable.subscribe { (event) inprint(event)
}
//dispos'им нашу одноразовую подписку
subscription.dispose()

There are more nice correct option.


//создаем сумку "утилизации"let bag = DisposeBag()
//создали массив значенийlet array = [1, 2, 3]
//создали observable из массива значенийlet observable = Observable<Int>.from(array)
//создали подписку на observable
_ = observable.subscribe { (event) inprint(event)
}.disposed(by: bag)

This way we add our subscription to the recycling bag or to DisposeBag .
What is it for? If you, using a subscription, do not add it to the bag or do not explicitly call dispose , or, in extreme cases, do not bring the observable to an end in any way , then most likely you will get a memory leak. You will use DisposeBag very often in your work with RxSwift.


Operators


In functional reactive programming (FRP further) there are many built-in operators for transforming observable elements. There is a site rxmarbles , on it you can see the work and the effect of all operators, well, we still consider some of them.


Map


The map operator is used very often and I think that it is familiar to many, with its help we transform all the elements obtained.
Example:


let bag = DisposeBag()
let array = [1, 2, 3]
let observable = Observable<Int>.from(array).map { $0 * 2 }
_ = observable.subscribe { (e) in
    print(e)
}.disponsed(by: bag)

What we get in the console:


next(2)
next(4)
next(6)  
completed

We take each element of the sequence and create a new resultant sequence. To make it clearer what is happening it is better to record in more detail.


let bag = DisposeBag()
let observable = Observable<Int>.from(array)
//создаем новый observablelet transformObservable = observable.map { $0 * 2 }
_ = transformObservable.subscribe { (e) inprint(e)
}.disposed(by: bag)

What is "$ 0"?


$ 0 is the element name by default, we can use abbreviated and full notation in the methods, most commonly abbreviated notation is used.


//сокращенная формаlet transformObservable = observable.map { $0 * 2 }
 //полная формаlet transformObservable = observable.map { (element) -> Intinreturn element * 2
}

Agree that the abbreviated form is much easier, right?


Filter


The filter operator allows us to filter the data emitted by our observable, that is, when we subscribe, we will not receive values ​​that we do not need.
Example:


let bag = DisposeBag()
let array = [1, 2, 3, 4, 5 , 6, 7]
//создаем observable из массиваlet observable = Observable<Int>.from(array)
//применяем функцию filter, сохраняя результат в новый observablelet filteredObservable = observable.filter { $0 > 2 }
//подписка_ = filteredObservable.subscribe { (e) inprint(e)
}.disposed(by: bag)

What do we get to the console?


next(3)
next(4)  
next(5)
... 
completed

As we see, in the console we have only those values ​​that satisfy our conditions.


By the way, operators can be combined, that's what it would look like if we wanted to apply both the filtering operator and the map operator right away .


let bag = DisposeBag()
let array = [1, 2, 3, 4, 5 , 6, 7]
let observable = Observable<Int>.from(array)
let filteredAndMappedObservable = observable
    .filter { $0 > 2 }
    .map { $0 * 2 }
_ = filteredAndMappedObservable.subscribe { (e) in
    print(e)
}.disposed(by: bag)

Console:


next(6)
next(8)
next(10)
next(12)
next(14)
completed

Distinct


Another excellent operator, which is associated with filtering, the disctinct operator allows you to skip only the changed data, it is best to immediately turn to an example and everything will become clear.


let bag = DisposeBag()
letarray = [1, 1, 1, 2, 3, 3, 5, 5, 6]
let observable = Observable<Int>.from(array)
let filteredObservable = observable.distinctUntilChanged()        
_ = filteredObservable.subscribe { (e) in
    print(e)
}.disposed(by: bag)

In the console we get the following:


next(1)
next(2)
next(3)
next(5)
next(6)
completed

that is, if the current element of the sequence is identical to the previous one, then it is skipped and this happens until an element different from the previous one appears, it is very convenient when working with the UI, namely with the table, if we received data , the same as we have now, then you should not reload the table.


TakeLast


A very simple takeLast operator , we take the nth number of elements from the end.


let bag = DisposeBag()
letarray = [1, 1, 1, 2, 3, 3, 5, 5, 6]
let observable = Observable<Int>.from(array)
let takeLastObservable = observable.takeLast(1)        
_ = takeLastObservable.subscribe { (e) in
    print(e)
}.disposed(by: bag)

In the console we get the following:


next(6)
completed

Throttle and Interval


Then I decided to take 2 operators at once, simply because with the help of the second one, it is easy to show the work of the first one.


The throttle operator allows you to take a pause between capturing the transmitted values, it’s a very simple example, you write a reactive application, use the search string and don’t want to reload the table or go to the server every time you enter the data, so you use throttle and thus say that Do you want to take user data every 2 seconds (for example, you can put any interval) and reduce resource consumption for unnecessary processing, how does it work and is described in the code? See below for an example.


let bag = DisposeBag()
//observable будет генерировать значение каждые 0.5 секунды с шагом 1 начиная от 0let observable = Observable<Int>.interval(0.5, scheduler: MainScheduler.instance)
let throttleObservable = observable.throttle(1.0, scheduler: MainScheduler.instance)
_ = takeLastObservable.subscribe { (event) inprint("throttle \(event)")
}.disposed(by: bag)

The console will:


throttle next(0)
throttle next(2)
throttle next(4)
throttle next(6)
...

The interval statement causes the observable to generate values ​​every 0.5 seconds in increments of 1 starting at 0, this is the simple timer for Rx. It turns out once the values ​​are generated every 0.5 seconds, then 2 values ​​are generated per second, simple arithmetic, and the throttle operator waits a second and takes the last value.


Debounce


Debounce is very similar to the previous statement, but a little more smarter, in my opinion. The debounce operator waits for the nth amount of time and if there were no changes from the start of its timer, it takes the last value, but if we send the value, the timer will restart again. This is very useful for the situation described in the previous example, the user enters data, we wait for him to finish (if the user is idle for a second or a half), and then we begin to perform some actions. Therefore, if we simply change the operator in the previous code, we will not get the values ​​to the console, because debounce will wait a second, but every 0.5 seconds will receive a new value and restart its timer, so we will not get anything. Let's see an example.


let bag = DisposeBag()
let observable = Observable<Int>.interval(1.5, scheduler: MainScheduler.instance)
let debounceObservable = observable.debounce(1.0, scheduler: MainScheduler.instance)
_ = debounceObservable.subscribe({ (e) inprint("debounce \(e)")
}).disposed(by: bag)

At this stage, I propose to finish with the operators, there are a lot of them in the RxSwift framework, it cannot be said that all of them are very necessary in everyday life, but you still need to know about their existence, therefore it is advisable to read the full list of operators on the rxmarbles website .


Scheduler


A very important topic that I would like to touch on in this article is the scheduler. Scheduler, let us run our observable on certain threads and they have their own subtleties. We begin, there are 2 types to install observable scheduler - [observeOn] () and [subscribeOn] ().


SubscribeOn


SubscribeOn is responsible for the flow in which the entire observable process will be executed until its events reach the handler (subscriber).


Observeon


As you can guess, observeOn is responsible for the stream in which the events received by the subscriber will be processed.


This is a very cool thing, because we can very easily put the download of something from the network into the background stream, and when getting the result, go to the main stream and somehow affect the UI.


Let's see how this works with an example:


let observable = Observable<Int>.create { (observer) -> Disposable inprint("thread observable -> \(Thread.current)")
    observer.onNext(1)
    observer.onNext(2)
    return Disposables.create()
}.subscribeOn(ConcurrentDispatchQueueScheduler(qos: .background))
_ = observable
    .observeOn(MainScheduler.instance)
    .subscribe({ (e) inprint("thread -> \(Thread.current)")
        print(e)
})

In the console, we get:


thread observable -> <NSThread: 0x604000465040>{number = 3, name = (null)}
thread -> <NSThread: 0x60000006f6c0>{number = 1, name = main}
next(1)
thread -> <NSThread: 0x60000006f6c0>{number = 1, name = main}
next(2)

We see that the observable was created in the background stream, and we were processing the data in the main stream. This is useful when working with a network for example:


let rxRequest = URLSession.shared.rx.data(request: URLRequest(url: URL(string: "http://jsonplaceholder.typicode.com/posts/1")!)).subscribeOn(ConcurrentDispatchQueueScheduler(qos: .background))
_ = rxRequest
    .observeOn(MainScheduler.instance)
    .subscribe { (event) inprint("данные \(event)")
        print("thread \(Thread.current)")
}

Thus, the request will be executed in the background thread, and all response processing will occur in main. At this stage, it is too early to say that the URLSession suddenly drew a method for the rx , this will be discussed later, this code was cited as an example of using Scheduler , by the way, we will get the following answer to the console.


curl -X GET 
"http://jsonplaceholder.typicode.com/posts/1" -i -v
Success (305ms): Status 200
**данные next(292 bytes)**
thread -> <NSThread: 0x600000072580>{number = 1, name = main}
данные completed
thread -> <NSThread: 0x600000072580>{number = 1, name = main}

In the final, let's see what else data came to us, for this we will have to perform a check in order not to start parsing the message completed accidentally.


_ = rxRequest
    .observeOn(MainScheduler.instance)
    .subscribe { (event) inif (!event.isCompleted && event.error == nil) {
            let json = try? JSONSerialization.jsonObject(with: event.element!, options: [])
            print(json!)
        }
        print("data -> \(event)")
        print("thread -> \(Thread.current)")
}

We check that event is not an observable shutdown message and not an error from it, although it was possible to implement a different subscription method and process all these types of events separately, but you can do it yourself, and we will receive the following in the console.


curl -X GET 
"http://jsonplaceholder.typicode.com/posts/1" -i -v
Success (182ms): Status 200
{
    body = "quia et suscipit\nsuscipit recusandae consequuntur expedita et cum\nreprehenderit molestiae ut ut quas totam\nnostrum rerum est autem sunt rem eveniet architecto";
    id = 1;
    title = "sunt aut facere repellat provident occaecati excepturi optio reprehenderit";
    userId = 1;
}
data -> next(292 bytes)
thread -> <NSThread: 0x60400006c6c0>{number = 1, name = main}
data -> completed
thread -> <NSThread: 0x60400006c6c0>{number = 1, name = main}

Data received :-)


Subjects


We turn to the hot, namely from the "cold" or "passive" observable to the "hot" or "active" observable, which are called subjects. If before that our observables started their work only after subscribing to them and you had a question in your head "well, and why do I need all this?", Then Subjects always work and always send the received data.


Like this? In the case of observable, we went to the clinic, went to the evil granny onthe reception the receptionist came up and asked which office to go to, then babulotion answered us, in the case of subjects, babulotation stands and listens to the schedule and condition of the doctors at the hospital and as soon as it receives information about the movement of any doctor immediately says this, ask something is useless to babulence, we can just walk up, listen to it, leave, and she will continue to say, something is carried away with comparisons, let's get to the code.


Create one subject and 2 subscribers, create the first one right after the subject, send the subject a value, and then create the second and send a couple more values.


let subject = PublishSubject<Int>()
subject.subscribe { (event) inprint("первый подписчик \(event)")
}
subject.onNext(1)
_ = subject.subscribe { (event) inprint("второй подписчик \(event)")
}
subject.onNext(2)
subject.onNext(3)
subject.onNext(4)

What will we see in the console? correctly, the first managed to get the first event, and the second did not.


первый подписчик next(1)
первый подписчик next(2)
второй подписчик next(2)
первый подписчик next(3)
второй подписчик next(3)
первый подписчик next(4)
второй подписчик next(4)

Already more suited to your understanding of reactive programming?
Subjects come in several forms, they all differ in how they send values.


PublishSubject is the easiest, no matter what it is, it just sends to all subscribers what has come to it and forgets about it.


ReplaySubject - but this is the most important one; when creating, we give it a buffer size (how many values ​​it will memorize), as a result it stores the last n values ​​in memory and sends them immediately to a new subscriber. 


let subject = ReplaySubject<Int>.create(bufferSize: 3)
subject.subscribe { (event) inprint("первый подписчик \(event)")
}
subject.onNext(1)
subject.subscribe { (event) inprint("второй подписчик \(event)")
}
subject.onNext(2)
subject.onNext(3)
subject.subscribe { (event) inprint("третий подписчик \(event)")
}
subject.onNext(4)

We look in the console


первый подписчик next(1)
второй подписчик next(1)
первый подписчик next(2)
второй подписчик next(2)
первый подписчик next(3)
второй подписчик next(3)
третий подписчик next(1)
третий подписчик next(2)
третий подписчик next(3)
первый подписчик next(4)
второй подписчик next(4)
третий подписчик next(4)

BehaviorSubject is not so naughty as the previous one, it has a starting value and it remembers the last value and sends it immediately after the subscriber’s subscription.


let subject = BehaviorSubject<Int>(value: 0)
subject.subscribe { (event) inprint("первый подписчик \(event)")
}
subject.onNext(1)
subject.subscribe { (event) inprint("второй подписчик \(event)")
}
subject.onNext(2)
subject.onNext(3)
subject.subscribe { (event) inprint("третий подписчик \(event)")
}
subject.onNext(4)

Console


первый подписчик next(0)
первый подписчик next(1)
второй подписчик next(1)
первый подписчик next(2)
второй подписчик next(2)
первый подписчик next(3)
второй подписчик next(3)
третий подписчик next(3)
первый подписчик next(4)
второй подписчик next(4)
третий подписчик next(4)

Conclusion


It was an introductory article, written so that you know the basics and could later build on them. In the following articles, we will look at how to work with RxSwift with iOS UI components, creating extensions for UI components.


Not RxSwift'om one


Reactive programming is implemented not only in the RxSwift library, there are several implementations, here are the most popular ones ReacktiveKit / Bond , ReactiveSwift / ReactiveCocoa . They all have small differences in the implementation under the hood, but in my opinion it is better to start your knowledge of “reactive” with RxSwift, as it is the most popular among them and it will have more answers in great Google , but after understand the essence of this concept, you can choose the library to your taste and color.
Article author: Grechikhin Pavel

Tags:

Also popular now: