Next error complete rxjs

Name already in use

angular / aio / content / guide / observables.md

• Go to file T
• Go to line L
• Copy path
• Copy permalink

Copy raw contents

Copy raw contents

Using observables to pass values

Observables provide support for passing messages between parts of your application. They are used frequently in Angular and are a technique for event handling, asynchronous programming, and handling multiple values.

The observer pattern is a software design pattern in which an object, called the subject, maintains a list of its dependents, called observers, and notifies them automatically of state changes. This pattern is similar (but not identical) to the publish/subscribe design pattern.

Observables are declarative —that is, you define a function for publishing values, but it is not executed until a consumer subscribes to it. The subscribed consumer then receives notifications until the function completes, or until they unsubscribe.

An observable can deliver multiple values of any type —literals, messages, or events, depending on the context. The API for receiving values is the same whether the values are delivered synchronously or asynchronously. Because setup and teardown logic are both handled by the observable, your application code only needs to worry about subscribing to consume values, and when done, unsubscribing. Whether the stream was keystrokes, an HTTP response, or an interval timer, the interface for listening to values and stopping listening is the same.

Because of these advantages, observables are used extensively within Angular, and for application development as well.

Basic usage and terms

As a publisher, you create an Observable instance that defines a subscriber function. This is the function that is executed when a consumer calls the subscribe() method. The subscriber function defines how to obtain or generate values or messages to be published.

To execute the observable you have created and begin receiving notifications, you call its subscribe() method, passing an observer. This is a JavaScript object that defines the handlers for the notifications you receive. The subscribe() call returns a Subscription object that has an unsubscribe() method, which you call to stop receiving notifications.

Here’s an example that demonstrates the basic usage model by showing how an observable could be used to provide geolocation updates.

A handler for receiving observable notifications implements the Observer interface. It is an object that defines callback methods to handle the three types of notifications that an observable can send:

An observer object can define any combination of these handlers. If you don’t supply a handler for a notification type, the observer ignores notifications of that type.

An Observable instance begins publishing values only when someone subscribes to it. You subscribe by calling the subscribe() method of the instance, passing an observer object to receive the notifications.

In order to show how subscribing works, we need to create a new observable. There is a constructor that you use to create new instances, but for illustration, we can use some methods from the RxJS library that create simple observables of frequently used types:

Here’s an example of creating and subscribing to a simple observable, with an observer that logs the received message to the console:

Alternatively, the subscribe() method can accept callback function definitions in line, for next , error , and complete handlers. For example, the following subscribe() call is the same as the one that specifies the predefined observer:

In either case, a next handler is required. The error and complete handlers are optional.

NOTE:
A next() function could receive, for instance, message strings, or event objects, numeric values, or structures, depending on context. As a general term, we refer to data published by an observable as a stream. Any type of value can be represented with an observable, and the values are published as a stream.

Use the Observable constructor to create an observable stream of any type. The constructor takes as its argument the subscriber function to run when the observable’s subscribe() method executes. A subscriber function receives an Observer object, and can publish values to the observer’s next() method.

For example, to create an observable equivalent to the of(1, 2, 3) above, you could do something like this:

To take this example a little further, we can create an observable that publishes events. In this example, the subscriber function is defined inline.

Now you can use this function to create an observable that publishes keydown events:

A typical observable creates a new, independent execution for each subscribed observer. When an observer subscribes, the observable wires up an event handler and delivers values to that observer. When a second observer subscribes, the observable then wires up a new event handler and delivers values to that second observer in a separate execution.

Sometimes, instead of starting an independent execution for each subscriber, you want each subscription to get the same values —even if values have already started emitting. This might be the case with something like an observable of clicks on the document object.

Multicasting is the practice of broadcasting to a list of multiple subscribers in a single execution. With a multicasting observable, you don’t register multiple listeners on the document, but instead re-use the first listener and send values out to each subscriber.

