DEV Community: Barış BALLI

RxJS 7 - Subjects

Barış BALLI — Thu, 06 Apr 2023 08:11:22 +0000

Here are my detailed notes from the fantastic course created by Jurek Wozniak, called RxJS 7 and Observables: Introduction.

In order to understand the concept please consider buying the course yourself, these notes are simply subsidiaries

Subjects

Subjects allow us to multicast notifications to multiple observers, because of it is using a shared resource, it is a hot observable.

Subjects work like event emitters, they emit events and all the observers that are subscribed to it, react to those multicasted events.

Subjects has 2 natures: observable and observer. They can be subscribe-able like normal observables. But at the same time can be used .next() .error() and .complete() like an observer.

Subjects in Action

import { fromEvent, Subject } from 'rxjs';
import { map } from 'rxjs/operators';

const emitButton = document.querySelector('button#emit');
const inputElement: HTMLInputElement = document.querySelector('#value-input');
const subscribeButton = document.querySelector('button#subscribe');

const value$ = new Subject<string>();

fromEvent(emitButton, 'click').subscribe(() => value$.next(inputElement.value));

fromEvent(subscribeButton, 'click').subscribe(() => {
  console.log('New Subscription');
  value$.subscribe((value) => console.log(value));
});

After we subscribe we could emit to multiple observers.

We could also emit a value directly by passing the subject, there is no need to specifically call next in subject.

fromEvent(emitButton, 'click')
  .pipe(map(() => inputElement.value))
  .subscribe(value$);

BehaviourSubject - Concept

Let’s say we have a subject and 2 observers,

Subject is emitting match scores and 2 observers since they followed the match since the beginning know the score (2:1).

Let’s say we have one more observer that has joined, the new observer will not have an opportunity to know the score until a new notification is emitted.

In order to solve this there is another subject type called BehaviourSubject, this subject emits the latest notification to the newly joined observers, when they join.

import { BehaviorSubject, fromEvent, Subject } from "rxjs";
import { withLatestFrom } from "rxjs/operators";

const loggedInSpan: HTMLElement = document.querySelector('span#logged-in');
const loginButton: HTMLElement = document.querySelector('button#login');
const logoutButton: HTMLElement = document.querySelector('button#logout');
const printStateButton: HTMLElement = document.querySelector('button#print-state');

const isLoggedIn$ = new BehaviorSubject<boolean>(false);

fromEvent(loginButton, 'click').subscribe(() => isLoggedIn$.next(true));
fromEvent(logoutButton, 'click').subscribe(() => isLoggedIn$.next(false));

// Navigation bar
isLoggedIn$.subscribe(
  isLoggedIn => loggedInSpan.innerText = isLoggedIn.toString()
);

// Buttons
isLoggedIn$.subscribe(isLoggedIn => {
  logoutButton.style.display = isLoggedIn ? 'block' : 'none';
  loginButton.style.display = !isLoggedIn ? 'block' : 'none';
});

fromEvent(printStateButton, 'click').pipe(
  withLatestFrom(isLoggedIn$)
).subscribe(
  ([event, isLoggedIn]) => console.log('User is logged in:', isLoggedIn)
);

We see that for the BehaviourSubjects we need to provide an initial value

Let's keep in touch

Hey, thanks a lot if you read it so far.
I want you to keep in touch for more sweet content.
Please subscibe in Dev.to and other channels (🦸‍♂️ especially twitter)

Twitter 🐦 https://twitter.com/barisbll_dev
Linkedin 📎 https://www.linkedin.com/in/barisbll-dev/
Github 👨🏻‍💻 https://github.com/barisbll
Medium 📰 https://medium.com/@barisbll

RxJS 7 - Pipeable Operators

Barış BALLI — Wed, 29 Mar 2023 08:24:02 +0000

Here are my detailed notes from the fantastic course created by Jurek Wozniak, called RxJS 7 and Observables: Introduction.

In order to understand the concept please consider buying the course yourself, these notes are simply subsidiaries

Pipeable Operators

Operator Stacking

Pipe operators make some operations before an event actually reaches to an observer.

[Event Source → operator 1 → operator 2 → operator 3 → …] → observer

Important thing here is observer only have access the last value after all those operators.

This works just like javascript array utility functions.

Note: Applying a pipeable operator creates a new observable with some additional logic, it doesn’t modify the existing observable.

filter

When a value passed to the filter function it will either pass to (other operators or observable) or it will be discarded.

Even though filter function filters notifications emitted by the next, it always passes error and complete

We can call pipe method from any observable and then we can call any operator we want.

import { Observable } from 'rxjs';
import { filter } from 'rxjs/operators';

interface NewsItem {
  category: 'Bussiness' | 'Sports';
  content: string;
}

