I wanted to use rxjs for the first time but am a bit stucked 'cause it doesn't behave exactly like I want it to: In my scenario I want to create an observable from a promise. But I want the promise only being called once (not on every subscription) and I want it not being called on creation time (defer the call to the first subscription).

First I tried this:

var source = Rx.Observable.fromPromise(_this.getMyPromise())

which causes a call to the getMyPromise function right on creation time. This is not satisfying because at that time I don't know if the source really will be used.

Then I tried:

var source = Rx.Observable.defer(function() { return _this.getMyPromise() })

which causes a call to the getMyPromise function each time a new subscription is being made to source. This makes way too many unnecessary calls to the web server. The Rx.Observable.create function seems to have the same issue.

So what is left or what am I missing?

I wanted to use rxjs for the first time but am a bit stucked 'cause it doesn't behave exactly like I want it to: In my scenario I want to create an observable from a promise. But I want the promise only being called once (not on every subscription) and I want it not being called on creation time (defer the call to the first subscription).

First I tried this:

var source = Rx.Observable.fromPromise(_this.getMyPromise())

which causes a call to the getMyPromise function right on creation time. This is not satisfying because at that time I don't know if the source really will be used.

Then I tried:

var source = Rx.Observable.defer(function() { return _this.getMyPromise() })

which causes a call to the getMyPromise function each time a new subscription is being made to source. This makes way too many unnecessary calls to the web server. The Rx.Observable.create function seems to have the same issue.

So what is left or what am I missing?

• javascript
• promise
• rxjs

• 1 So you only want it to be called once, but what do you want it to do if it is called after it is already pleted? – John Commented Mar 11, 2016 at 9:52

2 Answers 2

.shareReplay() does this, e.g.:

var source = Rx.Observable.defer(function() { return _this.getMyPromise() }).shareReplay();

If you're using rxjs5, you'll want to read: Pattern for shareReplay(1) in RxJS5

In answer to your ment below, I can think of a fairly straightforward extension to the above logic that will do what you want, but it has a caveat. Let's say the events you want to use to trigger a "refresh" are represented in a stream, s$, then you could do something like:

var source = Rx.Observable.of({}).concat(s$)
    .flatMapLatest(function() {
        return Rx.Observable.defer(function() {
            return _this.getMyPromise()
        })
    })
    .shareReplay(1)

What we have here is a stream starting with a dummy object to get things rolling, followed by a stream consisting of your refresh events. Each of these is projected into a new observable created from a fresh invocation of your getMyPromise method, and the whole thing is flattened into a single stream. Finally, we keep the shareReplay logic so we only actually make calls when we should.

The caveat is that this will only work properly if there's always at least one subscriber to the source (the first subscription after all others are disposed will run the promise again, and will receive both the previously-cached value and the result of the promise it caused to run).

Here is an answer that does not require at least one subscriber at the source at all times using a simple helper:

var _p = null;
var once = function() { return _p || (_p = _this.getMyPromise());

var source = Rx.Observable.defer(once);

Or if you're using lodash, you can _.memoize your getMyPromise and get this automatically.