Making Observable "awaitable"

[benlesh]

This would make observables "thennable", and therefore "awaitable".

await myObservable;

I think the internals would just be

function then(success, failure) {
  return this.toPromise().then(success, failure);
}

Nice and simple. I want to put this together and experiment with it soon.

[kwonoj]

❤️ : I would love if I can use async / await. (though browser support is still far away..)

[benlesh]

Gross side effect: people can just use them as promises.

[benlesh]

I'm going to close this for now... The problem I see with it is anything that wants to check if something is Observable or a Promise would be forced to check Observable first or they'll be thrust down the promise path.

I think toPromise and forEach are adequate enough for me.

[benlesh]

Reopening this for discussion. cc @jayphelps

[jayphelps]

Internally at Netflix we have numerous SDKs and we're doing a bit of standardization on some of them. One of the things that came up was having a consistent return type for async APIs. Observables are the obvious choice because they work for 1 or many items since some of the APIs are indeed streams but others just return a single value (request -> response) and they are cancellable.

While fielding this option, there was some minor pushback on "forcing" all dev consumers to "learn Rx" (even though you certainly don't have to...just .subscribe(callback) and do everything imperatively..) In these cases, the devs wished we would return a Promise instead, but others wished we would always return Observables for 1. consistency, 2. cancellability, 3. composability.

So the currently planned solution was to make the Observables we returned a "PromiseObservable" which just means that it is an Observable that supports [some of] the Promise API too. So when consumers are calling an API that is request -> response they can just treat the return as a Promise if they choose. Obviously that includes using async/await.

So, since we find this ability useful, we thought others might find it useful as well, particularly people who are also dealing with multiple consumers who may or may not have experience with Observables.

Caveats

We cannot support the same Promise.prototype.catch function on the Observable itself, before stream$.then(), because RxJS already has a .catch() operator with different semantics. They can however do stream$.then(success).catch(failure) because the return value of of .then() is a real Promise.

It also might make the intrinsic differences between Promises and Observables more of a footgun when the Observable indeed emits more than one value. The most gruesome example of which is a stream that never ends by design (like most websockets)...which you can demo with:

async function main() {
  const response = await Observable.never().toPromise();
  alert('NEVER CALLED!!');
}

[benlesh]

cc @jhusain

[kwonoj]

While I'm generally in favor, also curious if this can be separate module so consumer can explicitly choose to use or not, instead of exposing this availability out of box.

[jayphelps]

@kwonoj do you mean a separate npm package or using our existing operator patching mechanism?

import 'rxjs/add/operator/then';

I forgot to call that out, but I was thinking it be an operator you need to add (unless you include all), not built-in to the prototype like subscribe is.

[kwonoj]

separate npm package or using our existing operator patching mechanism

: both would be fine, but my initial thought was npm package, as complete separation and consumer literally choose to use it. In any case not to be patched by default at least.

[jayphelps]

Actually--thinking about this further..it's not really an operator. It subscribes and does not return an Observable. So it would need to be namespaced somewhere other than "rxjs/add/operator/" I suggest.

[jayphelps]

@kwonoj re: npm package, that's certainly the fallback option that is acceptable but I want to make it clear this discussion is around whether it belongs in core or not. i.e. does it provide enough benefit to justify being included.

[kwonoj]

makes sense. I personally consider this is kind of non-core functionally as I'm seeing this is somewhat similar to interop from Observable, reason asked if it could be separated from main codebase. Course it's personal view though.

[jayphelps]

@kwonoj that's basically what we're looking for, personal views. 💃 We definitely don't want to shoehorn stuff we need that others don't.

[david-driscoll]

I can see the value in it, but there are definitely some foot gun scenarios I can see happening. Checking for promise first, observables never ending, etc.

If this "adapter" was something that was specifically opt in, as in it is not included as part of the default import, then I think it would be a a great edition. At least then the consumer chooses if they want to the ability to shoot themselves in the foot or not.

[david-driscoll]

As far bikeshedding goes what about ...

import 'rxjs/add/adapter/promise';
or
import 'rxjs/add/interop/promise';

[benlesh]

I'm almost in favor of adding this, and removing toPromise, the more I think about it. The only reason to use toPromise is for interop, and this is much more ergonomic.

This is nice:

await source.takeUntil(timer(5000)).toArray();

[jayphelps]

@Blesh there's still utility in .toPromise(), like if someone wanted to expose or provide a true 100% Promise to some library or API. Especially since Observables cannot support the Promise version of .catch()

EDIT: although, technically speaking the Promise API seems to allow you to invoke promise.then().then(callback) which means stream$.then() without any args could support the same, which basically would serve the same purpose of .toPromise() but obviously be far less obvious.

[benlesh]

Especially since Observables cannot support the Promise version of .catch()

This is the strongest argument against making observable thennable that I've read. Although Rx catch supports returning promises... Hmmm

[benlesh]

I spoke with @jhusain about this offline. He brought up a valid point that I think @domenic made originally, which is that Promises can have side effects at creation time. In other words:

let sideEffect = false;
const p = new Promise((resolve) => {
  sideEffect = true;
  resolve('wee');
});
console.log(sideEffect); // true

where:

let sideEffect = false;
let o = new Observable((observer) => {
  sideEffect = true;
  observer.next('wee');
  observer.complete();
});
console.log(sideEffect); // false

At best, that means that if we add a .then method to RxJS observables, it's a thennable... even if it's almost A+ compliant by virtue of the fact that .then would return an promise, and .catch is so semantically similar it shouldn't be hard to make it match the spec. The fact is that the eager nature of promises makes them fundamentally different from observables.

As such, it's probably best to keep the explicit .toPromise() call, or we could even bike-shed the name as something like .finalValue() or something, since .toPromise() doesn't really explain very clearly what it's doing.

[benlesh]

Another fundamental difference is that promises multicast and observables don't always do so.

so..

let source = new Observable(observer => {
  console.log('subscribed');
  observer.next('test');
  observer.complete();
});


// later on...

await source;   // logs "subscribed"
await source;   // logs "subscribed"
await source;   // logs "subscribed"

but...

let source = new Promise(resolve => {
  console.log('subscribed');
  resolve('test');
});  // logs "subscribed"


// later on...

await source;
await source;
await source;

Again, this is a fundamental difference that might surprise people that are treating this like a promise.

[jayphelps]

@Blesh can you clarify further this specific use case? AFAICT the side effect behavior remains the same because .then() would subscribe

console.clear();

let sideEffect = false;
let o = new Rx.Observable((observer) => {
  sideEffect = true;
  observer.next('wee');
  observer.complete();
})

o.toPromise().then(value => console.log('value', value));
console.log(sideEffect); // true

Or is this about Observables can be cold, but Promises cannot? I currently don't think that's justification for not providing interop, but I'm clearly biased. I'd be interested to hear real-world examples where this will bite someone. That may convince me otherwise, but then I'd vote to remove .toPromise() because it's providing the same functionality behind one extra invocation.

[domenic]

As such, it's probably best to keep the explicit .toPromise() call, or we could even bike-shed the name as something like .finalValue() or something, since .toPromise() doesn't really explain very clearly what it's doing.

I really like Dart's observable-esque "Streams" in this regard. See docs and in particular first/last/single.

[mattpodwysocki]

@domenic yeah, that was a thought to have a firstAsync, lastAsync etc return promises directly instead of wrappers for anything that returns a single value.

[benlesh]

Right now you could accomplish these things with .toPromise() (which is mostly equivalent to .last().toPromise()) and .first().toPromise(). both .last() and .first() allow lazy composition before getting a future/promise with toPromise()... however they have the added benefit of not losing the cancellation semantic if you want to just use them as observables.

I feel like the only tricky bit about toPromise is when the source observable gets more than one value it's not really apparent that it's waiting to emit the last value. I almost feel like toPromise() should error if there's more than one value or if no values are emitted... but I'm not sure. It's so rare that I use toPromise for anything, that I'd want to understand the use cases better before I was sure I'd want to change the behavior.

[jayphelps]

FYI my current implementation:

http://jsbin.com/tununi/edit?js,console

class PromisifyObservable extends Observable {
  constructor(source, operator) {
    super();

    if (source instanceof PromisifyObservable) {
      this.source = source;
    } else {
      this.source = Observable.fromPromise(source.toPromise());
    }

    this.operator = operator;
  }

  lift(operator) {
    return new PromisifyObservable(
      this.source, operator
    );
  }

  then() {
    return this.toPromise().then(...arguments);
  }
}


Observable.prototype.promisify = function promisify() {
  return new PromisifyObservable(this);
};

const fetchSomething = somethingId => {
  return Observable.ajax('https://api.github.com/search/repositories?q=test').promisify();
};

const request = fetchSomething(123);
request.then(first => {
  console.log('first', first.response.items.length);

  // only one ajax call is made, because promises are
  // hot and cache results
  request.map(second => second.response.items.length)
    .then(second => console.log('second', second));
});

Provides a observable.promisify() operator which is basically a type of multicast operator that opts into having Promise-like behavior.

I'm leaning towards this belonging as a community addon, not in core, because even though it is like one of the multicast operators, it adds that custom .then() operator only when it's promisified, which is novel behavior.

---

Contrary to my naive thinking before, this makes the original upstream observable non-cancellable. So that's a HUGE con....and arguably a deal breaker for me.

[benlesh]

Contrary to my naive thinking before, this makes the original upstream observable non-cancellable. So that's a HUGE con....and arguably a deal breaker for me.

Yep. Although cancellation tokens might help with this whenever they arrive.

[benlesh]

Resurrecting this.

Do we want people to be able to say await someObservable$;? or is it fine that they have to say await someObservable.toPromise()?

Given that promise cancellation has been abandoned, it looks like consuming anything via await will be non-cancellable anyhow.

In light of that, I feel like having a then() impl that is basically wrapping toPromise() is probably okay.

The only problem I can see with this is that toPromise is a last value implementation, where forEach currently sends undefined to the resulting promise's success path, but in the TC39 proposal it sends a completion value (which is currently being contested).

[jayphelps]

@Blesh has TC39 discussed this at all, at least for "future possibilities"?

[benlesh]

@jayphelps no they haven't, as it wouldn't pass muster. But I've discussed this with @jhusain who brought up a good point, since promises are by default multicast, the behavior of a promise if you await it in two different places would be different than the behavior of an observable if you await it in two different places.

I think we'll need to close this. I can't see this ever being something we can do realistically.

[david-driscoll]

Just as a side note. .NET does allow you to await an observable and the behavior as I understand it is very similar. Each await of the observable would act as a separate subscribe call.

That said its easy enough to await toPromise. The other odd side effect would be observables to always appear as if they were a promise which would be confusing.

[listepo-alterpost]

@benlesh @jachenry @david-driscoll any news of this for rxjs 6?

[benlesh]

@HideDev ... we can't really make Observable "thennable". So you'll need to use toPromise

The biggest problem is that promises are eager, where Observables are lazy and could have side effects.

Consider this:

let a = 0;
const someObservable = new Observable(observer => {
  const id = setTimeout(() => {
    observer.next(a++);
    observer.complete();
  });
  return () => clearTimeout(id);
});

let b = 0;
const somePromise = new Promise(resolve => {
  const id = setTimeout(() => {
    resolve(b++);
  });
});

// Let's assume observable is thennable here
async function consumer() {
    const a = await someObservable;
    const b = await somePromise;
    console.log({ a, b });
}

// then we call each multiple times:
consumer();
consumer();
consumer();

// results:

{ a: 0, b: 0 }
{ a: 1, b: 0 }
{ a: 2, b: 0 }

It would make for very confusing behavior.

Not to mention people that got it might treat it like a Promise imperatively, which isn't really desirable.

[pabl-o-ce]

@benlesh so at the end of the road for now a observable can't be awaitable my friend?

options can be: using async/await with .toPromise()?

is there a good practice to handle this type of things like async/ await does with promises... so far I read about generators but I'm not able to understand how handle with yield... anything that can bring me to the light side I would glad to heard about. 🧞‍♂️

[rgbkrk]

You can use await with toPromise, I use it with tests all the time.

[benlesh]

This is precisely the use case for toPromise.

[pabl-o-ce]

I 👀 read on some part about spawn / yield and I get confuse about it. 🌫

many thanks for the clarification. noble men 🍻

[amcdnl]

@benlesh - What about the way that C# handled it?

string result = await observable.FirstAsync();
string result = await observable.FirstOrDefaultAsync();
string result = await observable.LastAsync();
string result = await observable.LastOrDefaultAsync();
string result = await observable.SingleAsync();
string result = await observable.SingleOrDefaultAsync();

I realize that most of this could be accomplished right now with a long combination of pipeable operators but that defeats the point. There is a HUGE audience looking for a way to get the value of an observable (just google get value from observable) this could solve some of that (of course with some caveats).

We just got out of the age of callback hell but now it feels like observables are putting us back into that trend.

[david-driscoll]

We could do it like C#, so far we have chosen not to for a few reasons. The biggest reason is to avoid user confusion.

What happens when you do any of the following?

await Observable.never();  // Will never return
await Observable.empty();  // Will return immediately (but with a null/undefined value!).
await Observable.from([1,2,3,4,5]); // Returns 5
await Observable.fromEvent(window, 'click'); // Will never return

We're trying to avoid the footgun effect. If you have to explicitly "cast" an observable to make it awaitable then you have more of a chance to fully thinking through the problem you're trying to solve, and you won't get stuck in a place where you can't understand why your code just stops working entirely.

You can still do the following however, and they'll work as expected.

let result = await observable.first().toPromise();
result = await observable.defaultIfEmpty().first().toPromise();
result = await observable.last().toPromise();
result = await observable.defaultIfEmpty().last().toPromise();
result = await observable.single().toPromise();
result = await observable.defaultIfEmpty().single().toPromise();

[amcdnl]

1. Would the same thing not happen that would in C#?
2. I'm not suggesting making them magically awaitable, I'm merely suggesting a blessed shortcut method to do so. I think users specifically calling these operators will have to make a concise choice to use it and be aware of its caveats.
3. Yup, as I mentioned its all achievable now. In fact, that would even be longer if you were to use pippeable operators. Making it more simple is my suggestion.

[LinboLen]

the subscription's terminate function can have awaitable(toPromise) ?