Saturday, May 9, 2020

Angular 8 Tutorial: Observable and RXJS Examples


Angular 8 Tutorial:  Observable and RXJS Examples

Learning Angular 8 Observable and RXJS with examples that might be useful for building Angular application

In this Angular 8 tutorial, we will show you how to learn or understand Angular 8 Observable and RxJS by a few examples. Maybe you already know that Observable and RxJS use with HttpClient. This time, we will show you the example of Angular Observable and RxJS with HttpClient, data exchange between components, Async pipe, Router, and Reactive Forms.

Table of Contents:

Angular Observable use as an interface to handle a variety of common asynchronous operations such as send observable data from child to parent component by defining custom events, handle AJAX or HTTP requests and responses, listen and respond user input in Angular Router and Forms. RxJS (Reactive Extensions for JavaScript) is a library for reactive programming using observables that makes it easier to compose asynchronous or callback-based code. The following tools, frameworks, and libraries or modules are required for this tutorial.

  1. Node.js
  2. Angular 8
  3. Terminal (Linux/Mac) or Node Command Line (Windows)
  4. Text Editor or IDE (We use VSCode)


Preparation

To make all examples runnable, we need to prepare an Angular 8 application to implements all Angular Observable and RxJS. We are creating an Angular 8 app using Angular CLI which needs Node.js and NPM to install or update it. Make sure you have to install Node.js and NPM before installing or updating Angular CLI. To install or update an Angular CLI type this command.

sudo npm install -g @angular/cli

Next, create a new Angular 8 app using Angular CLI by type this command.

ng new angular-observable-rxjs

That command will create a new Angular 8 app with the name `angular-observable-rxjs` and pass all questions as default then the Angular CLI will automatically install the required NPM modules. After finished, go to the newly created Angular 8 folder then run the Angular 8 app for the first time.

cd ./angular-observable-rxjs
ng serve --open

Using that "--open" parameters will automatically open the Angular 8 in your default web browser. Here's the Angular 8 default page look like.

Angular 8 Tutorial:  Observable and RXJS Examples - Home Page


Basic Angular Observable and RxJS

In Angular Observable, there are a publisher and subscriber. The publisher can create an Observable instance that defines a subscriber function. The subscriber is receiving notification by executing the observable using subscribe() method and stop receiving the notification using the unsubscribe() method. To practice a Basic Angular Observable and RxJS we will use existing `src/app/app.component.ts`. Open and edit that file to implement these basic examples.

Observe Timer Example

This example demonstrates the basic usage model by showing how the RxJS timer works by subscribing it and stopped after unsubscribing it. Add this import of RxJS timer.

import { timer } from 'rxjs';

Add a constant variable after the imports that declare the RxJS timer.

const source = timer(1000, 2000);

Subscribe the RxJS timer then print out to Javascript console.

const subscribe = source.subscribe(val => console.log(val));

Add a function that will unsubscribe the RxJS timer.

setTimeout(() => { subscribe.unsubscribe(); }, 10000);

Now, you will see the timer print the increased steps and stopped after 10 seconds.

Angular 8 Tutorial:  Observable and RXJS Examples - Output Console

Basic Subscribing using Observer

This example shows an example of subscribing using Observer. Add this RxJS `of` operator.

import { of } from 'rxjs';

RxJS `of` operator used to emit a variable amount of values in a sequence and then emits a complete notification. Next, declare a constant variable before the Angular @Component that contains the sequence of observable string that will emit using RxJS `of` operator.

const myObservable = of('apple', 'orange', 'grappe');

Add a constant observer object variable after the above constant.

const myObserver = {
  next: (x: string) => console.log('Observer got a next value: ' + x),
  error: (err: string) => console.error('Observer got an error: ' + err),
  complete: () => console.log('Observer got a complete notification'),
};

Inside the class, the body adds a constructor that will execute the observable by subscribing to the above observer object.

