123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108 |
- import { tryCatch } from '../util/tryCatch';
- import { errorObject } from '../util/errorObject';
- import { OuterSubscriber } from '../OuterSubscriber';
- import { subscribeToResult } from '../util/subscribeToResult';
- /**
- * Ignores source values for a duration determined by another Observable, then
- * emits the most recent value from the source Observable, then repeats this
- * process.
- *
- * <span class="informal">It's like {@link auditTime}, but the silencing
- * duration is determined by a second Observable.</span>
- *
- * <img src="./img/audit.png" width="100%">
- *
- * `audit` is similar to `throttle`, but emits the last value from the silenced
- * time window, instead of the first value. `audit` 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 by calling the `durationSelector` function with the
- * source value, which returns the "duration" Observable. When the duration
- * Observable emits a value or completes, 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.
- *
- * @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.audit(ev => Rx.Observable.interval(1000));
- * result.subscribe(x => console.log(x));
- *
- * @see {@link auditTime}
- * @see {@link debounce}
- * @see {@link delayWhen}
- * @see {@link sample}
- * @see {@link throttle}
- *
- * @param {function(value: T): SubscribableOrPromise} durationSelector A function
- * that receives a value from the source Observable, for computing the silencing
- * duration, returned as an Observable or a Promise.
- * @return {Observable<T>} An Observable that performs rate-limiting of
- * emissions from the source Observable.
- * @method audit
- * @owner Observable
- */
- export function audit(durationSelector) {
- return function auditOperatorFunction(source) {
- return source.lift(new AuditOperator(durationSelector));
- };
- }
- class AuditOperator {
- constructor(durationSelector) {
- this.durationSelector = durationSelector;
- }
- call(subscriber, source) {
- return source.subscribe(new AuditSubscriber(subscriber, this.durationSelector));
- }
- }
- /**
- * We need this JSDoc comment for affecting ESDoc.
- * @ignore
- * @extends {Ignored}
- */
- class AuditSubscriber extends OuterSubscriber {
- constructor(destination, durationSelector) {
- super(destination);
- this.durationSelector = durationSelector;
- this.hasValue = false;
- }
- _next(value) {
- this.value = value;
- this.hasValue = true;
- if (!this.throttled) {
- const duration = tryCatch(this.durationSelector)(value);
- if (duration === errorObject) {
- this.destination.error(errorObject.e);
- }
- else {
- const innerSubscription = subscribeToResult(this, duration);
- if (innerSubscription.closed) {
- this.clearThrottle();
- }
- else {
- this.add(this.throttled = innerSubscription);
- }
- }
- }
- }
- clearThrottle() {
- 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);
- }
- }
- notifyNext(outerValue, innerValue, outerIndex, innerIndex) {
- this.clearThrottle();
- }
- notifyComplete() {
- this.clearThrottle();
- }
- }
- //# sourceMappingURL=audit.js.map
|