I’ve been rereading Angular official documentation lately, and I’ve found some interesting things. Reminds me of heilan House’s advertising slogan: every time there is a new experience.

Observables and Observer

The basic RXJS concepts, Observables and Observers, have confused us a lot.

In fact, look at the following simple line of code to understand the relationship:

observable.subscribe(observer);
Copy the code

An Observable is a data source, a producer, and a subscriber to be subscribed. An observer is an observer, a data user, and a data consumer.

This code returns a Subscription object, representing a Subscription occurrence, a Subscription process, and an instantiation of the Subscription object.

An observer is an object with three callbacks, each corresponding to three Observable notification types (Next, Error, and complete). Callbacks do not have to provide three at a time. If we provide only one callback as an argument, SUBSCRIBE takes the argument we provide as a callback handler for next.

Observable.subscribe () is a Subscription object. Subscription is an Observable execution that can be cleaned. The most common method for this object is the unsubscribe method. It also has the Add method that allows us to cancel multiple subscriptions.

I have an analogy to put this relationship in perspective: there is now a milk producer whose milk is delicious, delicious and cheap. It advertises on TELEVISION and anyone can call its phone to order milk. The milkman is an Observable and the citizen is an Observer. If citizens subscribe to the milk merchant, they receive the milk when next succeeds, and it is their business to drink the milk. Citizens do not care about how the milk is produced and delivered (e.g. database, TCP/IP protocol for HTTP process, handshake process, etc.). The milk delivery process may encounter an error, and after success, the milk will be marked as complete.

These are the basics of basics, and even though they sometimes confuse us, we have a rough idea of what they mean. And the following Subject, many people around feel that they do not quite understand.

We are often confused about the concept of Subject in RXJS and when to use it. In fact, at the very beginning of the official documentation, the simple HERO tutorial has a paragraph like this:

Subject is actually a very useful and elegant thing. In the official RXJS documentation, we can see the introduction of Subject:

Subjects are the only way to share any Observable execution with multiple observers. The translation of this official document is obviously somewhat awkward. To put it simply, a Subject is both an Observable and an observer.

We know that for an Observable, each observer has its own independent execution of subscribers. What does that mean? For example, if a person orders milk, there will be a milk delivery process; An Http request occurs when a Get request in a Service is subscribed.

This pattern is not suitable for some of our cases, for example, where input change may occur in milliseconds. How many subscriptions do we need to create in this way? Or milk merchants do promotional activities on Double Eleven, and suddenly there is an order produced in a second. If we still use one delivery, how many milk delivery procedures do we need?

Well, you’re smart enough to think of multiple milk deliveries. This is what the official RXJS documentation says: a Subject is like an Observable, but can be multicast to multiple observers. Subject also acts like EventEmitters, maintaining a registry of multiple listeners.

Here’s a look at the Hero app:

private searchTerms = new Subject<string>(); 
search(term: string): void { 
   this.searchTerms.next(term); 
}
Copy the code

SearchTerms is now a continuous stream of input field values. The value can be continuously issued, and can use subscribe continuously get value. And all of this is made very simple by Subject.

Subject still have three more commonly used subclasses: ReplaySubject, AsyncSubject, BehaviorSubject. What do they mean? What’s the difference?

Subject

Code examples:

import { Subject, BehaviorSubject, ReplaySubject, AsyncSubject } from 'rxjs';

let subject1: Subject<number> = new Subject<number>();
subject1.next(100);
subject1.subscribe(res=>{
      console.log('SubjectA:'+res);
 })
subject1.subscribe(res=>{
      console.log('SubjectB:'+res);
 })
subject1.next(200);
subject1.next(300);
Copy the code

Execution Result:

 SubjectA:200
 SubjectB:200
 SubjectA:300
 SubjectB:300
Copy the code

As you can see, the Subject can only receive the values emitted by the data source after subscribing. Subject1.next (100) has not yet been subscribed, so the result will not be printed. What if we wanted to get the data at any point after the subscriber was created? This is where it comes in.

BehaviorSubject

Code examples:

let subject2: BehaviorSubject<number> = new BehaviorSubject<number>(0);
subject2.next(100);
subject2.subscribe(res=>{
     console.log('Behavior-SubjectA:'+res);
});
subject2.next(200);
subject2.subscribe(res=>{
     console.log('Behavior-SubjectB:'+res);
});
subject2.next(300);
Copy the code

