Merge pull request #3667 from cartant/issue-3605-stable

fix(delay): fix memory leak (in stable)

spec/operators/delay-spec.ts

@@ -1,4 +1,7 @@
+import { expect } from 'chai';
+import * as sinon from 'sinon';
 import * as Rx from '../../dist/package/Rx';
+import { delay, repeatWhen, skip, take, tap } from '../../dist/package/operators';
 import marbleTestingSignature = require('../helpers/marble-testing'); // tslint:disable-line:no-require-imports
 
 declare const { asDiagram, time };
@@ -147,4 +150,33 @@ describe('Observable.prototype.delay', () => {
 
     expectObservable(result).toBe(expected);
   });
-});
+
+  it('should unsubscribe scheduled actions after execution', () => {
+    let subscribeSpy: any = null;
+    const counts: number[] = [];
+
+    const e1 =       cold('a|');
+    const expected =      '--a-(a|)';
+    const duration = time('-|');
+    const result = e1.pipe(
+      repeatWhen(notifications => {
+        const delayed = notifications.pipe(delay(duration, rxTestScheduler));
+        subscribeSpy = sinon.spy(delayed['source'], 'subscribe');
+        return delayed;
+      }),
+      skip(1),
+      take(2),
+      tap({
+        next() {
+          const [[subscriber]] = subscribeSpy.args;
+          counts.push(subscriber._subscriptions.length);
+        },
+        complete() {
+          expect(counts).to.deep.equal([1, 1]);
+        }
+      })
+    );
+
+    expectObservable(result).toBe(expected);
+  });
+});

src/operators/delay.ts

@@ -96,6 +96,7 @@ class DelaySubscriber<T> extends Subscriber<T> {
       const delay = Math.max(0, queue[0].time - scheduler.now());
       this.schedule(state, delay);
     } else {
+      this.unsubscribe();
       source.active = false;
     }
   }