const newsFeed$ = new Observable<NewsItem>((subscriber) => {
  setTimeout(() => {
    subscriber.next({ category: 'Bussiness', content: 'A' });
  }, 1000);
  setTimeout(() => {
    subscriber.next({ category: 'Sports', content: 'B' });
  }, 3000);
  setTimeout(() => {
    subscriber.next({ category: 'Bussiness', content: 'C' });
  }, 4000);
  setTimeout(() => {
    subscriber.next({ category: 'Sports', content: 'D' });
  }, 5000);
  setTimeout(() => {
    subscriber.next({ category: 'Bussiness', content: 'E' });
  }, 6000);
});

newsFeed$
  .pipe(filter((item) => item.category === 'Sports'))
  .subscribe((item) => console.log(item));

If we want we can extract only sport feed and use it in anywhere we want

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

interface NewsItem {
  category: 'Business' | 'Sports';
  content: string;
}

const newsFeed$ = new Observable<NewsItem>(subscriber => {
  setTimeout(() => 
    subscriber.next({ category: 'Business', content: 'A' }), 1000);
  setTimeout(() => 
    subscriber.next({ category: 'Sports', content: 'B' }), 3000);
  setTimeout(() => 
    subscriber.next({ category: 'Business', content: 'C' }), 4000);
  setTimeout(() => 
    subscriber.next({ category: 'Sports', content: 'D' }), 6000);
  setTimeout(() => 
    subscriber.next({ category: 'Business', content: 'E' }), 7000);
});

const sportsNewsFeed$ = newsFeed$.pipe(
  filter(item => item.category === 'Sports')
);

newsFeed$.subscribe(
  item => console.log(item)
);

map

map function works really similar to js map function

import { forkJoin } from "rxjs";
// Mike is from New Delhi and likes to eat pasta.

import { ajax } from "rxjs/ajax";
import { map } from "rxjs/operators";

const randomFirstName$ = ajax<any>('https://random-data-api.com/api/name/random_name').pipe(
  map(ajaxResponse => ajaxResponse.response.first_name)
);

const randomCapital$ = ajax<any>('https://random-data-api.com/api/nation/random_nation').pipe(
  map(ajaxResponse => ajaxResponse.response.capital)
);

const randomDish$ = ajax<any>('https://random-data-api.com/api/food/random_food').pipe(
  map(ajaxResponse => ajaxResponse.response.dish)
);

forkJoin([randomFirstName$, randomCapital$, randomDish$]).subscribe(
  ([firstName, capital, dish]) =>
    console.log(`${firstName} is from ${capital} and likes to eat ${dish}.`)
);

tap

tab operator works like a spy and allow us to make some side effect without interacting with the notifications.

Because of in map, the value we return becomes the input of the next chained function sometimes for debug purposes for example we can pass a tap() function in the middle and spy what is happening

import { of } from "rxjs";
import { filter, map, tap } from "rxjs/operators";

of(1, 7, 3, 6, 2).pipe(
  filter(value => value > 5),
  map(value => value * 2),
  tap({
    next: value => console.log('Spy:', value)
  }),
).subscribe(value => console.log('Output:', value));

But of couse if we would use map function with curly brances and return values with return keyword, just before returning them we could also console log.

But keep in mid that tap operator is more than just a console log, it is there for making any side effects therefore it is quite a useful tool.

One more thing…

Starting from RxJS 7.3.0, the tap() operator can do even more. You can see when the Subscription starts and ends at the level of the tap() operator.

On top of the ‘next’, ‘error’ and ‘complete’ handlers which we have discussed above, the tap() operator introduces three more handlers:

subscribe — called when a new Subscription is made,
unsubscribe — called when the Subscription is closed by unsubscribing (calling unsubscribe); this will NOT be executed when ‘error’ or ‘complete’ is emitted,
finalize — called when the Subscription ends, no matter what the cause: it can be either unsubscribing, an error or a complete notification; all will cause the ‘finalize’ handler to be called.

...
  tap({
    subscribe: () => { console.log('New Subscription'); },
    unsubscribe: () => { console.log('Unsubscribed'); },
    finalize: () => { console.log('Subscription ended'); }
  }),
...

debounceTime

debounceTime delays the emitted notifications and only emits the lastly emitted value after a certain threshold.

For example we have multiple notifications coming let’s say 5 notifications per second and we are only interested in with the last value. We can have such a scenario when we listen for input field, the user writes fastly and when the user stops for a bit we finally emit the value and work on it.

In this way we can avoid the excesive processing.

import { fromEvent } from 'rxjs';
import { map, debounceTime } from 'rxjs/operators';

const sliderInput = document.querySelector('input#slider');

