Creating An Observable with RxJS

[00:03] Hi. Let's get started with creating our own Rx Observable. To start with, I've already included Rx -- in this case, rx.all but rx.lite will do -- and I've added this console clear mostly so we don't junk up our console each time we test our Observable.

[00:22] Let's create our first Observable. To do that, we call Rx Observable.Create. It takes a function that gets to it an argument of observer. Now, let's subscribe to it so we can test it. Our first argument is our Next handler. For that, I'm just going to log out the Next value.

[00:56] Second argument is the Error handler, and the last argument is our Completion handler. If I run this, you'll see it doesn't do anything. Doesn't even error. That's because this is a valid Observable. This is the equivalent of Rx Observable.Never, which returns an Observable that never emits a value and never completes.

[01:33] Let's have an emitter value. To do that, we call our observer with Observer OnNext, and we pass to it a value. When we run it, we see it emits a value but it does not complete. To get it to complete, we call "Observer OnCompleted.'

[01:56] When I run it, you'll see that it emits done. Let's do something asynchronous. To do that, I'm just going to call "setTimeout" inside of my Observable creation function, and I'll get it to delay of about one second.

[02:23] We can see what's happening. I'm going to add some logging for when the Observable is started and for when the setTimeout handler has been hit. We can see it's started. It hit our timeout, it emits a value and then it completes.

[02:51] One feature of Observables is that you can dispose of them. What happens if I dispose of our Observable subscription before our other timeout completes? To do this in about a half-second, I'm going to call "Dispose" on our Disposable return from our subscription.

[03:18] When I run this, it shows started timeout hit, but you notice it doesn't emit anything. The timeout hit is actually a problem. It is more desirable, and by design, better for our Observable to actually cancel this timeout when it was disposed of.

[03:37] To enable that, I'm going to get our Timeout id, and from our Observable creation function, I'm going to return a function. The function I return from this is our Disposal handler. It's going to be called when you call "Dispose" on the subscription.

[03:58] Inside of this, I'm going to call clearTimeout with our id. When I run this, it says, "Started," but it never hits our Timeout handler. That's because it cleared the timeout. Just to show you, I'll add some logging here.

"[04:17] I run, Disposal called, never hits our timeout." One last important feature of creating an Observable. Let's say I was doing some calculation inside of here, and it was important to handle an error. If I were to create an error inside of this and I run it, you'll notice I get an Uncaught exception.

[04:59] Ideally, we want to hit our Error handler. In order to do that, what you want to do is you want to add a try-catch inside of an appropriate place in your Observable creation function. And you want to call Observer OnError, pass your error to it.

[05:27] When I run it, we see that it actually hits our error logging and then calls "The Disposal method." This is very important. If you do not handle your errors inside of your Observable, it won't clean your Observable up. Any event handlers you might have will still fire or may still fire, or worse, it will just break your application because you have an Unhandled exception.

[05:56] Wrapping up. To create an Observable, you call Rx Observable.Create and you pass to it a function that gives you an observer. On which you can call OnNext to emit a value, OnError to emit an error and OnCompleted to complete the Observable.

[06:15] It's important to add a try-catch around any area of your Observable that you think may throw an error, and then pass that error into your OnError handler. Finally, do any cleanup for your Observable inside of your returned function.

[06:32] You can also return a Disposable or a CompositeDisposable if you have any Observable subscriptions inside of your custom Observable. Observable creation is generally what's done at each step in your operator chain off of Observables.

[06:48] Hopefully egghead gives you a better idea of the anatomy of an Observable.