The Problem
If an observable is running synchronously, then the callback that is given to subscribe is executed before subscribe returns. The result is that the following code gives an error. (sub is not initialized)
const sub = from([1,2,3,4,5]).subscribe(x => {
if(x > 3) sub.unsubscribe();
console.log(x);
});
A Nieve Solution
If we force the values of our stream into the event loop, we no longer have this problem. Subscribe will always return before the lambda is called.
const sub = from([1,2,3,4,5]).pipe(
delay(0)
).subscribe(x => {
if(x > 3) sub.unsubscribe();
console.log(x);
});
This, however, strikes me as a bad idea. If for no other reason than performance. Though it also makes the run-order less deterministic (Which Browser?, NodeJS?).
The Idiomatic RxJS Solution
Don't unsubscribe yourself, let an operator do that for you
const unsub = new Subject();
from([1,2,3,4,5]).pipe(
takeUntil(unsub)
).subscribe(x => {
if(x > 3) {
unsub.next();
unsub.complete();
}
console.log(x);
});
The problem here is that we need to create the entire apparatus that is a subject in order to accomplish a very specific goal. It's like buying a truck in order to get the wheel. It doesn't scale well. Finally, just like calling unsubscribe() yourself, It's also still mixing imperative and functional javascript.
The Same Problem at a Bigger Scale
Consider an operator that takes a list of observables and emits values only from the observables earlier in the list than any previous emissions.
Here is this operator done with the event loop.
function prefer<T, R>(...observables: Observable<R>[]): Observable<R>{
return new Observable(observer => {
const subscrptions = new Array<Subscription>();
const unsub = (index) => {
for(let i = index; i < subscrptions.length; i++){
subscrptions[i].unsubscribe();
}
}
observables.map(stream => stream.pipe(
delay(0)
)).forEach((stream, index) =>
subscrptions.push(stream.subscribe(payload => {
observer.next(payload);
unsub(index + 1);
subscrptions.length = index + 1;
}))
);
return { unsubscribe: () => unsub(0) }
})
}
and then without the event loop and without unsubscribe().
function prefer<T, R>(...observables: Observable<R>[]): Observable<R>{
return defer(() => {
const wUnsub = observables.map((stream, index) => ({
stream: stream.pipe(
map(payload => ({index, payload}))
),
unsub: new Subject()
}));
const unsub = (index) => {
for(let i = index; i < wUnsub.length; i++){
wUnsub[i].unsub.next();
wUnsub[i].unsub.complete();
}
}
return merge(...wUnsub.map(build => build.stream.pipe(
takeUntil(build.unsub)
))).pipe(
tap(({index}) => {
unsub(index + 1);
wUnsub.length = index + 1;
}),
map(({payload}) => payload),
finalize(() => unsub(0))
);
});
}
Also here's the operator in use
prefer(
interval(10000).pipe(
take(5),
map(_ => "Every 10s")
),
interval(5000).pipe(map(_ => "Every 5s")),
interval(1000).pipe(map(_ => "Every 1s")),
interval(250).pipe(map(_ => "Every 1/4s"))
).subscribe(console.log);
Imagine using this operator at scale. It's relatively easy to understand that the memory footprint of the first approach is much smaller than the second approach (O(n) vs O(n*n) memory usage).
Finally; The Question
Since (in javascript) synchronous code runs to completion before any other code runs, it doesn't seem to make sense to be able to access an observable's subscription before the synchronous section of that subscription has returned. Yet, as a means to abort a stream early, it seems that being able to access a stream's subscription early might have benefits (at the very least in memory).
Is there a (relatively) elegant way to instrument an Observable to work around these problems?