Observable Anatomy

An observable’s subscribe method has the following signature

  1. stream.subscribe(fnValue, fnError, fnComplete)

The first one is being demonstrated below fnValue

  1. let stream$ = Rx.Observable.create((observer) => {
  2.   observer.next(1)
  3. });
  5. stream$.subscribe((data) => {
  6.   console.log('Data', data);
  7. })
  9. // 1

When observer.next(<value>) is being called the fnValue is being invoked.

The second callback fnError is the error callback and is being invoked by the following code, i.e observer.error(<message>)

  1. let stream$ = Rx.Observable.create((observer) => {
  2.    observer.error('error message');
  3. })
  5. stream$.subscribe(
  6.    (data) => console.log('Data', data)),
  7.    (error) => console.log('Error', error)

Lastly we have the fnComplete and it should be invoked when a stream is done and has no more values to emit. It is triggered by a call to observer.complete() like so:

  1. let stream$ = Rx.Observable.create((observer) => {
  2.    // x calls to observer.next(<value>)
  3.    observer.complete();
  4. })

Unsubscribe

So far we have been creating an irresponsible Observable, irresponsible in the sense that it doesn’t clean up after itself. So let’s look at how to do that:

  1. let stream$ = new Rx.Observable.create((observer) => {
  2.   let i = 0;
  3.   let id = setInterval(() => {
  4.     observer.next(i++);
  5.   },1000)
  7.   return function(){
  8.     clearInterval( id );
  9.   }
  10. })
  12. let subscription = stream$.subscribe((value) => {
  13.   console.log('Value', value)
  14. });
  16. setTimeout(() => {
  17.   subscription.unsubscribe() // here we invoke the cleanup function
  19. }, 3000)

So ensure that you

  • Define a function that cleans up
  • Implicitely call that function by calling subscription.unsubscribe()