constructor() {
  myObservable.subscribe(myObserver);
}

That observable subscribing example will show output like this.

Observer got a next value: apple
app.component.ts:11 Observer got a next value: orange
app.component.ts:11 Observer got a next value: grappe
app.component.ts:13 Observer got a complete notification
core.js:38780 Angular is running in the development mode. Call enableProdMode() to enable the production mode.
client:52 [WDS] Live Reloading enabled.

Observable with Constructor Example

We can use the observable constructor to create an observable stream of any type. The observable’s subscribe() executes with the constructor argument. The Observer object received by a subscriber function then publishes values using the observer's next() method. Add or modify the RxJS import to add Observer function.

import { Observable } from 'rxjs';

Add a function that runs the sequence of string synchronously then unsubscribe after values completed delivered.

function sequenceSubscriber(observer) {
  observer.next('Apple');
  observer.next('Orange');
  observer.next('Grappe');
  observer.complete();

  return {unsubscribe() {}};
}

Instantiate Observable that will deliver the above sequence of string.

const sequence = new Observable(sequenceSubscriber);

Execute the observable by print out the string sequence to the Javascript console.

sequence.subscribe({
  next(msg) { console.log(msg); },
  complete() { console.log('Finished sequence'); }
});

The Observable that Publishes Events Example

This example is taken from the official Angular docs with a little modification to work in `src/app/app.component.ts`. First, add an HTML input element to the existing `src/app/app.component.html`.

<div>
  <input type="text" id="yourname" placeholder="Type your name here"/>
</div>

Back to `src/app/app.component.ts`, modify this import to add Angular AfterViewInit lifecycle interface.

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

Implement that AfterViewInit in the main class.

export class AppComponent implements AfterViewInit {
...
}

Add a function that executes the Observable of the event from the target element.

fromEvent(target: HTMLInputElement, eventName: string) {
  return new Observable((observer) => {
    const handler = (e: unknown) => observer.next(e);

    target.addEventListener(eventName, handler);

    return () => {
      target.removeEventListener(eventName, handler);
    };
  });
}

Add ngAfterViewInit() function call the above function to create an observable that publishes key-down events.

ngAfterViewInit() {
  const ESC_KEY = 27;
  const nameInput = document.getElementById('yourname') as HTMLInputElement;
  this.fromEvent(nameInput, 'keydown')
  .subscribe((e: KeyboardEvent) => {
    if (e.keyCode === ESC_KEY) {
      nameInput.value = '';
    }
  });
}

Now, every ESC key push in the HTML input element, it will clear the input value.


Observable and RxJS with HttpClient

Almost our Angular tutorial involves REST API requests using Angular HttpClient, Observable, and RxJS. So, this example is a combination of these Angular features. This time we will put all in `src/app/app.component.html`, in the real world, you can separate the REST API access using their own services. First, open and edit `src/app/app.module.ts` then add this import of HttpClientModule that is part of @angular/common/http.

import { HttpClientModule } from '@angular/common/http';

Add it to @NgModule import array after the BrowserModule.

imports: [
  BrowserModule,
  HttpClientModule
],

Back to `src/app/app.component.ts` then add these imports of the required Angular HttpClient, RxJS, and Observable.

import { Observable, of, throwError } from 'rxjs';
import { HttpClient, HttpHeaders, HttpErrorResponse } from '@angular/common/http';
import { catchError, tap, map } from 'rxjs/operators';

Add these constants before the `@Component`.

const httpOptions = {
  headers: new HttpHeaders({'Content-Type': 'application/json'})
};
const apiUrl = 'http://localhost:3000/api/v1/products';

Inject `HttpClient` module to the constructor.

constructor(private http: HttpClient) { }

Add the error handler function that returns as an Observable.

private handleError<T> (operation = 'operation', result?: T) {
  return (error: any): Observable<T> => {

    // TODO: send the error to remote logging infrastructure
    console.error(error); // log to console instead

    // Let the app keep running by returning an empty result.
    return of(result as T);
  };
}