When creating an observable you should determine how you want that observable to be used and whether or not you want to multicast its values.

Let’s look at an example that counts from 1 to 3, with a one-second delay after each number emitted.

Notice that if you subscribe twice, there will be two separate streams, each emitting values every second. It looks something like this:

Changing the observable to be multicasting could look something like this:

Multicasting observables take a bit more setup, but they can be useful for certain applications. Later we will look at tools that simplify the process of multicasting, allowing you to take any observable and make it multicasting.

Because observables produce values asynchronously, try/catch will not effectively catch errors. Instead, you handle errors by specifying an error callback on the observer. Producing an error also causes the observable to clean up subscriptions and stop producing values. An observable can either produce values (calling the next callback), or it can complete, calling either the complete or error callback.

Error handling (and specifically recovering from an error) is covered in more detail in a later section.

Источник

Introduction

RxJS is a library for composing asynchronous and event-based programs by using observable sequences. It provides one core type, the Observable, satellite types (Observer, Schedulers, Subjects) and operators inspired by Array#extras (map, filter, reduce, every, etc) to allow handling asynchronous events as collections.

Think of RxJS as Lodash for events.

ReactiveX combines the Observer pattern with the Iterator pattern and functional programming with collections to fill the need for an ideal way of managing sequences of events.

The essential concepts in RxJS which solve async event management are:

• Observable: represents the idea of an invokable collection of future values or events.
• Observer: is a collection of callbacks that knows how to listen to values delivered by the Observable.
• Subscription: represents the execution of an Observable, is primarily useful for cancelling the execution.
• Operators: are pure functions that enable a functional programming style of dealing with collections with operations like map , filter , concat , flatMap , etc.
• Subject: is the equivalent to an EventEmitter, and the only way of multicasting a value or event to multiple Observers.
• Schedulers: are centralized dispatchers to control concurrency, allowing us to coordinate when computation happens on e.g. setTimeout or requestAnimationFrame or others.

First examples

Normally you register event listeners.

Using RxJS you create an observable instead.

Purity

What makes RxJS powerful is its ability to produce values using pure functions. That means your code is less prone to errors.

Normally you would create an impure function, where other pieces of your code can mess up your state.

Using RxJS you isolate the state.

The scan operator works just like reduce for arrays. It takes a value which is exposed to a callback. The returned value of the callback will then become the next value exposed the next time the callback runs.

RxJS has a whole range of operators that helps you control how the events flow through your observables.

This is how you would allow at most one click per second, with plain JavaScript:

Values

You can transform the values passed through your observables.

Here’s how you can add the current mouse x position for every click, in plain JavaScript:

Observable

Observables are lazy Push collections of multiple values. They fill the missing spot in the following table:

Example. The following is an Observable that pushes the values 1 , 2 , 3 immediately (synchronously) when subscribed, and the value 4 after one second has passed since the subscribe call, then completes:

To invoke the Observable and see these values, we need to subscribe to it:

Which executes as such on the console:

Pull versus Push

Pull and Push are two different protocols that describe how a data Producer can communicate with a data Consumer.

What is Pull? In Pull systems, the Consumer determines when it receives data from the data Producer. The Producer itself is unaware of when the data will be delivered to the Consumer.

Every JavaScript Function is a Pull system. The function is a Producer of data, and the code that calls the function is consuming it by «pulling» out a single return value from its call.

ES2015 introduced generator functions and iterators ( function* ), another type of Pull system. Code that calls iterator.next() is the Consumer, «pulling» out multiple values from the iterator (the Producer).

What is Push? In Push systems, the Producer determines when to send data to the Consumer. The Consumer is unaware of when it will receive that data.

Promises are the most common type of Push system in JavaScript today. A Promise (the Producer) delivers a resolved value to registered callbacks (the Consumers), but unlike functions, it is the Promise which is in charge of determining precisely when that value is «pushed» to the callbacks.