fromEvent<any>(sliderInput, 'input')
  .pipe(
    debounceTime(500),
    map((event) => event.target['value'])
  )
  .subscribe((value) => console.log(value));

In this example instead of emitting thousands of notifications it just emits 3, which is an amazing performance gain.

catchError

catchError function provides a fallback observable in case of error. It automatically pass next() and complete() functions it doesn’t interest with them.

If an error notification occurs instead of passing it to the next chained function it passes it to an observable.

And this observable maybe directly emits next() and complete() notifications, then those notifications are passed to the next chained functions.

import { Observable, of } from 'rxjs';
import { catchError } from 'rxjs/operators';

const failingHttpRequest$ = new Observable((subscriber) => {
  setTimeout(() => {
    subscriber.error(new Error('Timeout'));
  }, 3000);

  return () => console.log('Teardown');
});

console.log('App started');

failingHttpRequest$
  .pipe(
    catchError((error) =>
      of(`Fallback value and err message: ${error.message}`)
    )
  )
  .subscribe({
    next: (value) => console.log(value),
    error: (err) => console.log(err.message),
    complete: () => console.log('Completed'),
  });

We see that error didn’t pass to the other chained functions instead, we just create a new observable with *of* and passed the next() function also the error message.

We can also see that after of function sends a completed notification it also emits a completed notification to our main pipe.

Sometimes we may want to hide the error message or not want to do anything when error occurs, for those cases rxjs has an EMPTY observable, this observable as the name suggests is empty and just emits complete() notification.

import { EMPTY, Observable, of } from "rxjs";
import { catchError } from "rxjs/operators";

const failingHttpRequest$ = new Observable(subscriber => {
  setTimeout(() => {
    subscriber.error(new Error('Timeout'));
  }, 3000);
});

console.log('App started');

failingHttpRequest$.pipe(
  catchError(error => EMPTY)
).subscribe({
  next: value => console.log(value),
  complete: () => console.log('Completed')
});

Flattening Operators

flattening operators work like catch error but for next notification.

Flattening operator react to a next notification by subscribing a new observable, as long as source emit new values flattening operator will keep subscribing to new observables.

Flattening Operators Static Example

In the flattening operators first source emit a value and with that value a new observable is getting created, and it also starts to emit new values, then if we have a complete notification coming from the newly created observable, they are not emitted to the main notifications.

let’s make an example with concatMap

import { Observable, of } from 'rxjs';
import { concatMap } from 'rxjs/operators';

const source$ = new Observable((subscriber) => {
  setTimeout(() => subscriber.next('A'), 2000);
  setTimeout(() => subscriber.next('B'), 3000);
});

console.log('App has started');
source$
  .pipe(concatMap((value) => of(1, 2)))
  .subscribe((value) => console.log(value));

As we can see we totally mapped the values ‘A’ and ‘B’ to those observables.

Even though for the static numbers, it is not a really efficient use case, if we can handle creation of hot observables etc. These flattening operators can be quite useful.

Flattening Operators - Dynamic HTTP Request

Let’s this time make a ***************concatMap*************** example that will send an http request with the value provided

import { fromEvent } from "rxjs";
import { ajax } from "rxjs/ajax";
import { concatMap, map } from "rxjs/operators";

const endpointInput: HTMLInputElement = document.querySelector('input#endpoint');
const fetchButton = document.querySelector('button#fetch');

fromEvent(fetchButton, 'click').pipe(
  map(() => endpointInput.value),
  concatMap(value =>
    ajax(`https://random-data-api.com/api/${value}/random_${value}`)
  )
).subscribe(
  value => console.log(value)
);

Flattening Operators - Error Handling #1

Even though the complete notifications are not passed to the main stream, the error notifications does!

So we need to handle tem carefully in order to not kill our main pipeline

import { EMPTY, fromEvent } from "rxjs";
import { ajax } from "rxjs/ajax";
import { catchError, concatMap, map } from "rxjs/operators";

const endpointInput: HTMLInputElement = document.querySelector('input#endpoint');
const fetchButton = document.querySelector('button#fetch');

fromEvent(fetchButton, 'click').pipe(
  map(() => endpointInput.value),
  concatMap(value =>
    ajax(`https://random-data-api.com/api/${value}/random_${value}`)
  ),
  catchError(() => EMPTY)
).subscribe({
  next: value => console.log(value),
  error: err => console.log('Error:', err),
  complete: () => console.log('Completed')
});

We have a problem with this approach, even though we catch the error, we catch it in the main pipeline and convert it to a complete notification this also kills the main pipeline and we cannot keep searching.

Flattening Operators - Error Handling #2

In order to solve the above problem we need to chain catchError to the internal observable

import { EMPTY, fromEvent, of } from 'rxjs';
import { ajax } from 'rxjs/ajax';
import { catchError, concatMap, map } from 'rxjs/operators';