Add the functions to get data from the REST API and return the response as Observable. It also, extract or read data using the Pipe method and tap operator and catch the error by calling the error handler. RxJS pipe method is an Observable method to make data reading easier. RxJS Tap operator uses to perform a side effect for every emission on the source Observable but returns an Observable that is identical to the source.

getProducts(): Observable<any[]> {
  return this.http.get<any[]>(apiUrl)
    .pipe(
      tap(product => console.log('fetched products')),
      catchError(this.handleError('getProducts', []))
    );
}

getProduct(id: number): Observable<any> {
  const url = `${apiUrl}/${id}`;
  return this.http.get<any>(url).pipe(
    tap(_ => console.log(`fetched product id=${id}`)),
    catchError(this.handleError<any>(`getProduct id=${id}`))
  );
}

To run that function simply call the subscribe method and put the result to the variable that declared before the constructor.

data: any[] = [];

constructor(private http: HttpClient) {
  this.getProducts()
  .subscribe((res: any) => {
    this.data = res;
    console.log(this.data);
  }, err => {
    console.log(err);
  });
}


Data Exchange between Components

This example shows data exchange between components using service. Data emitted by service that subscribes by components. Component able to change data and data changes received by another component. For that, add a service and two components.

ng g service Shared
ng g component Acomp
ng g component Bcomp

Open and edit `src/app/shared.service.ts` then add this import of RxJS BehaviorSubject. BehaviorSubject is a Subject that requires an initial value and emits its current value to new subscribers.

import { BehaviorSubject } from 'rxjs';

Declare a variable before the constructor that instantiates BehaviorSubject with object data. Also, a variable that converts BehaviorSubject as Observable.

private dataSource = new BehaviorSubject({name: 'Maradona'});
currentData = this.dataSource.asObservable();

Add a function to change the data of BehaviorSubject.

changeData(data: any) {
  this.dataSource.next(data);
}

Next, open and edit `src/app/acomp/acomp.component.ts` then add these imports of Router and SharedService.

import { Router } from '@angular/router';
import { SharedService } from '../shared.service';

Inject that module to the constructor.

constructor(private router: Router, private sharedData: SharedService) { }

Declare a variable before the constructor to hold the received data.

data: any;

Subscribe the observable data from the service inside the NgOnInit function.

ngOnInit() {
  this.sharedData.currentData.subscribe(data => this.data = data);
}

Add a function to change the shared data then emit to its subscriber then navigate to the other component to see the data changes.

changeData() {
  this.sharedData.changeData({name: 'Eric Cantona'});
  this.router.navigate(['/bcomp']);
}

Next, open the `src/app/acomp/acomp.component.html` then replace all HTML tags with these tags.

<p>acomp works!</p>
<p>{{data.name}}</p>
<p>
  <button (click)="changeData()">Send Data</button>
</p>

Do a similar way for the other component, open and edit `src/app/bcomp/bcomp.component.ts` then replace all codes with these.

import { Component, OnInit, Output, EventEmitter } from '@angular/core';
import { Router } from '@angular/router';
import { SharedService } from '../shared.service';

@Component({
  selector: 'app-bcomp',
  templateUrl: './bcomp.component.html',
  styleUrls: ['./bcomp.component.scss']
})
export class BcompComponent implements OnInit {

  data: any;

  constructor(private router: Router, private sharedData: SharedService) { }

  ngOnInit() {
    this.sharedData.currentData.subscribe(data => this.data = data);
  }

  changeData() {
    this.sharedData.changeData({name: 'Romario Varia'});
    this.router.navigate(['/acomp']);
  }

}

Next, open and edit `src/app/bcomp/bcomp.component.html` then replace all HTML tags with these.

<p>acomp works!</p>
<p>{{data.name}}</p>
<p>
  <button (click)="changeData()">Send Data</button>
</p>


