feat(sample): readd sample operator

Add an implementation of `sample` that behaves like RxJS 4

BREAKING CHANGE: `sample` behavior returned to RxJS 4 behavior

spec/operators/sample-spec.js

@@ -0,0 +1,180 @@
+/* globals describe, it, expect, hot, expectObservable,, expectSubscriptions */
+var Rx = require('../../dist/cjs/Rx.KitchenSink');
+var Observable = Rx.Observable;
+
+describe('Observable.prototype.sample', function () {
+  it('should get samples when the notifier emits', function () {
+    var e1 =   hot('----a-^--b----c----d----e----f----|          ');
+    var e1subs =         '^                           !          ';
+    var e2 =   hot(      '-----x----------x----------x----------|');
+    var e2subs =         '^                           !          ';
+    var expected =       '-----b----------d----------f|          ';
+
+    expectObservable(e1.sample(e2)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+    expectSubscriptions(e2.subscriptions).toBe(e2subs);
+  });
+
+  it('should sample nothing if source has not nexted yet', function () {
+    var e1 =   hot('----a-^-------b----|');
+    var e1subs =         '^            !';
+    var e2 =   hot(      '-----x-------|');
+    var e2subs =         '^            !';
+    var expected =       '-------------|';
+
+    expectObservable(e1.sample(e2)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+    expectSubscriptions(e2.subscriptions).toBe(e2subs);
+  });
+
+  it('should not complete when the notifier completes, nor should it emit', function () {
+    var e1 =   hot('----a----b----c----d----e----f----');
+    var e1subs =   '^                                 ';
+    var e2 =   hot('------x-|                         ');
+    var e2subs =   '^       !                         ';
+    var expected = '------a---------------------------';
+
+    expectObservable(e1.sample(e2)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+    expectSubscriptions(e2.subscriptions).toBe(e2subs);
+  });
+
+  it('should complete only when the source completes, if notifier completes early', function () {
+    var e1 =   hot('----a----b----c----d----e----f---|');
+    var e1subs =   '^                                !';
+    var e2 =   hot('------x-|                         ');
+    var e2subs =   '^       !                         ';
+    var expected = '------a--------------------------|';
+
+    expectObservable(e1.sample(e2)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+    expectSubscriptions(e2.subscriptions).toBe(e2subs);
+  });
+
+  it('should allow unsubscribing explicitly and early', function () {
+    var e1 =   hot('----a-^--b----c----d----e----f----|          ');
+    var unsub =          '              !                        ';
+    var e1subs =         '^             !                        ';
+    var e2 =   hot(      '-----x----------x----------x----------|');
+    var e2subs =         '^             !                        ';
+    var expected =       '-----b---------                        ';
+
+    expectObservable(e1.sample(e2), unsub).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+    expectSubscriptions(e2.subscriptions).toBe(e2subs);
+  });
+
+  it('should not break unsubscription chains when result is unsubscribed explicitly', function () {
+    var e1 =   hot('----a-^--b----c----d----e----f----|          ');
+    var e1subs =         '^             !                        ';
+    var e2 =   hot(      '-----x----------x----------x----------|');
+    var e2subs =         '^             !                        ';
+    var expected =       '-----b---------                        ';
+    var unsub =          '              !                        ';
+
+    var result = e1
+      .mergeMap(function (x) { return Observable.of(x); })
+      .sample(e2)
+      .mergeMap(function (x) { return Observable.of(x); });
+
+    expectObservable(result, unsub).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+    expectSubscriptions(e2.subscriptions).toBe(e2subs);
+  });
+
+  it('should only sample when a new value arrives, even if it is the same value', function () {
+    var e1 =   hot('----a----b----c----c----e----f----|  ');
+    var e1subs =   '^                                 !  ';
+    var e2 =   hot('------x-x------xx-x---x----x--------|');
+    var e2subs =   '^                                 !  ';
+    var expected = '------a--------c------c----e------|  ';
+
+    expectObservable(e1.sample(e2)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+    expectSubscriptions(e2.subscriptions).toBe(e2subs);
+  });
+
+  it('should raise error if source raises error', function () {
+    var e1 =   hot('----a-^--b----c----d----#                    ');
+    var e1subs =         '^                 !                    ';
+    var e2 =   hot(      '-----x----------x----------x----------|');
+    var e2subs =         '^                 !                    ';
+    var expected =       '-----b----------d-#                    ';
+
+    expectObservable(e1.sample(e2)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+    expectSubscriptions(e2.subscriptions).toBe(e2subs);
+  });
+
+  it('should completes if source does not emits', function () {
+    var e1 =   hot('|');
+    var e2 =   hot('------x-------|');
+    var expected = '|';
+    var e1subs =   '(^!)';
+    var e2subs =   '(^!)';
+
+    expectObservable(e1.sample(e2)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+    expectSubscriptions(e2.subscriptions).toBe(e2subs);
+  });
+
+  it('should raise error if source throws immediately', function () {
+    var e1 =   hot('#');
+    var e2 =   hot('------x-------|');
+    var expected = '#';
+    var e1subs =   '(^!)';
+    var e2subs =   '(^!)';
+
+    expectObservable(e1.sample(e2)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+    expectSubscriptions(e2.subscriptions).toBe(e2subs);
+  });
+
+  it('should raise error if notification raises error', function () {
+    var e1 =   hot('--a-----|');
+    var e2 =   hot('----#');
+    var expected = '----#';
+    var e1subs =   '^   !';
+    var e2subs =   '^   !';
+
+    expectObservable(e1.sample(e2)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+    expectSubscriptions(e2.subscriptions).toBe(e2subs);
+  });
+
+  it('should not completes if source does not complete', function () {
+    var e1 =   hot('-');
+    var e1subs =   '^              ';
+    var e2 =   hot('------x-------|');
+    var e2subs =   '^             !';
+    var expected = '-';
+
+    expectObservable(e1.sample(e2)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+    expectSubscriptions(e2.subscriptions).toBe(e2subs);
+  });
+
+  it('should sample only until source completes', function () {
+    var e1 =   hot('----a----b----c----d-|');
+    var e1subs =   '^                    !';
+    var e2 =   hot('-----------x----------x------------|');
+    var e2subs =   '^                    !';
+    var expected = '-----------b---------|';
+
+    expectObservable(e1.sample(e2)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+    expectSubscriptions(e2.subscriptions).toBe(e2subs);
+  });
+
+  it('should complete sampling if sample observable completes', function () {
+    var e1 =   hot('----a----b----c----d-|');
+    var e1subs =   '^                    !';
+    var e2 =   hot('|');
+    var e2subs =   '(^!)';
+    var expected = '---------------------|';
+
+    expectObservable(e1.sample(e2)).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+    expectSubscriptions(e2.subscriptions).toBe(e2subs);
+  });
+});

src/CoreOperators.ts

@@ -62,6 +62,7 @@ export interface CoreOperators<T> {
   repeat?: (count?: number) => Observable<T>;
   retry?: (count?: number) => Observable<T>;
   retryWhen?: (notifier: (errors: Observable<any>) => Observable<any>) => Observable<T>;
+  sample?: (notifier: Observable<any>) => Observable<T>;
   scan?: <R>(project: (acc: R, x: T) => R, acc?: R) => Observable<R>;
   share?: () => Observable<T>;
   single?: (predicate?: (value: T, index: number) => boolean) => Observable<T>;

src/Observable.ts

@@ -246,6 +246,7 @@ export class Observable<T> implements CoreOperators<T>  {
   repeat: (count?: number) => Observable<T>;
   retry: (count?: number) => Observable<T>;
   retryWhen: (notifier: (errors: Observable<any>) => Observable<any>) => Observable<T>;
+  sample: (notifier: Observable<any>) => Observable<T>;
   scan: <R>(accumulator: (acc: R, x: T) => R, seed?: T | R) => Observable<R>;
   share: () => Observable<T>;
   single: (predicate?: (value: T, index: number) => boolean) => Observable<T>;

src/Rx.KitchenSink.ts

@@ -102,6 +102,7 @@ import './add/operator/reduce';
 import './add/operator/repeat';
 import './add/operator/retry';
 import './add/operator/retryWhen';
+import './add/operator/sample';
 import './add/operator/scan';
 import './add/operator/share';
 import './add/operator/single';

src/Rx.ts

@@ -74,6 +74,7 @@ import './add/operator/reduce';
 import './add/operator/repeat';
 import './add/operator/retry';
 import './add/operator/retryWhen';
+import './add/operator/sample';
 import './add/operator/scan';
 import './add/operator/share';
 import './add/operator/single';

src/add/operator/sample.ts

@@ -0,0 +1,5 @@
+import {Observable} from '../../Observable';
+import {sample} from '../../operator/sample';
+Observable.prototype.sample = sample;
+
+export var _void: void;

src/operator/sample.ts

@@ -0,0 +1,56 @@
+import {Observable} from '../Observable';
+import {Operator} from '../Operator';
+import {Subscriber} from '../Subscriber';
+
+export function sample<T>(notifier: Observable<any>): Observable<T> {
+  return this.lift(new SampleOperator(notifier));
+}
+
+class SampleOperator<T, R> implements Operator<T, R> {
+  constructor(private notifier: Observable<any>) {
+  }
+
+  call(subscriber: Subscriber<R>) {
+    return new SampleSubscriber(subscriber, this.notifier);
+  }
+}
+
+class SampleSubscriber<T> extends Subscriber<T> {
+  private lastValue: T;
+  private hasValue: boolean = false;
+
+  constructor(destination: Subscriber<T>, private notifier: Observable<any>) {
+    super(destination);
+    this.add(notifier._subscribe(new SampleNotificationSubscriber(this)));
+  }
+
+  _next(value: T) {
+    this.lastValue = value;
+    this.hasValue = true;
+  }
+
+  notifyNext() {
+    if (this.hasValue) {
+      this.hasValue = false;
+      this.destination.next(this.lastValue);
+    }
+  }
+}
+
+class SampleNotificationSubscriber<T> extends Subscriber<T> {
+  constructor(private parent: SampleSubscriber<T>) {
+    super(null);
+  }
+
+  _next() {
+    this.parent.notifyNext();
+  }
+
+  _error(err: any) {
+    this.parent.error(err);
+  }
+
+  _complete() {
+    //noop
+  }
+}