RxJS Advanced – Responsive programming with Observable types

This blog post is an introduction to the RxJS advanced series of responsive programming, cold, warm, and Hot Observables. This introduction is not a copy of some technical terms and apis, but a generalization of some of the most difficult concepts, combined with examples and real business scenarios, to give you a real grasp of how they are used in our daily development. If you encounter any UNFAMILIAR APIS in this article, check out the official RxJS documentation for yourself.

RxJS has always been the bane of the Angular technology stack. Many newcomers said that they were still at a loss after reading the RxJS Api documentation. There were a lot of awkward professional terms and difficult concepts, and there were also a lot of questions in their mind during the process of learning and using:

• Why use RxJS? What are its advantages?
• Why does Subject subscribe in advance? What about Http requests?
• How to achieve page to page communication?
• How to achieve global interface reuse and avoid repeated requests?
• .

Today we will take these doubts together to study RxJS in depth.

Responsive programming

One of the reasons RxJS has a high barrier to entry is that it requires a shift from familiar imperative programming to responsive programming thinking. So we need to understand what responsive programming is. Let’s first look at the definitions of reactive and imperative programming:

Reactive programming is Imperative programming, a declarative programming paradigm oriented to data flow and change propagation. It is a programming paradigm that describes the behavior required by a computer

You may not be able to understand the definition, but let’s take the following assignment operations as an example:

    a = b + c;
    m = (b + c) / n; 
    a = b + c - d + e + f + .... + n;
Copy the code

So this first expression, the value of a is equal to b plus c. When b and c are fixed, there’s nothing wrong with writing this. But if B and C are variable values, how can the value of A be updated synchronously? The second expression, m, is also associated with b and C values. After b and C values are updated, not only a value needs to be updated, but also M value needs to be updated synchronously. The third expression, the value of A is jointly determined by a set of mutable values, how can the value of A be updated synchronously?

In normal imperative programming, the only way to implement this requirement is to manually perform an assignment after each change. A = b + c, m = (b + c)/ n, m = (b + c)/ n, m = (b + C)/ n The result is a lot of duplicate code, and state and dependency maintenance becomes very difficult.

So can we get rid of the manual update process and realize the automatic update of A? The answer is yes, with reactive programming. In reactive programming, we think of data changes as data streams composed of data one by one. In the above example, we simply subscribe to the data streams of B and C, and then calculate the data streams of A or M. The calculation logic is automatically invoked and the value of A or M is updated when data b or C changes.

The formula function we use in Excel is also responsive programming. The figure below illustrates the function of using Excel to count the sum of data of multiple grids. In Excel, select the grid B10, enter = in the formula section, and then select the grid B1 to B9 with the mouse to complete a responsive programming. Then, no matter what number I fill in B1 through B9, the value in the B10 cell automatically becomes the sum of the values of all cells B1 through B9. In other words, B10 “responds” to the changes in the values of these cells.

We can implement a simple Excel automatic calculation function by ourselves. According to the input value, it will automatically calculate the sum and average value, as shown below:

For a more intuitive comparison of imperative programming and RxJS differences and advantages and disadvantages, we implement both methods: the first method: imperative programming

<h2>Imperative programming -Excel spreadsheet</h2>
<table>
  <tr>
    <td>Value of 1</td>
    <td>
      <input type="number" #input1 [ngModel] ="number1" (ngModelChange) ="onValue1Change($event)">
    </td>
  </tr>
  <tr>
    <td>Value 2</td>
    <td>
      <input type="number" #input2 [ngModel] ="number2" (ngModelChange) ="onValue2Change($event)">
    </td>
  </tr>
  <tr>
    <td>The value 3</td>
    <td>
      <input type="number" #input3 [ngModel] ="number3" (ngModelChange) ="onValue3Change($event)">
    </td>
  </tr>
  <tr>
    <td>A total of</td>
    <td>
      <input type="number" [value] ="sum">
    </td>
  </tr>
  <tr>
    <td>The mean</td>
    <td>
      <input type="number" [value] ="average">
    </td>
  </tr>
  <tr>
    <td>operation</td>
    <td>
      <button (click) ="reset()">empty</button>
    </td>
  </tr>
</table>
Copy the code

export class ExcelTableComponent {
  public number1: number;
  public number2: number;
  public number3: number;
  public sum: number;
  public average: number;

