How to create Observable from function?

Question

I want to call a function (synchronously) and then use its return value as an initial emission (subsequently chaining some other operators on resulting observable).

I want to invoke this function during subscription, so I can't just use Observable.of(() => getSomeValue()). I've seen bindCallback (previously fromCallback) but I don't think it can be used for this task (correct me if I'm wrong). I've seen start static operator in v4 docs but apparently it is not implemented in v5 (and no indication that its on the way). RxJava also has fromCallable operator that does exactly that afaik.

Only way I could think of is like this:

Observable.create((observer: Observer<void>) => {
  let val = getSomeValue();
  observer.next(val);
  observer.complete();
})

which I think does just that. But this just seems so complicated for simple thing that should probably have been like Observable.fromFunction(() => getSomeValue()) And what if I want to run it asynchronously, like start operator does? How can I do this in current verion of RxJS?

Answer 1

I tend to avoid any explicit use of Observable.create where ever possible, because generally it is a source of bugs to have to manage not just your event emission but also your teardown logic.

You can use Observable.defer instead. It accepts a function that returns an Observable or an Observable-like thing (read: Promise, Array, Iterators). So if you have a function that returns an async thing it is as easy as:

Observable.defer(() => doSomethingAsync());

If you want this to work with a synchronous result then do:

Observable.defer(() => Observable.of(doSomethingSync()));

Note: That like create this will rerun the function on each subscription. This is different then say the result of Observable.bindCallback which stores the function call result without re-executing the function. So if you need that sort of behavior you will need to use the appropriate multicasting operator.