RxJS introduces Observables, a new Push system for JavaScript. An Observable is a Producer of multiple values, «pushing» them to Observers (Consumers).

• A Function is a lazily evaluated computation that synchronously returns a single value on invocation.
• A generator is a lazily evaluated computation that synchronously returns zero to (potentially) infinite values on iteration.
• A Promise is a computation that may (or may not) eventually return a single value.
• An Observable is a lazily evaluated computation that can synchronously or asynchronously return zero to (potentially) infinite values from the time it’s invoked onwards.

Observables as generalizations of functions

Contrary to popular claims, Observables are not like EventEmitters nor are they like Promises for multiple values. Observables may act like EventEmitters in some cases, namely when they are multicasted using RxJS Subjects, but usually they don’t act like EventEmitters.

Observables are like functions with zero arguments, but generalize those to allow multiple values.

Consider the following:

We expect to see as output:

You can write the same behavior above, but with Observables:

And the output is the same:

This happens because both functions and Observables are lazy computations. If you don’t call the function, the console.log(‘Hello’) won’t happen. Also with Observables, if you don’t «call» it (with subscribe ), the console.log(‘Hello’) won’t happen. Plus, «calling» or «subscribing» is an isolated operation: two function calls trigger two separate side effects, and two Observable subscribes trigger two separate side effects. As opposed to EventEmitters which share the side effects and have eager execution regardless of the existence of subscribers, Observables have no shared execution and are lazy.

Subscribing to an Observable is analogous to calling a Function.

Some people claim that Observables are asynchronous. That is not true. If you surround a function call with logs, like this:

You will see the output:

And this is the same behavior with Observables:

And the output is:

Which proves the subscription of foo was entirely synchronous, just like a function.

Observables are able to deliver values either synchronously or asynchronously.

What is the difference between an Observable and a function? Observables can «return» multiple values over time, something which functions cannot. You can’t do this:

Functions can only return one value. Observables, however, can do this:

With synchronous output:

But you can also «return» values asynchronously:

• func.call() means «give me one value synchronously«
• observable.subscribe() means «give me any amount of values, either synchronously or asynchronously«

Anatomy of an Observable

Observables are created using Rx.Observable.create or a creation operator, are subscribed to with an Observer, execute to deliver next / error / complete notifications to the Observer, and their execution may be disposed. These four aspects are all encoded in an Observable instance, but some of these aspects are related to other types, like Observer and Subscription.

Core Observable concerns:

• Creating Observables
• Subscribing to Observables
• Executing the Observable
• Disposing Observables

Creating Observables

Rx.Observable.create is an alias for the Observable constructor, and it takes one argument: the subscribe function.

The following example creates an Observable to emit the string ‘hi’ every second to an Observer.

Observables can be created with create , but usually we use the so-called creation operators, like of , from , interval , etc.

In the example above, the subscribe function is the most important piece to describe the Observable. Let’s look at what subscribing means.

Subscribing to Observables

The Observable observable in the example can be subscribed to, like this:

It is not a coincidence that observable.subscribe and subscribe in Observable.create(function subscribe(observer) <. >) have the same name. In the library, they are different, but for practical purposes you can consider them conceptually equal.

This shows how subscribe calls are not shared among multiple Observers of the same Observable. When calling observable.subscribe with an Observer, the function subscribe in Observable.create(function subscribe(observer) <. >) is run for that given Observer. Each call to observable.subscribe triggers its own independent setup for that given Observer.

Subscribing to an Observable is like calling a function, providing callbacks where the data will be delivered to.

This is drastically different to event handler APIs like addEventListener / removeEventListener . With observable.subscribe , the given Observer is not registered as a listener in the Observable. The Observable does not even maintain a list of attached Observers.

A subscribe call is simply a way to start an «Observable execution» and deliver values or events to an Observer of that execution.

Executing Observables