  public onValue1Change(value: number) :void {
    this.number1 = value;
    this.onValueChange();
  }

  public onValue2Change(value: number) :void {
    this.number2 = value;
    this.onValueChange();
  }

  public onValue3Change(value: number) :void {
    this.number3 = value;
    this.onValueChange();
  }

  public reset(): void {
    this.number1 = null;
    this.number2 = null;
    this.number3 = null;
    this.onValueChange();
  }

  private onValueChange(): void {
    this.updateSum();
    this.updateAverage();
  }

  private updateSum(): void {
    if (!this.isNullOrUndefined(this.number1)
      && !this.isNullOrUndefined(this.number2)
      && !this.isNullOrUndefined(this.number3)) {
      this.sum = this.number1 + this.number2 + this.number3;
    } else {
      this.sum = null; }}private updateAverage(): void {
    if (!this.isNullOrUndefined(this.sum)) {
      this.average = this.sum / 3;
    } else {
      this.average = null; }}private isNullOrUndefined(val: number | undefined | null) :boolean {
    return typeof val === 'undefined' || val === null; }}Copy the code

The second way: RxJS implementation

<h2>Responsive programming -Excel spreadsheet</h2>
<table>
  <tr>
    <td>Value of 1</td>
    <td>
      <input type="number" #input1 [ngModel] ="number1$ | async" (ngModelChange) ="number1$.next($event)">
    </td>
  </tr>
  <tr>
    <td>Value 2</td>
    <td>
      <input type="number" #input2 [ngModel] ="number2$ | async" (ngModelChange) ="number2$.next($event)">
    </td>
  </tr>
  <tr>
    <td>The value 3</td>
    <td>
      <input type="number" #input3 [ngModel] ="number3$ | async" (ngModelChange) ="number3$.next($event)">
    </td>
  </tr>
  <tr>
    <td>A total of</td>
    <td>
      <input type="number" [value] ="sum$ | async">
    </td>
  </tr>
  <tr>
    <td>The mean</td>
    <td>
      <input type="number" [value] ="average$ | async">
    </td>
  </tr>
  <tr>
    <td>operation</td>
    <td>
      <button (click) ="reset()">empty</button>
    </td>
  </tr>
</table>
Copy the code

@Component({
  selector: 'app-excel-table-stream'.templateUrl: './excel-table-stream.component.html'.styleUrls: ['./excel-table-stream.component.scss']})export class ExcelTableStreamComponent implements OnInit {
  public number1$ = new Subject<number> ();public number2$ = new Subject<number> ();public number3$ = new Subject<number> ();public sum$: Observable<number>;
  public average$: Observable<number>;

  public ngOnInit() {
    this.sum$ = combineLatest(
      this.number1$.asObservable(),
      this.number2$.asObservable(),
      this.number3$.asObservable()
    )
      .pipe(
        map(([number1, number2, number3]: number[]) = > number1 + number2 + number3)
      );

    this.average$ = this.sum$
      .pipe(
        map((sum: number) = > sum / 3)); }public reset(): void {
    this.number2$.next();
    this.number1$.next();
    this.number3$.next(); }}Copy the code

It is clear from the comparison of the above example that in the usual imperative programming, I need to manually recalculate the associated dependencies every time the value of the input field changes. In reactive programming, we just send the changed value. The calculation and update after the value changes are performed automatically by the subscriber. In this way, the logic of the producer and the subscriber is separated and the logic is clear.

After reading the examples above, let’s answer the first question at the beginning of this article: Why use RxJS? What are its advantages? RxJS is based on the idea of responsive programming, which treats all the changeable things on a web page, such as DOM manipulation, user input, Http requests, data updates, and so on, as data streams. It is well suited to the scenario of dealing with these changing data flows, because it can manipulate these abstract data flows like an array, decouple the process of production and subscription processing values, and realize automatic update of dependencies after data changes. Now do you see why we use RxJS?

Cold, warm, and hot Observable

In RxJS, we can classify observables into three types according to their different performance characteristics: cold, warm and hot Observables. It is difficult but crucial for beginners to understand these three types. So let’s learn more about them. First, let’s list the characteristics of these three Observables.

