benlesh
commented
1 year ago So, it's a problem... but it's a bit of an edge case. The problem usually comes from composition where people are creating "inner subscriptions" for things like custom-implemented operators that subscribe to observables to "flatten them" like concatMap et al.
To illustrate the issue, I'll just use a function (not like the whole pipeable operator thing from RxJS):
/**
* A function that flattens an observable of observables
*/
function flatten<T>(sources: Observable<Observable<T>>) {
return new Observable<T>(subscriber => {
let outerComplete = false;
const innerSubscriptions = new Set<Subscription>();
const sourceSubscription = sources.subscribe({
next: (innerSource) => {
const innerSubscription = innerSource.subscribe({
next: value => subscriber.next(value),
error: err => subscriber.error(err),
complete: () => {
innerSubscriptions.delete(innerSubscription);
if (outerComplete && innerSubscriptions.size === 0) {
subscriber.complete();
}
}
})
/*** THE PROBLEM EXISTS IN THIS TEMPORAL SPACE ***/
innerSubscriptions.add(innerSubscription);
},
error: err => subscriber.error(err),
complete: () => {
outerComplete = true;
if (innerSubscriptions.size === 0) {
subscriber.complete();
}
}
});
return () => {
sourceSubscription.unsubscribe();
for (const subscription of innerSubscriptions) {
subscription.unsubscribe();
}
innerSubscriptions.clear();
}
})
}
function take<T>(source: Observable<T>, count: number) {
return new Observable<T>(subscriber => {
let counter = 0
const subscription = source.subscribe({
next: (value) => {
if (counter < count) {
subscriber.next(value);
if (++counter === count) {
subscriber.complete(); // NOTE: causes teardown
}
}
},
error: err => subscriber.error(err),
complete: () => subscriber.complete(),
})
return () => subscription.unsubscribe() // teardown
})
}
const firehose = new Observable<number>(subscriber => {
let n = 0;
while (!subscriber.closed) {
subscriber.next(n++);
}
});And then we use it like this:
const source = Observable.from(['start'])
const flattened = flatten(source);
const threeValues = take(flattened, 3);
// Logs three values, then locks the thread! :)
threeValues.subscribe(console.log);The problem is that since we have to wait for the inner observable's subscribe call to complete and return before we get a handle we can use to tear it down, we can't flip the inner observable "closed" in time. It still only emits the 3 values, but that while (!subscriber.closed) loop in the inner subscription will never stop.
RxJS has been handling this internally by composing subscriptions through chaining Subscriber during construction of the Subscriber. However that's mostly an internal implementation detail.
There is an RFC right now about moving to a more token-like API that resolves this issue (and a couple of other more RxJS-related issues)... although in RxJS's case we're leveraging Subscriber and Subscription as an "owner", I intentionally left signal out of it for now. I suspect that's where this should head.
The only problems I have with AbortSignal for RxJS usage:
- They're SUPER slow for registering and unregistering event handlers (this is probably solvable at a runtime level)
- They're not very ergonomic when it comes to creating "child subscriptions" so to speak. That would be the act of creating new
AbortControllers that "follow" an existingAbortSignal.
But again, I think those are solvable issues.
I'm 100% amenable to providing APIs in RxJS-land that utilize AbortSignal, but only really if it's to match something that lands in the platform.
domfarolino
mk-pmb
A 2019 attempt at reviving the old TC39 proposal involved an API simplification that incidentally allowed Observables to handle the "synchronous firehose of data" problem. That is, it allows a subscriber to unsubscribe (via AbortController, in that proposal) while an Observable is synchronously pushing perhaps a "firehose" of data to the next() handler before
subscribe()returns (see this example).My understanding is that neither RxJS nor zen-observable currently support this, but if we're generalizing the API for the platform maybe we should consider it? Designing Observables such that they handle this seems nice, as they can unsubscribe via token based cancelation in the next() handler and even before subscribing is complete if the producer pushes "enough" (or "too much") data.
On the other hand, it could get a little clunky. With this approach,
subscribe()returns nothing and the assumption is that your only interaction with a "subscription" is via the AbortController that you must create before subscribing. That's not terrible, but it could get worse if we ever wanted to give more functionality to a "subscription" beyond just cancellation. In that world we might need to introduce aSubscriptionobject thatsubscribe()returns, but then you'd have two objects to keep track of for any given subscription: the AbortController that you created before callingsubscribe()and then theSubscriptionreturned fromsubscribe().That, plus the fact that two major Observable libraries today don't support this (IIUC) might mean there is not enough appetite to cater our design to it. Thoughts?