The code inside Observable.create(function subscribe(observer) <. >) represents an «Observable execution», a lazy computation that only happens for each Observer that subscribes. The execution produces multiple values over time, either synchronously or asynchronously.

There are three types of values an Observable Execution can deliver:

• «Next» notification: sends a value such as a Number, a String, an Object, etc.
• «Error» notification: sends a JavaScript Error or exception.
• «Complete» notification: does not send a value.

Next notifications are the most important and most common type: they represent actual data being delivered to an Observer. Error and Complete notifications may happen only once during the Observable Execution, and there can only be either one of them.

These constraints are expressed best in the so-called Observable Grammar or Contract, written as a regular expression:

In an Observable Execution, zero to infinite Next notifications may be delivered. If either an Error or Complete notification is delivered, then nothing else can be delivered afterwards.

The following is an example of an Observable execution that delivers three Next notifications, then completes:

Observables strictly adhere to the Observable Contract, so the following code would not deliver the Next notification 4 :

It is a good idea to wrap any code in subscribe with try / catch block that will deliver an Error notification if it catches an exception:

Disposing Observable Executions

Because Observable Executions may be infinite, and it’s common for an Observer to want to abort execution in finite time, we need an API for canceling an execution. Since each execution is exclusive to one Observer only, once the Observer is done receiving values, it has to have a way to stop the execution, in order to avoid wasting computation power or memory resources.

When observable.subscribe is called, the Observer gets attached to the newly created Observable execution. This call also returns an object, the Subscription :

The Subscription represents the ongoing execution, and has a minimal API which allows you to cancel that execution. Read more about the Subscription type here. With subscription.unsubscribe() you can cancel the ongoing execution:

When you subscribe, you get back a Subscription, which represents the ongoing execution. Just call unsubscribe() to cancel the execution.

Each Observable must define how to dispose resources of that execution when we create the Observable using create() . You can do that by returning a custom unsubscribe function from within function subscribe() .

For instance, this is how we clear an interval execution set with setInterval :

Just like observable.subscribe resembles Observable.create(function subscribe() <. >) , the unsubscribe we return from subscribe is conceptually equal to subscription.unsubscribe . In fact, if we remove the ReactiveX types surrounding these concepts, we’re left with rather straightforward JavaScript.

The reason why we use Rx types like Observable, Observer, and Subscription is to get safety (such as the Observable Contract) and composability with Operators.

Observer

What is an Observer? An Observer is a consumer of values delivered by an Observable. Observers are simply a set of callbacks, one for each type of notification delivered by the Observable: next , error , and complete . The following is an example of a typical Observer object:

To use the Observer, provide it to the subscribe of an Observable:

Observers are just objects with three callbacks, one for each type of notification that an Observable may deliver.

Observers in RxJS may also be partial. If you don’t provide one of the callbacks, the execution of the Observable will still happen normally, except some types of notifications will be ignored, because they don’t have a corresponding callback in the Observer.

The example below is an Observer without the complete callback:

When subscribing to an Observable, you may also just provide the callbacks as arguments, without being attached to an Observer object, for instance like this:

Internally in observable.subscribe , it will create an Observer object using the first callback argument as the next handler. All three types of callbacks may be provided as arguments:

Subscription

What is a Subscription? A Subscription is an object that represents a disposable resource, usually the execution of an Observable. A Subscription has one important method, unsubscribe , that takes no argument and just disposes the resource held by the subscription. In previous versions of RxJS, Subscription was called «Disposable».

A Subscription essentially just has an unsubscribe() function to release resources or cancel Observable executions.

Subscriptions can also be put together, so that a call to an unsubscribe() of one Subscription may unsubscribe multiple Subscriptions. You can do this by «adding» one subscription into another:

When executed, we see in the console:

Subscriptions also have a remove(otherSubscription) method, in order to undo the addition of a child Subscription.

Subject

What is a Subject? An RxJS Subject is a special type of Observable that allows values to be multicasted to many Observers. While plain Observables are unicast (each subscribed Observer owns an independent execution of the Observable), Subjects are multicast.

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

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.

