Rxjs: Add completion value

[benlesh]

In order to comply with the the spec 100%, we need to have the complete notification pass a value.

There will undoubtedly be some confusion around this when it comes to some combinators. So...

1. member operators should always return an observable that passes through the source observable's completion value.
2. static operators that join multiple observables will drop all completion values.

So let's assume I have this observable:

let source = new Observable((observer) => {
  observer.next(1);
  observer.next(2);
  observer.complete('done');
});

Assuming the call is source.anyOperatorYouWantHere() the completion value should always be "done". As the source of the operation would pass that value through.

If the call is a static join operator like: Observable.combineLatest(source, someOtherSource) the completion value would be undefined, because nothing will be passed and all completion values would be dropped.

There will, of course, be discussion of the idea of "completion operators", but I strongly view the use of completion values as an anti-pattern. In fact, it should be prominently documented that using completion values is an anti-pattern.

[trxcllnt]

What happens to the completion value from an Observable that has been split into many Observables (via window, groupBy, etc.)? What should the flattening operators (merge, concat, switch, etc.) do with the completion values from the flattened inner Observables? This is quite a troubling development.

[staltz]

There will, of course, be discussion of the idea of "completion operators", but I strongly view the use of completion values as an anti-pattern. In fact, it should be prominently documented that using completion values is an anti-pattern.

I'm on the same train. observer.complete('done') is so dumb.

[paulpdaniels]

I guess I would start from the other side, what are the use cases for this? Is there a compelling set of applications that are having to hack around _not_ having this value currently? Or is it just a symmetry thing?

[benlesh]

I guess I would start from the other side, what are the use cases for this? Is there a compelling set of applications that are having to hack around not having this value currently?

I've asked, and I haven't been presented with much. Most of this exists as a means to support transforming an iterable into an observable, I guess.

[benlesh]

What happens to the completion value from an Observable that has been split into many Observables (via window, groupBy, etc.)?

Dropped. Anywhere it's ambiguous, we're just going to drop it. People really shouldn't be using it. Prior art for the act of dropping completion values: for..of and iterable.

[benlesh]

This is no longer blocking release.

The reason being that we can add a completion value at a later time and it's not a breaking change.

[farantesrodrigues]

Hi guys, i just want to add a use case for this. In the API I have server side events sending objects to frontend as soon as it can but I also collect them all and send in a promise once fulfilled. In the frontend I have an angular 2 app with an ngFor subscribed to SSE events stream but want to update it once I get the complete callback with all the values. There might be other ways to do it but it would be much cleaner to just create an observable that passes next on each SSE stream event and completes passing the values of the fulfilled promise when they're available.

[felixfbecker]

This doesn't achieve symmetry with Iterators - just like Observables, Iterators emit n values or throw a single error. There is no completion value, only a completion event (done: true).
What does have a return value is a _Generator_. But I would argue that the return value of generators is not really there for the use case of streams, but for the use case of implementing coroutines. Generators also support other stuff that don't map to Observables, like throwing _into_ and returning values_to_ the Generator yield.

The biggest question I have is: Will fromPromise() return an Observable with just a completion value? It would make sense if you think about it. I always found it a bit weird that a Promise<void> will convert to an Observable that emits a single undefined. It means that to chain multiple Promise-returning calls in an Observable stream you have to use mergeMap. But if you ever call a function in that chain that returns a multi-item Observable, and ignore its elements with .ignoreElements(), you cannot continue the chain with mergeMap anymore and instead have to do a weird dance with .concat(Observable.defer(() => ...)). This is especially dangerous when refactoring code from Promises to Observables, because suddenly functions don't emit a single undefined anymore but no items at all and your mergeMap chain broke silently.

---

@farantesrodrigues why can't you just use the aggregation of all values you received through SSE?

[staltz]

Ben, how about we just close this issue? The spec currently has https://github.com/tc39/proposal-observable#observer

interface Observer {
    start(subscription : Subscription);
    next(value);
    error(errorValue);
    complete();
}

[benlesh]

YAY!

[samal-rasmussen]

I'm happy for you guys that the observable spec went in the same direction that you guys went with rxjs :)

So seeing as we won't be getting values on complete, I'd like to ask how you guys handle an operation that gives progress updates and then a final result value? This would in my mind map perfectly to sending progress values on next and the the final value on complete. I can come up with two other ways to implement this that I don't think are as elegant, but maybe you know better than me :)

1. Wrap the sent value in a {value, type} tuple object, where value is the value you want to send, and type is to denote whether it is a progress update or the final result.
2. Return two observables, one for the progress updates, and one for the final value.

[felixfbecker]

@samal84 Observables fit best to emit 0..n values of the same type. This is why you for example can get an Observable for a single event from an event emitter. Your use case sounds more like you have two events: a final result value, and progress reports that - I assume - contain no result values, but only something a percentage. Observables work well if the final result value is a direct aggregation of all the "progress" events, i.e. if the progress events are chunks of the final result. For your use case a NodeJS EventEmitter maybe would be better (which you can easily get an Observable from for a specific event) or a pattern where you pass a ProgressReporter object as a parameter.

[samal-rasmussen]

Yeah the use case I'm thinking of is where the progress values would be something like a percentage telling how much work has been done so far. And I'm working on an in browser app, so no nodejs.

[felixfbecker]

Node's EventEmitter implementation is on NPM as a browser-compatible package

[david-driscoll]

Without using something different, two observables would make sense.

You could also have an observable that looks like...

let o: Observable<T><{ complete: false; progress: number; } | { complete: true; value: T; }>;
o.subscribe(x => {
    if (x.complete) { // something with value } else { // something with progress }
});
// could also filter
let progress = o.filter(x => !x.complete);
let final = o.filter(x => x.complete).take(1);

[samal-rasmussen]

Yeah two observables is my second option that I presented, and that first example of yours is basically what I meant with option 1. Just that you added some lovely use of Typescript union types and type guards. Pretty awesome actually :)

Hopefully the take(1) in the second example of yours would redundant, as the source should call complete after sending the last value.

[felixfbecker]

.find(predicate) is also equivalent to .filter(predicate).take(1)

[lock[bot]]

This thread has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.