Creating Observables

So now we know what observables are, but how do we create them? There are a lot of different ways to create new observables. We can create them from scratch, from a literal, from some variable, from promises, from events, etc. Let's go over some examples.

From Scratch

The most direct way to create an Observable is via its create method. Let's see what that looks like.

const myObservable = Observable.create(subscriber => {
    subscriber.next(1);
    subscriber.next(2);
    subscriber.next(3);
});

Here we see that the create method takes in a function that will be passed a reference to the subscriber (also called the observer). By calling the subscribers next function, we notify it of new items on the stream.

We can subscribe to our new observable like we could any other observable (because there is no difference between them):

const myObservable = Observable.create(subscriber => {
    subscriber.next(1);
    subscriber.next(2);
    subscriber.next(3);
});

myObservable.subscribe(x => console.log(x));

Try Me

If you were to run the above, you would see 1, 2, and 3 printed out to the console as you would expect.

You can do any logic you want, even calling external functions, inside the body of the function that you pass to create. For example, lets introduce a timeout.

const myObservable = Observable.create(subscriber => {
    subscriber.next(1);
    subscriber.next(2);
    subscriber.next(3);
    setTimeout(() => subscriber.next(4), 1000);
});

myObservable.subscribe(x => console.log(x));

Try Me

If you were to run that, you would see 1, 2, and 3 just as before, but after 1 second passes, you would see 4 printed out to the console.

In addition to next, a subscriber has two other methods available: error and complete that we should call when an error arises or when no more values will be emitted respectively.

Let's see what that looks like:

const myObservable = Observable.create(subscriber => {
    subscriber.next(1);
    subscriber.next(2);
    subscriber.next(3);
    subscriber.complete();
});

myObservable.subscribe(
    x => console.log(x),
    err => console.log(err),
    () => console.log('completed')
);

Try Me

If you were to run the above, you would see 1, 2, 3, and completed printed to the console. Typically you wrap logic in a try catch inside the created observable to handle errors:

const myObservable = Observable.create(subscriber => {
    try {
        subscriber.next(1);
        subscriber.next(2);
        subscriber.next(3);
        throw('An Error')
        subscriber.complete();
    } catch(err) {
        subscriber.error(err);
    }
});

myObservable.subscribe(
    x => console.log(x),
    err => console.log(err),
    () => console.log('completed')
);

myObservable
    .catch(err => console.log('Handled in catch', err))
    .subscribe(
        x => console.log(x),
        err => console.log(err),
        () => console.log('completed'));

Try Me

On the second subscription above, the catch would ...catch the error instead of the onError function inside the subscriber. You will often see this catch method used instead of passing a second function to subscribe.

Thats pretty much it for manually creating one.

Using the `of` Method

If you want to create an observable that simply emits a sequence and completes, you can use the of method. In our create example all we did was emit three numbers and complete. It would get pretty tiresome if we had to do this every time we want to just emit a few things and stop. This is the problem of tries to solve. Let's look at what that would look like compared to the create method:

const myObservable = Observable.create(subscriber => {
    subscriber.next(1);
    subscriber.next(2);
    subscriber.next(3);
    subscriber.complete();
});

const ofObservable = Observable.of(1, 2, 3);

These two observables are functionally identical. If we were to subscribe to it like this:

ofObservable.subscribe(
    x => console.log(x),
    err => console.log(err),
    () => console.log('completed'));

We would see 1, 2, 3, completed printed out to the console.

Try Me

Using the `fromPromise` Method

We have another creation method that converts promises into observables. Explaining promises is outside the scope of this documentation, but all we need to know is that a promise is a thing that will have a value at some point in the future. That is to say it will resolve at some point in the future, either with an error or with a value.

The DOM has a great API for demonstrating this called fetch. Fetch will make a get request to the URL you pass and return a promise. That promise will either resolve with a response object, or will be an error.

Let's look at how we would use fetch with fromPromise:

const myObservable = Observable.fromPromise(fetch('https://null.jsbin.com'));

The response object has a couple of properties, one of which is the status code. So a subscribe to this observable would look something like this:

myObservable.subscribe(
    x => console.log(x.status),
    err => console.log(err),
    () => console.log('completed'));

Try Me

If you were to run that, you would see either the status code or the error printed out to the console, followed by completed. An observable created with fromPromise will only ever emit one item because promises only ever represent one value.

Using the `from` Method

We also have a vanilla from method available that will detect if what is passed in is an array or a promise, and handle it with either of or fromPromise respectively.

const numbers = [ 1, 2, 3 ];
const myArrObservable = Observable.from(numbers);
const myPromObservable = Observable.from(fetch('https://null.jsbin.com'));

Try Me

Creating Observables