Internally to the Subject, subscribe does not invoke a new execution that delivers values. It simply registers the given Observer in a list of Observers, similarly to how addListener usually works in other libraries and languages.

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.

In the example below, we have two Observers attached to a Subject, and we feed some values to the Subject:

With the following output on the console:

Since a Subject is an Observer, this also means you may provide a Subject as the argument to the subscribe of any Observable, like the example below shows:

Which executes as:

With the approach above, we essentially just converted a unicast Observable execution to multicast, through the Subject. This demonstrates how Subjects are the only way of making any Observable execution be shared to multiple Observers.

There are also a few specializations of the Subject type: BehaviorSubject , ReplaySubject , and AsyncSubject .

Multicasted Observables

A «multicasted Observable» passes notifications through a Subject which may have many subscribers, whereas a plain «unicast Observable» only sends notifications to a single Observer.

A multicasted Observable uses a Subject under the hood to make multiple Observers see the same Observable execution.

Under the hood, this is how the multicast operator works: Observers subscribe to an underlying Subject, and the Subject subscribes to the source Observable. The following example is similar to the previous example which used observable.subscribe(subject) :

multicast returns an Observable that looks like a normal Observable, but works like a Subject when it comes to subscribing. multicast returns a ConnectableObservable , which is simply an Observable with the connect() method.

The connect() method is important to determine exactly when the shared Observable execution will start. Because connect() does source.subscribe(subject) under the hood, connect() returns a Subscription, which you can unsubscribe from in order to cancel the shared Observable execution.

Reference counting

Calling connect() manually and handling the Subscription is often cumbersome. Usually, we want to automatically connect when the first Observer arrives, and automatically cancel the shared execution when the last Observer unsubscribes.

Consider the following example where subscriptions occur as outlined by this list:

1. First Observer subscribes to the multicasted Observable
2. The multicasted Observable is connected
3. The next value 0 is delivered to the first Observer
4. Second Observer subscribes to the multicasted Observable
5. The next value 1 is delivered to the first Observer
6. The next value 1 is delivered to the second Observer
7. First Observer unsubscribes from the multicasted Observable
8. The next value 2 is delivered to the second Observer
9. Second Observer unsubscribes from the multicasted Observable
10. The connection to the multicasted Observable is unsubscribed

To achieve that with explicit calls to connect() , we write the following code:

If we wish to avoid explicit calls to connect() , we can use ConnectableObservable’s refCount() method (reference counting), which returns an Observable that keeps track of how many subscribers it has. When the number of subscribers increases from 0 to 1 , it will call connect() for us, which starts the shared execution. Only when the number of subscribers decreases from 1 to 0 will it be fully unsubscribed, stopping further execution.

refCount makes the multicasted Observable automatically start executing when the first subscriber arrives, and stop executing when the last subscriber leaves.

Below is an example:

Which executes with the output:

The refCount() method only exists on ConnectableObservable, and it returns an Observable , not another ConnectableObservable.

BehaviorSubject

One of the variants of Subjects is the BehaviorSubject , which has a notion of «the current value». It stores the latest value emitted to its consumers, and whenever a new Observer subscribes, it will immediately receive the «current value» from the 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.

In the following example, the BehaviorSubject is initialized with the value 0 which the first Observer receives when it subscribes. The second Observer receives the value 2 even though it subscribed after the value 2 was sent.

ReplaySubject

A ReplaySubject is similar to a BehaviorSubject in that it can send old values to new subscribers, but it can also record a part of the Observable execution.

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

When creating a ReplaySubject , you can specify how many values to replay:

You can also specify a window time in milliseconds, besides of the buffer size, to determine how old the recorded values can be. In the following example we use a large buffer size of 100 , but a window time parameter of just 500 milliseconds.