Observable and RxJS in Router

In this example, we will show you how to convert Router events as Observable. RxJS filter() use to look for events of interest and subscribe to them in order to make decisions based on the sequence of events in the navigation process. Still using the previously created components, open and edit `src/app/acomp/acomp.component.ts` then add or modify these imports of @angular/router Router, NavigationStart, RxJS filter, and Observable.

import { Router, NavigationStart } from '@angular/router';
import { filter } from 'rxjs/operators';
import { Observable } from 'rxjs';

Inject the constructor with Router and add to the constructor body the Router events that converted as Observable.

constructor(private router: Router) {
  this.navStart = router.events.pipe(
    filter(evt => evt instanceof NavigationStart)
  ) as Observable<NavigationStart>;
}

Add NavigationStart subscriber to NgOnInit function.

ngOnInit() {
  this.navStart.subscribe(evt => console.log('Navigation Started!'));
}

Next, the example of ActivateRoute that injected router service that makes use of observables to get information about a route path and parameters. Open and edit `src/app/bcomp/bcomp.component.ts` then add/modify these imports of @angular/router ActivatedRoute.

import { ActivatedRoute } from '@angular/router';

Inject above ActivatedRoute to the constructor.

constructor(private activatedRoute: ActivatedRoute) { }

Add ActivatedRoute URL subscribing to the NgOnInit function.

ngOnInit() {
  this.activatedRoute.url
    .subscribe(url => console.log('The URL changed to: ' + url));
}


Observable and RxJS in Reactive Forms

This example will show you how Observable detect the FormControl value changes. The Observable is used by the property in ReactiveForms. For that, we will register the ReactiveForm by open and edit `src/app/app.module.ts` then add/modify this import of ReactiveFormsModule, and FormModule.

import { FormsModule, ReactiveFormsModule } from '@angular/forms';

Register the above modules to `@NgModule` imports.

imports: [
  BrowserModule,
  HttpClientModule,
  FormsModule,
  ReactiveFormsModule,
  AppRoutingModule
],

Next, open and edit `src/app/app.component.ts` then add this import of @angular/forms FormGroup and FormControl.

import { FormGroup, FormControl } from '@angular/forms';

Add these variables of FormGroup and log for every FormControl value change.

inputChangeLog: string[] = [];
inputForm: FormGroup;

Add a function that detects the value changes of FormControl then saves to log array variable.

logInputChange() {
  const nameControl = this.inputForm.get('name');
  nameControl.valueChanges.forEach(
    (value: string) => {
      this.inputChangeLog.push(value);
      console.log(this.inputChangeLog);
    }
  );
}