const endpointInput: HTMLInputElement =
  document.querySelector('input#endpoint');
const fetchButton = document.querySelector('button#fetch');

fromEvent(fetchButton, 'click')
  .pipe(
    map(() => endpointInput.value),
    concatMap((value) =>
      ajax(`https://random-data-api.com/api/${value}/random_${value}`).pipe(
        catchError((error) => of(`Could not fetch data: ${error}`))
      )
    )
  )
  .subscribe({
    next: (value) => console.log(value),
    error: (err) => console.log('Error:', err),
    complete: () => console.log('Completed'),
  });

Flattening Operators - Concurrency - concatMap

concatMap operators do not pass to the next observable until the first observable completes.

This can be a good feature since uncompleted observables would create a memory leak in long term.

concatMap always guarantees that observables will be executed in the entered order. And this is super important (think multiple httpRequests or mouse positions for example)

Flattening Operators - switchMap

switchMap provides us almost the same solution as the concatMap but with one big difference.

If for example a new value is emitted while the previous observable still didn’t provide complete notification, switchMap automatically cancels the previous observable and creates the new observable with the new value.

It creates unimportant memory leaks because program cleans this memory after the cancel operation

This way it stops memory leaks and also can act really fast, based on the situation.

Flattening Operators - mergeMap

mergeMap is also like concatMap, but it allows multiple open observables, it doesn’t cancel the previous observable like switchMap, it allows them to work concurrently,

and this is sometimes, the exact thing we need.

Let's keep in touch

Hey, thanks a lot if you read it so far.
I want you to keep in touch for more sweet content.
Please subscibe in Dev.to and other channels (🦸‍♂️ especially twitter)

Twitter 🐦 https://twitter.com/barisbll_dev
Linkedin 📎 https://www.linkedin.com/in/barisbll-dev/
Github 👨🏻‍💻 https://github.com/barisbll
Medium 📰 https://medium.com/@barisbll

RxJS 7 - Creation Functions

Barış BALLI — Mon, 20 Mar 2023 09:28:12 +0000

Here are my detailed notes from the fantastic course created by Jurek Wozniak, called RxJS 7 and Observables: Introduction.

In order to understand the concept please consider buying the course yourself, these notes are simply subsidiaries

Creation Functions

Until now we always created the observable internal logic manually, but this is not at ideal approach since we usually have quite complicated observable logic.

Rxjs comes with many built-in creation functions. Creation functions create an observable with certain popular behaviour. We already saw an ajax function that makes an http request inside.

of - How Creation Functions Work

*of*, creation function takes an array of values, then emits all those values with next(), and when it comes to the last item, after emitting with next() it completes the observable with complete()

import { of } from 'rxjs';

of('Alice', 'Ben', 'Charlie').subscribe({
  next: (value) => console.log(value),
  complete: () => console.log('Completed'),
});

We can also easily make our own custom of function

import { Observable, of } from "rxjs";

ourOwnOf('Alice', 'Ben', 'Charlie').subscribe({
  next: value => console.log(value),
  complete: () => console.log('Completed')
});

function ourOwnOf(...args: string[]): Observable<string> {
  return new Observable<string>(subscriber => {
    for(let i = 0; i < args.length; i++) {
      subscriber.next(args[i]);
    }
    subscriber.complete();
  });
}

from

from works like *of*, but instead of having seperate values from receive one array, promise or an iterable.

When it takes a promise if it resolves from calls next with resolved value, if it receives rejects from calls error with rejected value.

And of course as always at the end emits a complete notification (also with promises.)

import { from } from "rxjs";

const somePromise = new Promise((resolve, reject) => {
  // resolve('Resolved!');
  reject('Rejected!');
});

const observableFromPromise$ = from(somePromise);

observableFromPromise$.subscribe({
  next: value => console.log(value),
  error: err => console.log('Error:', err),
  complete: () => console.log('Completed')
});

We can also pass observables to from (even though it creates an observable with it 😀 )

fromEvent

fromEvent is a creation function that emits events from various event sources.

Some of those event sources

Dom EventTarget
Node.js Event Emitter
jQuery Events

for example for the *******************fromEvent(button, click()),*******************

subscribe() will work like addEventListener()

unsubscribe() will work like removeEventListener()

Because actually rxjs will use those methods for us

import { fromEvent } from 'rxjs';

const triggerButton = document.querySelector('button#trigger');

fromEvent<PointerEvent>(triggerButton, 'click').subscribe((event) =>
  console.log(event.type, event.x, event.y)
);

If we want to build it ourselves we could do it like this (with additional setTimeout)

import { fromEvent, Observable } from "rxjs";

const triggerButton = document.querySelector('button#trigger');

