A representation of any set of values over any amount of time. This is the most basic building block of RxJS.

class Observable<T> implements Subscribable {
  static create: Function
  static if: typeof iif
  static throw: typeof throwError
  constructor(subscribe?: (this: Observable<T>, subscriber: Subscriber<T>) => TeardownLogic)
  _isScalar: boolean
  source: Observable<any>
  operator: Operator<any, T>
  lift<R>(operator: Operator<T, R>): Observable<R>
  subscribe(observerOrNext?: PartialObserver<T> | ((value: T) => void), error?: (error: any) => void, complete?: () => void): Subscription
  _trySubscribe(sink: Subscriber<T>): TeardownLogic
  forEach(next: (value: T) => void, promiseCtor?: PromiseConstructorLike): Promise<void>
  pipe(...operations: OperatorFunction<any, any>[]): Observable<any>
  toPromise(promiseCtor?: PromiseConstructorLike): Promise<T>
}

Static Properties

Property	Type	Description
create	Function	Creates a new cold Observable by calling the Observable constructor
if	typeof iif
throw	typeof throwError

Constructor

constructor(subscribe?: (this: Observable<T>, subscriber: Subscriber<T>) => TeardownLogic)

Parameters

subscribe

Optional. Default is undefined. the function that is called when the Observable is initially subscribed to. This function is given a Subscriber, to which new values can be nexted, or an error method can be called to raise an error, or complete can be called to notify of a successful completion.

Properties

Property	Type	Description
_isScalar	boolean	Internal implementation detail, do not use directly.
source	Observable<any>
operator	Operator<any, T>

Methods

lift<R>(operator: Operator<T, R>): Observable<R>

Creates a new Observable, with this Observable as the source, and the passed operator defined as the new observable's operator.

Parameters

operator

the operator defining the operation to take on the observable

Returns

Observable<R>: a new observable with the Operator applied

Use it when you have all these Observables, but still nothing is happening.

subscribe is not a regular operator, but a method that calls Observable's internal subscribe function. It might be for example a function that you passed to Observable's constructor, but most of the time it is a library implementation, which defines what will be emitted by an Observable, and when it be will emitted. This means that calling subscribe is actually the moment when Observable starts its work, not when it is created, as it is often the thought.

Apart from starting the execution of an Observable, this method allows you to listen for values that an Observable emits, as well as for when it completes or errors. You can achieve this in two of the following ways.

The first way is creating an object that implements Observer interface. It should have methods defined by that interface, but note that it should be just a regular JavaScript object, which you can create yourself in any way you want (ES6 class, classic function constructor, object literal etc.). In particular do not attempt to use any RxJS implementation details to create Observers - you don't need them. Remember also that your object does not have to implement all methods. If you find yourself creating a method that doesn't do anything, you can simply omit it. Note however, if the error method is not provided, all errors will be left uncaught.

The second way is to give up on Observer object altogether and simply provide callback functions in place of its methods. This means you can provide three functions as arguments to subscribe, where the first function is equivalent of a next method, the second of an error method and the third of a complete method. Just as in case of Observer, if you do not need to listen for something, you can omit a function, preferably by passing undefined or null, since subscribe recognizes these functions by where they were placed in function call. When it comes to error function, just as before, if not provided, errors emitted by an Observable will be thrown.

Whichever style of calling subscribe you use, in both cases it returns a Subscription object. This object allows you to call unsubscribe on it, which in turn will stop the work that an Observable does and will clean up all resources that an Observable used. Note that cancelling a subscription will not call complete callback provided to subscribe function, which is reserved for a regular completion signal that comes from an Observable.

Remember that callbacks provided to subscribe are not guaranteed to be called asynchronously. It is an Observable itself that decides when these functions will be called. For example of by default emits all its values synchronously. Always check documentation for how given Observable will behave when subscribed and if its default behavior can be modified with a scheduler.

Example

Subscribe with an Observer

import { of } from 'rxjs';

const sumObserver = {
  sum: 0,
  next(value) {
    console.log('Adding: ' + value);
    this.sum = this.sum + value;
  },
  error() {
    // We actually could just remove this method,
    // since we do not really care about errors right now.
  },
  complete() {
    console.log('Sum equals: ' + this.sum);
  }
};

of(1, 2, 3) // Synchronously emits 1, 2, 3 and then completes.
  .subscribe(sumObserver);

// Logs:
// "Adding: 1"
// "Adding: 2"
// "Adding: 3"
// "Sum equals: 6"

Subscribe with functions

import { of } from 'rxjs'

let sum = 0;

of(1, 2, 3).subscribe(
  value => {
    console.log('Adding: ' + value);
    sum = sum + value;
  },
  undefined,
  () => console.log('Sum equals: ' + sum)
);

// Logs:
// "Adding: 1"
// "Adding: 2"
// "Adding: 3"
// "Sum equals: 6"

Cancel a subscription

import { interval } from 'rxjs';

const subscription = interval(1000).subscribe(
  num => console.log(num),
  undefined,
  () => {
    // Will not be called, even when cancelling subscription.
    console.log('completed!');
  }
);

setTimeout(() => {
  subscription.unsubscribe();
  console.log('unsubscribed!');
}, 2500);

// Logs:
// 0 after 1s
// 1 after 2s
// "unsubscribed!" after 2.5s

_trySubscribe(sink: Subscriber<T>): TeardownLogic

Parameters

sink	Type: Subscriber.

Returns

TeardownLogic

forEach(next: (value: T) => void, promiseCtor?: PromiseConstructorLike): Promise<void>

Parameters

next	a handler for each value emitted by the observable
promiseCtor	Optional. Default is undefined. a constructor function used to instantiate the Promise

Returns

Promise<void>: a promise that either resolves on observable completion or rejects with the handled error