# Series: RxJS - Getting Started (15 Parts) — Luca Del Puppo

> Complete 15-part technical series "RxJS - Getting Started" by Luca Del Puppo (https://delpuppo.net/blog/series/rxjs-getting-started/). Ordered chronologically from Part 1 to Part 15.

- **Canonical Series URL**: https://delpuppo.net/blog/series/rxjs-getting-started/
- **Hashnode Series Mirror**: https://blog.delpuppo.net/series/rxjs-getting-started
- **Blog Index LLMs.txt**: https://delpuppo.net/blog/llms.txt

## Series Curriculum

1. [Part 1: RxJS - Observable](https://delpuppo.net/blog/rxjs-getting-started/rxjs-observable/) (Sep 6, 2021) — Hi Guys, Today I want to start a little series about Reactive Programming with RxJS. I start from the... ([Article LLMs.txt](https://delpuppo.net/blog/rxjs-getting-started/rxjs-observable/llms.txt))
2. [Part 2: RxJS - Observer and Subscription](https://delpuppo.net/blog/rxjs-getting-started/rxjs-observer-and-subscription/) (Sep 8, 2021) — Welcome back guys, today the topics of the article are Observers and Subscriptions. ... ([Article LLMs.txt](https://delpuppo.net/blog/rxjs-getting-started/rxjs-observer-and-subscription/llms.txt))
3. [Part 3: RxJS - Operators](https://delpuppo.net/blog/rxjs-getting-started/rxjs-operators/) (Sep 10, 2021) — Hi Guys and welcome back, today I introduce the concepts of Operators. Operators are functions. Isn't... ([Article LLMs.txt](https://delpuppo.net/blog/rxjs-getting-started/rxjs-operators/llms.txt))
4. [Part 4: RxJS - Marble Diagrams](https://delpuppo.net/blog/rxjs-getting-started/rxjs-marble-diagrams/) (Sep 13, 2021) — Hi guy, before continuing with the operators, today I want to explain the Marble Diagrams. The Marble... ([Article LLMs.txt](https://delpuppo.net/blog/rxjs-getting-started/rxjs-marble-diagrams/llms.txt))
5. [Part 5: RxJS - Creation Operators](https://delpuppo.net/blog/rxjs-getting-started/rxjs-creation-operators/) (Sep 15, 2021) — Welcome back guys, today I'll start to show you the Creation operators in RxJS. The Creation... ([Article LLMs.txt](https://delpuppo.net/blog/rxjs-getting-started/rxjs-creation-operators/llms.txt))
6. [Part 6: RxJS - Filtering Operators](https://delpuppo.net/blog/rxjs-getting-started/rxjs-filtering-operators/) (Sep 17, 2021) — Welcome back Guys, Today we'll start with the Filtering Operators. As you can imagine these... ([Article LLMs.txt](https://delpuppo.net/blog/rxjs-getting-started/rxjs-filtering-operators/llms.txt))
7. [Part 7: RxJS - Transformation Operators](https://delpuppo.net/blog/rxjs-getting-started/rxjs-transformation-operators/) (Sep 20, 2021) — Welcome back, Today I'll speak about the Transformation Operators. These operators are used to modify... ([Article LLMs.txt](https://delpuppo.net/blog/rxjs-getting-started/rxjs-transformation-operators/llms.txt))
8. [Part 8: RxJS - Join Operators](https://delpuppo.net/blog/rxjs-getting-started/rxjs-join-operators/) (Sep 22, 2021) — Hi Guys and Welcome Back, Today I'll speak about the Join Operators. Some of these operators are... ([Article LLMs.txt](https://delpuppo.net/blog/rxjs-getting-started/rxjs-join-operators/llms.txt))
9. [Part 9: RxJS - Utility Operators](https://delpuppo.net/blog/rxjs-getting-started/rxjs-utility-operators/) (Sep 24, 2021) — Welcome back Guys, Today I'll show you some utility operators. Let's start :) tap Used to... ([Article LLMs.txt](https://delpuppo.net/blog/rxjs-getting-started/rxjs-utility-operators/llms.txt))
10. [Part 10: RxJS - Conditional & Mathematical Operators](https://delpuppo.net/blog/rxjs-getting-started/rxjs-conditional-mathematical-operators/) (Sep 27, 2021) — Hi Guys and welcome back, today I'll illustrate you two simple types of the pipeable operators:... ([Article LLMs.txt](https://delpuppo.net/blog/rxjs-getting-started/rxjs-conditional-mathematical-operators/llms.txt))
11. [Part 11: RxJS - Error Handling Operators](https://delpuppo.net/blog/rxjs-getting-started/rxjs-error-handling-operators/) (Sep 29, 2021) — One of the best practice in our work is handle the errors, so today I'll show you some operators in... ([Article LLMs.txt](https://delpuppo.net/blog/rxjs-getting-started/rxjs-error-handling-operators/llms.txt))
12. [Part 12: RxJS - Multicast Operator](https://delpuppo.net/blog/rxjs-getting-started/rxjs-multicast-operator/) (Oct 1, 2021) — Hi Folk 👋, in the previous articles we've seen that when we subscribe to an observable, the... ([Article LLMs.txt](https://delpuppo.net/blog/rxjs-getting-started/rxjs-multicast-operator/llms.txt))
13. [Part 13: RxJS - Custom Operators](https://delpuppo.net/blog/rxjs-getting-started/rxjs-custom-operators/) (Oct 4, 2021) — Hi Folks 👋 after many articles in which I have illustrated the different types of operators already... ([Article LLMs.txt](https://delpuppo.net/blog/rxjs-getting-started/rxjs-custom-operators/llms.txt))
14. [Part 14: RxJS - Subjects](https://delpuppo.net/blog/rxjs-getting-started/rxjs-subjects/) (Oct 6, 2021) — What's a Subject Luca? And why today do you want to speak about subjects? Is there a mistake in the... ([Article LLMs.txt](https://delpuppo.net/blog/rxjs-getting-started/rxjs-subjects/llms.txt))
15. [Part 15: RxJS - Marble Testing](https://delpuppo.net/blog/rxjs-getting-started/rxjs-marble-testing/) (Oct 8, 2021) — Hello Folks 👋! Today it's time to speak about testing and how to test the code when we use... ([Article LLMs.txt](https://delpuppo.net/blog/rxjs-getting-started/rxjs-marble-testing/llms.txt))

---

## Full Series Content

### Part 1 of 15: RxJS - Observable

- **URL**: https://delpuppo.net/blog/rxjs-getting-started/rxjs-observable/
- **Published**: Sep 6, 2021
- **Tags**: javascript, typescript, rxjs

Hi Guys,
Today I want to start a little series about Reactive Programming with [RxJS](https://rxjs.dev/).
I start from the basic core concepts behind of this Programming Paradigm to move after and show the potentialities of the RxJS library.

The idea of this series was born some weeks ago during a refactor in an Angular application, because I noticed a lot of errors in the codebase about RxJS. The problem behind this code unfortunately is that the developers begin using the angular framework without the basics of the RxJS library and the Reactive Programming. So I decided to write some articles to show the basics behind the RxJS library and Reactive Programming in general.

A little preamble before passing on the topic of this article: RxJS is a library for composing asynchronous and event-based programs by using observable sequences. So RxJS and Reactive Programming are not the solution for all your problems, but they are good solutions in contexts where asynchronous and events are the stars. I've preferred to clarify that because sometimes programmers believe that but after having introduced the Reactive Programming, they come across with other problems because the library and the paradigm are very simple but at the same time it's easy to get hurt if you don't know how it works.

The last info about the series: the code is written in Typescript, but you can use RxJS also in vanilla javascript if you prefer.

Now let's start with the first argument, the star behind the Reactive Programming, the *Observable*.

Observables (the one who is observed) are one of the key concepts behind Reactive Programming along with Observers and Subscribers (those who observes, controls).
Observables can be a stream or a collection of datas, but you could imagine an observable like a lazy Push collection of multiple values.
To understand better the concept let me show you an example

```ts
import { Observable, Subscriber } from "rxjs";

const observable = new Observable<string>((subscriber: Subscriber<string>) => {
  subscriber.next("Hello World");
  subscriber.error(new Error("Something went wrong!"));
  subscriber.complete();
});
```

As you can see the Observable is a class that accept a subscriber (a callback function).
This subscriber has 3 main possible action:
1. **next**: The next method emits the value passing as parameter to all the subscriptions, so the system can react accordingly.
2. **error**: The error method emits an error during the execution of the observable.
3. **complete**: The complete method sets the observable closed. When this happens, all the future methods (next and error) emitted for the closed observable will be ignored.

To better understand here's another example, with its result in the console

```ts
import { Observable, Observer, Subscriber } from "rxjs";

const observer: Observer<string> = {
  next: (value: string) => console.log("next", value),
  error: (error: Error) => console.error("error", error),
  complete: () => console.log("complete!"),
};

const observable = new Observable<string>((subscriber: Subscriber<string>) => {
  subscriber.next("Hello");
  subscriber.next("World");

  subscriber.complete();

  // this will never be logged

  subscriber.error(new Error("Something went wrong!"));

  subscriber.next("Hello");
  subscriber.next("World");
});

observable.subscribe(observer);
```
```sh
next Hello
next World
complete!
```
_p.s. don't pay attention to the observer in this moment, to simplify it think that when the subscriber calls the next method, the next function in the observer will be called and the same goes for the error and complete methods_

You can notice how the subscriber calls the next method twice: first with "Hello" and after with "World" and the result is logged into the console. Next, the subscriber calls the complete method and it is registered in the console too. After that, the subscriber calls the error method and the next method twice, but in the console nothing happens. This behaviour is due to the fact that the observable is ended by the complete method so the observable no longer emits any events.
When we complete the observable, it's important to remember that all the next methods called (next, error or complete) are ignored.

That's all for now.
You can find the example at this [in the GitHub repository](https://github.com/Puppo/rxjs-getting-started/tree/01-observable)

See you soon guys!

---

### Part 2 of 15: RxJS - Observer and Subscription

- **URL**: https://delpuppo.net/blog/rxjs-getting-started/rxjs-observer-and-subscription/
- **Published**: Sep 8, 2021
- **Tags**: javascript, typescript, rxjs

Welcome back guys,
today the topics of the article are **Observers** and **Subscriptions**.

## Observer
An observer is a consumer of values. An observer is something that is interested to the emitted values by the observable. In RxJS an Observer is simply a set of callbacks (next, error, complete). 
Here's a simple example of an observer
```ts
import { Observer } from "rxjs";

const observer: Observer<string> = {
  next: (value: string) =>
    console.log(`[observer] next`, value),
  error: (error: Error) =>
    console.error(`[observer] error`, error),
  complete: () =>
    console.log(`[observer] complete!`),
};
```
As you can see an observer has three callbacks, one for each type of notification that an Observable may emit. Every callback can react in the base of the observable event notification.
An observer could have not all the three callbacks defined because maybe it doesn't want to react to all the notifications. In these cases, you can create an observer with the only callbacks you need.

```ts
import {
  NextObserver,
  ErrorObserver,
  CompletionObserver,
} from "rxjs";

const nextObserver: NextObserver<string> = {
  next: (value: string) =>
    console.log(`[nextObserver] next`, value),
};

const errorObserver: ErrorObserver<string> = {
  error: (error: Error) =>
    console.error(`[errorObserver] error`, error),
};

const completeObserver: CompletionObserver<string> = {
  complete: () =>
    console.log(`[completeObserver] complete!`),
};
```
It's possible to define a NextObserver without the next property but indicating only the body of the next method, in this case, the observer by default is of the type NextObserver.
```
const defaultNextObserver: (value: string) => void = (value: string) =>
    console.log(`${new Date().toISOString()} - [defaultNextObserver] next`, value)
```

## Subscription
A Subscription is an actor that decides when an Observable must be listened and when we can stop listening to it.
Until an observable is not subscribed in your code nothing happens, but as soon as you create a subscription, the magic starts.
A Subscription in RxJS is an object created using the method "subscribe" and it has one main method: "unsubscribe"; this method allows you to stop listening the observable event.
_In your code is **important** to call the "unsubscribe" when you no longer need the subscription, this prevent problems as the memory leaks._
An example of a Subscription

```ts
import { Subscription } from "rxjs";

const observer = (value: string) => console.log(`[unsubscribe method] next`, value)

const subscription: Subscription = observable.subscribe(observer);
subscription.unsubscribe();
```
Another pretty feature in RxJS Subscription is the "add" method; this method allows you to collect more subscriptions inside of one Subscription instance and after that, you can unsubscribe all the subscriptions at once time.
```ts
import { Subscription } from "rxjs";

const subscription: Subscription = observable.subscribe((value: string) =>
  console.log(
    `[unsubscribe 1 method] next`,
    value
  )
);
subscription.add(
  observable.subscribe((value: string) =>
    console.log(
      `[unsubscribe 2 method] next`,
      value
    )
  )
);

subscription.unsubscribe();
```
In some cases when you call the unsubscribe method you need to run some special code in your observable: RxJS let us do this using a special syntax inside of the observable declaration. When you create the observable you can return a function that the library invokes in the future during the unsubscription. Below a simple example to understand better the problem and the solution:

```ts
import {
  Observable,
  Subscriber,
  Subscription,
} from "rxjs";

const observableWithCallback = new Observable<string>(
  (subscriber: Subscriber<string>) => {
    let count = 0;
    const id = setInterval(() => {
      subscriber.next(`Count: ${++count}`);
    }, 300);

    return () => {
      console.log("On UnSubscription");
      clearInterval(id);
    };
  }
);

const subscriptionObservableWithCallback: Subscription = observableWithCallback.subscribe({
  next: (value: string) =>
    console.log(`[observableWithCallback] Next: ${value}`),
});
setTimeout(() => {
  subscriptionObservableWithCallback.unsubscribe();
}, 3000);

```

```console
[observableWithCallback] Next: Count: 1
[observableWithCallback] Next: Count: 2
[observableWithCallback] Next: Count: 3
[observableWithCallback] Next: Count: 4
[observableWithCallback] Next: Count: 5
[observableWithCallback] Next: Count: 6
[observableWithCallback] Next: Count: 7
[observableWithCallback] Next: Count: 8
[observableWithCallback] Next: Count: 9
On UnSubscription
```
Ok, guys, that's all for today.
In the next article we'll see the last fundamental of RxJS: the *Operator* and later, we can move to see the features of this Library.

You can find the examples of this article [in the GitHub repository](https://github.com/Puppo/rxjs-getting-started/tree/02-observer-and-subscription).

See you soon guys!

---

### Part 3 of 15: RxJS - Operators

- **URL**: https://delpuppo.net/blog/rxjs-getting-started/rxjs-operators/
- **Published**: Sep 10, 2021
- **Tags**: javascript, typescript, rxjs

Hi Guys and welcome back, today I introduce the concepts of **Operators**.
Operators are functions. Isn't it easy? Operators are functions of two types in RxJS: _Creation_ or _Pipeable_.

## Creation
Creation operators are simple functions and their scope is to create new observables.

```ts
import { of } from "rxjs";

of(1, 2, 3, 4).subscribe(x => console.log("[of] result", x));
```

```console
[of] result 1
[of] result 2
[of] result 3
[of] result 4
```
The most commons creation operators are: ajax, empty, from, fromEvent, interval, of, throwError, timer, combineLatest, concat, forkJoin, merge and zip.

In the next article I'll deepen these types of operators, today I just make an overview of the operator's types.

## Pipeable
Pipeable operators are functions that take an Observable as input and return another observable. These functions are pure, so the input observable does not change but the function returns a new one. The main scopes of these operators are: transform, filter, and work with the input observable.
An example of pipeable operators:
```ts
import { Observable } from "rxjs";
import { map, tap } from "rxjs/operators";

new Observable<number>(observer => {
  observer.next(1);
  observer.next(2);
  observer.next(3);
  observer.next(4);

  observer.complete();
})
  .pipe(
    map(val => val * 2),
    tap(res => {
      console.log("[pipeable tap]", res);
    })
  )
  .subscribe();
```
```
[pipeable tap] 2
[pipeable tap] 4
[pipeable tap] 6
[pipeable tap] 8
```
The pipeable operators are divided by scopes and the scopes are: Transformation, Filtering, Join, Multicasting, Error Handling, Utility, Conditional and Boolean and Mathematical and Aggregate.

As you can understand you can combine Creation operators with Pipeable operators to manage your business logic like here

```ts
import { timer } from "rxjs";
import { take } from "rxjs/operators";

timer(0, 1000)
  .pipe(take(10))
  .subscribe(x => console.log("[timer] result", x));
```
```
[timer] result 0
[timer] result 1
[timer] result 2
[timer] result 3
[timer] result 4
[timer] result 5
[timer] result 6
[timer] result 7
[timer] result 8
[timer] result 9
```

With this last article, I have introduced all the pillars to the base of the RxJS library and the Reactive Programming. In the next article, we'll start to explore the operator's world.

That's all guys,
See you soon!

_[in the GitHub repository](https://github.com/Puppo/rxjs-getting-started/tree/03-operators) you can find the code of this article._

---

### Part 4 of 15: RxJS - Marble Diagrams

- **URL**: https://delpuppo.net/blog/rxjs-getting-started/rxjs-marble-diagrams/
- **Published**: Sep 13, 2021
- **Tags**: javascript, typescript, rxjs

Hi guy,
before continuing with the operators, today I want to explain the _Marble Diagrams_.
The Marble Diagrams is a timeline where you can illustrate the state of your observable during its execution.
The actors in this diagram are timeline and values(circle).
The timeline is used to represent the time during the execution of the observable though the circles indicate the values emitted.
But let me show you an example:
![map marble diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/kcktws04ceojvqfmwp24.jpg)

This example is based on this code
```ts
import { Observable } from "rxjs";
import { map } from "rxjs/operators";

const source$ = new Observable<number>(observer => {
  let count = 0;
  const id = setInterval(() => {
    if (count++ < 3) {
      observer.next(count);
    } else {
      clearInterval(id);
      observer.complete();
    }
  }, 1000);
});

source$.pipe(map(value => value * 2)).subscribe({
  next: console.log,
});
```
As you can notice, in the diagram there are two timelines, one for the source and the other for the map operator.
In the first timeline you can see when the source emits the value, in the second timeline you can see the result of the transformation after the execution of the map operator.

To build a Marble diagram you need to keep in mind some easy rules: there is always a timeline that represents the source observable, there are N timelines as many operators as you need to display, every timeline illustrates the state of the values after the execution of the operator indicated in the timeline and finally, you need to use a circle to represent the values.

This tool is very convenient to illustrate the transformation of the observable during its execution and it helps us to have an image of the state of the observable execution.

In addition to the Marble Diagram you can use the Marble Testing to test the execution of you Observable.
The Marble testing uses a special format to represent the timeline and the value during the execution, but I will speak about it in the future.

To reinforce the Marble Diagram concept let me show you another example

```ts
import { Observable } from "rxjs";
import { delay, map } from "rxjs/operators";

const source$ = new Observable<number>(observer => {
  let count = 0;
  const id = setInterval(() => {
    if (count++ < 3) {
      observer.next(count);
    } else {
      clearInterval(id);
      observer.complete();
    }
  }, 1000);
});

source$
  .pipe(
    map(value => value * 2),
    delay(1500)
  )
  .subscribe({
    next: console.log,
  });
```
![Marble Diagram Map+Delay](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/6t41ih83gfqy7w8x3fmt.jpg)

In this example you can see how the observable in the first operator doubles the value and then it waits 1.5 seconds before emitting the result.
To represent this case the marble diagram has 3 timelines, one with the source, one with the map operator and one with the delay operator. Every timeline indicates the value during the execution of its operator so you can see the behaviour of this implementation.

It's all from the marble diagram.

See you soon guys!

---

### Part 5 of 15: RxJS - Creation Operators

- **URL**: https://delpuppo.net/blog/rxjs-getting-started/rxjs-creation-operators/
- **Published**: Sep 15, 2021
- **Tags**: javascript, typescript, rxjs

Welcome back guys,
today I'll start to show you the Creation operators in RxJS.

The Creation Operators are used to create new observables. They are divided into Creation Operators and Join Creation Operators.
The main difference between them consists in the fact that the Join Creation Operators create observable from other observables instead the Creation Operators create observables from objects which differ from observable.

## [Creation Operators](https://rxjs.dev/guide/operators#creation-operators-2)

* **[of](https://rxjs.dev/api/index/function/of)**

> The Of operator is used to convert an argument to an observable.
```ts
import { of } from "rxjs";

of([1, 2, 3], 4, 5, 6).subscribe({
  next: res => console.log("[of]", res),
  complete: () => console.log("[of] complete"),
});
```
```console
[of] [ 1, 2, 3 ]
[of] 4
[of] 5
[of] 6
[of] complete
```
![of Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/f7yid35mp6bd5zfigy1t.jpg)
As you can see, this operator takes a list of arguments and converts them to values of the observable.

* **[from](https://rxjs.dev/api/index/function/from)**

> This operator creates an Observable from an Array, an array-like object, a Promise, an iterable object, or an Observable-like object.

__from array__
```ts
from([1, 2, 3]).subscribe(res => console.log("[from]", res));
```
```console
[from] 1
[from] 2
[from] 3
```
![from Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/9rg9xh5qute698rgk4de.jpg)
In this case we converted an array to an observable. 
The from operator converts all the items in the array to values of the observable. It's common to use this operator when we need to convert the items of an array to the values to react our system.

__from promise__
```ts
from(
  new Promise(res => setTimeout(() => res("Hello from promise"), 3000))
).subscribe(res => console.log(new Date().toLocaleTimeString(), `[from]`, res));
```
```console
13:47:52 [from] Hello from promise
```
In this case we converted a promise to an observable. It is common to convert a promise to an observable, or because we have multiple subscribers which need to react to this promise or because we need to use these promise inside of a pipe chain.

__from iterator__

```ts
const wait = async (time: number) => new Promise(res => setTimeout(res, time));
async function* hello() {
  yield "Hello";
  await wait(1000);
  yield "from";
  await wait(1000);
  yield "iterator";
}
from(hello()).subscribe({
  next: res => console.log(new Date().toLocaleTimeString(), `[from]`, res),
  complete: () =>
    console.log(new Date().toLocaleTimeString(), `[from] complete`),
});
```
```console
13:48:42 [from] Hello
13:48:43 [from] from
13:48:44 [from] iterator
13:48:44 [from] complete
```
Here instead, the from operator converts an iterator to an observable. This is common when we need to react to every item of the iterator or when we need to use the generator inside of a pipe chain. 

* **[fromEvent](https://rxjs.dev/api/index/function/fromEvent)**

> The fromEvent operator is usage to emit an observable in base of an event

Some examples: when a user clicks to a document element in HTML/JS or when a system emits an [event](https://nodejs.org/api/events.html) in node.js eco system.
Here a practise example
```ts
import * as events from "events";
import { fromEvent } from "rxjs";

console.log(new Date().toLocaleTimeString(), "[fromEvent] start");

const em = new events.EventEmitter();
fromEvent(em, "custom-event").subscribe({
  next: res => console.log(new Date().toLocaleTimeString(), "[fromEvent]", res),
  complete: () =>
    console.log(new Date().toLocaleTimeString(), "[fromEvent] complete"),
});

setTimeout(() => {
  em.emit("custom-event", "Hello from event emitter");
}, 3000);
```
```console
14:05:06 [fromEvent] start
14:05:09 [fromEvent] Hello from event emitter
```
![fromEvent Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/wv3wnvejxhylwlpsymsv.jpg)

* **[EMPTY](https://rxjs.dev/api/index/const/EMPTY)**

> Creates an Observable that emits no items to the Observer and immediately emits a complete notification.
```ts
import { EMPTY } from "rxjs";

EMPTY.subscribe({
  next: res => console.log("[EMPTY]", res),
  complete: () => console.log("[EMPTY] complete"),
});
```
```console
[EMPTY] complete
```
![EMPTY Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/db8tb1m32lap04p8o0ed.jpg)
This operator is used for scheduling the emission of the complete notification.

* **[timer](https://rxjs.dev/api/index/function/timer)**

> Creates an observable that will wait for a specified time period, or exact date, before emitting the number 0.
```ts
import { timer } from "rxjs";

console.log(new Date().toLocaleTimeString(), "[timer] start");
timer(1000).subscribe({
  next: res => console.log(new Date().toLocaleTimeString(), "[timer]", res),
  complete: () =>
    console.log(new Date().toLocaleTimeString(), "[timer] complete"),
});
```
```console
14:14:34 [timer] start
14:14:35 [timer] 0
14:14:35 [timer] complete
```
![timer Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/hxxfkyum2jegoeu5c8sx.jpg)
This operator is similar to the setTimeout function in JS. It creates an observable and emits only a value after a time indicated as argument.

* **[interval](https://rxjs.dev/api/index/function/interval)**

> Creates an Observable that emits sequential numbers every specified interval of time, on a specified SchedulerLike.
```ts
import { interval } from "rxjs";

console.log(new Date().toLocaleTimeString(), "[interval] start");
interval(1000).subscribe({
  next: res => console.log(new Date().toLocaleTimeString(), "[interval]", res),
  complete: () =>
    console.log(new Date().toLocaleTimeString(), "[interval] complete"),
});
```
```console
14:15:10 [interval] start
14:15:11 [interval] 0
14:15:12 [interval] 1
14:15:13 [interval] 2
14:15:14 [interval] 3
14:15:15 [interval] 4
...
...
...
```
![interval Marble Diagram Text](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/8lpv70f4h6m96xvxck5a.jpg)
This operator is similar to the setInterval function in JS. It creates an observable and emits a sequential number every specific time indicated as argument.
_N.B. it start from ZERO and not from ONE_

**timer as special interval**
Sometimes you need to create an interval with a specific period but you need to start it with a period different from the interval period. In these cases it is common to use the timer operator that accepts a period as second argument.
So you can indicate the time before emitting the first value as first argument and the period of the interval as second argument.
```ts
timer(1000, 1000).subscribe({
  next: res =>
    console.log(new Date().toLocaleTimeString(), "[timer as interval]", res),
  complete: () =>
    console.log(
      new Date().toLocaleTimeString(),
      "[timer as interval] complete"
    ),
});
```
```console
14:25:56 [timer as interval] start
14:26:00 [timer as interval] 0
14:26:01 [timer as interval] 1
14:26:02 [timer as interval] 2
...
...
```

## [Join Creation Operators](https://rxjs.dev/guide/operators#join-creation-operators)

Before starting, I clarify that all the examples use these three observable as inputs.

```ts
import { from, Observable } from "rxjs";

async function* hello() {
  const wait = async (time: number) =>
    new Promise(res => setTimeout(res, time));
  yield "Hello";
  await wait(1000);
  yield "from";
  await wait(500);
  yield "iterator";
}

export const iterator$ = from(hello());
export const arrayFrom$ = from(["Hello", "from", "array"]);
export const arrayOfWithDelay$ = new Observable<number>(subscriber => {
  let counter = 10;
  const id = setInterval(() => {
    if (counter > 0) {
      subscriber.next(counter--);
    } else {
      clearInterval(id);
      subscriber.complete();
    }
  }, 500);
});
```

* **[combineLatest](https://rxjs.dev/api/index/function/combineLatest)**

> Combines multiple Observables to create an Observable whose values are calculated from the latest values of each of its input Observables.

```ts
import { combineLatest } from "rxjs";
import { arrayFrom$, arrayOfWithDelay$, iterator$ } from "../sources";

console.log(new Date().toLocaleTimeString(), `[combineLatest] start`);

combineLatest([iterator$, arrayFrom$, arrayOfWithDelay$]).subscribe({
  next: res =>
    console.log(new Date().toLocaleTimeString(), `[combineLatest]`, res),
  complete: () =>
    console.log(new Date().toLocaleTimeString(), `[combineLatest] complete`),
});
```
```console
12:38:22 [combineLatest] start
12:38:22 [combineLatest] [ 'Hello', 'array', 10 ]
12:38:23 [combineLatest] [ 'from', 'array', 10 ]
12:38:23 [combineLatest] [ 'from', 'array', 9 ]
12:38:23 [combineLatest] [ 'iterator', 'array', 9 ]
12:38:23 [combineLatest] [ 'iterator', 'array', 8 ]
12:38:24 [combineLatest] [ 'iterator', 'array', 7 ]
12:38:24 [combineLatest] [ 'iterator', 'array', 6 ]
12:38:25 [combineLatest] [ 'iterator', 'array', 5 ]
12:38:25 [combineLatest] [ 'iterator', 'array', 4 ]
12:38:26 [combineLatest] [ 'iterator', 'array', 3 ]
12:38:26 [combineLatest] [ 'iterator', 'array', 2 ]
12:38:27 [combineLatest] [ 'iterator', 'array', 1 ]
12:38:27 [combineLatest] complete
```
![combineLatest Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/0ixjrhbs6o70piea384a.jpg)
In this example, you can see how this operator emits an array of value every time an observable emits one value.
It's important to remember that the operator emits the first value when all the dependence observables emit the first value.
As you can see, the result of the combineLatest operator is an array where items respect the order of the observables in the declaration.

* **[forkJoin](https://rxjs.dev/api/index/function/forkJoin)**

> Accepts an Array of ObservableInput or a dictionary Object of ObservableInput and returns an Observable that emits either an array of values in the exact same order as the passed array, or a dictionary of values in the same shape as the passed dictionary.

```ts
import { forkJoin } from "rxjs";
import { arrayFrom$, arrayOfWithDelay$, iterator$ } from "../sources";

console.log(new Date().toLocaleTimeString(), `[forkJoin] start`);

forkJoin([iterator$, arrayFrom$, arrayOfWithDelay$]).subscribe({
  next: res => console.log(new Date().toLocaleTimeString(), `[forkJoin]`, res),
  complete: () =>
    console.log(new Date().toLocaleTimeString(), `[forkJoin] complete`),
});
```
```console
14:38:58 [forkJoin] start
14:39:04 [forkJoin] [ 'iterator', 'array', 1 ]
14:39:04 [forkJoin] complete
```
![forkJoin Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/633cbiicb2ag0n6u1lm9.jpg)
forkJoin is similar to the combineLatest operator, the difference is that the forkJoin operator emits only one value when all the observables are completed. In simple words, the forkJoin operator emits only the last value of the combineLatest operator.

* **[concat](https://rxjs.dev/api/index/function/concat)**

> Creates an output Observable which sequentially emits all values from the first given Observable and then moves on to the next.

```ts
import { concat } from "rxjs";
import { arrayFrom$, arrayOfWithDelay$, iterator$ } from "../sources";

console.log(new Date().toLocaleTimeString(), `[concat] start`);

concat(iterator$, arrayFrom$, arrayOfWithDelay$).subscribe({
  next: res => console.log(new Date().toLocaleTimeString(), `[concat]`, res),
  complete: () =>
    console.log(new Date().toLocaleTimeString(), `[concat] complete`),
});
```
```console
14:44:23 [concat] start
14:44:23 [concat] Hello
14:44:24 [concat] from
14:44:24 [concat] iterator
14:44:24 [concat] Hello
14:44:24 [concat] from
14:44:24 [concat] array
14:44:25 [concat] 10
14:44:25 [concat] 9
14:44:26 [concat] 8
14:44:26 [concat] 7
14:44:27 [concat] 6
14:44:27 [concat] 5
14:44:28 [concat] 4
14:44:28 [concat] 3
14:44:29 [concat] 2
14:44:29 [concat] 1
14:44:30 [concat] complete
```
![concat Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/cmyi538qlfm44raobivz.jpg)
As you can see, this operator emits all the values of the observables in sequences.
Concat unlike of combineLatest does not run all observables in concurrency, but it runs the observables in sequence, it starts from the first and doesn't go to the next until the current is not completed.

* **[merge](https://rxjs.dev/api/index/function/merge)**

> Creates an output Observable which concurrently emits all values from every given input Observable.

```ts
import { merge } from "rxjs";
import { arrayFrom$, arrayOfWithDelay$, iterator$ } from "../sources";

console.log(new Date().toLocaleTimeString(), `[merge] start`);

merge(iterator$, arrayFrom$, arrayOfWithDelay$).subscribe({
  next: res => console.log(new Date().toLocaleTimeString(), `[merge]`, res),
  complete: () =>
    console.log(new Date().toLocaleTimeString(), `[merge] complete`),
});
```
```console
14:58:48 [merge] start
14:58:48 [merge] Hello
14:58:48 [merge] from
14:58:48 [merge] array
14:58:48 [merge] Hello
14:58:48 [merge] 10
14:58:49 [merge] from
14:58:49 [merge] 9
14:58:49 [merge] iterator
14:58:49 [merge] 8
14:58:50 [merge] 7
14:58:50 [merge] 6
14:58:51 [merge] 5
14:58:51 [merge] 4
14:58:52 [merge] 3
14:58:52 [merge] 2
14:58:53 [merge] 1
14:58:53 [merge] complete
```
![merge Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/1m8e0zqtn8ymn7nwgrq3.jpg)
The merge operator is similar to the concat operator unlike that the merge operator runs all observables in concurrency mode, so, in this case, all the observables start together and every time an observable emits a value the merge operator emits this last value.

* **[race](https://rxjs.dev/api/index/function/race)**

> Returns an observable that mirrors the first source observable to emit an item.

```ts
import { race } from "rxjs";
import { arrayFrom$, arrayOfWithDelay$, iterator$ } from "../sources";

console.log(new Date().toLocaleTimeString(), `[race] start`);

race([iterator$, arrayFrom$, arrayOfWithDelay$]).subscribe({
  next: res => console.log(new Date().toLocaleTimeString(), `[race]`, res),
  complete: () =>
    console.log(new Date().toLocaleTimeString(), `[race] complete`),
});
```
```console
15:09:03 [race] start
15:09:03 [race] Hello
15:09:03 [race] from
15:09:03 [race] array
15:09:03 [race] complete
```
![race Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/c6xo26df399om1tfj3lc.jpg)
This operator is particular, it emits the first observable that emits the first value. In other words, it takes the faster observable and ignores the others.

* **[zip](https://rxjs.dev/api/index/function/zip)**

> Combines multiple Observables to create an Observable whose values are calculated from the values, in order, of each of its input Observables.

```ts
import { zip } from "rxjs";
import { arrayFrom$, arrayOfWithDelay$, iterator$ } from "../sources";

console.log(new Date().toLocaleTimeString(), `[zip] start`);

zip([iterator$, arrayFrom$, arrayOfWithDelay$]).subscribe({
  next: res => console.log(new Date().toLocaleTimeString(), `[zip]`, res),
  complete: () =>
    console.log(new Date().toLocaleTimeString(), `[zip] complete`),
});
```
```console
15:09:27 [zip] start
15:09:27 [zip] [ 'Hello', 'Hello', 10 ]
15:09:28 [zip] [ 'from', 'from', 9 ]
15:09:28 [zip] [ 'iterator', 'array', 8 ]
15:09:28 [zip] complete
```
![zip Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/4wk2khoruej9urxlm1de.jpg)
This operator may seem strange, but it can use to combine in order the values of difference observables.
In this example we have 3 observables:
- iterator$: `['Hello', 'from', 'iterator', '!']`
- arrayFrom$: `['Hello', 'from', 'array', '!']`
- arrayOfWithDelay$: `[10, 9, 8, 7, 6, 5, 4, 3, 2, 1]`

With the zip operator we combine the values in order of them index:

- `['Hello', 'Hello', 10]`
- `['from', 'from', 9]`
- `['iterator', 'array', 8]`

As you can see, the operator stops emitting values at the index of the first completed observable.

That's it for today.
You can find the code of these examples at this [in the GitHub repository](https://github.com/Puppo/rxjs-getting-started/tree/05-creation-operators)

See you soon Guys!

---

### Part 6 of 15: RxJS - Filtering Operators

- **URL**: https://delpuppo.net/blog/rxjs-getting-started/rxjs-filtering-operators/
- **Published**: Sep 17, 2021
- **Tags**: javascript, typescript, rxjs

Welcome back Guys,

Today we'll start with the Filtering Operators. As you can imagine these operators are used to filter our sources.
Well, let's get started.

* **[auditTime](https://rxjs.dev/api/operators/auditTime)**

> Ignores source values for duration milliseconds, then emits the most recent value from the source Observable, then repeats this process.

```ts
import { Observable } from "rxjs";
import { auditTime } from "rxjs/operators";

const input$ = new Observable<number>(subscriber => {
  let count = 0;
  const id = setInterval(() => {
    if (count < 10) {
      subscriber.next(++count);
    } else {
      clearInterval(id);
      subscriber.complete();
    }
  }, 1000);
});

const result = input$.pipe(auditTime(2000));

result.subscribe({
  next: x =>
    console.log(
      `${new Date().toLocaleTimeString()} - [auditTime result]: ${x}`
    ),
});
```
```console
11:15:25 - [auditTime result]: 2
11:15:27 - [auditTime result]: 4
11:15:29 - [auditTime result]: 6
11:15:31 - [auditTime result]: 8
11:15:33 - [auditTime result]: 10
```
![auditTime Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/r9mlb6aumdr7nfxe3im2.jpg)
This operator is particular: for the indicated time it ignores the emitted values, and when the time is passed, it emits the last value emitted from the source. It can be used when we have a source that emits a lot of values but we need to get only the value in a specific interval, for example, we have a temperature sensor that emits the temperature every second but we need to take the value every minute, using auditTime we can get the last value of every minute.

* **[debounceTime](https://rxjs.dev/api/operators/debounceTime)**

> Emits a notification from the source Observable only after a particular time span has passed without another source emission.

```ts
import { Observable } from "rxjs";
import { debounceTime, tap } from "rxjs/operators";

const timesInSecond = [1, 0.5, 3, 1.5, 3, 1];

const input$ = new Observable<number>(subscriber => {
  let count = 0;
  (function next() {
    const seconds = timesInSecond[count++];
    setTimeout(() => {
      subscriber.next(seconds);

      if (count > 5) {
        subscriber.complete();
        return;
      }

      next();
    }, seconds * 1000);
  })();
});

const result = input$.pipe(
  tap(x =>
    console.log(
      `${new Date().toLocaleTimeString()} - [before debounceTime]: ${x}`
    )
  ),
  debounceTime(2000)
);
result.subscribe({
  next: x =>
    console.log(
      `${new Date().toLocaleTimeString()} - [debounceTime result]: ${x}`
    ),
});
```
```console
09:44:29 - [before debounceTime]: 1
09:44:29 - [before debounceTime]: 0.5
09:44:31 - [debounceTime result]: 0.5
09:44:32 - [before debounceTime]: 3
09:44:34 - [before debounceTime]: 1.5
09:44:36 - [debounceTime result]: 1.5
09:44:37 - [before debounceTime]: 3
09:44:38 - [before debounceTime]: 1
09:44:38 - [debounceTime result]: 1
```
![debounceTime Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/p75fu3jxsav4zs9a769d.jpg)
After receiving a value, this operator waits for the indicated time before emitting the value. If during this time a new value is emitted, the operator ignores the previous value and waits again, otherwise, if nothing happens it emits the value.
For example, this operator is used to manage the autocomplete text boxes that search items from an API; with this operator, we can prevent calling the API on every click on the keyboard but we can wait until the user finish writing some letters before executing the search.

* **[throttleTime](https://rxjs.dev/api/operators/throttleTime)**

> Emits a value from the source Observable, then ignores subsequent source values for duration milliseconds, then repeats this process.

```ts
import { Observable } from "rxjs";
import { tap, throttleTime } from "rxjs/operators";

const timesInSecond = [1, 0.5, 3, 1.5, 3, 1];

const input$ = new Observable<number>(subscriber => {
  let count = 0;
  (function next() {
    const seconds = timesInSecond[count++];
    setTimeout(() => {
      subscriber.next(seconds);

      if (count > 5) {
        subscriber.complete();
        return;
      }

      next();
    }, seconds * 1000);
  })();
});


const result = input$.pipe(
  tap(x =>
    console.log(
      `${new Date().toLocaleTimeString()} - [before throttleTime]: ${x}`
    )
  ),
  throttleTime(2000)
);
result.subscribe({
  next: x =>
    console.log(
      `${new Date().toLocaleTimeString()} - [throttleTime result]: ${x}`
    ),
});
```
```console
10:41:46 - [before throttleTime]: 1
10:41:46 - [throttleTime result]: 1
10:41:46 - [before throttleTime]: 0.5
10:41:49 - [before throttleTime]: 3
10:41:49 - [throttleTime result]: 3
10:41:51 - [before throttleTime]: 1.5
10:41:54 - [before throttleTime]: 3
10:41:54 - [throttleTime result]: 3
10:41:55 - [before throttleTime]: 1
```
![throttleTime Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/6egl8d2xxbve6ixn4uqc.jpg)
After receiving a value, this operator emits it and then waits for the indicated time. If during that time the source emits other values this operator ignores them, it repeats this process for life.
This operator is commonly used to prevent multiple clicks in some buttons or to prevent many executions of an action in a small period.

* **[distinct](https://rxjs.dev/api/operators/distinct)**

> Returns an Observable that emits all items emitted by the source Observable that are distinct by comparison from previous items.

```ts
import { Observable } from "rxjs";
import { distinct } from "rxjs/operators";

const input$ = new Observable<number | string>(subscriber => {
  let count = 0;
  const array = [1, 1, "1", 2, 2, 3, 3];
  const id = setInterval(() => {
    if (count < array.length) {
      subscriber.next(array[count++]);
    } else {
      clearInterval(id);
      subscriber.complete();
    }
  }, 1000);
});

input$.pipe(distinct()).subscribe({
  next: x =>
    console.log(`${new Date().toLocaleTimeString()} - [distinct]: ${x}`),
});
```
```console
12:22:58 - [distinct]: 1
12:23:00 - [distinct]: 1
12:23:01 - [distinct]: 2
12:23:03 - [distinct]: 3
```
![distinct Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/gr6koje1mt14qtsh7tl9.jpg)
This operator emits the value only if the value is different from the last value emitted. It's important to remember that the distinct operator checks the equality using the strict equal operator (===). It's also possible, if your source emits an object, to indicate the property used to check the equality `distinct(p => p.id)`.

* **[distinctUntilChanged](https://rxjs.dev/api/operators/distinctUntilChanged)**

```ts
import { Observable } from "rxjs";
import { distinctUntilChanged } from "rxjs/operators";

type State = { id: number; value: string };

const createValue = (id: number): State => ({ id, value: String(id) });
const array = [
  createValue(1),
  createValue(1),
  createValue(1),
  createValue(2),
  createValue(2),
  createValue(2),
  createValue(3),
  createValue(3),
  createValue(3),
  createValue(4),
  createValue(4),
  createValue(4),
];

const input$ = new Observable<State>(subscriber => {
  let count = 0;
  const id = setInterval(() => {
    if (count < array.length) {
      subscriber.next(array[count++]);
    } else {
      clearInterval(id);
      subscriber.complete();
    }
  }, 1000);
});

input$
  .pipe(distinctUntilChanged((prev, curr) => prev.id === curr.id))
  .subscribe({
    next: x =>
      console.log(
        `${new Date().toLocaleTimeString()} - [distinctUntilChanged]`,
        x
      ),
  });
```
```console
11:45:39 - [distinctUntilChanged] { id: 1, value: '1' }
11:45:42 - [distinctUntilChanged] { id: 2, value: '2' }
11:45:45 - [distinctUntilChanged] { id: 3, value: '3' }
11:45:48 - [distinctUntilChanged] { id: 4, value: '4' }
```
![distinctUntilChanged Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/7ozfifbk15f49apiaihf.jpg)
This operator is similar to the distinct operator but you can indicate a function that accepts the previous and the current item as arguments and indicates the algorithm to determinate if the two items are equals.

* **[distinctUntilKeyChanged](https://rxjs.dev/api/operators/distinctUntilKeyChanged)**

> Returns an Observable that emits all items emitted by the source Observable that are distinct by comparison from the previous item, using a property accessed by using the key provided to check if the two items are distinct.

```ts
import { Observable } from "rxjs";
import { distinctUntilKeyChanged } from "rxjs/operators";

type State = { id: number; value: string };

const createValue = (id: number): State => ({ id, value: String(id) });
const array = [
  createValue(1),
  createValue(1),
  createValue(1),
  createValue(2),
  createValue(2),
  createValue(2),
  createValue(3),
  createValue(3),
  createValue(3),
  createValue(4),
  createValue(4),
  createValue(4),
];

const input$ = new Observable<State>(subscriber => {
  let count = 0;
  const id = setInterval(() => {
    if (count < array.length) {
      subscriber.next(array[count++]);
    } else {
      clearInterval(id);
      subscriber.complete();
    }
  }, 1000);
});

input$.pipe(distinctUntilKeyChanged("id")).subscribe({
  next: x =>
    console.log(
      `${new Date().toLocaleTimeString()} - [distinctUntilKeyChanged]`,
      x
    ),
});
```
```console
11:46:12 - [distinctUntilKeyChanged] { id: 1, value: '1' }
11:46:15 - [distinctUntilKeyChanged] { id: 2, value: '2' }
11:46:18 - [distinctUntilKeyChanged] { id: 3, value: '3' }
11:46:21 - [distinctUntilKeyChanged] { id: 4, value: '4' }
```
![distinctUntilKeyChanged Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/jasmb16womm884k7jekw.jpg)
This operator is similar to the distinct operator but you can indicate the name of the property used to compare the equality.

* **[filter](https://rxjs.dev/api/operators/filter)**

> Filter items emitted by the source Observable by only emitting those that satisfy a specified predicate.

```ts
import { Observable } from "rxjs";
import { filter } from "rxjs/operators";

const input$ = new Observable<number>(subscriber => {
  let count = 0;
  const id = setInterval(() => {
    if (count < 10) {
      subscriber.next(++count);
    } else {
      clearInterval(id);
      subscriber.complete();
    }
  }, 1000);
});

input$.pipe(filter(x => x % 2 === 0)).subscribe({
  next: x => console.log(`${new Date().toLocaleTimeString()} - [filter]: ${x}`),
});
```
```console
11:46:43 - [filter]: 2
11:46:45 - [filter]: 4
11:46:47 - [filter]: 6
11:46:49 - [filter]: 8
11:46:51 - [filter]: 10
```
![filter Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/rfq2mjcx3b80zg4i05c2.jpg)
As you can imagine, this operator accepts a function that has an argument (the current item) and returns a boolean that indicates if the value can be emitted or must be ignored.

* **[first](https://rxjs.dev/api/operators/first)**

> Emits only the first value (or the first value that meets some condition) emitted by the source Observable.

```ts
import { Observable } from "rxjs";
import { first } from "rxjs/operators";

const input$ = new Observable<number>(subscriber => {
  let count = 0;
  const id = setInterval(() => {
    if (count < 10) {
      subscriber.next(++count);
    } else {
      clearInterval(id);
      subscriber.complete();
    }
  }, 1000);
});

console.log(`${new Date().toLocaleTimeString()} - [first] start`)
input$.pipe(first()).subscribe({
  next: x => console.log(`${new Date().toLocaleTimeString()} - [first]: ${x}`),
  complete: () => console.log(`${new Date().toLocaleTimeString()} - [first] complete`),
});
```
```console
09:47:15 - [first] start
09:47:16 - [first]: 1
09:47:16 - [first] complete
```
![first Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/ea26ick6pwkfh9u7enla.jpg)
This operator takes the first value and ignores the others.
It's important to remember that when the first operator emits the value it completes the subscription too.

* **[last](https://rxjs.dev/api/operators/last)**

> Returns an Observable that emits only the last item emitted by the source Observable. It optionally takes a predicate function as a parameter, in which case, rather than emitting the last item from the source Observable, the resulting Observable will emit the last item from the source Observable that satisfies the predicate.

```ts
import { Observable } from "rxjs";
import { last } from "rxjs/operators";

const input$ = new Observable<number>(subscriber => {
  let count = 0;
  const id = setInterval(() => {
    if (count < 10) {
      subscriber.next(++count);
    } else {
      clearInterval(id);
      subscriber.complete();
    }
  }, 1000);
});

console.log(`${new Date().toLocaleTimeString()} - [last] start`)
input$.pipe(last()).subscribe({
  next: x => console.log(`${new Date().toLocaleTimeString()} - [last]: ${x}`),
  complete: () => console.log(`${new Date().toLocaleTimeString()} - [last] complete`),
});
```
```console
09:48:14 - [last] start
09:48:25 - [last]: 10
09:48:25 - [last] complete
```
![last Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/xpk5q4s7v7faaash52jj.jpg)
This operator takes the last value and ignores the others.

* **[skip](https://rxjs.dev/api/operators/skip)**

> Returns an Observable that skips the first count items emitted by the source Observable.

```ts
import { Observable } from "rxjs";
import { skip } from "rxjs/operators";

const input$ = new Observable<number>(subscriber => {
  let count = 0;
  const id = setInterval(() => {
    if(count < 5) subscriber.next(count++);
    else {
      clearInterval(id);
      subscriber.complete();
    } 
  }, 1000);
  return () => {
    clearInterval(id);
    subscriber.complete();
  };
});

input$.pipe(skip(2)).subscribe({
  next: x => console.log(`${new Date().toLocaleTimeString()} - [skip]: ${x}`),
  complete: () =>
    console.log(`${new Date().toLocaleTimeString()} - [skip]: complete`),
});
```
```console
10:33:41 - [skip]: 2
10:33:42 - [skip]: 3
10:33:43 - [skip]: 4
10:33:44 - [skip]: complete
```
![skip Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/3p7gj5gz1wtbt665409w.jpg)
As you can see, this operator skips the first x items we indicate, if we write `skip(1)` the operator skips the first value, if we write `skip(2)` the operator skips the first two values and so on.

* **[skipLast](https://rxjs.dev/api/operators/skipLast)**

> Skip a specified number of values before the completion of an observable.

```ts
import { Observable } from "rxjs";
import { skipLast } from "rxjs/operators";

const input$ = new Observable<number>(subscriber => {
  let count = 0;
  const id = setInterval(() => {
    if (count < 5) subscriber.next(count++);
    else {
      clearInterval(id);
      subscriber.complete();
    }
  }, 1000);
});

console.log(`${new Date().toLocaleTimeString()} - [skipLast]: start`)

input$.pipe(skipLast(3)).subscribe({
  next: x => console.log(`${new Date().toLocaleTimeString()} - [skipLast]: ${x}`),
  complete: () =>
    console.log(`${new Date().toLocaleTimeString()} - [skipLast]: complete`),
});
```
```console
10:33:58 - [skipLast]: start
10:34:02 - [skipLast]: 0
10:34:03 - [skipLast]: 1
10:34:04 - [skipLast]: complete
```
![skipLast Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/3ljibsfj1i20fwjwq1db.jpg)
This operator skips the last x items we indicate, if we write `skipLast(1)` the operator skips the last value, if we write `skipLast(2)` the operator skips the last two values and so on.

* **[skipUntil](https://rxjs.dev/api/operators/skipUntil)**

> Returns an Observable that skips items emitted by the source Observable until a second Observable emits an item.

```ts
import { Observable } from "rxjs";
import { skipUntil } from "rxjs/operators";

const input$ = new Observable<number>(subscriber => {
  let count = 0;
  const id = setInterval(() => {
    if (count < 5) subscriber.next(count++)
    else {
      clearInterval(id);
      subscriber.complete();
    }
  }, 1000);
});

const untilInput$ = new Observable<void>(subscriber => {
  setTimeout(() => {
    console.log(
      `${new Date().toLocaleTimeString()} - untilInput$ emit`
    );
    subscriber.next();
    subscriber.complete();
  }, 5000);
});

input$.pipe(skipUntil(untilInput$)).subscribe({
  next: x =>
    console.log(`${new Date().toLocaleTimeString()} - [skipUntil]: ${x}`),
  complete: () =>
    console.log(`${new Date().toLocaleTimeString()} - [skipUntil]: complete`),
});
```
```console
10:34:37 - untilInput$ emit
10:34:37 - [skipUntil]: 4
10:34:38 - [skipUntil]: complete
```
![skipUntil Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/fpej4jgcs8b68g2k0x4l.jpg)
This operator skips all the values until another observable doesn't emit a value.

* **[skipWhile](https://rxjs.dev/api/operators/skipWhile)**

> Returns an Observable that skips all items emitted by the source Observable as long as a specified condition holds true, but emits all further source items as soon as the condition becomes false.

```ts
import { Observable } from "rxjs";
import { skipWhile } from "rxjs/operators";

const input$ = new Observable<number>(subscriber => {
  let count = 0;
  const id = setInterval(() => {
    if (count < 5) subscriber.next(count++);
    else {
      subscriber.next(count++);
      subscriber.complete();
    }
  }, 1000);
  return () => {
    clearInterval(id);
  };
});

input$.pipe(skipWhile(val => val < 3)).subscribe({
  next: x =>
    console.log(`${new Date().toLocaleTimeString()} - [skipWhile]: ${x}`),
  complete: () =>
    console.log(`${new Date().toLocaleTimeString()} - [skipWhile]: complete`),
});
```
```console
10:36:34 - [skipWhile]: 3
10:36:35 - [skipWhile]: 4
10:36:36 - [skipWhile]: 5
10:36:36 - [skipWhile]: complete
```
![skipWhile Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/dt6ile46ejbf9aqtnrqw.jpg)
This operator skips all the values ​​until the predicate isn't true, after the first true result, the skipWhile operator doesn't eval anymore the predicate and emits all the values.

* **[take](https://rxjs.dev/api/operators/take)**

> Emits only the first count values emitted by the source Observable.

```ts
import { Observable } from "rxjs";
import { take } from "rxjs/operators";

const input$ = new Observable<number>(subscriber => {
  let count = 0;
  const id = setInterval(() => {
    subscriber.next(count++);
  }, 1000);
  return () => {
    clearInterval(id);
    subscriber.complete();
  };
});

input$.pipe(take(2)).subscribe({
  next: x => console.log(`${new Date().toLocaleTimeString()} - [take]: ${x}`),
  complete: () =>
    console.log(`${new Date().toLocaleTimeString()} - [take]: complete`),
});
```
```console
10:39:39 - [take]: 0
10:39:40 - [take]: 1
10:39:40 - [take]: complete
```
![take Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/hsxljrplhxsxq01r4ajw.jpg)
The take operator is the opposite of the skip operator, if the skip ignores x values, the take operator emits x values.
So if you write `take(1)` you get only the first value, if you write `take(2)` you get only the first two values.
It's important to remember that when the operator emits the last value completes the subscription too.

* **[takeLast](https://rxjs.dev/api/operators/takeLast)**

> Waits for the source to complete, then emits the last N values from the source, as specified by the count argument.

```ts
import { Observable } from "rxjs";
import { takeLast } from "rxjs/operators";

const input$ = new Observable<number>(subscriber => {
  let count = 0;
  const id = setInterval(() => {
    if (count < 5) subscriber.next(count++);
    else {
      clearInterval(id);
      subscriber.complete();
    }
  }, 1000);
});

console.log(`${new Date().toLocaleTimeString()} - [takeLast]: start`)

input$.pipe(takeLast(3)).subscribe({
  next: x => console.log(`${new Date().toLocaleTimeString()} - [takeLast]: ${x}`),
  complete: () =>
    console.log(`${new Date().toLocaleTimeString()} - [takeLast]: complete`),
});
```
```console
10:40:08 - [takeLast]: start
10:40:14 - [takeLast]: 2
10:40:14 - [takeLast]: 3
10:40:14 - [takeLast]: 4
10:40:14 - [takeLast]: complete
```
![takeLast Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/2qbefu68sm4vtmxclfqm.jpg)
This operator takes the last x items we indicate, if we write takeLast(1) the operator takes the last value, if we write takeLast(3) the operator takes the last three values, and so on.

* **[takeUntil](https://rxjs.dev/api/operators/takeUntil)**

> Emits the values emitted by the source Observable until a notifier Observable emits a value.

```ts
import { Observable } from "rxjs";
import { takeUntil } from "rxjs/operators";

const input$ = new Observable<number>(subscriber => {
  let count = 0;
  const id = setInterval(() => {
    if (count < 5) subscriber.next(++count);
    else {
      clearInterval(id);
      subscriber.complete();
    }
  }, 1000);
});

const untilInput$ = new Observable<void>(subscriber => {
  setTimeout(() => {
    console.log(
      `${new Date().toLocaleTimeString()} - untilInput$ emit`
    );
    subscriber.next();
    subscriber.complete();
  }, 2500);
});

input$.pipe(takeUntil(untilInput$)).subscribe({
  next: x =>
    console.log(`${new Date().toLocaleTimeString()} - [takeUntil]: ${x}`),
  complete: () =>
    console.log(`${new Date().toLocaleTimeString()} - [takeUntil]: complete`),
});
```
```console
10:40:38 - [takeUntil]: 1
10:40:39 - [takeUntil]: 2
10:40:42 - untilInput$ emit
10:40:42 - [takeUntil]: complete
```
![takeUntil Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/5dchari3mm6r4tc26r9l.jpg)
This operator takes all the values until another observable doesn't emit a value, after that it complete the subscription.

* **[takeWhile](https://rxjs.dev/api/operators/takeWhile)**

> Emits values emitted by the source Observable so long as each value satisfies the given predicate, and then completes as soon as this predicate is not satisfied.

```ts
import { Observable } from "rxjs";
import { takeWhile } from "rxjs/operators";

const input$ = new Observable<number>(subscriber => {
  let count = 0;
  const id = setInterval(() => {
    subscriber.next(count++);
  }, 1000);
  return () => {
    clearInterval(id);
    subscriber.complete();
  };
});

input$.pipe(takeWhile(val => val < 3)).subscribe({
  next: x =>
    console.log(`${new Date().toLocaleTimeString()} - [takeWhile]: ${x}`),
  complete: () =>
    console.log(`${new Date().toLocaleTimeString()} - [takeWhile]: complete`),
});
```
```console
10:41:10 - [takeWhile]: 0
10:41:11 - [takeWhile]: 1
10:41:12 - [takeWhile]: 2
10:41:13 - [takeWhile]: complete
```
![takeWhile Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/nrpqei8a9kvbjqjoz3kq.jpg)
This operator emits all the values ​​until the predicate is true, after that the takeWhile operator completes the subscription.


Ok Guys, I think that today it's enough for the Filter Operators.
You can find all the code of this article [in the GitHub repository](https://github.com/Puppo/rxjs-getting-started/tree/06-filtering-operators).

See you in the next article.
Bye Bye!

---

### Part 7 of 15: RxJS - Transformation Operators

- **URL**: https://delpuppo.net/blog/rxjs-getting-started/rxjs-transformation-operators/
- **Published**: Sep 20, 2021
- **Tags**: javascript, typescript, rxjs

Welcome back,
Today I'll speak about the Transformation Operators. These operators are used to modify the value received.
But cut the chatter, and let's start.

* **[map](https://rxjs.dev/api/operators/map)**

> Applies a given project function to each value emitted by the source Observable, and emits the resulting values as an Observable.

```ts
import { interval } from 'rxjs';
import { map, take, tap } from 'rxjs/operators';

const source$ = interval(1000, ).pipe(
    take(5),
    tap(val => {
        console.log(`${new Date().toLocaleTimeString()}: Generated`, val);
    }),
)

source$.pipe(
    map(value => Math.pow(value, 2))
)
.subscribe({
    next: value => console.log(`${new Date().toLocaleTimeString()}: map`, value),
    complete: () => console.log(`${new Date().toLocaleTimeString()}: map: complete`),
})
```
```console
16:58:51: Generated 0
16:58:51: map 0
16:58:52: Generated 1
16:58:52: map 1
16:58:53: Generated 2
16:58:53: map 4
16:58:54: Generated 3
16:58:54: map 9
16:58:55: Generated 4
16:58:55: map 16
16:58:55: map: complete
```
The map operator is like the map function of the arrays.
It transforms the value using the function passed as an argument.

* **[mapTo](https://rxjs.dev/api/operators/mapTo)**

> Emits the given constant value on the output Observable every time the source Observable emits a value.

```ts
import { interval } from 'rxjs';
import { mapTo, take, tap } from 'rxjs/operators';

const source1$ = interval(1000).pipe(
    take(5),
    tap(val => {
        console.log(`${new Date().toLocaleTimeString()}: Source1 Generated`, val);
    }),
)

console.log(`${new Date().toLocaleTimeString()}: mapTo: start`)
source1$.pipe(
    mapTo(1000)
)
.subscribe({
    next: value => console.log(`${new Date().toLocaleTimeString()}: mapTo`, value),
    complete: () => console.log(`${new Date().toLocaleTimeString()}: mapTo: complete`),
})
```
```console
17:03:01: mapTo: start
17:03:02: Source1 Generated 0
17:03:02: mapTo 1000
17:03:03: Source1 Generated 1
17:03:03: mapTo 1000
17:03:04: Source1 Generated 2
17:03:04: mapTo 1000
17:03:05: Source1 Generated 3
17:03:05: mapTo 1000
17:03:06: Source1 Generated 4
17:03:06: mapTo 1000
17:03:06: mapTo: complete
```
The operator is similar to the map operator, but it returns a fixed value that does not depend on the source.


Now I start to talk about the operators: concat, exhaust, merge and switch.
These operators are similar to the others but they have some minimum differences that change their behaviour, and if you choose the wrong implementations you may not have the aspected result.

* **[concatMap](https://rxjs.dev/api/operators/concatMap)**

> Projects each source value to an Observable which is merged in the output Observable, in a serialized fashion waiting for each one to complete before merging the next.

```ts
import { interval, Observable, of } from 'rxjs';
import { concatMap, delay, take, tap } from 'rxjs/operators';

const powWithDelay$ = (value: number): Observable<number> => of(Math.pow(value, 2)).pipe(delay(2000))

const source$ = interval(1000).pipe(
    take(5),
    tap(val => {
        console.log(`${new Date().toLocaleTimeString()}: Generated`, val);
    }),
)

source$.pipe(
    concatMap(value => powWithDelay$(value))
)
.subscribe({
    next: value => console.log(`${new Date().toLocaleTimeString()}: concatMap`, value),
    complete: () => console.log(`${new Date().toLocaleTimeString()}: concatMap: complete`),
})
```
```console
17:54:07: Generated 0
17:54:08: Generated 1
17:54:09: Generated 2
17:54:09: concatMap 0
17:54:10: Generated 3
17:54:11: Generated 4
17:54:11: concatMap 1
17:54:13: concatMap 4
17:54:15: concatMap 9
17:54:17: concatMap 16
17:54:17: concatMap: complete
```
This operator is used to concatenate different observables.
The first source starts to emit the values, and the concatMap operator runs another observable for every value. The result values of the last observables are emitted as results of the operator concat.
You should remember that the concatMap operator runs only one value at a time. You can notice from the previous examples that the values 0,1 and 2 are emitted before the result of the first concatMap (0), and after that, you can notice the emission of the values 3 and 4 before the result of the second emitted value. The particular behaviour of the concatMap operator is better visible after the last generated value from the first source (4) because we can see all the concatMap results emitted every 2 seconds from each other.

Here's the marble diagram to explain better the behaviour:
![ConcatMap Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/7wmkcr2jyuj0rk4x1hsl.jpg)

* **[concatMapTo](https://rxjs.dev/api/operators/concatMapTo)**

> Projects each source value to the same Observable which is merged multiple times in a serialized fashion on the output Observable.

```ts
import { interval } from 'rxjs';
import { concatMapTo, skip, take, tap } from 'rxjs/operators';

const source1$ = interval(1000, ).pipe(
    skip(2),
    take(2),
    tap(val => {
        console.log(`${new Date().toLocaleTimeString()}: Source1 Generated`, val);
    }),
)

const source2$ = interval(2000, ).pipe(
    take(2),
    tap(val => {
        console.log(`${new Date().toLocaleTimeString()}: Source2 Generated`, val);
    }),
)

console.log(`${new Date().toLocaleTimeString()}: concatMapTo: start`)
source1$.pipe(
    concatMapTo(source2$)
)
.subscribe({
    next: value => console.log(`${new Date().toLocaleTimeString()}: concatMapTo`, value),
    complete: () => console.log(`${new Date().toLocaleTimeString()}: concatMapTo: complete`),
})
```
```console
18:12:28: concatMapTo: start
18:12:31: Source1 Generated 2
18:12:32: Source1 Generated 3
18:12:33: Source2 Generated 0
18:12:33: concatMapTo 0
18:12:35: Source2 Generated 1
18:12:35: concatMapTo 1
18:12:37: Source2 Generated 0
18:12:37: concatMapTo 0
18:12:39: Source2 Generated 1
18:12:39: concatMapTo 1
18:12:39: concatMapTo: complete
```
This operator is like the concatMap apart from the fact that it returns another observable that does not depend on the value received.
It can be used when an observable emit a value and we need to run another observable.
For example, we have a source that is a timer, and on every tick, we need to call an API. If we use the concatMapTo we can implement this solution easier.

* **[exhaustMap](https://rxjs.dev/api/operators/exhaustMap)**

> Projects each source value to an Observable which is merged in the output Observable only if the previous projected Observable has completed.

```ts
import { interval, Observable, of } from 'rxjs';
import { delay, exhaustMap, take, tap } from 'rxjs/operators';

const powWithDelay$ = (value: number): Observable<number> => of(Math.pow(value, 2)).pipe(delay(2000))

const source$ = interval(1000, ).pipe(
    take(5),
    tap(val => {
        console.log(`${new Date().toLocaleTimeString()}: Generated`, val);
    }),
)

source$.pipe(
    exhaustMap(value => powWithDelay$(value))
)
.subscribe({
    next: value => console.log(`${new Date().toLocaleTimeString()}: exhaustMap`, value),
    complete: () => console.log(`${new Date().toLocaleTimeString()}: exhaustMap: complete`),
})
```
```console
18:17:47: Generated 0
18:17:48: Generated 1
18:17:49: Generated 2
18:17:49: exhaustMap 0
18:17:50: Generated 3
18:17:51: Generated 4
18:17:52: exhaustMap 9
18:17:52: exhaustMap: complete
```

The exhaustMap operator syntactically is equal to the concat operator, but it has a different behavior: this operator during the execution of a projection ignores all other values received until the execution is not ended. If you see the result in the console of the before example, you can notice that the power of the values 1, 2, and 4 has never been shown.

The marble diagram of the exhaustMap operator in the previous example:
![ExhaustMap Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/o5heddkiwdmbj3uyehot.jpg)

* **[mergeMap](https://rxjs.dev/api/operators/mergeMap)**

> Projects each source value to an Observable which is merged in the output Observable.

```ts
import { interval, Observable, of } from 'rxjs';
import { delay, mergeMap, take, tap } from 'rxjs/operators';

const powWithDelay$ = (value: number): Observable<number> => of(Math.pow(value, 2)).pipe(delay(2000))

const source$ = interval(1000, ).pipe(
    take(5),
    tap(val => {
        console.log(`${new Date().toLocaleTimeString()}: Generated`, val);
    }),
)

source$.pipe(
    mergeMap(value => powWithDelay$(value))
)
.subscribe({
    next: value => console.log(`${new Date().toLocaleTimeString()}: mergeMap`, value),
    complete: () => console.log(`${new Date().toLocaleTimeString()}: mergeMap: complete`),
})
```
```console
18:18:16: Generated 0
18:18:17: Generated 1
18:18:18: Generated 2
18:18:18: mergeMap 0
18:18:19: mergeMap 1
18:18:19: Generated 3
18:18:20: mergeMap 4
18:18:20: Generated 4
18:18:21: mergeMap 9
18:18:22: mergeMap 16
18:18:22: mergeMap: complete
```
The mergeMap operator is syntactically like concat and exhaust but it executes the projection as soon as a value arrives. This behaviour can open some strange cases, for example, it's not guaranteed the order of the result values with the order of the source values.
Let me give an example: we have a source that emits some values every 500ms, in the mergeMap operator we call an http API; the first value calls the API and the response arrives after 1.5seconds, the second value calls the same api but the response arrives after 200ms. The mergeMap operator in this case emits the result of the second value primarily and the result of the first one secondly.
As you can imagine, in some cases this behaviour can be wrong or unexpected, so, if you need to use the mergeMap operator, remember what I said before and ask yourself if it can produce some side effects or if it is the right solution of your problems.

Here is the MergeMap marble Diagram of the previous example:
![MergeMap Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/z72t3ufmngyub6m1okpn.jpg)


* **[mergeMapTo](https://rxjs.dev/api/operators/mergeMapTo)**

> Projects each source value to the same Observable which is merged multiple times in the output Observable.

```ts
import { interval } from 'rxjs';
import { mergeMapTo, skip, take, tap } from 'rxjs/operators';

const source1$ = interval(1000, ).pipe(
    skip(2),
    take(2),
    tap(val => {
        console.log(`${new Date().toLocaleTimeString()}: Source1 Generated`, val);
    }),
)

const source2$ = interval(2000, ).pipe(
    take(2),
    tap(val => {
        console.log(`${new Date().toLocaleTimeString()}: Source2 Generated`, val);
    }),
)

console.log(`${new Date().toLocaleTimeString()}: mergeMap: start`)
source1$.pipe(
    mergeMapTo(source2$)
)
.subscribe({
    next: value => console.log(`${new Date().toLocaleTimeString()}: mergeMap`, value),
    complete: () => console.log(`${new Date().toLocaleTimeString()}: mergeMap: complete`),
})
```
```console
18:18:53: mergeMap: start
18:18:56: Source1 Generated 2
18:18:57: Source1 Generated 3
18:18:58: Source2 Generated 0
18:18:58: mergeMap 0
18:18:59: Source2 Generated 0
18:18:59: mergeMap 0
18:19:00: Source2 Generated 1
18:19:00: mergeMap 1
18:19:01: Source2 Generated 1
18:19:01: mergeMap 1
18:19:01: mergeMap: complete
```
This operator is like the mergeMap apart from the fact that it returns another observable that does not depend on the value received.
It can be used when an observable emits a value and we need to run another observable.
In simple words, it's like the concatMapTo with the behaviour of the mergeMap operator behind the scene.

* **[switchMap](https://rxjs.dev/api/operators/switchMap)**

> Projects each source value to an Observable which is merged in the output Observable, emitting values only from the most recently projected Observable.

```ts
import { interval, Observable, of } from 'rxjs';
import { delay, switchMap, take, tap } from 'rxjs/operators';

const powWithDelay$ = (value: number): Observable<number> => of(Math.pow(value, 2)).pipe(delay(2000))

const source$ = interval(1000, ).pipe(
    take(5),
    tap(val => {
        console.log(`${new Date().toLocaleTimeString()}: Generated`, val);
    }),
)

source$.pipe(
    switchMap(value => powWithDelay$(value))
)
.subscribe({
    next: value => console.log(`${new Date().toLocaleTimeString()}: switchMap`, value),
    complete: () => console.log(`${new Date().toLocaleTimeString()}: switchMap: complete`),
})
```
```console
18:19:16: Generated 0
18:19:17: Generated 1
18:19:18: Generated 2
18:19:19: Generated 3
18:19:20: Generated 4
18:19:22: switchMap 16
18:19:22: switchMap: complete
```
The switchMap operator is syntactically like concat, exhaust and merge. It executes the projection as soon as a value arrives, but when a new value arrives, if the projection of the previous value is in execution, it kills it and starts the execution of the projection for the new value.
It can be used for example for searching the data of an autocomplete input. When the user types a new letter and emits a new value, with this operator we can stop the previous search if it's on execution and start the new one.

Here is the SwitchMap marble Diagram of the previous example:
![SwitchMap Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/homky2m86l0gfq0ssrar.jpg)

* **[switchMapTo](https://rxjs.dev/api/operators/switchMapTo)**

> Projects each source value to the same Observable which is flattened multiple times with switchMap in the output Observable.

```ts
import { interval } from 'rxjs';
import { skip, switchMapTo, take, tap } from 'rxjs/operators';

const source1$ = interval(1000, ).pipe(
    skip(2),
    take(2),
    tap(val => {
        console.log(`${new Date().toLocaleTimeString()}: Source1 Generated`, val);
    }),
)

const source2$ = interval(2000, ).pipe(
    take(2),
    tap(val => {
        console.log(`${new Date().toLocaleTimeString()}: Source2 Generated`, val);
    }),
)

console.log(`${new Date().toLocaleTimeString()}: switchMap: start`)
source1$.pipe(
    switchMapTo(source2$)
)
.subscribe({
    next: value => console.log(`${new Date().toLocaleTimeString()}: switchMap`, value),
    complete: () => console.log(`${new Date().toLocaleTimeString()}: switchMap: complete`),
})
```
```console
18:19:38: switchMap: start
18:19:41: Source1 Generated 2
18:19:42: Source1 Generated 3
18:19:44: Source2 Generated 0
18:19:44: switchMap 0
18:19:46: Source2 Generated 1
18:19:46: switchMap 1
18:19:46: switchMap: complete
```
This operator is like the switchMap apart from the fact that it returns another observable that does not depend on the value received.
It can be used when an observable emits a value and we need to run another observable.
In simple words, it's like the concatMapTo or the mergeMapTo with the behaviour of the switchMap operator behind the scene.


_with this operator, we have finish the explanation of the 4 "special" operators (concat, exhaust, merge and switch). As you can see they are similar to each other but if you choose the wrong one it can produces some unexpected side effects.
I hope these examples can help you in future to choose the right operator for you_


* **[pairwise](https://rxjs.dev/api/operators/pairwise)**

> Groups pairs of consecutive emissions together and emits them as an array of two values.

```ts
import { of } from 'rxjs';
import { pairwise } from 'rxjs/operators';

const source$ = of(1, 2, 3, 4, 5);

source$.pipe(
    pairwise()
)
.subscribe({
    next: value => console.log(`${new Date().toLocaleTimeString()}: pairwise`, value),
})
```
```console
18:20:02: pairwise [ 1, 2 ]
18:20:02: pairwise [ 2, 3 ]
18:20:02: pairwise [ 3, 4 ]
18:20:02: pairwise [ 4, 5 ]
```
This operator is used to get as result a tuple where in the first index there is the previous value and in the second index there is the current value. As you can imagine, if your source emits only one value the pairwise operator will never emit a value.


* **[scan](https://rxjs.dev/api/operators/scan)**

> Useful for encapsulating and managing state. Applies an accumulator (or "reducer function") to each value from the source after an initial state is established -- either via a seed value (second argument), or from the first value from the source.

```ts
import { of } from 'rxjs';
import { scan } from 'rxjs/operators';

const source$ = of(1, 2, 3, 4, 5);

source$.pipe(
    scan((acc, curr) => acc + curr, 0)
)
.subscribe({
    next: value => console.log(`${new Date().toLocaleTimeString()}: scan`, value),
})
```
```console
18:20:26: scan 1
18:20:26: scan 3
18:20:26: scan 6
18:20:26: scan 10
18:20:26: scan 15
```
This operator is similar to the [reduce](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Array/Reduce) method of the array. We can create an accumulator and every time a new value is emitted from the source we can update this accumulator and return it as result.

Ok, that's all from the Transformation Operators, I hope this examples can help you in future and I hope you have achieved a clear idea of how and when to use these operators.

[in the GitHub repository](https://github.com/Puppo/rxjs-getting-started/tree/07-transformation-operators) you can find all the code examples.

See you soon Guys
Bye Bye!

---

### Part 8 of 15: RxJS - Join Operators

- **URL**: https://delpuppo.net/blog/rxjs-getting-started/rxjs-join-operators/
- **Published**: Sep 22, 2021
- **Tags**: javascript, typescript, rxjs

Hi Guys and Welcome Back,
Today I'll speak about the Join Operators. Some of these operators are similar to other operators shown in the [Creation Operators](https://dev.to/this-is-learning/rxjs-creation-operators-6bh) but they are used in different contexts, ok let's start!

* **[combineLatestAll](https://rxjs.dev/api/operators/combineLatestAll)**

> Flattens an Observable-of-Observables by applying combineLatest when the Observable-of-Observables completes.

```ts
import { interval, timer } from 'rxjs';
import { combineLatestAll, map, take } from 'rxjs/operators';

const A_CHAR_CODE = 65
const createSource = (index: number) =>
    interval((index + 1) * 1000).pipe(
        take(3),
        map(value => `${String.fromCharCode(A_CHAR_CODE + index)}${String.fromCharCode(A_CHAR_CODE + value)}`)
    )

console.log(`${new Date().toLocaleTimeString()}: combineLatestAll start`)
const source$ = timer(0, 1000).pipe(
    take(3)
);
source$.pipe(
    map(createSource),
    combineLatestAll()
).subscribe({
    next: val => console.log(`${new Date().toLocaleTimeString()}: combineLatestAll`, val)
});
```
```console
16:22:07: combineLatestAll start
16:22:12: combineLatestAll [ 'AB', 'BA', 'CA' ]
16:22:12: combineLatestAll [ 'AC', 'BA', 'CA' ]
16:22:13: combineLatestAll [ 'AC', 'BB', 'CA' ]
16:22:15: combineLatestAll [ 'AC', 'BB', 'CB' ]
16:22:15: combineLatestAll [ 'AC', 'BC', 'CB' ]
16:22:18: combineLatestAll [ 'AC', 'BC', 'CC' ]
```
![CombineLatestAll Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/lv89hgb4oy8qmn6v2sid.jpg)
As you can see, this operator is the brother of the combineLatest operator discussed in a previous article [RxJS - Creation Operators](https://dev.to/this-is-learning/rxjs-creation-operators-6bh).
Its job is to convert an Observable of Observables to an array of values, where each item of this array is a value of an observable.
When we use this operator we need to remember that the first value emitted by the operator is when all the observables have emitted at least one value.
This operator is used when in the pipe chain we have an operator that emits a list of observables; using this operator we can flat the values in a single array including all the values emitted by these observables. 

* **[concatAll](https://rxjs.dev/api/operators/concatAll)**

> Converts a higher-order Observable into a first-order Observable by concatenating the inner Observables in order.

```ts
import { interval, timer } from 'rxjs';
import { concatAll, map, take } from 'rxjs/operators';

const createSource = (index: number) => interval(index * 1000).pipe(
        take(3),
        map(value => `${index}-${value + 1}`)
    )

console.log(`${new Date().toLocaleTimeString()}: concatAll start`)
const source$ = timer(0, 1000).pipe(
    take(3)
);
source$.pipe(
    map(v => createSource(v + 1)),
    concatAll()
).subscribe({
    next: val => console.log(`${new Date().toLocaleTimeString()}: concatAll`, val)
});
```
```console
15:38:29: concatAll start
15:38:30: concatAll 1-1
15:38:31: concatAll 1-2
15:38:32: concatAll 1-3
15:38:34: concatAll 2-1
15:38:36: concatAll 2-2
15:38:38: concatAll 2-3
15:38:41: concatAll 3-1
15:38:44: concatAll 3-2
15:38:47: concatAll 3-3
```
![ConcatAll Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/v4dt0rys57q6ek219o8b.jpg)
The concatAll operator concatenates an Observable of Observables and emits all the observables in sequence.
It's important to remember that when we use this operator, the observables are running one after the other, so the operator executes the first observable until it isn't completed, then it passes to the second observable, and so on.
This operator is used when we have in a pipe chain an operator that returns a list of observables and we need to execute them in sequence.

* **[exhaustAll](https://rxjs.dev/api/operators/exhaustAll)**

> Converts a higher-order Observable into a first-order Observable by dropping inner Observables while the previous inner Observable has not yet completed.

```ts
import { interval, timer } from 'rxjs';
import { exhaustAll, map, take } from 'rxjs/operators';

const createSource = (index: number) => interval(index * 1000).pipe(
        take(3),
        map(value => `${index}-${value + 1}`)
    )

console.log(`${new Date().toLocaleTimeString()}: exhaustAll start`)
const source$ = timer(0, 2500).pipe(
    take(3)
);
source$.pipe(
    map(v => createSource(v + 1)),
    exhaustAll()
).subscribe({
    next: val => console.log(`${new Date().toLocaleTimeString()}: exhaustAll`, val)
});
```
```console
16:27:05: exhaustAll start
16:27:07: exhaustAll 1-1
16:27:08: exhaustAll 1-2
16:27:09: exhaustAll 1-3
16:27:14: exhaustAll 3-1
16:27:17: exhaustAll 3-2
16:27:20: exhaustAll 3-3
```

![ExhaustAll Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/66slkqzo681v4c1uiqey.jpg)

* **[mergeAll](https://rxjs.dev/api/operators/mergeAll)**

> Converts a higher-order Observable into a first-order Observable which concurrently delivers all values that are emitted on the inner Observables.

```ts
import { interval, timer } from 'rxjs';
import { map, mergeAll, take } from 'rxjs/operators';

const createSource = (index: number) => interval(index * 1000).pipe(
        take(3),
        map(value => `${index}-${value + 1}`),
    )

console.log(`${new Date().toLocaleTimeString()}: mergeAll start`)
const source$ = timer(0, 1000).pipe(
    take(3),
);
source$.pipe(
    map(v => createSource(v + 1)),
    mergeAll()
).subscribe({
    next: val => console.log(`${new Date().toLocaleTimeString()}: mergeAll`, val)
});
```
```console
15:41:46: mergeAll start
15:41:47: mergeAll 1-1
15:41:48: mergeAll 1-2
15:41:49: mergeAll 2-1
15:41:49: mergeAll 1-3
15:41:51: mergeAll 3-1
15:41:51: mergeAll 2-2
15:41:53: mergeAll 2-3
15:41:54: mergeAll 3-2
15:41:57: mergeAll 3-3
```

![MergeAll Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/4q65kbrb8rsedypkjszo.jpg)

* **[switchAll](https://rxjs.dev/api/operators/switchAll)**

> Converts a higher-order Observable into a first-order Observable producing values only from the most recent observable sequence

```ts
import { interval, timer } from 'rxjs';
import { map, switchAll, take } from 'rxjs/operators';

const createSource = (index: number) => interval(index * 1000).pipe(
        take(3),
        map(value => `${index}-${value + 1}`)
    )

console.log(`${new Date().toLocaleTimeString()}: switchAll start`)
const source$ = timer(0, 1000).pipe(
    take(3)
);
source$.pipe(
    map(v => createSource(v + 1)),
    switchAll()
).subscribe({
    next: val => console.log(`${new Date().toLocaleTimeString()}: switchAll`, val)
});
```
```console
15:42:48: switchAll start
15:42:49: switchAll 1-1
15:42:53: switchAll 3-1
15:42:56: switchAll 3-2
15:42:59: switchAll 3-3
```

![SwitchAll Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/phayj8tj7dsg3xzjc9c8.jpg)

* **[startWith](https://rxjs.dev/api/operators/startWith)**

> Returns an observable that, at the moment of subscription, will synchronously emit all values provided to this operator, then subscribe to the source and mirror all of its emissions to subscribers.

```ts
import { of } from 'rxjs';
import { startWith } from 'rxjs/operators';

const source1$ = of(1, 2, 3).pipe(
    startWith(1000)
)

source1$.subscribe({
    next: val => console.log(`${new Date().toLocaleTimeString()}: startWith`, val)
});
```
```console
15:43:47: startWith 1000
15:43:47: startWith 1
15:43:47: startWith 2
15:43:47: startWith 3
```

* **[withLatestFrom](https://rxjs.dev/api/operators/withLatestFrom)**

> Combines the source Observable with other Observables to create an Observable whose values are calculated from the latest values of each, only when the source emits.

```ts

import { interval } from 'rxjs';
import { take, withLatestFrom } from 'rxjs/operators';

const source1$ = interval(1000).pipe(
    take(5)
)
const source2$ = interval(2500).pipe(
    take(5)
)

console.log(`${new Date().toLocaleTimeString()}: withLatestFrom start`)
source1$.pipe(
    withLatestFrom(source2$)
).subscribe({
    next: val => console.log(`${new Date().toLocaleTimeString()}: withLatestFrom`, val)
});
```
```console
15:44:11: withLatestFrom start
15:44:14: withLatestFrom [ 2, 0 ]
15:44:15: withLatestFrom [ 3, 0 ]
15:44:16: withLatestFrom [ 4, 1 ]
```

Ok guys, that's all for today!
You can find the code of this article [in the GitHub repository](https://github.com/Puppo/rxjs-getting-started/tree/08-join-operators)

See you soon!

---

### Part 9 of 15: RxJS - Utility Operators

- **URL**: https://delpuppo.net/blog/rxjs-getting-started/rxjs-utility-operators/
- **Published**: Sep 24, 2021
- **Tags**: javascript, typescript, rxjs

Welcome back Guys,
Today I'll show you some utility operators.
Let's start :)
 
* **[tap](https://rxjs.dev/api/operators/tap)**

> Used to perform side-effects for notifications from the source observable

```ts
import { of } from 'rxjs'
import { tap } from 'rxjs/operators'


of('a', 'b', 'c')
  .pipe(
      tap({
          next: x => console.log(`tap: ${x}`),
          complete: () => console.log('tap: complete'),
          error: err => console.log(`tap: error: ${err}`)
      })
  )
  .subscribe()
```
```console
tap: a
tap: b
tap: c
tap: complete
```
This operator helps us to get some info about the observable during its execution. This operator is equal to the subscriber and it uses three methods to get info: next, complete, and error.
What could happen in these three methods? Exactly what you want :) You can run a side-effect or log the values. Usually I prefer logging info in this operator and not run side effects because at times side effects are difficult to test.

* **[delay](https://rxjs.dev/api/operators/delay)**

> Delays the emission of items from the source Observable by a given timeout or until a given Date.

```ts
import { of } from 'rxjs'
import { delay, tap } from 'rxjs/operators'


of('a', 'b', 'c')
  .pipe(
      tap(x => console.log(`${new Date().toLocaleTimeString()} tap before delay: ${x}`)),
      delay(1000),
      tap(x => console.log(`${new Date().toLocaleTimeString()} tap after delay: ${x}`)),
  )
  .subscribe()
```
```console
17:08:26 tap before delay: a
17:08:26 tap before delay: b
17:08:26 tap before delay: c
17:08:27 tap after delay: a
17:08:27 tap after delay: b
17:08:27 tap after delay: c
```
![delay Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/3a3bbb7mnlaunt47ke9o.jpg)
This operator is used to wait some specific time before emitting the value.

* **[timestamp](https://rxjs.dev/api/operators/timestamp)**

> Attaches a timestamp to each item emitted by an observable indicating when it was emitted

```ts
import { of } from 'rxjs'
import { timestamp } from 'rxjs/operators'

of('a', 'b', 'c')
  .pipe(
      timestamp()
  )
  .subscribe(console.log)
```
```console
{ value: 'a', timestamp: 1629385746523 }
{ value: 'b', timestamp: 1629385746528 }
{ value: 'c', timestamp: 1629385746528 }
```
This operator is used to attach a timestamp to each emitted item. The timestamp indicates the time when the value was emitted.
This operator can be helpful during debugging or if we need to have info about the time emission of the value.

* **[timeout](https://rxjs.dev/api/operators/timeout)**

> Errors if Observable does not emit a value in given time span.

```ts
import { Observable, throwError } from 'rxjs';
import { timeout } from 'rxjs/operators';

const source$ = new Observable<number>(subscriber => {
    let id: NodeJS.Timeout | undefined
    (function next(count = 0) {
        if (count > 10) {
            subscriber.complete();
            return;
        }
        id = setTimeout(() => {
            subscriber.next(count)
            next(count + 1)
        }, Math.random() > 0.9 ? 2000 : 1000);
    })()

    return () => {
        if (id) {
            clearTimeout(id);
        }
    }

})

source$
  .pipe(
      timeout({ each: 1500, with: info => throwError(() => new Error(`Timeout ${1500}ms: info: ${JSON.stringify(info)}`)) }),
  )
  .subscribe({
      next: console.log,
      error: error => {
          console.error(`Something Wrong!`)
          console.error(error.message)
      }
  })
```
```console
0
1
2
3
4
5
6
7
8
Something Wrong!
Timeout 1500ms: info: {"meta":null,"lastValue":null,"seen":9}
```
This operator checks the time of the execution of the observable, if the value is not emitted within the timeout time the operator throws an error.

* **[toArray](https://rxjs.dev/api/operators/toArray)**

> Collects all source emissions and emits them as an array when the source completes.

```ts
import { of } from 'rxjs'
import { toArray } from 'rxjs/operators'


of('a', 'b', 'c', 'd', 'e', 'f', 'g', 'h')
  .pipe(
      toArray()
  )
  .subscribe(console.log)
```
```console
[
  'a', 'b', 'c',
  'd', 'e', 'f',
  'g', 'h'
]
```
![toArray Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/btr18a4jnjfusx3dke3a.jpg)
This operator converts a sequence of values in one array, where all the array's items are the emitted values in sequence.

Ok Guys, from the Utilities Operators is all.
[in the GitHub repository](https://github.com/Puppo/rxjs-getting-started/tree/09-utility-operators) you can find all the code of this article.

See you soon,
Bye Bye!

---

### Part 10 of 15: RxJS - Conditional & Mathematical Operators

- **URL**: https://delpuppo.net/blog/rxjs-getting-started/rxjs-conditional-mathematical-operators/
- **Published**: Sep 27, 2021
- **Tags**: javascript, typescript, rxjs

Hi Guys and welcome back,
today I'll illustrate you two simple types of the pipeable operators: Conditional Operators and Mathematical Operators.
No time to waste, let's start.

## Conditional Operators
These operators are useful to check if there are values in the observables or to find some specific value in them. Some of these operators are similar to some array methods, with the difference that they work with observables and not with arrays.

* **[defaultIfEmpty](https://rxjs.dev/api/operators/defaultIfEmpty)**

> Emits a given value if the source Observable completes without emitting any next value, otherwise mirrors the source Observable.

```ts
import { EMPTY, Observer } from "rxjs";
import { defaultIfEmpty } from "rxjs/operators";

const observer: Observer<number> = {
    next: x => console.log('value', x),
    error: err => console.error('error', err),
    complete: () => console.log('complete'),
};

EMPTY.pipe(
    defaultIfEmpty(10)
).subscribe(observer);
```
```console
value 10
complete
```
![defaultIfEmpty Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/mc6m5h39hc2ico12qsnw.jpg)
This operator, as you can see, permits us to receive a default value if the observable does not emit any value.

* **[every](https://rxjs.dev/api/operators/every)**

> Returns an Observable that emits whether or not every item of the source satisfies the condition specified.

```ts
import { Observer, of } from "rxjs";
import { every } from "rxjs/operators";

const observer: Observer<boolean> = {
    next: x => console.log('value', x),
    error: err => console.error('error', err),
    complete: () => console.log('complete'),
};

of(1,2,3,4,5,6,7,8,9).pipe(
    every(val => val < 10)
).subscribe(observer);
```
```console
value true
complete
```
![every Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/o6u37lq95xx255qhqtkz.jpg)
This operator is like the [every method](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Array/every?retiredLocale=it) in the arrays.
We can use it when we need to check if all the values of our observable meet a condition.

* **[find](https://rxjs.dev/api/operators/find)**

> Emits only the first value emitted by the source Observable that meets some condition.

```ts
import { Observer, of } from "rxjs";
import { find } from "rxjs/operators";

const observer: Observer<number | undefined> = {
    next: x => console.log('value', x),
    error: err => console.error('error', err),
    complete: () => console.log('complete'),
};

of(1,2,3,4,5,6,7,8,9).pipe(
    find(val => val === 5)
).subscribe(observer);
```
```console
value 5
complete
```
![find Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/gs72e66yjrbl0ek5f73n.jpg)
This operator is like the [find method](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Array/find) in the arrays.
We can use it to find a value that meets a condition in our observable. It's important to remember that when the operator finds a value that matches our condition it completes the observable.

* **[findIndex](https://rxjs.dev/api/operators/findIndex)**

> Emits only the index of the first value emitted by the source Observable that meets some condition.

```ts
import { Observer, of } from "rxjs";
import { findIndex } from "rxjs/operators";

const observer: Observer<number> = {
    next: x => console.log('value', x),
    error: err => console.error('error', err),
    complete: () => console.log('complete'),
};

of(1,2,3,4,5,6,7,8,9).pipe(
    findIndex(val => val === 5)
).subscribe(observer);
```
```console
value 4
complete
```
![findIndex Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/0b6ykgw3cc7yjqm3dxnt.jpg)
This operator is like the [findIndex method](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Array/findIndex) in the arrays.
We can use it to find the index of a value that meets a condition in our observable. It's important to remember that when the operator finds a value that matches our condition it completes the observable.

* **[isEmpty](https://rxjs.dev/api/operators/isEmpty)**

> Emits false if the input Observable emits any values, or emits true if the input Observable completes without emitting any values.

```ts
import { EMPTY, Observer } from "rxjs";
import { isEmpty } from "rxjs/operators";

const observer: Observer<boolean> = {
    next: x => console.log('value', x),
    error: err => console.error('error', err),
    complete: () => console.log('complete'),
};

EMPTY.pipe(
    isEmpty()
).subscribe(observer);
```
```console
value true
complete
```
![isEmpty Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/vdmg3vr5zoazs8b06b1k.jpg)
This operator is used to check if an observable has emitted at least a value.


## Mathematical Operators
These operators are used to retrieve some special values in our observables or to reduce the values.

* **[count](https://rxjs.dev/api/operators/count)**

> Counts the number of emissions on the source and emits that number when the source is completed.

```ts
import { Observer, of } from "rxjs";
import { count } from "rxjs/operators";

const observer: Observer<number | undefined> = {
    next: x => console.log('value', x),
    error: err => console.error('error', err),
    complete: () => console.log('complete'),
};

of("a", "b", "c", "d", "e").pipe(
    count()
).subscribe(observer);
```
```console
value 5
complete
```
![count Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/gvscx0007cr4zd70wixz.jpg)
This operator is used to count the values of our observables. 

* **[max](https://rxjs.dev/api/operators/max)**

> The Max operator operates on an Observable that emits numbers (or items that can be compared with a provided function), and when source Observable completes it emits a single item: the item with the largest value.

```ts
import { Observer, of } from "rxjs";
import { max } from "rxjs/operators";

const observer: Observer<string> = {
    next: x => console.log('value', x),
    error: err => console.error('error', err),
    complete: () => console.log('complete'),
};

of("a", "b", "e", "d", "c").pipe(
    max()
).subscribe(observer);
```
```console
value e
complete
```
![max Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/d2s1ckomsrg8m0e04i04.jpg)
This operator is used to get the maximum value emitted by our  observables.

* **[min](https://rxjs.dev/api/operators/min)**

> The Min operator operates on an Observable that emits numbers (or items that can be compared with a provided function), and when source Observable completes it emits a single item: the item with the smallest value.

```ts
import { Observer, of } from "rxjs";
import { min } from "rxjs/operators";

const observer: Observer<string> = {
    next: x => console.log('value', x),
    error: err => console.error('error', err),
    complete: () => console.log('complete'),
};

of("a", "b", "e", "d", "c").pipe(
    min()
).subscribe(observer);
```
```console
value a
complete
```
![min Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/ptc7qfdls3pysimfberg.jpg)
This operator is used to get the minimum value emitted by our  observables.

* **[reduce](https://rxjs.dev/api/operators/reduce)**

> Applies an accumulator function over the source Observable, and returns the accumulated result when the source completes, given an optional seed value.

```ts
import { Observer, of } from "rxjs";
import { reduce } from "rxjs/operators";

const observer: Observer<number> = {
    next: x => console.log('value', x),
    error: err => console.error('error', err),
    complete: () => console.log('complete'),
};

of(1,2,3,4,5,6,7,8,9).pipe(
    reduce((acc, curr) => acc + curr, 0)
).subscribe(observer);
```
```console
value 45
complete
```
![reduce Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/3l39f71fzjh6xmkxlniv.jpg)
This operator is like the [reduce method](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Array/Reduce) of the array.
It can be used to reduce all the emitted values. The reduced algorithm has to be implemented by us.


Ok, That's all for today.
I hope these simple operators can help you in the future as they did me.

See you in the next article!

_You can find the code of this article [in the GitHub repository](https://github.com/Puppo/rxjs-getting-started/tree/11-condition-and-mathematical-operators)._

---

### Part 11 of 15: RxJS - Error Handling Operators

- **URL**: https://delpuppo.net/blog/rxjs-getting-started/rxjs-error-handling-operators/
- **Published**: Sep 29, 2021
- **Tags**: javascript, typescript, rxjs

One of the best practice in our work is handle the errors, so today I'll show you some operators in RxJS to handle the errors.
Let's start :)

* **[catchError](https://rxjs.dev/api/operators/catchError)**

> Catches errors on the observable to be handled by returning a new observable or throwing an error.

```ts
import { of } from 'rxjs';
import { catchError, map } from 'rxjs/operators';
 
of('a', 'b', 'c', 'd', 1, 3).pipe(
    map((n: any) => n.toUpperCase()),
    catchError(err => {
        console.error(err.message);
        return of('A', 'B', 'C', 'D')
    }),
  )
  .subscribe(x => console.log(x));
```
```console
A
B
C
D
n.toUpperCase is not a function
A
B
C
D
```
This operator is used to catch the errors during the execution of the observables.
This operator helps us to handle the errors and prevent a bad flow of our users.

* **[retry](https://rxjs.dev/api/operators/retry)**

```ts
import { of } from 'rxjs';
import { map, retry } from 'rxjs/operators';

function toString(val: { toString: () => string }): string | never {
    console.log('toString of', val);
    if (Math.random() > 0.6)
        return val.toString()

    console.log('toString of', val, 'failed');
    throw new Error('toString failed')
}
 
of(1, 2, 3, 4, 5, 6, 7, 8, 9, 0).pipe(
    map(toString),
    retry(3)
  )
  .subscribe({
      next: x => console.log(x),
      error: error => console.error(error.message),
  });
```
```console
toString of 1
toString of 1 failed
toString of 1
toString of 1 failed
toString of 1
1
toString of 2
toString of 2 failed
toString of 1
toString of 1 failed
toString failed
```
This operator is used to retry the execution of an observable if it raises an error. We can indicate the max number of retries. If during the execution an observable raises a number of errors greater than the indicated value, the retry operator doesn't try anymore but goes out and raise the error.

* **[retryWhen](https://rxjs.dev/api/operators/retryWhen)**

> Returns an Observable that mirrors the source Observable with the exception of an error. If the source Observable calls error, this method will emit the Throwable that caused the error to the Observable returned from notifier. If that Observable calls complete or error then this method will call complete or error on the child subscription. Otherwise this method will resubscribe to the source Observable.

```ts
import { interval, timer } from 'rxjs';
import { delayWhen, filter, map, retryWhen, tap } from 'rxjs/operators';
 
interval(1000).pipe(
  map(val => {
    if (val > 5) {
      throw val;
    }
    return val;
  }),
  retryWhen(errors =>
    errors.pipe(
      tap(val => console.log(`Value ${val} was too high!`)),
      filter((_, index) => index < 3),
      delayWhen(val => timer(val * 1000))
    )
  )
).subscribe({
    next: x => console.log(x),
    error: error => console.error(error.message),
});
```
```console
0
1
2
3
4
5
Value 6 was too high!
0
1
2
3
4
5
Value 6 was too high!
0
1
2
3
4
5
Value 6 was too high!
0
1
2
3
4
5
Value 6 was too high!
```
This operator is used to retry the execution of an observable and it allows us to indicate the strategy of the retry.

I hope you have all the means now to better handle errors in your code.
It's all for today.
You can find all the code of this article in the repository

See you soon,
Bye Bye!

---

### Part 12 of 15: RxJS - Multicast Operator

- **URL**: https://delpuppo.net/blog/rxjs-getting-started/rxjs-multicast-operator/
- **Published**: Oct 1, 2021
- **Tags**: javascript, typescript, rxjs

Hi Folk 👋,
in the previous articles we've seen that when we subscribe to an observable, the observable restarts every time and do not remember the last value emitted.
In some cases, this behaviour can not be the right solution, so today I'll show you how to share the values using the Multicast Operators.

* **[share](https://rxjs.dev/api/operators/share)**

> Returns a new Observable that multicasts (shares) the original Observable. As long as there is at least one Subscriber this Observable will be subscribed and emitting data. When all subscribers have unsubscribed it will unsubscribe from the source Observable. Because the Observable is multicasting it makes the stream hot.

```ts
/**
marble share
{
    source a:           +--0-1-2-3-4-#
    operator share:     {
        +--0-1-2-3-4-#
        ......+2-3-4-#
    }
}
*/
import { interval } from 'rxjs';
import { share, take, tap } from 'rxjs/operators';
 
const source1 = interval(1000)
.pipe(
    take(5),
    tap((x: number) => console.log('Processing: ', x)),
    share()
);

source1.subscribe({
    next: x => console.log('subscription 1: ', x),
    complete: () => console.log('subscription 1 complete'),
});

setTimeout(() => {
    source1.subscribe({
        next: x => console.log('subscription 2: ', x),
        complete: () => console.log('subscription 2 complete'),
    });
}, 3000);


setTimeout(() => {
    source1.subscribe({
        next: x => console.log('subscription 3: ', x),
        complete: () => console.log('subscription 3 complete'),
    });
}, 7000);
```
```console
Processing:  0
subscription 1:  0
Processing:  1
subscription 1:  1
Processing:  2
subscription 1:  2
subscription 2:  2
Processing:  3
subscription 1:  3
subscription 2:  3
Processing:  4
subscription 1:  4
subscription 2:  4
subscription 1 complete
subscription 2 complete
Processing:  0
subscription 3:  0
Processing:  1
subscription 3:  1
Processing:  2
subscription 3:  2
Processing:  3
subscription 3:  3
Processing:  4
subscription 3:  4
subscription 3 complete
```
![share Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/7kq3wgsd7p6daltyihrd.jpg)
This operator can help us when we need to share the value of an observable during its execution. But what does it mean? It means that the first subscription starts the observable and all the next subscriptions that subscribe to this observable do not run a new instance of the observable but they receive the same values of the first subscription, thus losing all the previous values emitted before their subscription.
It's important to remember that when the observable is completed, and another observer subscribes itself to the observable, the shared operator resets the observable and restarts its execution from the beginning.
Anyway, sometimes our code needs to prevent the restarting of our observables, but what can we do in these cases?
It's simple! The share operator exposes us some options: resetOnComplete, resetOnError, resetOnRefCountZero, and each of these options can help us to prevent the resetting of the observables in different cases. These options can work or with a simple boolean value that enables or disables the behaviour, or we can pass a notifier factory that returns an observable which grants more fine-grained control over how and when the reset should happen.
The resetOnComplete option prevents the resetting after the observable's completion. So, if it is enabled when another observer subscribes to an observable already completed this observer receives immediately the complete notification.
The resetOnError option prevents the resetting of the observable after an error notification.
The resetOnRefCountZero option works with the number of observers subscribed instead. It prevents the resetting if there aren't any observer subscribed. To better understand, if all the subscriptions of our observable are unsubscribed, and this option is enabled, the observable isn't reset. otherwise, if this option is disabled, the observable restarts from the beginning at the next subscription.
Here's an example using the resetOnRefCountZero option.

```ts
import { interval, timer } from 'rxjs';
import { share, take } from 'rxjs/operators';
 
const source = interval(1000).pipe(take(3), share({ resetOnRefCountZero: () => timer(1000) }));
 
const subscriptionOne = source.subscribe(x => console.log('subscription 1: ', x));
setTimeout(() => subscriptionOne.unsubscribe(), 1300);

setTimeout(() => source.subscribe(x => console.log('subscription 2: ', x)), 1700);

setTimeout(() => source.subscribe(x => console.log('subscription 3: ', x)), 5000);
```
```console
subscription 1:  0
subscription 2:  1
subscription 2:  2
subscription 3:  0
subscription 3:  1
subscription 3:  2
```
![shared with resetOnRefCountZero option Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/q6ebecgs8rcymrtwz8is.jpg)

* **[shareReplay](https://rxjs.dev/api/operators/shareReplay)**

> Share source and replay specified number of emissions on subscription.

```ts
import { interval } from 'rxjs';
import { shareReplay, take, tap } from 'rxjs/operators';

const obs$ = interval(1000);
const shared$ = obs$.pipe(
  take(4),
  tap(console.log),
  shareReplay(3)
);
shared$.subscribe(x => console.log('sub A: ', x));

setTimeout(() => {
  shared$.subscribe(y => console.log('sub B: ', y));
}, 3500);
```
```console
0
sub A:  0
1
sub A:  1
2
sub A:  2
sub B:  0
sub B:  1
sub B:  2
3
sub A:  3
sub B:  3
```
![shareReplay Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/kud1ghis4m39geeqsuvf.jpg)
In some cases, when we share the values between multiple observers, if an observer subscribes to an already started observable, we also need to replay all the previous already emitted values. To resolve this problem we can use the shareReplay operator.
This operator shares the emitted values and if another observer subscribes to the observable it replays the previous values.
The number of values replayed can be configured: by default all the values already emitted are emitted again, but we can also indicate or a maximum number of elements to remember or a maximum time length.

Ok guys, that's all for today.
If you are interested in trying the code of this article, you can find it [in the GitHub repository](https://github.com/puppo/rxjs-getting-started/tree/12-multicasting-operators).

In the next article, I'll show you how to create your custom operators.

See you soon!

---

### Part 13 of 15: RxJS - Custom Operators

- **URL**: https://delpuppo.net/blog/rxjs-getting-started/rxjs-custom-operators/
- **Published**: Oct 4, 2021
- **Tags**: javascript, typescript, rxjs

Hi Folks 👋
after many articles in which I have illustrated the different types of operators already included in the RxJS library, today I show you how to create your own operators.
You can create a new operator in two ways: mixing the different operators illustrated in the previous articles or from scratch.
In common cases, when we need to create a new operator, all the behaviours that we need already exist in the operators exposed by the RxJS library. In these cases, we can use the "pipe" function to combine together the operators. With the "pipe" function you can create a new function that will be the new operator, and inside this function you can create the logic of your operator.
Here's an example
```ts
import { delay, interval, Observable, pipe, take, tap, UnaryFunction } from 'rxjs';

function takeLogAndDelay<T>(takeNumber: number, message: string, time: number): UnaryFunction<Observable<T>, Observable<T>> {
  return pipe(
    tap(x => console.log(message, x)),
    take(takeNumber),
    delay(time),
  );
}

interval(1000).pipe(
    takeLogAndDelay(10, 'Source', 2000)
).subscribe();
```
```console
Source 0
Source 1
Source 2
Source 3
Source 4
Source 5
Source 6
Source 7
Source 8
Source 9
```

As you can see, in the last example I created a new operator called "takeLogAndDelay". This operator combines together three operators: tap, take and delay.
As you can see, it's easy to create a new operator of this type and you just need to flow these rules:
- create a new function with the name of the new operator
- return the pipe function
- combine the operators in the pipe function

In 95% of your cases, you can resolve your problem and create a new operator using this method, while in the 5% of the cases not managed by this method you can create a new operator from scratch.
Below an example of the delay operator created from scratch.
```ts
import { Observable, of } from 'rxjs';
import { tap } from 'rxjs/operators';

function delay<T>(delayInMs: number) {
  return (observable: Observable<T>) =>
    new Observable<T>((subscriber) => {
      const allTimerIDs = new Set<NodeJS.Timeout>();
      let hasCompleted = false;
      const subscription = observable.subscribe({
        next(value) {
          const timerID = setTimeout(() => {
            subscriber.next(value);
            allTimerIDs.delete(timerID);
            if (hasCompleted && allTimerIDs.size === 0) {
              subscriber.complete();
            }
          }, delayInMs);

          allTimerIDs.add(timerID);
        },
        error: subscriber.error,
        complete() {
          hasCompleted = true;
          if (allTimerIDs.size === 0) {
            subscriber.complete();
          }
        },
      });

      return () => {
        subscription.unsubscribe();
        allTimerIDs.forEach((timerID) => clearTimeout(timerID));
      };
    });
}

of(1, 2, 3).pipe(
    tap((value) => console.log(new Date().toLocaleTimeString(), "before", value)),
    delay(3000)
).subscribe({
    next: (value) => console.log(new Date().toLocaleTimeString(), "after", value),
});
```
```console
22:11:01 before 1
22:11:01 before 2
22:11:01 before 3
22:11:04 after 1
22:11:04 after 2
22:11:04 after 3
```
As you can see, we can create a new operator and manage all the behaviours needed for our logic using the Observable constructor. It's rare to use this approach but if you find yourself in front of a case not yet solved by the existing operators, you can rely on it.

Ok guys, that's all for today.

See you soon
Bye bye 👋

---

### Part 14 of 15: RxJS - Subjects

- **URL**: https://delpuppo.net/blog/rxjs-getting-started/rxjs-subjects/
- **Published**: Oct 6, 2021
- **Tags**: javascript, typescript, rxjs

What's a Subject Luca? And why today do you want to speak about subjects? Is there a mistake in the publishing? Or are you simply drunk?
![Jack Sparrow Meme](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/hnf6ewbhahoqn1e0jpsv.jpg)

No guys, don't worry, It's the topic of the day :)
I think that after this long trip in the RxJS world and after Observables, Observers, Subscriptions, and a number of Operators that I don't even remember, it's time to show you the Subjects, a powerful type of Observable, but not only.
Let's start from the definition:
> A Subject is like an Observable, but can multicast to many Observers. Subjects are like EventEmitters: they maintain a registry of many listeners.

![Balto meme](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/t15rdmpxvomkdyscj2bd.jpg)
Ok, I'm joking :)
But let's continue with two notes present in the official documentation.
- **Every Subject is an Observable**
Given a Subject, you can subscribe to it, providing an Observer, which will start receiving values normally. From the perspective of the Observer, it cannot tell whether the Observable execution is coming from a plain unicast Observable or a Subject.
- **Every Subject is an Observer**
It is an object with the methods next(v), error(e), and complete(). To feed a new value to the Subject, just call next(theValue), and it will be multicasted to the Observers registered to listen to the Subject.

Ok, but what does it means?
It means that you can create a Subject, subscribe to it with infinitive observers, multicast the values with great performance, emit new values using a simple method called `next`, emit errors using a method called `error` and complete your subject with a simple method called `complete`.
The icing on that cake is that all these things can be done with a clean code and they are not enclosed into the body of a function as on the Observable constructor.

But let me show you an example.
```ts
import { Subject } from 'rxjs';
 
const subject = new Subject<number>();
 
subject.subscribe({
  next: (v) => console.log(new Date().toLocaleTimeString(), `observerA: ${v}`),
  complete: () => console.log(new Date().toLocaleTimeString(), 'observerA: complete')
});
 
subject.next(1);

setTimeout(() => {
  subject.subscribe({
  next: (v) => console.log(new Date().toLocaleTimeString(), `observerB: ${v}`),
  complete: () => console.log(new Date().toLocaleTimeString(), 'observerB: complete')
});
}, 2000);

setTimeout(() => {
  subject.next(2);
}, 3000);

setTimeout(() => {
  subject.complete();
}, 4000);
```
```console
18:52:23 observerA: 1
18:52:26 observerA: 2
18:52:26 observerB: 2
18:52:27 observerA: complete
18:52:27 observerB: complete
```
![Subject Example](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/sqouh2wvh9x13wis2u57.jpg)

As you can see, with a Subject we can emit new values, subscribe to its changes, and complete the subject.
I recommend you to try this special observable because it can be powerful in many cases.

The RxJS library does not stop its power here. It exposes us four different types of subjects: Subject, BehaviorSubject, ReplaySubject and AsyncSubject.
These four types have 4 different behaviour that can help us to resolve different cases. Let me show you these differences.

* **[subject](https://rxjs.dev/guide/subject#subject)**

> A Subject is like an Observable, but can multicast to many Observers. Subjects are like EventEmitters: they maintain a registry of many listeners.

```ts
import { Subject } from 'rxjs';
 
const subject = new Subject<number>();
 
subject.subscribe({
  next: (v) => console.log(new Date().toLocaleTimeString(), `observerA: ${v}`),
  complete: () => console.log(new Date().toLocaleTimeString(), 'observerA: complete')
});
 
subject.next(1);

setTimeout(() => {
    subject.subscribe({
        next: (v) => console.log(new Date().toLocaleTimeString(), `observerB: ${v}`),
        complete: () => console.log(new Date().toLocaleTimeString(), 'observerB: complete')
    });
}, 1000);

setTimeout(() => {
  subject.next(2);
}, 3000);

setTimeout(() => {
  subject.complete();
}, 4000);
```
```console
19:14:41 observerA: 1
19:14:44 observerA: 2
19:14:44 observerB: 2
19:14:45 observerA: complete
19:14:45 observerB: complete
```
![Subject Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/jxl0aj6oe8db0h1taph2.jpg)

This subject can be used when we need to emit values and notify some of these. 
This subject doesn't have an initial value and if an observer subscribes to this type of subject it receives only the values emitted after its subscription.

_N.B. In some cases, we don't need to send a value when we call the next method of the subject, but we just need to call the method to notify some of these. In these cases, we can use the [Void Subject](https://rxjs.dev/guide/subject#void-subject), a special subject that does not need a value._


* **[behaviorsubject](https://rxjs.dev/guide/subject#behaviorsubject)**

> BehaviorSubjects are useful for representing "values over time". For instance, an event stream of birthdays is a Subject, but the stream of a person's age would be a BehaviorSubject.

```ts
import { BehaviorSubject } from 'rxjs';
 
const subject = new BehaviorSubject<number>(0);
 
subject.subscribe({
  next: (v) => console.log(new Date().toLocaleTimeString(), `observerA: ${v}`),
  complete: () => console.log(new Date().toLocaleTimeString(), 'observerA: complete')
});
 
subject.next(1);

setTimeout(() => {
    subject.subscribe({
        next: (v) => console.log(new Date().toLocaleTimeString(), `observerB: ${v}`),
        complete: () => console.log(new Date().toLocaleTimeString(), 'observerB: complete')
    });
}, 1000);

setTimeout(() => {
  subject.next(2);
}, 3000);

setTimeout(() => {
  subject.complete();
}, 4000);
```
```console
19:15:57 observerA: 0
19:15:57 observerA: 1
19:15:58 observerB: 1
19:16:00 observerA: 2
19:16:00 observerB: 2
19:16:01 observerA: complete
19:16:01 observerB: complete
```
![BehaviorSubject Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/579vwh9m6d6xwwpxzch8.jpg)
The BehaviorSubject is a subject that requires an initial value. When an observer subscribes to a BehaviorSubject it immediately receives the last emitted value and then it waits for the future values.

* **[replaysubject](https://rxjs.dev/guide/subject#replaysubject)**

> A ReplaySubject records multiple values from the Observable execution and replays them to new subscribers.

```ts
import { ReplaySubject } from 'rxjs';
 
const subject = new ReplaySubject<number>();
 
subject.subscribe({
  next: (v) => console.log(new Date().toLocaleTimeString(), `observerA: ${v}`),
  complete: () => console.log(new Date().toLocaleTimeString(), 'observerA: complete')
});
 
subject.next(1);

setTimeout(() => {
  subject.next(2);
}, 1000);

setTimeout(() => {
    subject.subscribe({
        next: (v) => console.log(new Date().toLocaleTimeString(), `observerB: ${v}`),
        complete: () => console.log(new Date().toLocaleTimeString(), 'observerB: complete')
    });
}, 2000);

setTimeout(() => {
  subject.next(3);
}, 3000);

setTimeout(() => {
  subject.complete();
}, 4000);
```
```console
19:17:37 observerA: 1
19:17:38 observerA: 2
19:17:39 observerB: 1
19:17:39 observerB: 2
19:17:40 observerA: 3
19:17:40 observerB: 3
19:17:41 observerA: complete
19:17:41 observerB: complete
```
![ReplaySubject Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/t6yfl4usnju1kusvvizk.jpg)
The ReplaySubject is a subject that doesn't require an initial value but when an observer subscribes to it, it receives immediately all the values already emitted and after that, it waits for the future values.

* **[asyncsubject](https://rxjs.dev/guide/subject#asyncsubject)**

> The AsyncSubject is a variant where only the last value of the Observable execution is sent to its observers, and only when the execution completes.

```ts
import { AsyncSubject } from 'rxjs';
 
const subject = new AsyncSubject<number>();
 
subject.subscribe({
  next: (v) => console.log(new Date().toLocaleTimeString(), `observerA: ${v}`),
  complete: () => console.log(new Date().toLocaleTimeString(), 'observerA: complete')
});
 
subject.next(1);

setTimeout(() => {
    subject.subscribe({
        next: (v) => console.log(new Date().toLocaleTimeString(), `observerB: ${v}`),
        complete: () => console.log(new Date().toLocaleTimeString(), 'observerB: complete')
    });
}, 1000);

setTimeout(() => {
  subject.next(2);
}, 3000);

setTimeout(() => {
  subject.complete();
}, 4000);
```
```console
19:19:01 observerA: 2
19:19:01 observerB: 2
19:19:01 observerA: complete
19:19:01 observerB: complete
```
![AsyncSubject Marble Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/7t3ybrtpia5gcppgru40.jpg)
The AsyncSubject is a subject that does not emit values ​​until it is completed, at the time of its completion it emits the last value executed.


These are the differences between these subjects, it's important to remember these differences because if you choose the wrong subject you could do unexpected mistakes.

Before ending the article let me show you a best practice to expose our subject to the consumers. If you need to hide to your consumers that your Observable is a Subject, you need to expose the subject using its "asObservable" method. This method returns the subject as an Observable, so the consumer can't call the "next" and "complete" methods.

Ok guys, that's all for today, I recommend you to play with these subject to understand better their power and characteristics.

See you soon
Bye Bye

---

### Part 15 of 15: RxJS - Marble Testing

- **URL**: https://delpuppo.net/blog/rxjs-getting-started/rxjs-marble-testing/
- **Published**: Oct 8, 2021
- **Tags**: javascript, typescript, rxjs

Hello Folks 👋!
Today it's time to speak about testing and how to test the code when we use RxJS.
Before moving our focus to the testing, I need to explain two different types of Observables: cold and hot.

## Hot and Cold
When you create an Observable, you can create a hot or a cold observable. This characteristic depends on where the producer of this Observable is set; when the producer is created by the Observable, the observable is "cold", instead when the producer is created out from the observable it is defined "hot".
Let me shed some light on all these words.
**What's a producer?**
The producer is the source of the values of our observable.
**Cold Observables**
Cold Observables are functions that create the producer and manage it for all its life. The cold observable is in charge of these things:
1. create the producer
2. activate the producer
3. start listening to the producer
4. unicast
5. close the producer

Here's an example:
```ts
const source = new Observable((observer) => {
  const socket = new WebSocket('ws://someurl');
  socket.addEventListener('message', (e) => observer.next(e));
  return () => socket.close();
});
```
**Hot Observables**
An observable is “hot” if its underlying producer is either created or activated outside of subscription.
1. shares a reference to a producer
2. starts listening to the producer
3. multicast (usually)

Here's an example: 
```ts
const socket = new WebSocket('ws://someurl');
const source = new Observable((observer) => {
  socket.addEventListener('message', (e) => observer.next(e));
});
```

Ok, I think that now you have an idea of the difference between these two types and I can move to the topic of the day, but if you are interested in deepen this argument [in this reference](https://benlesh.medium.com/hot-vs-cold-observables-f8094ed53339) is a post about Hot and Cold written by [Ben Lesh](https://twitter.com/BenLesh).

## Marble Testing
To test our code in RxJS we use Marble testing. A method that combines the [Marble Diagram](https://dev.to/this-is-learning/rxjs-marble-diagrams-4jmg) with the code, and allows us to represent the behaviours of our observables and translate them to something that the Testing Framework can understand.

### Marble Syntax
As you can imagine, the marble testing has its own syntax to represent the behaviours and here you can find its rules:
- ` ` whitespace: horizontal whitespace is ignored, and can be used to help vertically align multiple marble diagrams.
- `-` frame: 1 "frame" of virtual time passing (see above description of frames).
- `[0-9]+[ms|s|m]` time progression: the time progression syntax lets you progress virtual time by a specific amount. It's a number, followed by a time unit of ms (milliseconds), s (seconds), or m (minutes) without any space between them, e.g. a 10ms b.
- `|` complete: The successful completion of an observable. This is the observable producer signaling complete().
- `#` error: An error terminating the observable. This is the observable producer signaling error().
- `[a-z0-9]` e.g. 'a' any alphanumeric character: Represents a value being emitted by the producer signaling next().
- `()` sync groupings: When multiple events need to be in the same frame synchronously, parentheses are used to group those events.
- `^` subscription point: (hot observables only) shows the point at which the tested observables will be subscribed to the hot observable. This is the "zero frame" for that observable, every frame before the ^ will be negative. Negative time might seem pointless, but there are in fact advanced cases where this is necessary, usually involving ReplaySubjects.

### [TestScheduler](https://rxjs.dev/api/testing/TestScheduler)
The TestScheduler is the magician that translates the Marble Syntax to something that the Test Framework can understand. It's important to remember that we cannot use the TestScheduler to test our code all the time. For example, if the code consumes a Promise we cannot use it but we should use a traditional method.
The TestScheduler exposes us some APIs that we can use to write our tests but we start from its initialization.
```ts
import { TestScheduler } from 'rxjs/testing';

describe('Marble Testing', () => {
    let testScheduler: TestScheduler;

    beforeEach(() => {
        testScheduler = new TestScheduler((actual, expected) => {
            expect(actual).toEqual(expected);
        });
    });
});
```
_N.B. all the examples use [jest](https://jestjs.io/) as Testing Framework_

As you can see, it's important to create a new TestScheduler for every test, this allows us to have a new instance for every test and create a clean case. But the weird thing in this code is the body of the code passing to the TestScheduler. This code is particular if you are confident with any test framework because in this function we have already indicated the expectations of the test, but we haven't written one yet. This, because the TestScheduler exposes some helpers to test the code and these helpers call the function indicated in the constructor of the  TestScheduler to check the failure or the success of the test.

But let's see a concrete example:
```ts
import { TestScheduler } from 'rxjs/testing';

describe('Marble Testing', () => {
    let testScheduler: TestScheduler;

    beforeEach(() => {
        testScheduler = new TestScheduler((actual, expected) => {
            expect(actual).toEqual(expected);
        });
    });
    
    it('test', () => {
        testScheduler.run((helpers) => {
            const { cold, expectObservable } = helpers;
            const source$ = cold('-a-b-c|');
            const expected = '-a-b-c|';
        
            expectObservable(source$).toBe(expected);
        });
    });

});
```
In this example, we created a cold Observable that emits 3 values: a, b and c. Using the expectObservable helper we can test our observable by comparing it with the expectation passed to the toBe method.
Now let's add some dynamism to our tests and see how to pass values to our observable inside of the tests.
```ts
import { TestScheduler } from 'rxjs/testing';

describe('Marble Testing', () => {
    let testScheduler: TestScheduler;

    beforeEach(() => {
        testScheduler = new TestScheduler((actual, expected) => {
            expect(actual).toEqual(expected);
        });
    });

    it('test with values', () => {
        testScheduler.run((helpers) => {
            const { cold, expectObservable } = helpers;
            const source$ = cold('-a-b-c|', { a: 1, b: 2, c: 3 });
            const expected = '-a-b-c|';
        
            expectObservable(source$).toBe(expected, { a: 1, b: 2, c: 3 });
        });
    });
});
```
In this example, you can see that we can pass another argument to the cold function. This argument is an object where the fields are the correspondents of the value passed in the marble string, so if you use a, b, and c in the marble string you have to use a, b, and c as fields of your argument. The values of these fields are the values used by the test and emitted by the observable. The same goes for the toBe method, it accepts another argument where we can pass the expected result values.
Another important concept when you test your observables is the time, in these cases, it's possible to specify after how much time an observable emits a value or after how much time a value is expected. Here, an example using the `concatMap` operator combined with the `delay` operator that delays the result by 100ms.
```ts
import { concatMap, delay, of } from 'rxjs';
import { TestScheduler } from 'rxjs/testing';

describe('Marble Testing', () => {
    let testScheduler: TestScheduler;

    beforeEach(() => {
        testScheduler = new TestScheduler((actual, expected) => {
            expect(actual).toEqual(expected);
        });
    });
    
    it('test', () => {
        testScheduler.run((helpers) => {
            const { cold, expectObservable } = helpers;
            const source$ = cold('-a-b-c|');
            const final$ = source$.pipe(concatMap(val => of(val).pipe(delay(100))));
            const expected = '- 100ms a 99ms b 99ms (c|)';
            expectObservable(final$).toBe(expected);
        });
    });

});
```
In the previous examples I showed you how to test cold Observables, but you can test hot observables too.
One of the important concepts, when you create a hot Observable, is that you can indicate when the observables are subscribed, and you can do that using the `^` character.
When you indicate the subscription you might remember that your results start from the subscription and the values emitted before are ignored by the test. You can see an example below.
```ts
import { TestScheduler } from 'rxjs/testing';

describe('Marble Testing', () => {
    let testScheduler: TestScheduler;

    beforeEach(() => {
        testScheduler = new TestScheduler((actual, expected) => {
            expect(actual).toEqual(expected);
        });
    });
    
    it('test', () => {
        testScheduler.run((helpers) => {
            const { hot, expectObservable } = helpers;
            const source$ = hot('-a-b-^-c|');
            const expected = '--c|';
            expectObservable(source$).toBe(expected);
        });
    });

});
```
The last test case that I'll show you today is how to test the subscriptions. Sometimes could be necessary to test when an observable is subscribed and for how much time.
Let me use an example: we have two observables combined together using a concat operator, in this case, we need to test if the first observable is subscribed and when it is completed we need to check if the second observable is subscribed.
While you are before these cases, you need to use the `expectSubscriptions` helper. This helper allows you to check the subscriptions of an observable and detects when the observable is subscribed and when is unsubscribed.
Here you can find the example explained above.
```ts
import { concat } from 'rxjs';
import { TestScheduler } from 'rxjs/testing';

describe('Marble Testing', () => {
    let testScheduler: TestScheduler;

    beforeEach(() => {
        testScheduler = new TestScheduler((actual, expected) => {
            expect(actual).toEqual(expected);
        });
    });
    
    it('test subscriptions', () => {
        testScheduler.run((helpers) => {
            const { cold, expectObservable, expectSubscriptions } = helpers;
            const source1$ = cold('-a-b-c|');
            const source2$ = cold('-d-e-f|');
            const final$ = concat(source1$, source2$);

            const expected = '-a-b-c-d-e-f|';
            const expectedSubscriptionsSource1 = '^-----!';
            const expectedSubscriptionsSource2 = '------^-----!';

            expectObservable(final$).toBe(expected);
            expectSubscriptions(source1$.subscriptions).toBe(expectedSubscriptionsSource1);
            expectSubscriptions(source2$.subscriptions).toBe(expectedSubscriptionsSource2);
        });
    });

});
```

Ok guys, I think that's enough for today. [in the GitHub repository](https://github.com/Puppo/rxjs-getting-started/tree/15-testing) you can find the code of this article.
But before I let you go, I have bad news unfortunately, this is the last article of the series, so today there will be some special regards.
I hope you enjoyed this series and you understood better the RxJS world.
Thanks for having read the series and if you have any doubts don't hesitate to contact me, I will be happy to help you.

I will be back soon with new contents, see you soon guys, stay subscribed, and bye-bye.

---