// const subscription = fromEvent<MouseEvent>(triggerButton, 'click').subscribe(
//   event => console.log(event.type, event.x, event.y)
// );

const triggerClick$ = new Observable<MouseEvent>(subscriber => {
  const clickHandlerFn = event => {
    console.log('Event callback executed');
    subscriber.next(event);
  };

  triggerButton.addEventListener('click', clickHandlerFn);

  return () => {
    triggerButton.removeEventListener('click', clickHandlerFn);
  };
});

const subscription = triggerClick$.subscribe(
  event => console.log(event.type, event.x, event.y)
);

setTimeout(() => {
  console.log('Unsubscribe');
  subscription.unsubscribe();
}, 5000);

We have to return a cleanup function to leave the used resource (in our case event listener), in order to avoid memory leaks

As we can see rxjs creation functions make event listening much easier.

timer

timer works as an setTimeout function inside an observable

import { timer } from 'rxjs';

timer(200).subscribe({
  next: (value) => console.log(value),
  complete: () => console.log('Completed'),
});

As we can see it automatically emits the value 0 after finishes, and calls complete we can also see what happens inside

import { Observable, timer } from "rxjs";

console.log('App started');

const timer$ = new Observable<number>(subscriber => {
  const timeoutId = setTimeout(() => {
    console.log('Timeout!');
    subscriber.next(0);
    subscriber.complete();
  }, 2000);

  return () => clearTimeout(timeoutId);
});

const subscription = timer$.subscribe({
  next: value => console.log(value),
  complete: () => console.log('Completed')
});

setTimeout(() => {
  subscription.unsubscribe();
  console.log('Unsubscribe');
}, 1000);

interval

interval works just like the timer, this time instead of receiving just the value 0, we have in interval values starting from 0 and incremented by one every once in a dedicated interval time.

forkJoin - Handle Multiple Http Calls

forkJoin works different than other functions we talked previously.

forkJoin takes multiple observables and wait until ***********************************all the observables emit their value.*********************************** After receiving all the values from those observables forkJoin emits an array of responses.

This can be handy if we want to make multiple http requests and make an operation just after receiving all the answers.

import { forkJoin } from 'rxjs';

import { ajax } from 'rxjs/ajax';

const randomName$ = ajax('https://random-data-api.com/api/name/random_name');

const randomNation$ = ajax(
  'https://random-data-api.com/api/nation/random_nation'
);

const randomFood$ = ajax('https://random-data-api.com/api/food/random_food');

// randomName$.subscribe((ajaxResponse) =>
//   console.log(ajaxResponse.response.first_name)
// );
// randomNation$.subscribe((ajaxResponse) =>
//   console.log(ajaxResponse.response.capital)
// );
// randomFood$.subscribe((ajaxResponse) =>
//   console.log(ajaxResponse.response.dish)
// );

forkJoin([randomName$, randomNation$, randomFood$]).subscribe(
  ([nameAjax, nationAjax, foodAjax]) => console.log(`${nameAjax.response.first_name} is from ${nationAjax.response.capital} and likes to eat ${foodAjax.response.dish}.`)
);

If we would subscribe to each observable ourselves it would emit values in random order,

Thanks to forkJoin we can have all those values at once and format them to a specific format like this.

But let’s say we have such an observable that calls next multiple times before actually completes the action in that case what will be returned?

In that case the last values emitted via next before the complete notification are returned as response. And all the previous values emitted before those last ones are discarded.

And what happens when an error notificatios is emitted by one of the observables in the forkJoin function? Of course it automatically ends the subscription with error

import { forkJoin, Observable } from "rxjs";

const a$ = new Observable(subscriber => {
  setTimeout(() => {
    subscriber.next('A');
    subscriber.complete();
  }, 5000);

  return () => {
    console.log('A teardown');
  };
});

const b$ = new Observable(subscriber => {
  setTimeout(() => {
    subscriber.error('Failure!');
  }, 3000);

  return () => {
    console.log('B teardown');
  };
});

forkJoin([a$, b$]).subscribe({
  next: value => console.log(value),
  error: err => console.log('Error:', err)
});

We can see that teardown function is called as soon as error notification is received in both of the observables.

Combine Latest - Reacting to Multiple Input Changes

combineLatest also takes an array of observables but instead of emitting just the last values before complete notification, it emits a new array of resolved values each time one of the observables emit a value (starting with first set of all resolved values)

import { combineLatest, fromEvent } from "rxjs";

const temperatureInput = document.getElementById('temperature-input');
const conversionDropdown = document.getElementById('conversion-dropdown');
const resultText = document.getElementById('result-text');

const temperatureInputEvent$ = fromEvent(temperatureInput, 'input');
const conversionInputEvent$ = fromEvent(conversionDropdown, 'input');

