fix(sample): source is now subscribed to before the notifier

fixes #2075

spec/operators/sample-spec.ts

@@ -1,4 +1,6 @@
 import * as Rx from '../../dist/cjs/Rx';
+import { expect } from 'chai';
+
 declare const {hot, asDiagram, expectObservable, expectSubscriptions};
 
 const Observable = Rx.Observable;
@@ -29,6 +31,21 @@ describe('Observable.prototype.sample', () => {
     expectSubscriptions(e2.subscriptions).toBe(e2subs);
   });
 
+  it('should behave properly when notified by the same observable as the source (issue #2075)', () => {
+    const item$ = new Rx.Subject();
+    const results = [];
+
+    item$
+      .sample(item$)
+      .subscribe(value => results.push(value));
+
+    item$.next(1);
+    item$.next(2);
+    item$.next(3);
+
+    expect(results).to.deep.equal([1, 2, 3]);
+  });
+
   it('should sample nothing if source has nexted after all notifications, but notifier does not complete', () => {
     const e1 =   hot('----a-^------b-----|');
     const e1subs =         '^            !';

src/operator/sample.ts

@@ -49,7 +49,10 @@ class SampleOperator<T> implements Operator<T, T> {
   }
 
   call(subscriber: Subscriber<T>, source: any): TeardownLogic {
-    return source._subscribe(new SampleSubscriber(subscriber, this.notifier));
+    const sampleSubscriber = new SampleSubscriber(subscriber);
+    const subscription = source._subscribe(sampleSubscriber);
+    subscription.add(subscribeToResult(sampleSubscriber, this.notifier));
+    return subscription;
   }
 }
 
@@ -62,11 +65,6 @@ class SampleSubscriber<T, R> extends OuterSubscriber<T, R> {
   private value: T;
   private hasValue: boolean = false;
 
-  constructor(destination: Subscriber<any>, notifier: Observable<any>) {
-    super(destination);
-    this.add(subscribeToResult(this, notifier));
-  }
-
   protected _next(value: T) {
     this.value = value;
     this.hasValue = true;