Introduction to RxJS

What is RxJS?

Pre-knowledge point

• Reactive programming is programming with asynchronous data streams. A common example of asynchronous data flow involves Event buses. Create a Data stream with anything that includes these events, listen on it, and respond.
• Focus only on events that business logic depends on each other rather than implementation details
• It is suitable for a large number of data-related events, especially for high real-time requirements

• A stream is an uninterrupted sequence sorted by time. It produces three different types of events: values, errors, and signals of completion. By defining event handlers for these three, you can catch these events asynchronously
• Each stream has multiple methods that, when called, return a new stream based on the original stream. The original stream is not modified to ensure immutability
• Data streams support chain calls, and you can combine different functions to process streams, create and filter different streams. Even one or more streams can serve as inputs to another stream. You can merge two data streams. You can filter a data stream and get a data stream containing the events you are interested in. You can map values from one data stream to another

var iterable = [1, 2];
var iterator = iterable[Symbol.iterator]();
iterator.next(); // => { value: "1".done: false}
iterator.next(); // => { value: "2".done: false}
iterator.next(); // => { value: undefined, done: true}Copy the code

1. Promises are essentially observables, and you can use fromPromises to turn promises into Observables
2. But promise.then () returns only one value, and observables can return multiple values
3. A Promise is either resolve or reject, and only responds once. An Observable can respond multiple times
4. Promises can’t be canceled, and an Observable can unsubscribe by calling unsubscribe()

Problem solved

• Unification of synchronous and asynchronous
• Composable data change process
• Precise binding of data and view
• Automatic recalculation after a condition change

The core concept

var observable = Rx.Observable
// Create an Observable with the create method
// The callback takes an observer argument, that is, the observer role
	.create(function(observer) {
		observer.next('hi');
		observer.next('world');

        setTimeout((a)= > {
			observer.next('This is an asynchronous operation.');
		}, 30)})// Subscribe to this Observable
// Only after subscribing will observer methods be called and notified when the stream Observable changes
// You can also unsubscribe by calling unsubscribe()
console.log('start')
var subscription = observable.subscribe(function(value) {
	console.log(value);
})
console.log('end')
setTimeOut((a)= > {
  subscription.unsubscribe()
}, 5000)


// The program will output in turn
'start'
"hi"
'world'
'end'
'This is an asynchronous operation.'Copy the code

Rx.Observable.of(2)
             .map(v= > v * 2) 
             .subscribe(v= > console.log('output:' + v));
// output:4Copy the code

var Observer = {
    next(value) { /* Process the value */ },
    error(error) { /* Handle exceptions */ },
    complete() { /* Process the completed state */}}; Next (): receives Observable values error(): not used in iteratorstry catchThe Observer receives an error with an error method (optional) complete(): Fires an action when no new data is sent (optional)Copy the code

• Each Subject is an Observable that can subscribe to it. From the perspective of the Observer, it can’t distinguish its execution environment from the single-path push of a common Observable or the multi-path push based on Subject
• Each Subject can also be an Observer, since it also consists of next, error, and complete methods. By calling next, the Subject multiplexes the current value to all observers registered on it

// Create an Observable that outputs one number per second and ends with three
var source = Rx.Observable.interval(1000).take(3);

// Define two observers
var observerA = {
    next: value= > console.log('A next: ' + value),
    error: error= > console.log('A error: ' + error),
    complete: (a)= > console.log('A complete! ')}var observerB = {
    next: value= > console.log('B next: ' + value),
    error: error= > console.log('B error: ' + error),
    complete: (a)= > console.log('B complete! ')}// Create a subject -- a special Observable
var subject = new Rx.Subject()

// observerA Subscribe Subject
subject.subscribe(observerA)

// The Subject subscribes to Observable as an observer
source.subscribe(subject);

setTimeout((a)= > {
    subject.subscribe(observerB);
}, 1000);

/ / output:
// "A next: 0"
// "A next: 1"
// "B next: 1"
// "A next: 2"
// "B next: 2"
// "A complete!"
// "B complete!"A and B are independent of each otherCopy the code

• Queue – When recursing, queue blocks the recursion to avoid unnecessary performance loss
• Asap — as soon as possible, expressed as setTimeout is 0. Used mostly for observables that never unsubscribe, such as polling
• Async – Specifies that an Observable executes asynchronously, using setInterval.
• AnimationFrame — Window. The requestAnimationFrame API implementation, suitable for high frequency trigger UI animation

var observable = Rx.Observable.create(function (observer) {
    observer.next(1);
    observer.next(2);
    observer.next(3);
    observer.complete();
});

console.log('before subscribe');
observable.observeOn(Rx.Scheduler.async) // It was synchronous, but now it is asynchronous
.subscribe({
    next: (value) = > { console.log(value); },
    error: (err) = > { console.log('Error: ' + err); },
    complete: (a)= > { console.log('complete'); }});console.log('after subscribe');

// "before subscribe"
// "after subscribe"
/ / 1
/ / 2
/ / 3
// "complete"Copy the code

Application scenarios

1. There’s a lot of data, a lot of manipulation of data
2. The data displayed is a combination of multiple data, such as task, corresponding owner, tag, and so on
3. Updates to the same data may come from different originators
4. New data should require the same data processing rules as the original

1. Data is obtained by caching and asynchronously
2. Each data flow pipe is combined, and the superposition of the streams is the final data
3. Get and subscribe together, so you don’t need to know where the data comes from, right
4. The current and future data merges are then processed using the same API to ensure the same rules for data

other

data

• RxJS 5
• The introduction to Reactive Programming you' ve been missing
• Flowing Data – Use RxJS to construct data logic for complex single-page applications