From a08232be6dcab74e94cfbb17cc5138050bcd6ddb Mon Sep 17 00:00:00 2001 From: Nicholas Jamieson Date: Mon, 7 Sep 2020 00:56:45 +1000 Subject: [PATCH] fix(ReplaySubject): don't buffer next if stopped (#5696) * test(ReplaySubject): add failing tests * fix(ReplaySubject): don't buffer next if stopped * chore: remove VSCode-added lodash import Oh, FFS, now I've done it. --- spec/subjects/ReplaySubject-spec.ts | 28 ++++++++++++++++++++++++++++ src/internal/ReplaySubject.ts | 22 ++++++++++++---------- 2 files changed, 40 insertions(+), 10 deletions(-) diff --git a/spec/subjects/ReplaySubject-spec.ts b/spec/subjects/ReplaySubject-spec.ts index 834706b017..ba611ad649 100644 --- a/spec/subjects/ReplaySubject-spec.ts +++ b/spec/subjects/ReplaySubject-spec.ts @@ -351,4 +351,32 @@ describe('ReplaySubject', () => { expect(results).to.deep.equal([3, 4, 5, 'done']); }); + + it('should not buffer nexted values after complete', () => { + const results: (number | string)[] = []; + const subject = new ReplaySubject(); + subject.next(1); + subject.next(2); + subject.complete(); + subject.next(3); + subject.subscribe({ + next: value => results.push(value), + complete: () => results.push('C'), + }); + expect(results).to.deep.equal([1, 2, 'C']); + }); + + it('should not buffer nexted values after error', () => { + const results: (number | string)[] = []; + const subject = new ReplaySubject(); + subject.next(1); + subject.next(2); + subject.error(new Error('Boom!')); + subject.next(3); + subject.subscribe({ + next: value => results.push(value), + error: () => results.push('E'), + }); + expect(results).to.deep.equal([1, 2, 'E']); + }); }); diff --git a/src/internal/ReplaySubject.ts b/src/internal/ReplaySubject.ts index 63f35c3fc7..18214a8b34 100644 --- a/src/internal/ReplaySubject.ts +++ b/src/internal/ReplaySubject.ts @@ -65,21 +65,23 @@ export class ReplaySubject extends Subject { } private nextInfiniteTimeWindow(value: T): void { - const _events = this._events; - _events.push(value); - // Since this method is invoked in every next() call than the buffer - // can overgrow the max size only by one item - if (_events.length > this._bufferSize) { - _events.shift(); + if (!this.isStopped) { + const _events = this._events; + _events.push(value); + // Since this method is invoked in every next() call than the buffer + // can overgrow the max size only by one item + if (_events.length > this._bufferSize) { + _events.shift(); + } } - super.next(value); } private nextTimeWindow(value: T): void { - this._events.push({ time: this._getNow(), value }); - this._trimBufferThenGetEvents(); - + if (!this.isStopped) { + this._events.push({ time: this._getNow(), value }); + this._trimBufferThenGetEvents(); + } super.next(value); }