The Techies' pub

Angular 2 – Working with Observables

Hope you are all well. In this post let’s talk about Observables. Now observables are mentioned whenever people talk about reactive programming. Why is that so ? Well, when an application work’s synchronously; or rather when the events in an application are in sync then the states of an application can be predicted. But in case of asynchronous events, it becomes difficult. Observables really do a great job in such scenarios.

Enough; show me some code.

Let’s get started.

A screencast of this post:

Let’s scaffold our application using the ng cli tool.

If you haven’t installed it already; kindly give the below command in a terminal to get the ng-cli tool. (Trust me; it’s really useful).

1

npm install-gangular-cli

Once you have installed the cli tool simply run the below command

1

ng new<appname>

This command will generate an application for us. Now open up this app in your editor of choice and navigate to the <appname>.component.ts file.

Here the appname I have used is observablestudy. So my file would be in the directory src -> app -> observablestudy.component.ts

Open up the file and add the below code.

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

import{Component}from'@angular/core';

import{Observable}from'rxjs/Observable';

import'rxjs/Rx';

@Component({

moduleId:module.id,

selector:'observablestudy-app',

templateUrl:'observablestudy.component.html',

styleUrls:['observablestudy.component.css']

})

exportclassObservablestudyAppComponent{

privatedata:any;

constructor(){

this.data=newObservable(observer=>{

setTimeout(()=>{

observer.next(21);

},1000);

setTimeout(()=>{

observer.next(22);

},2000);

setTimeout(()=>{

observer.complete();

},3000);

});

let subscriber=this.data.subscribe(

value=>console.log(value),

err=>console.log(err),

()=>console.log('done')

);

}

}

Please note that I am importing all the operators using import ‘rxjs/Rx’. You could simply import only the specific operator if needed.

Now if you look at the above code. I have created a very simple observable and to simulate a stream I have used timeout functions to send data at short intervals.

Then a subscriber subscribes to that observable to access the data sent therein.

Let’s look at some operators.

Map & Flatmap

You might already be familiar with the map operator while playing around with arrays in Javascript.

Add the below code to your observablestudy.component.ts file

1

2

3

let newobs=Observable.interval(1000).take(5).map(x=>x*5);

let newsub=newobs.subscribe(x=>console.log(x));

Now this is a very simple line of code. I have just created an observable which sends 5 values with an interval of 1 second. I have also used a map operator to transform the values (multiplying them by 5 every time).

I also have a subscriber which just prints the values on a console.If I run this code I would get an output as shown below.

As you can see the values are mapped to observables. Now what do we do in such cases? Or simply what should we do when we need to combine two streams. We could simply use flatMap. Now modify the code again as shown below.

As you can see flatMap not only maps each value into an observable but also subscribes to that observable resulting this in this output. This would be highly useful when you deal with multiple streams under a single observable.

Cold & Hot Observables:

Okay, All the observables which we have been seeing so far are cold ones.

Look at the below code.

1

2

3

4

5

6

7

let obs=Observable.interval(1000).take(7);

let firstsub=obs.subscribe(x=>console.log('first sub: '+x));

setTimeout(()=>{

let secondsub=obs.subscribe(x=>console.log('second sub: '+x));

},3000)

Now this observable will just generate 7 values i.e., from 0 to 6 with an interval of 1 second. We have two subscribers. secondsub has a timeout of 3 seconds which simply means that it subscribes 3 seconds after the firstsub. When I run this code I get the below output.

As you can see both the first and second subscribers act independently and both receive 7 values. Now what if we wanted them to act synchronously. If secondsub starts after 3 seconds it should start from the value 3 and not from 0. To do that we go for a hot observable.

Now tweak the code as shown below.

1

2

3

4

5

6

7

let obs=Observable.interval(1000).take(7).publish().refCount();

let firstsub=obs.subscribe(x=>console.log('first sub: '+x));

setTimeout(()=>{

let secondsub=obs.subscribe(x=>console.log('second sub: '+x));

},3000)

Now if you run the code this output will be shown

Now if you notice here the two subscribers have synced up. the firstsub receives 7 values whereas the secondsub has completely ignored 0,1,2 values which have appeared before it connected to the observable. This is a hot observable.

Apart from this there are several other operators that can be used along with Observables. You could reduce, scan, handle errors using catch, retry asynchronous operations etc. Observables are really excellent for asynchronous coding and making use of them would simplify your development process. To have a look at the complete operators available please click here.

Thanks for reading guys. Leave a comment below if you need help. Also feel free to ask me any queries by clicking here.