RxJS Tip: Understand the Terminology: Observable

To get the most from RxJS, it's important to understand its terminology and one of the key terms is Observable.

What Is an Observable?

An Observable is a collection of items over time. It is one of the key building blocks of RxJS.

A normal collection, such as an array, retains items you can access. An Observable doesn't retain items. You can only observe items as they are emitted.

In that respect, an Observable is like a drive-up window: The person at the window emits food items over time.

What Does an Observable Do?

An Observable doesn't do anything until a consumer subscribes. (More on subscriptions in a later post.)

When subscribed, the Observable begins emitting items or notifications to that consumer.

An Observable provides the following notifications:

next: The next item is emitted
error: An error occurred and no more items will be emitted
complete: No more items will be emitted

How Does an Observable Emit?

An Observable can emit items synchronously or asynchronously.

An Observable can emit one item and complete, such as the response returned from an asynchronous Http request.

An Observable can emit multiple items and complete.

An Observable can emit an infinite number of items, such as the location of each mouse move or key press.

Here is an oversimplified version of a marble diagram depicting two Observables.

The first Observable is "one and done", meaning it emits once and then completes. This is the type of Observable you'll get when using Angular's Http methods. In this example, the emitted returned response is an array of products.

The second Observable is "infinite", meaning it will continue to emit values until it's completed. In this example, it's emitting each key press.

How Do You Create an Observable?

With Angular, an Observable is automatically created and returned when using features, such as Http.

  products$ = this.http.get<Product[]>(this.productsUrl)
    .pipe(
      tap(data => console.log(JSON.stringify(data))),
      catchError(this.handleError)
    );

You can create your own Observable with the new keyword.

const source$ = new Observable();

However, this technique is not often used. In most cases, it's best to create an Observable using a creation function, discussed in a later post.

See the comments section of this post for a good example of when you would want to use new Observable() to wrap the functionality of a third-party API.

I hope that's clarified the meaning of the term Observable.

Comments

[Maxime]

const source$ = new Observable(); // Not best practice

I think you should clarify your thinking here because doing so can definitely be legit to wrap resources into observables and cleanup once the stream is erroring or complete. For example you could use this to create a cold observable which would add some 3D shape on a scene in webgl and whenever you unsubscribe from that observable you would run the code to remove it from the scene. Dealing with such an observable in a switchmap for example will make your life way easier and your code cleaner by encapsulating the creation and the removal of 3rd party objects/data without having to deal with both outside of this observable within some tap where you have to remember to do so

[Deborah Kurata]

Thank you for your comments. Could you provide a specific code example? I had understood the guidance to be to use a creation function instead of new Observable() where possible. And that we should only use new when creating a Subject/BehaviorSubject. But I'm happy to correct if my information is not accurate. Thanks!

[Maxime]

Sure thing :).

Imagine you're using a library like ngx-toastr: npmjs.com/package/ngx-toastr

where the API lets you create a toast message with the following API:

import { ToastrService } from 'ngx-toastr';

@Component({
  // ...
})
export class YourComponent {
  constructor(private toastr: ToastrService) {
    this.toastr.info(`Some message`, `Some title`);
  }
}

Now the code above will open a toast (and as soon as the default timeout is reached it'll be closed).

Let say that now, we have an HTTP call which may take super long, so we want the user to be aware of the status.

You could do the following:

import { ToastrService } from 'ngx-toastr';
import { tap } from 'rxjs/operators';

@Component({
  // ...
})
export class YourComponent {
  public res;

  constructor(private http: HttpClient, private toastr: ToastrService) {
    const toastRef = this.toastr.info(
      `Please wait as it may take a moment`,
      `Some work is being triggered`,
      {
        // the call may take longer than the toast timeout in this case
        // so we turn the timeout off
        disableTimeOut: true,
      }
    ).toastRef;

    this.http
      .get(
        // some request that'll take long
        `http://my-api.com`
      )
      .pipe(
        tap((res) => {
          this.res = res;

          this.toastRef.close();
        }),
        catchError(() => {
          this.toastRef.close();
        })
      )
      .subscribe();
  }
}

What you could do instead would be something like this:

import { ToastrService } from 'ngx-toastr';
import { Observable } from 'rxjs';
import { take, exhaustMap, shareReplay } from 'rxjs/operators';

@Component({
  // ...
})
export class YourComponent {
  private toastPleaseWait$: Observable<void> = new Observable<void>(
    (observer) => {
      const toastReference = this.toastr.info(
        `Please wait as it may take a moment`,
        `Some work is being triggered`,
        {
          // the call may take longer than the toast timeout in this case
          // so we turn the timeout off
          disableTimeOut: true,
        }
      );
      observer.next();

      // this callback will be closed whenever the stream is closed
      // and this is where we can perform some cleanup, in this case
      // closing the toast
      return () => toastReference.toastRef.close();
    }
  );

  // subscribe in the view with async to display the result
  public res$ = this.toastPleaseWait$.pipe(
    exhaustMap(() =>
      this.http.get(
        // some request that'll take long
        `http://my-api.com`
      )
    ),
    // make sure we close the stream so that toast$ is also
    // getting closed as soon as we've got our result and
    // therefore will run its cleanup callback
    take(1)
  );

  constructor(private http: HttpClient, private toastr: ToastrService) {}
}

The nice about this approach IMO is that the 3rd party resource manipulation which doesn't have a reactive public API is wrapped into a cold observable (which we could subscribe to multiple times) and no one down the stream has to worry about what was created and requires to be cleaned up.

It would also let us take advantage of Rxjs further if for example we were using a retry on the HTTP call. If one HTTP call was to start: the popup would show. As soon as it's completed or throws, the popup would be closed. If we do add retry(3) at the end of our stream, and the call was failing the first 2 times then succeeding the third one, we'd see:

• popup opening
• (fail) popup closing
• (automatic retry) popup opening
• (fail) popup closing
• (automatic retry) popup opening
• (success) popup closing

In conclusion, I'd say that when you have to work with APIs that are not exposing some reactive bindings you can isolate the creation and teardown logic of those by using new Observable(...) while operators like of, from etc would not be able to let you handle the cleanup.

Hope it's clear :)!

[Deborah Kurata]

Nice! Thanks for providing your example. I would have implemented this as an external Subject instead of wrapping the third-party API in a new Observable. This solution looks much cleaner. I'll adjust my comment. Thanks again!

[Chandler Baskins]

Thanks for sharing your knowledge!