combineLatest([temperatureInputEvent$, conversionInputEvent$]).subscribe(
  ([temperatureInputEvent, conversionInputEvent]) => {
    const temperature = Number(temperatureInputEvent.target['value']);
    const conversion = conversionInputEvent.target['value'];

    let result: number;
    if (conversion === 'f-to-c') {
      result = (temperature - 32) * 5/9;
    } else if (conversion === 'c-to-f') {
      result = temperature * 9/5 + 32;
    }

    resultText.innerText = String(result);
  }
);

As soon as we entered another value from one of two input fields above we will have a new emitted value.

Let's keep in touch

Hey, thanks a lot if you read it so far.
I want you to keep in touch for more sweet content.
Please subscibe in Dev.to and other channels (🦸‍♂️ especially twitter)

Twitter 🐦 https://twitter.com/barisbll_dev
Linkedin 📎 https://www.linkedin.com/in/barisbll-dev/
Github 👨🏻‍💻 https://github.com/barisbll
Medium 📰 https://medium.com/@barisbll

RxJS 7 - Types of Observables

Barış BALLI — Mon, 20 Mar 2023 09:27:05 +0000

Here are my detailed notes from the fantastic course created by Jurek Wozniak, called RxJS 7 and Observables: Introduction.

In order to understand the concept please consider buying the course yourself, these notes are simply subsidiaries

Types of Observables

If an observable has a defined set of possible values, or if its value is related to the interval it is in and changes based on defined rules we call it cold observables.

If an observable’s value based on user’s actions for example a dom event or mouse position we call it hot observable

An http request is also counted as a cold observable.

The main rule that seperates cold observables from the hot ones is, if each observation occurs independent from each other or not

It doesn’t have to contain the same events for each subscription, in our example we will send an http request to a random api, and even though the response we receive changes, because of the whole process was independent per each subscription we call it a cold observable

Cold Observable

import { ajax } from "rxjs/ajax";

const ajax$ = ajax<any>('https://random-data-api.com/api/name/random_name');

ajax$.subscribe(
  data => console.log('Sub 1:', data.response.first_name)
);

ajax$.subscribe(
  data => console.log('Sub 2:', data.response.first_name)
);

ajax$.subscribe(
  data => console.log('Sub 3:', data.response.first_name)
);

Hot Observables

In the hot observables, observables are connected to the same resource, therefore effected by each other.

For example a dom event like resizing a window would effect all the observer listening that observable

import { Observable } from 'rxjs';

const helloButton = document.querySelector('button#hello');

const helloClick$ = new Observable<PointerEvent>((subscriber) => {
  helloButton.addEventListener('click', (event: PointerEvent) => {
    subscriber.next(event);
  });
});

helloClick$.subscribe((event) =>
  console.log('Sub 1', event.type, event.x, event.y)
);

helloClick$.subscribe((event) =>
  console.log('Sub 2', event.type, event.x, event.y)
);

We can also subscribe another observer after a while

import { Observable } from "rxjs";

const helloButton = document.querySelector('button#hello');

const helloClick$ = new Observable<MouseEvent>(subscriber => {
  helloButton.addEventListener('click', (event: MouseEvent) => {
    subscriber.next(event);
  });
});

helloClick$.subscribe(
  event => console.log('Sub 1:', event.type, event.x, event.y)
);

setTimeout(() => {
  console.log('Subscription 2 starts');
  helloClick$.subscribe(
    event => console.log('Sub 2:', event.type, event.x, event.y)
  );
}, 5000);

Hot vs Cold

Cold: Produces the data inside

New subscriber → new data

*****Hot:***** Multicasts the data from a common source

All subscribers → common data

Observers can change their behaviour with time a hot observer can be a cold one

Let's keep in touch

Hey, thanks a lot if you read it so far.
I want you to keep in touch for more sweet content.
Please subscibe in Dev.to and other channels (🦸‍♂️ especially twitter)

Twitter 🐦 https://twitter.com/barisbll_dev
Linkedin 📎 https://www.linkedin.com/in/barisbll-dev/
Github 👨🏻‍💻 https://github.com/barisbll
Medium 📰 https://medium.com/@barisbll

RxJS 7 - Observable, Observer and Subscription

Barış BALLI — Mon, 20 Mar 2023 09:25:46 +0000

Here are my detailed notes from the fantastic course created by Jurek Wozniak, called RxJS 7 and Observables: Introduction.

In order to understand the concept please consider buying the course yourself, these notes are simply subsidiaries

Exercises: Observable, Observer and Subscription

Teardown, Completion Notification

Let’s say we want to send a complete notification after two seconds, and consume the complete notification that is emitted.

Also we want to make some cleanup when the teardown is happening. How can we do that?

import { Observable } from 'rxjs';