With the following output where the second Observer gets events 3 , 4 and 5 that happened in the last 500 milliseconds prior to its subscription:

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.

The AsyncSubject is similar to the last() operator, in that it waits for the complete notification in order to deliver a single value.

Operators

RxJS is mostly useful for its operators, even though the Observable is the foundation. Operators are the essential pieces that allow complex asynchronous code to be easily composed in a declarative manner.

What are operators?

Operators are methods on the Observable type, such as .map(. ) , .filter(. ) , .merge(. ) , etc. When called, they do not change the existing Observable instance. Instead, they return a new Observable, whose subscription logic is based on the first Observable.

An Operator is a function which creates a new Observable based on the current Observable. This is a pure operation: the previous Observable stays unmodified.

An Operator is essentially a pure function which takes one Observable as input and generates another Observable as output. Subscribing to the output Observable will also subscribe to the input Observable. In the following example, we create a custom operator function that multiplies each value received from the input Observable by 10:

Notice that a subscribe to output will cause input Observable to be subscribed. We call this an «operator subscription chain».

Instance operators versus static operators

What is an instance operator? Typically when referring to operators, we assume instance operators, which are methods on Observable instances. For instance, if the operator multiplyByTen would be an official instance operator, it would look roughly like this:

Instance operators are functions that use the this keyword to infer what is the input Observable.

Notice how the input Observable is not a function argument anymore, it is assumed to be the this object. This is how we would use such instance operator:

What is a static operator? Besides instance operators, static operators are functions attached to the Observable class directly. A static operator uses no this keyword internally, but instead relies entirely on its arguments.

Static operators are pure functions attached to the Observable class, and usually are used to create Observables from scratch.

The most common type of static operators are the so-called Creation Operators. Instead of transforming an input Observable to an output Observable, they simply take a non-Observable argument, like a number, and create a new Observable.

A typical example of a static creation operator would be the interval function. It takes a number (not an Observable) as input argument, and produces an Observable as output:

Another example of a creation operator is create , which we have been using extensively in previous examples. See the list of all static creation operators here.

However, static operators may be of different nature than simply creation. Some Combination Operators may be static, such as merge , combineLatest , concat , etc. These make sense as static operators because they take multiple Observables as input, not just one, for instance:

Marble diagrams

To explain how operators work, textual descriptions are often not enough. Many operators are related to time, they may for instance delay, sample, throttle, or debounce value emissions in different ways. Diagrams are often a better tool for that. Marble Diagrams are visual representations of how operators work, and include the input Observable(s), the operator and its parameters, and the output Observable.

In a marble diagram, time flows to the right, and the diagram describes how values («marbles») are emitted on the Observable execution.

Below you can see the anatomy of a marble diagram.

Throughout this documentation site, we extensively use marble diagrams to explain how operators work. They may be really useful in other contexts too, like on a whiteboard or even in our unit tests (as ASCII diagrams).

Choose an operator

Categories of operators

There are operators for different purposes, and they may be categorized as: creation, transformation, filtering, combination, multicasting, error handling, utility, etc. In the following list you will find all the operators organized in categories.

Creation Operators

Transformation Operators

Filtering Operators

Combination Operators

Multicasting Operators

Error Handling Operators

Utility Operators

Conditional and Boolean Operators

Mathematical and Aggregate Operators

Scheduler

What is a Scheduler? A scheduler controls when a subscription starts and when notifications are delivered. It consists of three components.

• A Scheduler is a data structure. It knows how to store and queue tasks based on priority or other criteria.
• A Scheduler is an execution context. It denotes where and when the task is executed (e.g. immediately, or in another callback mechanism such as setTimeout or process.nextTick, or the animation frame).
• A Scheduler has a (virtual) clock. It provides a notion of «time» by a getter method now() on the scheduler. Tasks being scheduled on a particular scheduler will adhere only to the time denoted by that clock.

A Scheduler lets you define in what execution context will an Observable deliver notifications to its Observer.

