Почему эта подписка RXJS никогда не закрыта? ⇐ Javascript

[Anonymous]

Я некоторое время подозревал утечки памяти и несвежие подписки RXJS в корпоративном программном обеспечении. Я решил расследовать, написав фрагмент ниже. Это очень просто. Во -первых, он загружает rxjs. Во -вторых, это Monkey Patches Specpibe Method, чтобы шпионить за ним и отслеживать, что происходит. В-третьих, я создаю следующую подписку: < /p>

Код: Выделить всё

rxjs.interval(1_000).pipe(rxjs.take(3)).subscribe()
< /code>
Это просто наблюдаемая, которая излучает одно значение каждую секунду и останавливается после трех значений. Моя Monkey Patched Method Specpibe раскрывает несколько вещей. Одним из них является то, что это не одна подписка, а две. Не очень удивительно, у нас есть труба, и поэтому каждый оператор (здесь только возьмите (3) 

) создаст наблюдаемую подписку на предыдущую (здесь внутренний (1_000) ).
Но это показывает другой гораздо более удивительный факт. После того, как три значения были излучены, одна подписка закрыта, но не другая. Это выглядит как «источник», поэтому интервал (1_000) остается активным навсегда. Вы можете попробовать сами, используя фрагмент ниже. Вы увидите, что обе подписки печатают первые три значения в консоли, а затем остается только один из них и продолжает печать без остановки. Я ожидаю, что обе подписки будут закрыты. Почему это не так?

Код: Выделить всё

// This script monkey patches Observable's subscribe method to spy on it

let subscriptionCount = 0;

const observables = [];
const subscriptions = [];
const values = [];

const originalSubscribe = rxjs.Observable.prototype.subscribe;

Error.stackTraceLimit = Infinity;

rxjs.Observable.prototype.subscribe = function(observerOrNext, error, complete) {
if (!this.id) {
this.id = observables.length;
observables.push(this);
}

const subscriptionId = subscriptionCount++;

const interceptValue = value => {
console.log(`Subscription ${subscriptionId}: ${value}`);

values[subscriptionId] = values[subscriptionId] || [];
values[subscriptionId].push(value);

if (subscriptions[subscriptionId]) {
subscriptions[subscriptionId].lastValueDate = Date.now();
}
};

let observer;
if (typeof observerOrNext === 'function') {
observer = value => {
interceptValue(value);
observerOrNext(value);
};
} else {
observer = {
next: value => {
interceptValue(value);
observerOrNext?.next?.(value);
},
error: err => {
observerOrNext?.error?.(err);
},
complete: () => {
observerOrNext?.complete?.();
}
};
}

const subscription = originalSubscribe.call(this, observer, error, complete);
subscription.id = subscriptionId;
subscription.observableId = this.id;
subscription.creationDate = Date.now();
subscription.trace = new Error().stack;
subscriptions.push(subscription);

return subscription;
}

// Here is the actual MRE

rxjs.interval(1_000).pipe(rxjs.take(3)).subscribe();< /code>

Подробнее здесь: https://stackoverflow.com/questions/795 ... ver-closed