Skip to content

Commit

Permalink
feat(auditTime): add higher-order lettable version of auditTime
Browse files Browse the repository at this point in the history
  • Loading branch information
benlesh committed Jun 25, 2017
1 parent e2daefe commit 9e963aa
Show file tree
Hide file tree
Showing 3 changed files with 55 additions and 59 deletions.
62 changes: 3 additions & 59 deletions src/operator/auditTime.ts
Original file line number Diff line number Diff line change
@@ -1,9 +1,7 @@
import { async } from '../scheduler/async';
import { Operator } from '../Operator';
import { IScheduler } from '../Scheduler';
import { Subscriber } from '../Subscriber';
import { Observable } from '../Observable';
import { Subscription, TeardownLogic } from '../Subscription';
import { auditTime as higherOrder } from '../operators';

/**
* Ignores source values for `duration` milliseconds, then emits the most recent
Expand Down Expand Up @@ -48,59 +46,5 @@ import { Subscription, TeardownLogic } from '../Subscription';
* @owner Observable
*/
export function auditTime<T>(this: Observable<T>, duration: number, scheduler: IScheduler = async): Observable<T> {
return this.lift(new AuditTimeOperator(duration, scheduler));
}

class AuditTimeOperator<T> implements Operator<T, T> {
constructor(private duration: number,
private scheduler: IScheduler) {
}

call(subscriber: Subscriber<T>, source: any): TeardownLogic {
return source.subscribe(new AuditTimeSubscriber(subscriber, this.duration, this.scheduler));
}
}

/**
* We need this JSDoc comment for affecting ESDoc.
* @ignore
* @extends {Ignored}
*/
class AuditTimeSubscriber<T> extends Subscriber<T> {

private value: T;
private hasValue: boolean = false;
private throttled: Subscription;

constructor(destination: Subscriber<T>,
private duration: number,
private scheduler: IScheduler) {
super(destination);
}

protected _next(value: T): void {
this.value = value;
this.hasValue = true;
if (!this.throttled) {
this.add(this.throttled = this.scheduler.schedule(dispatchNext, this.duration, this));
}
}

clearThrottle(): void {
const { value, hasValue, throttled } = this;
if (throttled) {
this.remove(throttled);
this.throttled = null;
throttled.unsubscribe();
}
if (hasValue) {
this.value = null;
this.hasValue = false;
this.destination.next(value);
}
}
}

function dispatchNext<T>(subscriber: AuditTimeSubscriber<T>): void {
subscriber.clearThrottle();
}
return higherOrder(duration, scheduler)(this);
}
51 changes: 51 additions & 0 deletions src/operators/auditTime.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
import { async } from '../scheduler/async';
import { IScheduler } from '../Scheduler';
import { audit } from './audit';
import { timer } from '../observable/timer';
import { MonoTypeOperatorFunction } from '../interfaces';

/**
* Ignores source values for `duration` milliseconds, then emits the most recent
* value from the source Observable, then repeats this process.
*
* <span class="informal">When it sees a source values, it ignores that plus
* the next ones for `duration` milliseconds, and then it emits the most recent
* value from the source.</span>
*
* <img src="./img/auditTime.png" width="100%">
*
* `auditTime` is similar to `throttleTime`, but emits the last value from the
* silenced time window, instead of the first value. `auditTime` emits the most
* recent value from the source Observable on the output Observable as soon as
* its internal timer becomes disabled, and ignores source values while the
* timer is enabled. Initially, the timer is disabled. As soon as the first
* source value arrives, the timer is enabled. After `duration` milliseconds (or
* the time unit determined internally by the optional `scheduler`) has passed,
* the timer is disabled, then the most recent source value is emitted on the
* output Observable, and this process repeats for the next source value.
* Optionally takes a {@link IScheduler} for managing timers.
*
* @example <caption>Emit clicks at a rate of at most one click per second</caption>
* var clicks = Rx.Observable.fromEvent(document, 'click');
* var result = clicks.auditTime(1000);
* result.subscribe(x => console.log(x));
*
* @see {@link audit}
* @see {@link debounceTime}
* @see {@link delay}
* @see {@link sampleTime}
* @see {@link throttleTime}
*
* @param {number} duration Time to wait before emitting the most recent source
* value, measured in milliseconds or the time unit determined internally
* by the optional `scheduler`.
* @param {Scheduler} [scheduler=async] The {@link IScheduler} to use for
* managing the timers that handle the rate-limiting behavior.
* @return {Observable<T>} An Observable that performs rate-limiting of
* emissions from the source Observable.
* @method auditTime
* @owner Observable
*/
export function auditTime<T>(duration: number, scheduler: IScheduler = async): MonoTypeOperatorFunction<T> {
return audit(() => timer(duration, scheduler));
}
1 change: 1 addition & 0 deletions src/operators/index.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
export { audit } from './audit';
export { auditTime } from './auditTime';
export { catchError } from './catchError';
export { concat } from './concat';
export { concatAll } from './concatAll';
Expand Down

0 comments on commit 9e963aa

Please sign in to comment.