In the example below, we take the usual simple Observable that emits values 1 , 2 , 3 synchronously, and use the operator observeOn to specify the async scheduler to use for delivering those values.

Which executes with the output:

Notice how the notifications got value. were delivered after just after subscribe , which is different to the default behavior we have seen so far. This is because observeOn(Rx.Scheduler.async) introduces a proxy Observer between Observable.create and the final Observer. Let’s rename some identifiers to make that distinction obvious in the example code:

The proxyObserver is created in observeOn(Rx.Scheduler.async) , and its next(val) function is approximately the following:

The async Scheduler operates with a setTimeout or setInterval , even if the given delay was zero. As usual, in JavaScript, setTimeout(fn, 0) is known to run the function fn earliest on the next event loop iteration. This explains why got value 1 is delivered to the finalObserver after just after subscribe happened.

The schedule() method of a Scheduler takes a delay argument, which refers to a quantity of time relative to the Scheduler’s own internal clock. A Scheduler’s clock need not have any relation to the actual wall-clock time. This is how temporal operators like delay operate not on actual time, but on time dictated by the Scheduler’s clock. This is specially useful in testing, where a virtual time Scheduler may be used to fake wall-clock time while in reality executing scheduled tasks synchronously.

Scheduler Types

The async Scheduler is one of the built-in schedulers provided by RxJS. Each of these can be created and returned by using static properties of the Scheduler object.

Using Schedulers

You may have already used schedulers in your RxJS code without explicitly stating the type of schedulers to be used. This is because all Observable operators that deal with concurrency have optional schedulers. If you do not provide the scheduler, RxJS will pick a default scheduler by using the principle of least concurrency. This means that the scheduler which introduces the least amount of concurrency that satisfies the needs of the operator is chosen. For example, for operators returning an observable with a finite and small number of messages, RxJS uses no Scheduler, i.e. null or undefined . For operators returning a potentially large or infinite number of messages, queue Scheduler is used. For operators which use timers, async is used.

Because RxJS uses the least concurrency scheduler, you can pick a different scheduler if you want to introduce concurrency for performance purpose. To specify a particular scheduler, you can use those operator methods that take a scheduler, e.g., from([10, 20, 30], Rx.Scheduler.async) .

Static creation operators usually take a Scheduler as argument. For instance, from(array, scheduler) lets you specify the Scheduler to use when delivering each notification converted from the array . It is usually the last argument to the operator. The following static creation operators take a Scheduler argument:

• bindCallback
• bindNodeCallback
• combineLatest
• concat
• empty
• from
• fromPromise
• interval
• merge
• of
• range
• throw
• timer

Use subscribeOn to schedule in what context will the subscribe() call happen. By default, a subscribe() call on an Observable will happen synchronously and immediately. However, you may delay or schedule the actual subscription to happen on a given Scheduler, using the instance operator subscribeOn(scheduler) , where scheduler is an argument you provide.

Use observeOn to schedule in what context will notifications be delivered. As we saw in the examples above, instance operator observeOn(scheduler) introduces a mediator Observer between the source Observable and the destination Observer, where the mediator schedules calls to the destination Observer using your given scheduler .

Instance operators may take a Scheduler as argument.

Time-related operators like bufferTime , debounceTime , delay , auditTime , sampleTime , throttleTime , timeInterval , timeout , timeoutWith , windowTime all take a Scheduler as the last argument, and otherwise operate by default on the Rx.Scheduler.async Scheduler.

Other instance operators that take a Scheduler as argument: cache , combineLatest , concat , expand , merge , publishReplay , startWith .

Notice that both cache and publishReplay accept a Scheduler because they utilize a ReplaySubject. The constructor of a ReplaySubjects takes an optional Scheduler as the last argument because ReplaySubject may deal with time, which only makes sense in the context of a Scheduler. By default, a ReplaySubject uses the queue Scheduler to provide a clock.

Источник