const $observable = new Observable((subscriber) => {
  subscriber.next('First');
  setTimeout(() => {
    subscriber.next('After one sec');
  }, 1000);
  setTimeout(() => {
    subscriber.next('After two sec');
    subscriber.complete();
  }, 2020);

  return () => {
    console.log('Teardown');
  };
});

const observer = $observable.subscribe({
  next: (value) => console.log(value),
  complete: () => console.log('Completed'),
});

We see for the first time that when we create a new observable, we can return a function and this function will be called at the teardown process.

This is really handy because in that step we can clean our logic and make the required changes.

Error Notification

This time let’s look at the logic what happens when an error is emitted

import { Observable } from 'rxjs';

const observable$ = new Observable<string>((subscriber) => {
  console.log('Observable executed');
  subscriber.next('Alice');
  subscriber.next('Ben');
  setTimeout(() => {
    subscriber.next('Charlie');
    subscriber.error(new Error('Something went wrong :('));
  }, 2000);

  return () => {
    console.log('Teardown');
  };
});

console.log('Before subscribe');
observable$.subscribe({
  next: (value) => console.log(value),
  complete: () => console.log('Completed'),
  error: (err: Error) => console.log(err.message),
});
console.log('After subscribe');

Unsubscribe

Let’s create an interval that emits a value every 2 second and we unsubscribe to it ater 7 seconds, and clean the interval using the teardown function

import { Observable } from "rxjs";

const interval$ = new Observable<number>(subscriber => {
  let counter = 1;

  const intervalId = setInterval(() => {
    console.log('Emitted', counter);
    subscriber.next(counter++);
  }, 2000);

  return () => {
    clearInterval(intervalId);
  };
});

const subscription = interval$.subscribe(value => console.log(value));

setTimeout(() => {
  console.log('Unsubscribe');
  subscription.unsubscribe();
}, 7000);

Let's keep in touch

Hey, thanks a lot if you read it so far.
I want you to keep in touch for more sweet content.
Please subscibe in Dev.to and other channels (🦸‍♂️ especially twitter)

Twitter 🐦 https://twitter.com/barisbll_dev
Linkedin 📎 https://www.linkedin.com/in/barisbll-dev/
Github 👨🏻‍💻 https://github.com/barisbll
Medium 📰 https://medium.com/@barisbll

RxJS 7 - Observables

Barış BALLI — Sun, 12 Mar 2023 08:11:09 +0000

Here are my detailed notes from the fantastic course created by Jurek Wozniak, called RxJS 7 and Observables: Introduction.

In order to understand the concept please consider buying the course yourself, these notes are simply subsidiaries

Observables - How Do They Work?

Array vs Stream

Observables are based around the idea of streams, that means the data can come in different parts of time and number of data can be theoretically infinite

The biggest different between arrays is, in arrays all the values inside are already known.

With the stream approach, we react to the things when they show up

For example we first see a lemon and we react to it or not,

then coconut etc.

We can for example have a stream for mouse positions, text input data change, http requests

Observable, Subscription, Observer, Key Elements

Observable is quite simple once an observable is executed it emits some notifications.

There are 3 types of notifications

Next
Error
Complete

For now we fill focus to the next notification which allows us to emit notifications.

Warm-Up Observable - Observable, Observer, Subscription

Observable doesn’t do anything by itself it just has some logic stored inside. We need to subscribe to it to make it work

Subscription should be closed when we are done with them in order to avoid memory leakage and unexpected behaviorus.

Subscriptions can be stopped in 2 ways.

It can stop automatically by the observable logic itself with error or complete notifications
We can cancel it by unsubscribing

import { Observable } from "rxjs";

const observable$ = new Observable<string>(subscriber => {
  console.log('Observable executed');
  subscriber.next('Alice');
  setTimeout(() => subscriber.next('Ben'), 2000);
  setTimeout(() => subscriber.next('Charlie'), 4000);
});

const subscription = observable$.subscribe(value => console.log(value));

setTimeout(() => {
  console.log('Unsubscribe');
  subscription.unsubscribe();
}, 3000);

One of the most important thing here is how we unsubscribe to observable by calling the subscription object.

We received subscription object as return value of subscribe function.

and we could also use observer like this

const observer = {
  next: (value) => console.log(value),
};

However since we just need to have a next function we used simple function version.

Usually this small - just function version is used in projects.

Multiple Subscriptions

What happens if 2 observers subscribe to the same observable.

Well, just as you would expect they work independently and execute through all observable stream

import { Observable } from 'rxjs';

const observable$ = new Observable<string>((subscriber) => {
  console.log('Observable executed');
  subscriber.next('Alice');
  setTimeout(() => subscriber.next('Ben'), 2000);
  setTimeout(() => subscriber.next('Charlie'), 4000);
});

