Difference between the methods .pipe() and .subscribe() on a RXJS observable

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 and an error happens, it will be thrown asynchronously. Errors thrown asynchronously cannot be caught using try/catch. Instead, use the onUnhandledError configuration option or use a runtime handler (like window.onerror or process.on('error)) to be notified of unhandled errors. Because of this, it’s recommended that you provide an error method to avoid missing thrown errors.

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 an Observer, if you do not need to listen for something, you can omit a function by passing undefined or null, since subscribe recognizes these functions by where they were placed in function call. When it comes to the error function, as with an Observer, if not provided, errors emitted by an Observable will be thrown asynchronously.

You can, however, subscribe with no parameters at all. This may be the case where you’re not interested in terminal events and you also handled emissions internally by using operators (e.g. using tap).

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.

Examples

Subscribe with an Observer

1. import { of } from ‘rxjs’;
2.  
3. const sumObserver = {
4. sum: 0,
5. next(value) {
6. console.log(‘Adding: ‘ + value);
7. this.sum = this.sum + value;
8. },
9. error() {
10. // We actually could just remove this method,
11. // since we do not really care about errors right now.
12. },
13. complete() {
14. console.log(‘Sum equals: ‘ + this.sum);
15. }
16. };
17.  
18. of(1, 2, 3) // Synchronously emits 1, 2, 3 and then completes.
19. .subscribe(sumObserver);
20.  
21. // Logs:
22. // ‘Adding: 1’
23. // ‘Adding: 2’
24. // ‘Adding: 3’
25. // ‘Sum equals: 6’

Subscribe with functions (deprecated)

1. import { of } from ‘rxjs’
2.  
3. let sum = 0;
4.  
5. of(1, 2, 3).subscribe(
6. value => {
7. console.log(‘Adding: ‘ + value);
8. sum = sum + value;
9. },
10. undefined,
11. () => console.log(‘Sum equals: ‘ + sum)
12. );
13.  
14. // Logs:
15. // ‘Adding: 1’
16. // ‘Adding: 2’
17. // ‘Adding: 3’
18. // ‘Sum equals: 6’

Cancel a subscription

1. import { interval } from ‘rxjs’;
2.  
3. const subscription = interval(1000).subscribe({
4. next(num) {
5. console.log(num)
6. },
7. complete() {
8. // Will not be called, even when cancelling subscription.
9. console.log(‘completed!’);
10. }
11. });
12.  
13. setTimeout(() => {
14. subscription.unsubscribe();
15. console.log(‘unsubscribed!’);
16. }, 2500);
17.  
18. // Logs:
19. // 0 after 1s
20. // 1 after 2s
21. // ‘unsubscribed!’ after 2.5s

reactivex.io/documentation/operators/subscribe.html A typical implementation of the Subscribe operator may accept one to three methods (which then constitute the observer), or it may accept an object (sometimes called an Observer or Subscriber) that implements the interface which includes those three methods:

onNext

An Observable calls this method whenever the Observable emits an item. This method takes as a parameter the item emitted by the Observable.

onError

An Observable calls this method to indicate that it has failed to generate the expected data or has encountered some other error. This stops the Observable and it will not make further calls to onNext or onCompleted. The onError method takes as its parameter an indication of what caused the error (sometimes an object like an Exception or Throwable, other times a simple string, depending on the implementation).

onCompleted

An Observable calls this method after it has called onNext for the final time, if it has not encountered any errors.

Difference between the methods .pipe() and .subscribe() on a RXJS observable

First, consider this function deposit() – It returns the Subscription object, becuase thats what is created when you call a .subscribe().

    deposit(account, amount){
    return this.http.get('url')
    .subscribe(res => {
        return res;
    }
}

And now with .pipe()

    deposit(account, amount){
    return this.http.get('url')
    .pipe(
        map(res => {
            return res;
        });
    );
}

In the second case, while using pipe, if you do not subscribe, nothing happens. pipe just combines several operators together. The second example return an Observable, but it does not execute. So if I want to actually get the emitted value of the Observable, then I have to use .subscirbe() after using .pipe()

The pipe method is for chaining observable operators, and the subscribe is for activating the observable and listening for emitted values.

The pipe method was added to allow webpack to drop unused operators from the final JavaScript bundle. It makes it easier to build smaller files.

Previous

Hot vs Cold Observables

Next

Route-Parameters