After reading the above table, you may still be unable to understand and remember the characteristics of these three Observables. It doesn’t matter, as you can see, I wrote several analogies on the last line of the table to help you understand the characteristics of these types of Observables. Let’s do it one by one.

Hot Observables are like the news feeds we watch. Since the news broadcast starts on time, it can be understood that the value is generated immediately after creation. So if we want to watch the whole thing, we have to watch it early or on time, or we’ll miss some of it. And the news broadcast is one-to-many, everyone shares an instance, there’s no progress bar, it’s all real time.

A cold Observable is like a TV show in a video app we watch all the time. It doesn’t play until you hit play, which is when you subscribe. And whenever you watch it, you can watch it from the beginning and not miss it, and subscribers get the same value. Because this platform has multiple accounts, each account serves only one entity, and each account keeps its own progress bar without interfering with each other. This corresponds to multiple instances of a cold Observable, the concept of unicast.

A warm Observable is between a cold Observable and a hot Observable. It determines its own behavior by counting. It’s like the offline courses we take. When none of the students show up, it continues to wait, even though it’s time to start classes. Until the first trainee arrives at the training site, the training begins. Once you’ve started, and before the class ends, people who come in later will miss part of the class because they’re late. But if some of the participants arrive after the course, they can arrange to attend the next round of training and still learn the whole thing. So the analogy: while a warm Observable is still sending values, a late Observable gets only the ones that it hasn’t missed. However, if the sending value is finished, the subscriber can also get the full value.

After understanding the characteristics of cold, warm, and hot Observables, we combine code examples to deepen our understanding of them respectively.

const source$ = interval(100)
  .pipe(
    take(3));setTimeout(() = > {
  source$.subscribe(res= > {
    console.log('observer1:', res);
  });
}, 500);

source$.subscribe(res= > {
  console.log('observer2:', res);
});
Copy the code

The above code is a common cold Observable, and most operators in RxJS generate cold observables. There are two observers for this cold Observable. The first one subscribes after 500ms, and the second one immediately. According to the characteristics of the cold Observable, all subscribers get the same value regardless of the time, so it is easy to infer its printed value as follows:

const source$ = (timer(0.100)
      .pipe(
        take(3),
        publish()
      ) as ConnectableObservable<number>);

    source$.connect();

    setTimeout(() = > {
      source$.subscribe(res= > {
        console.log('observer1:', res);
      });
    }, 100);

    setTimeout(() = > {
      source$.subscribe(res= > {
        console.log('observer2:', res);
      });
    }, 1000);
Copy the code

This code converts a cold Observable into a hot Observable using the multicast publish operator in RxJS. There are two observers for this hot Observable. The first one subscribes after 100ms, and the second one after 1000ms. According to the characteristics of the hot Observable, the value the subscriber gets depends on the subscription time. The following information is displayed:

const source$ = interval(100)
  .pipe(
    take(3),
    share()
  );

setTimeout(() = > {
  source$.subscribe(res= > {
    console.log('observer1:', res);
  });

  setTimeout(() = > {
    source$.subscribe(res= > {
      console.log('observer2:', res);
    });
  }, 200);
}, 200);

setTimeout(() = > {
  source$.subscribe(res= > {
    console.log('observer3:', res);
  });
}, 1000);
Copy the code

Warm Observables are a little more complicated to understand. This code converts cold Observables into warm Observables using the multicast share operator in RxJS. There are three observers for this warm Observable. The first observer subscribes after 200ms, the second observer subscribes after 200ms, and the third observer subscribes after 1000ms. At this point, the producer has finished sending values. According to the characteristics of a warm Observable, a producer sends a value from the first subscriber that subscribes, and gets a full value from any subscriber that has finished sending a value. So the first and third subscribers get all the values, and the second subscriber misses the first two values, so its final output looks like this:

After learning the above three Observable types, let’s answer the second question at the beginning of this article: Why does the Subject subscribe in advance? What about Http requests? Because Subject is a hot Observable, Http requests are cold Observables. Hot Observables need to subscribe in advance or they will miss the emitted values.

Due to the length problem, this blog post only talks about responsive programming and three Observable types, cold, warm and hot, and answers the first two questions at the beginning of the article. In the next blog post, we’ll cover Subject and some of its subclasses, as well as some real business scenarios involving the Observable type and use of Subject, and answer the last two questions raised at the beginning of this article.