console.log('Subscription 1 starts');
observable$.subscribe((value) => console.log('Subscription 1:', value));

setTimeout(() => {
  console.log('Subscription 2 starts');
  observable$.subscribe((value) => console.log('Subscription 2:', value));
}, 1000);

Output

Subscription 1 starts
Observable executed
Subscription 1:
Alice
Subscription 2 starts
Observable executed
Subscription 2:
Alice
Subscription 1:
Ben
Subscription 2:
Ben
Subscription 1:
Charlie
Subscription 2:
Charlie

Complete notification is sent only if all the messages are sent and it is telling us that it has nothing more to share.

What Next?

This is the second article of a series of articles I write about RxJS. Do not stop here! Continue reading.
See the first article

Let's keep in touch

Hey, thanks a lot if you read it so far.
I want you to keep in touch for more sweet content.
Please subscibe in Dev.to and other channels (🦸‍♂️ especially twitter)

Twitter 🐦 https://twitter.com/barisbll_dev
Linkedin 📎 https://www.linkedin.com/in/barisbll-dev/
Github 👨🏻‍💻 https://github.com/barisbll
Medium 📰 https://medium.com/@barisbll

RxJS 7 - Getting Started

Barış BALLI — Mon, 06 Mar 2023 20:15:49 +0000

Here are my detailed notes from the fantastic course created by Jurek Wozniak, called RxJS 7 and Observables: Introduction.

In order to understand the concept please consider buying the course yourself, these notes are simply subsidiaries

Note: This is the first article of a series of articles

Getting Started

Quick Start

Observables are like data or event generators with logic stored inside.

The data they generate can have various of sources like array of values, dom events, timers etc.

Anything you want can turned to be an observable

Let’s say we have an observable, now we need to start executing somehow to generate the data.

Coding Example

Observables are having $ sign at the end of their name as the naming practice, this way we can easily identify which variable is actually observable

First let’s see the functions we will use.

external.ts

import { Observable, of } from "rxjs";

export const name$ = of('Alice', 'Ben', 'Charlie');

export function storeDataOnServer(data: string): Observable<string> {
  return new Observable(subscriber => {
    setTimeout(() => {
      subscriber.next(`Saved successfully: ${data}`);
      subscriber.complete();
    }, 5000);
  });
}

export function storeDataOnServerError(data: string): Observable<string> {
  return new Observable(subscriber => {
    setTimeout(() => {
      subscriber.error(new Error('Failure!'));
    }, 5000);
  });
}

We can think observable like a function, when we call it, it emits some values.

index.ts

import { name$, storeDataOnServer, storeDataOnServerError } from './external';

name$.subscribe();

Here it probably emitted some data but because of we don’t have a handler we couldn’t catch it

import { name$, storeDataOnServer, storeDataOnServerError } from './external';

name$.subscribe((value) => console.log(value));

// Mike
// Brandon
// Alice

We can for example write a function that simulates data fetch from a remote server

storeDataOnServer('Some value').subscribe((value) => console.log(value));

We also have a version that acts like we are throwing an error

storeDataOnServerError('Some value').subscribe((value) => console.log(value));

This will fail our javascript since we didn’t handle the error .

For error handling in observables we have 2 versions

First the deprecated version (still a lot of projects written like this)

storeDataOnServerError('Some value').subscribe(
   (value) => console.log(value),
   (err) => console.log('Error when saving:', err.message)
 );

And the modern version the observable object

storeDataOnServerError('Some value').subscribe({
  next: (value) => console.log(value),
  error: (err) => console.log('Error when saving:', err.message),
});

The whole code:

import { name$, storeDataOnServer, storeDataOnServerError } from './external';

// name$.subscribe((value) => console.log(value));

// storeDataOnServer('Some value').subscribe((value) => console.log(value));
// storeDataOnServerError('Some value').subscribe((value) => console.log(value));

// storeDataOnServerError('Some value').subscribe(
//   (value) => console.log(value),
//   (err) => console.log('Error when saving:', err.message)
// );

storeDataOnServerError('Some value').subscribe({
  next: (value) => console.log(value),
  error: (err) => console.log('Error when saving:', err.message),
});

What Next?

This is the first article of a series of articles I write about RxJS. Do not stop here! Continue reading.
See the second article

Let's keep in touch

Hey, thanks a lot if you read it so far.
I want you to keep in touch for more sweet content.
Please subscibe in Dev.to and other channels (🦸‍♂️ especially twitter)

Twitter 🐦 https://twitter.com/barisbll_dev
Linkedin 📎 https://www.linkedin.com/in/barisbll-dev/
Github 👨🏻‍💻 https://github.com/barisbll
Medium 📰 https://medium.com/@barisbll