feat(operator): add sample and sampleTime

- sample is equivalent to Rx2/3 sample(observable)
- sampleTime is equivalent to Rx2/3 sample(number, scheduler)

closes #178

spec/operators/sample-spec.js

@@ -0,0 +1,26 @@
+/* globals describe, it, expect */
+var Rx = require('../../dist/cjs/Rx');
+var Observable = Rx.Observable;
+
+describe('Observable.prototype.sample', function () {
+  it('should get samples when the notifier emits', function (done) {
+    var expected = [1, 3, 5];
+    Observable.interval(100)
+      .sample(Observable.interval(220))
+      .take(3)
+      .subscribe(function (x) {
+        expect(x).toBe(expected.shift());
+      }, null, done);
+  }, 2000);
+
+  it('should not complete when the notifier completes, nor should it emit', function (done) {
+    Observable.interval(100)
+      .sample(Observable.timer(220))
+      .subscribe(function (x) {
+        expect(x).toBe(1);
+        setTimeout(done, 500);
+      }, null, function () {
+        throw 'should not be called';
+      });
+  });
+});

spec/operators/sampleTime-spec.js

@@ -0,0 +1,15 @@
+/* globals describe, it, expect */
+var Rx = require('../../dist/cjs/Rx');
+var Observable = Rx.Observable;
+
+describe('Observable.prototype.sampleTime', function () {
+  it('should get samples on a delay', function (done) {
+    var expected = [1, 3, 5];
+    Observable.interval(100)
+      .sampleTime(220)
+      .take(3)
+      .subscribe(function (x) {
+        expect(x).toBe(expected.shift());
+      }, null, done);
+  }, 2000);
+});

src/Observable.ts

@@ -178,6 +178,9 @@ export default class Observable<T>  {
   bufferTime: <T>(bufferTimeSpan: number, bufferCreationInterval?: number, scheduler?: Scheduler) => Observable<T[]>;
   bufferCount: <T>(bufferSize: number, startBufferEvery: number) => Observable<T[]>;
 
+  sample: <T>(notifier: Observable<any>) => Observable<T>;
+  sampleTime: <T>(delay: number, scheduler?: Scheduler) => Observable<T>;
+
   finally: (ensure: () => void, thisArg?: any) => Observable<T>;
   timeout: <T>(due: number|Date, errorToSend?: any, scheduler?: Scheduler) => Observable<T>;
   timeoutWith: <T>(due: number|Date, withObservable: Observable<any>, scheduler?: Scheduler) => Observable<T>;

src/Rx.ts

@@ -207,6 +207,12 @@ observableProto.bufferTime = bufferTime;
 observableProto.bufferToggle = bufferToggle;
 observableProto.bufferWhen = bufferWhen;
 
+import sample from './operators/sample';
+import sampleTime from './operators/sampleTime';
+
+observableProto.sample = sample;
+observableProto.sampleTime = sampleTime;
+
 var Scheduler = {
   nextTick,
   immediate

src/operators/sample.ts

@@ -0,0 +1,64 @@
+import Observable from '../Observable';
+import Operator from '../Operator';
+import Observer from '../Observer';
+import Subscriber from '../Subscriber';
+import Subscription from '../Subscription';
+
+export default 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(observer: Observer<R>) {
+    return new SampleSubscriber(observer, this.notifier);
+  }
+}
+
+class SampleSubscriber<T> extends Subscriber<T> {
+  private lastValue: T;
+  private hasValue: boolean = false;
+  private innerSubscription: Subscription<T>;
+
+  constructor(destination: Observer<T>, private notifier: Observable<any>) {
+    super(destination);
+    this.add(this.innerSubscription = notifier.subscribe(new SampleNoficationSubscriber(this)));
+  }
+
+  _next(value: T) {
+    this.lastValue = value;
+    this.hasValue = true;
+  }
+
+  notifyNext() {
+    if (this.hasValue) {
+      this.destination.next(this.lastValue);
+    }  
+  }
+
+  notifyComplete() {
+    this.remove(this.innerSubscription);
+    this.innerSubscription.unsubscribe();
+  }
+}
+
+
+class SampleNoficationSubscriber<T> extends Subscriber<T> {
+  constructor(private parent: SampleSubscriber<T>) {
+    super(null);
+  }
+
+  _next() {
+    this.parent.notifyNext();
+  }
+
+  _error(err: any) {
+    this.parent.error(err);
+  }
+
+  _complete() {
+    this.parent.notifyComplete();
+  }
+}

src/operators/sampleTime.ts

@@ -0,0 +1,46 @@
+import Observable from '../Observable';
+import Operator from '../Operator';
+import Observer from '../Observer';
+import Subscriber from '../Subscriber';
+import Subscription from '../Subscription';
+import Scheduler from '../Scheduler';
+import nextTick from '../schedulers/nextTick';
+
+export default function sampleTime<T>(delay: number, scheduler: Scheduler = nextTick): Observable<T> {
+  return this.lift(new SampleTimeOperator(delay, scheduler));
+}
+
+class SampleTimeOperator<T, R> implements Operator<T, R> {
+  constructor(private delay: number, private scheduler: Scheduler) {
+  }
+
+  call(observer: Observer<R>) {
+    return new SampleTimeSubscriber(observer, this.delay, this.scheduler);
+  }
+}
+
+class SampleTimeSubscriber<T> extends Subscriber<T> {
+  lastValue: T;
+  hasValue: boolean = false;
+
+  constructor(destination: Observer<T>, private delay: number, private scheduler: Scheduler) {
+    super(destination);
+    this.add(scheduler.schedule(delay, { subscriber: this }, dispatchNotification));
+  }
+
+  _next(value: T) {
+    this.lastValue = value;
+    this.hasValue = true;
+  }
+
+  notifyNext() {
+    if (this.hasValue) {
+      this.destination.next(this.lastValue);
+    }
+  }
+}
+
+function dispatchNotification<T>(state: { subscriber: SampleTimeSubscriber<T> }) {
+  state.subscriber.notifyNext();
+  (<any>this).schedule(state);
+}