feat(sampleTime): reimplement sampleTime with RxJS 4 behavior

BREAKING CHANGE: `sampleTime` now has the same behavior `sample(number, scheduler)` did in RxJS 4

spec/operators/sampleTime-spec.js

@@ -0,0 +1,109 @@
+/* globals describe, it, expect, expectObservable, expectSubscriptions, cold, hot, rxTestScheduler */
+var Rx = require('../../dist/cjs/Rx.KitchenSink');
+var Observable = Rx.Observable;
+
+describe('Observable.prototype.sampleTime', function () {
+ it('should get samples on a delay', function () {
+ var e1 = hot('----a-^--b----c----d----e----f----|');
+ var e1subs = '^ !';
+ var expected = '-----------c----------e-----|';
+ // timer -----------!----------!---------
+
+ expectObservable(e1.sampleTime(110, rxTestScheduler)).toBe(expected);
+ expectSubscriptions(e1.subscriptions).toBe(e1subs);
+ });
+
+ it('should sample nothing if new value has not arrived', function () {
+ var e1 = hot('----a-^--b----c--------------f----|');
+ var e1subs = '^ !';
+ var expected = '-----------c----------------|';
+ // timer -----------!----------!---------
+
+ expectObservable(e1.sampleTime(110, rxTestScheduler)).toBe(expected);
+ expectSubscriptions(e1.subscriptions).toBe(e1subs);
+ });
+
+ it('should sample ifnew value has arrived, even if it is the same value', function () {
+ var e1 = hot('----a-^--b----c----------c---f----|');
+ var e1subs = '^ !';
+ var expected = '-----------c----------c-----|';
+ // timer -----------!----------!---------
+
+ expectObservable(e1.sampleTime(110, rxTestScheduler)).toBe(expected);
+ expectSubscriptions(e1.subscriptions).toBe(e1subs);
+ });
+
+ it('should sample nothing if source has not nexted by time of sample', function () {
+ var e1 = hot('----a-^-------------b-------------|');
+ var e1subs = '^ !';
+ var expected = '----------------------b-----|';
+ // timer -----------!----------!---------
+
+ expectObservable(e1.sampleTime(110, rxTestScheduler)).toBe(expected);
+ expectSubscriptions(e1.subscriptions).toBe(e1subs);
+ });
+
+ it('should raise error if source raises error', function () {
+ var e1 = hot('----a-^--b----c----d----#');
+ var e1subs = '^ !';
+ var expected = '-----------c------#';
+ // timer -----------!----------!---------
+
+ expectObservable(e1.sampleTime(110, rxTestScheduler)).toBe(expected);
+ expectSubscriptions(e1.subscriptions).toBe(e1subs);
+ });
+
+ it('should allow unsubscribing explicitly and early', function () {
+ var e1 = hot('----a-^--b----c----d----e----f----|');
+ var unsub = ' ! ';
+ var e1subs = '^ ! ';
+ var expected = '-----------c----- ';
+ // timer -----------!----------!---------
+
+ expectObservable(e1.sampleTime(110, rxTestScheduler), unsub).toBe(expected);
+ expectSubscriptions(e1.subscriptions).toBe(e1subs);
+ });
+
+ it('should not break unsubscription chains when result is unsubscribed explicitly', function () {
+ var e1 = hot('----a-^--b----c----d----e----f----|');
+ var e1subs = '^ ! ';
+ // timer -----------!----------!---------
+ var expected = '-----------c----- ';
+ var unsub = ' ! ';
+
+ var result = e1
+ .mergeMap(function (x) { return Observable.of(x); })
+ .sampleTime(110, rxTestScheduler)
+ .mergeMap(function (x) { return Observable.of(x); });
+
+ expectObservable(result, unsub).toBe(expected);
+ expectSubscriptions(e1.subscriptions).toBe(e1subs);
+ });
+
+ it('should completes if source does not emits', function () {
+ var e1 = cold('|');
+ var e1subs = '(^!)';
+ var expected = '|';
+
+ expectObservable(e1.sampleTime(60, rxTestScheduler)).toBe(expected);
+ expectSubscriptions(e1.subscriptions).toBe(e1subs);
+ });
+
+ it('should raise error if source throws immediately', function () {
+ var e1 = cold('#');
+ var e1subs = '(^!)';
+ var expected = '#';
+
+ expectObservable(e1.sampleTime(60, rxTestScheduler)).toBe(expected);
+ expectSubscriptions(e1.subscriptions).toBe(e1subs);
+ });
+
+ it('should not completes if source does not complete', function () {
+ var e1 = cold('-');
+ var e1subs = '^';
+ var expected = '-';
+
+ expectObservable(e1.sampleTime(60, rxTestScheduler)).toBe(expected);
+ expectSubscriptions(e1.subscriptions).toBe(e1subs);
+ });
+});

src/Observable.ts

@@ -247,6 +247,7 @@ export class Observable<T> implements CoreOperators<T> {
  retry: (count?: number) => Observable<T>;
  retryWhen: (notifier: (errors: Observable<any>) => Observable<any>) => Observable<T>;
  sample: (notifier: Observable<any>) => Observable<T>;
+ sampleTime: (delay: number, scheduler?: Scheduler) => 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

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

src/Rx.ts

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

src/add/operator/sampleTime.ts

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

src/operator/sampleTime.ts

@@ -0,0 +1,46 @@
+import {Observable} from '../Observable';
+import {Operator} from '../Operator';
+import {Subscriber} from '../Subscriber';
+import {Scheduler} from '../Scheduler';
+import {asap} from '../scheduler/asap';
+
+export function sampleTime<T>(delay: number, scheduler: Scheduler = asap): 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(subscriber: Subscriber<R>) {
+ return new SampleTimeSubscriber(subscriber, this.delay, this.scheduler);
+ }
+}
+
+class SampleTimeSubscriber<T> extends Subscriber<T> {
+ lastValue: T;
+ hasValue: boolean = false;
+
+ constructor(destination: Subscriber<T>, private delay: number, private scheduler: Scheduler) {
+ super(destination);
+ this.add(scheduler.schedule(dispatchNotification, delay, { subscriber: this, delay }));
+ }
+
+ _next(value: T) {
+ this.lastValue = value;
+ this.hasValue = true;
+ }
+
+ notifyNext() {
+ if (this.hasValue) {
+ this.hasValue = false;
+ this.destination.next(this.lastValue);
+ }
+ }
+}
+
+function dispatchNotification<T>(state) {
+ let { subscriber, delay } = state;
+ subscriber.notifyNext();
+ (<any>this).schedule(state, delay);
+}