Initialize the FormGroup and call log input changes function inside the NgOnInit function (don't forget to add the OnInit module to this component).

ngOnInit() {
  this.inputForm = new FormGroup({
    name: new FormControl()
  });
  this.logInputChange();
}

Next, open and edit `src/app/app.component.html` then add the FormGroup and InputControl to the main <div>.

<div class="content" role="main">

  <form [formGroup]="inputForm">
    <input type="text" placeholder="Your Name" formControlName="name" />
  </form>

</div>

Now, every FormControl changes will be displayed in the browser console.

That it's for now, it just a few examples of Angular Observable and RxJS. You can get the full source codes from our GitHub.

If you don’t want to waste your time design your own front-end or your budget to spend by hiring a web designer then Angular Templates is the best place to go. So, speed up your front-end web development with premium Angular templates. Choose your template for your front-end project here.

Friday, May 8, 2020

Hot vs Cold Observables


Understanding the nature of hot and cold Observables is a crucial part to master Observables. Before we try to explore the topic through some practical examples, let’s read through the definition from the RxJS project itself:

Cold observables start running upon subscription, i.e., the observable sequence only starts pushing values to the observers when Subscribe is called. (…) This is different from hot observables such as mouse move events or stock tickers which are already producing values even before a subscription is active.

Ok, so far so good. Let’s see what that means in practice.

We start with a basic Observable that simply emits the number 1. We make two independent subscriptions to the same Observable and print out the output with a prefix so that we can tell them apart.

let obs = Rx.Observable.create(observer => observer.next(1));

obs.subscribe(v => console.log("1st subscriber: " + v));
obs.subscribe(v => console.log("2nd subscriber: " + v));

When we run this code we’ll see the following output in the console.

1st subscriber: 1
2nd subscriber: 1

Ok, cool. But the interesting question remains: is obs cold or hot? Let’s forget for a moment that we know how obs was created and imagine we would have obtained a reference to obs by calling getObservableFromSomewhere() instead. If that was the case, we wouldn’t be able to tell whether it’s cold or hot. And that’s one important thing to understand. It’s not always possible from the subscriber side to know whether you are dealing with a cold or hot Observable.

If we turn back to the definition that we cited in the beginning and think about what makes an Observable cold or hot, we can read between the lines and notice that if obs was cold it should produce fresh values upon subscription. But the pity is, with a value such as 1 we can’t easily tell whether it was created freshly upon subscription or not. So let’s replace 1 with Date.now() and see what happens.

If we run that code again we’ll notice that we actually get different output per subscription. Notice the change in the last digit.

1st subscriber: 1465990942935
2nd subscriber: 1465990942936

With this output it is clear that there must have been two calls to observer.next(Date.now()). In other words, the Observable started producing the values upon each subscription which makes it cold by definition.

Making Cold Observables Hot

Now that we know that our Observable is clearly cold, let’s try to warm it up a little.

let obs = Rx.Observable
            .create(observer => observer.next(Date.now()))
            .publish();

obs.subscribe(v => console.log("1st subscriber: " + v));
obs.subscribe(v => console.log("2nd subscriber: " + v));

obs.connect();

Let’s ignore the publish and connect operators for a moment and solely focus on the output.

1st subscriber: 1465994477014
2nd subscriber: 1465994477014

Clearly we can see that there must have been only one call to observer.next(Date.now()) with this setup as the numbers are identical. So, is it hot now? Well, kind of. Let’s put it that way: it’s warmer than a cold one but colder than a really hot one. It’s hot in the sense that there’s no new value producer/source (the thing calling observer.next(val)) created upon subscription. But it’s cold in the sense that it doesn’t start producing values before the subscriptions exist.

We can fix that by moving the connect call before the subscriptions.

let obs = Rx.Observable
            .create(observer => observer.next(Date.now()))
            .publish();
obs.connect();

obs.subscribe(v => console.log("1st subscriber: " + v));
obs.subscribe(v => console.log("2nd subscriber: " + v));

However, when we run this code we don’t see any output at all. And that’s expected behaviour because now obs is really really hot as in it produces values no matter if anyone listens or not. So by the time that our two subscribers start listening the value has long been pumped through.

In order to be able to understand the purpose of publish better, let’s use an Observable that will produce infinite values instead of just a single one.

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

obs.subscribe(v => console.log("1st subscriber:" + v));
setTimeout(()
  // delay for a little more than a second and then add second subscriber
  => obs.subscribe(v => console.log("2nd subscriber:" + v)), 1100);

The setup we use is slightly more complex from what we had before, so let’s break it down.

  1. We use interval(1000) to create an Observable that emits every second with an increasing index value starting at 0.
  2. We use publish to share the value producer across several subscriptions (one indicator of being hot!)
  3. We defer the second subscription by one second.

Let’s just ignore refCount for now as we’re going to explain it later.

We see the following output as we run the script.

1st subscriber:0
1st subscriber:1
2nd subscriber:1
1st subscriber:2
2nd subscriber:2
...

Clearly, we can see that the second subscriber is not starting over at 0. But that would have been the case if we didn’t use publish to share the value producing source across subscribers.

Understanding publishrefCount and connect

But how hot is obs in this scenario? Let’s make it visible by defering both subscriptions by another 2 seconds. If it’s really hot we shouldn’t see the numbers 0 and 1 at all because they would have been emitted before we start to listen, right?

Let’s try out and run this code instead.

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

setTimeout(() => {
  // delay both subscriptions by 2 seconds
  obs.subscribe(v => console.log("1st subscriber:" + v));
  setTimeout(
    // delay for a little more than a second and then add second subscriber
    () => obs.subscribe(
          v => console.log("2nd subscriber:" + v)), 1100);

},2000);

Interestingly we see exactly the same output as in the previous experiment which means we are dealing with an Observable that is rather warm than really hot. And that’s because of the way refCount works. The publish operator creates an ConnectableObservable which means it creates an Observable that shares one single subscription to the underlying source. However, the publish operator doesn’t subscribe to the underlying source just yet. It’s more like a gatekeeper that makes sure that subscriptions aren’t made to the underlying source but to the ConnectableObservable instead.

It’s the job of the connect operator to actually cause the ConnectableObservable to subscribe to the underlying source (the thing that produces values). In our case we’re using refCount which is an operator that builds up on connect and causes the ConnectableObservable to subscribe to the underlying source as soon as there is a first subscriber and to unsubscribe from it as soon as there’s no subscriber anymore. It simply keeps track of how many subscriptions are made to the ConnectableObservable.

And this explains why we see values starting at 0 with our last experiment. It’s because the subscription to the underlying source is only made once that we have a first subscriber to the ConnectableObservable.

If we want obs to be truly hot, we have to call connect ourselves early on.

let obs = Rx.Observable
            .interval(1000)
            .publish();
obs.connect();

setTimeout(() => {
  obs.subscribe(v => console.log("1st subscriber:" + v));
  setTimeout(
    () => obs.subscribe(v => console.log("2nd subscriber:" + v)), 1000);

},2000);

Notice how we get values starting from 2 when we run this code.

1st subscriber:2
1st subscriber:3
2nd subscriber:3

This means we now have a truly hot Observable that produces values no matter if someone listens or not.

When to use what

As we’ve seen the question whether an Observable is hot or cold is everything but black and white. In fact there are several strategies how values may be pushed to subscribers that we didn’t even touch on yet. In general we can say that we should be dealing with a hot Observable whenever we subscribe to something that is generating values no matter if someone is listening or not. When we subscribe to such a hot Observable, we don’t see past values but only new ones that were generated after our subscription.

A typical example of a hot observable are mousemove events. The mouse moves happen regardless if someone is listening or not. When we start listening for them, we only get future events.

Cold Observables on the other hand are the lazy ones. They only start producing values when someone subscribes. But then again, it’s really not a black and white thing. An iced cold Observable starts reproducing the values it emits independently with every new subscriber as we’ve seen in the examples above. But how should we call an Observable that only starts generating values as the first subscriber subscribes and then shares and reemits the exact same values to every new subscriber? Things get blurry and categorizing only by cold and hot doesn’t really cut it for every possible use case.

As a rule of thumb, when you have a cold Observable and you want multiple subscribers to it, and you don’t want them to cause regenerating the values but rather reusing existing values, you need to start thinking about publish and friends.

Caveat: Http with Observables

The Http service in Angular >= 2.x returns cold Observables and the implications may surprise us.

Consider this simple component that requests a contacts.json file from a server and renders the contacts as a list in the template.

...
@Component({
  ...
  template: `
    <ul>
      <li *ngFor="let contact of contacts | async">{{contact.name}}</li>
    </ul>
    `
  ...
})
export class AppComponent {
  contacts: Observable<Array<any>>;
  constructor (http: Http) {
    this.contacts = http.get('contacts.json')
                        .map(response => response.json().items);
  }
}

As we can see above, we’re exposing an Observable<Array<any>> to the template and subscribe to it from the template using the AsyncPipe.

The contacts.json contains a simple object with an items property holding the collection of contacts.

{
  "items": [
    { "name": "John Conner" },
    { "name": "Arnold Schwarzenegger" }
  ]
}

Now the interesting question is: What happens if we’d subscribe to it twice by adding another list to our template?

1st List
<ul>
  <li *ngFor="let contact of contacts | async">{{contact.name}}</li>
</ul>
2st List
<ul>
  <li *ngFor="let contact of contacts | async">{{contact.name}}</li>
</ul>

If we run that code and take a look at the network tab of our browser, we’ll notice that the browser fetches contacts.json twice! And that’s in stark contrast to what we’re used to when working with Promises for example. In fact, simply importing the toPromise operator and adding it as the last operator to our Observable causes the second request to vanish.

But let’s not fall back to Promises just yet. With all our knowledge regarding cold and hot Observables we should be able to simply fix our Observable so that it shares one single subscription to the underlying source - the Observable that issues the HTTP call.

this.contacts = http.get('contacts.json')
                    .map(response => response.json().items)
                    .publish()
                    .refCount();

That should do it, right? Well, almost. We can see that there’s no second request anymore. But there’s still a big issue that makes the code behave quite differently from what we are used with Promises.

Consider we’d want the second list to show up after a delay of 500ms. We could change the code to this.

@Component({
  ...
  template: `
    1st List
    <ul>
      <li *ngFor="let contact of contacts | async">{{contact.name}}</li>
    </ul>
    2st List
    <ul>
      <li *ngFor="let contact of contacts2 | async">{{contact.name}}</li>
    </ul>
    `,
  ...
})
export class AppComponent {
  contacts: Observable<Array<any>>;
  contacts2: Observable<Array<any>>;
  constructor (http: Http) {
    this.contacts = http.get('contacts.json')
                        .map(response => response.json().items)
                        .publish()
                        .refCount();

    setTimeout(() => this.contacts2 = this.contacts, 500);
  }
}

Notice that we are still using our one and only Observable but we are reexposing it as contacts2 after 500ms. But when we run the code, we notice that our second list isn’t showing up!

If you think about it, it makes perfect sense. With the use of publish we caused the Observable to share a single subscription to the underlying source - we made it hot. But maybe we made it a little too hot. Now when we have a second subscriber starting to listen after 500ms it will only get notified about new values that arrive after its subscription. What we rather want is that new subscribers see exactly the old values that were already emitted earlier. Fortunately we can have exactly that behaviour by using publishLast instead of publish.

this.contacts = http.get('contacts.json')
                    .map(response => response.json().items)
                    .publishLast()
                    .refCount();

Now when we run that code we see our second list after 500ms but we still don’t see a second request. In other words, we created an Observable that emits upon subscription with the data that was fetched with the first request.

Useful shortcut

Since using publish and refCount together is such a useful pattern there is an operator that combines them, and it’s called .share()

this.contacts = http.get('contacts.json')
  .map(response => response.json().items)
  .share();

Note: When using multiple async pipes on streams with default values, the .share() operator might cause problems:

The share() will publish the first value of the stream on the first subscription. The first async pipe will trigger that subscription and get that initial value. The second async pipe however will subscribe after that value has already been emitted and therefore miss that value.

The solution for this problem is the .shareReplay(1) operator, which will keep track of the previous value of the stream. That way all the async pipes will get the last value. Just like the share() operator is a shortcut for publish().refCount(), the shareReplay(1) operator is a shortcut for publishReplay(1).refCount(). We can see the difference between share() and shareReplay(1) in the following plunk. We should always pass 1 as the parameter value to the shareReplay function. Otherwise RxJS will keep track of all the values of that observable, when we only need the last one.

It’s important to note that this problem will only occur when using streams with initial values. The share operator would not pose problems with regular http requests for instance.

Whew! We came a long way. We hope this gives you a clearer picture of what the term hot vs cold actually means when it comes to Observables.