Execution Result:

 Behavior-SubjectA:100
 Behavior-SubjectA:200
 Behavior-SubjectB:200
 Behavior-SubjectA:300
 Behavior-SubjectB:300
Copy the code

As you can see, the BehaviorSubject saves the latest sent data, which is immediately used when subscribed. It then continues to receive new values. When the behavior-Subjecta is subscribed, the current value is 100, so 100 will be printed immediately, and then it will receive a new issue of 200; When the behavior-SubjectB is subscribed, it immediately prints the latest subject2 stream value of 200. And then A and B each receive 300.

Note that BehaviorSubject must have a default value. Because it has a concept of the latest value (current value). So what if we want to save all of our data, not just the latest values?

ReplaySubject

Code examples:

let subject3: ReplaySubject<number> = new ReplaySubject<number>();
subject3.next(100);
subject3.next(200);
subject3.subscribe(res=>{
      console.log('Replay-SubjectA:'+res);
});
subject3.next(300);
subject3.subscribe(res=>{
      console.log('Replay-SubjectB:'+res);
});
subject3.next(400);
Copy the code

Print result:

 Replay-SubjectA:100
 Replay-SubjectA:200
 Replay-SubjectA:300
 Replay-SubjectB:100
 Replay-SubjectB:200
 Replay-SubjectB:300
 Replay-SubjectA:400
 Replay-SubjectB:400
Copy the code

The ReplaySubject saves all the values and plays them back to the new subscriber.

The replay-subjecta will subscribe to a stream of 100,200 and then receive 300; The new replay-subjectb subscription will take place. The stream will contain 100,200,300 values, so it will print out the saved values in turn. And then A and B get 400.

The new ReplaySubject() parameter indicates the number of stores, i.e. BufferSize, for example, we set it to 1:

let subject3: ReplaySubject<number> = new ReplaySubject<number>(1);
Copy the code

The output will only store the data in one stream:

Replay-SubjectA:200
Replay-SubjectA:300
Replay-SubjectB:300
Replay-SubjectA:400
Replay-SubjectB:400
Copy the code

We can understand that these data streams are all saved and played back to the subscriber like a movie when a new subscription occurs.

AsyncSubject

The AsyncSubject sends the last value executed to the observer only when the Observable completes (complete()).

That is, it saves only the last piece of data in the stream and only sends it when the stream is complete.

let subject4: AsyncSubject<number> = new AsyncSubject<number>();
subject4.next(100);
subject4.subscribe(res => {
    console.log('Async-SubjectA:' + res);
});
subject4.next(200);
subject4.subscribe(res => {
    console.log('Async-SubjectB:' + res);
});
subject4.next(300);
subject4.subscribe(res => {
    console.log('Async-SubjectC:' + res);
});
subject4.complete();
subject4.next(400);
Copy the code

Execution Result:

 Async-SubjectA:300
 Async-SubjectB:300
 Async-SubjectC:300
Copy the code

As you can see, the data flow is 100,200,300 before complete. The last one is 300. All subscribers will only receive the last 300. After complete, the value will not be sent.

Additional knowledge

Finally: Subject also has a subclass of AnonymousSubject, but this subclass has very few usage scenarios. In the spirit of research curiosity, what is it?

Sometimes we take it for granted that Subject can be instantiated using create. In fact, subject.create () returns an instance of the AnonymousSubject object, and new Subject() returns an instance of the Subject object. AnonymousSubject doesn’t subscribe to itself like a normal Subject. It marks the data source, and when the subscription occurs, it directly connects the data source to the observer but does not track the progress of the subscription we created (that is, the data changes). Such as:

var timer$ = Rx.Observable.timer(1000, 2000);
var timerSubject = Rx.Subject.create(null, timer$);

var subscription1 = timerSubject.subscribe(n => console.log(n));
var subscription2 = timerSubject.subscribe(n => console.log(n));

setTimeout(() => timerSubject.unsubscribe(), 4000); / / not to take effectCopy the code

Subscription1 is actually subscribed to timer$directly, so it is not unsubscribe.

AnonymousSubject’s actual application scenarios are not described in the official documentation.