es6/operator/timeoutWith.js
import { async } from '../scheduler/async';
import { isDate } from '../util/isDate';
import { OuterSubscriber } from '../OuterSubscriber';
import { subscribeToResult } from '../util/subscribeToResult';
/* tslint:enable:max-line-length */
/**
* @param due
* @param withObservable
* @param scheduler
* @return {Observable<R>|WebSocketSubject<T>|Observable<T>}
* @method timeoutWith
* @owner Observable
*/
export function timeoutWith(due, withObservable, scheduler = async) {
let absoluteTimeout = isDate(due);
let waitFor = absoluteTimeout ? (+due - scheduler.now()) : Math.abs(due);
return this.lift(new TimeoutWithOperator(waitFor, absoluteTimeout, withObservable, scheduler));
}
class TimeoutWithOperator {
constructor(waitFor, absoluteTimeout, withObservable, scheduler) {
this.waitFor = waitFor;
this.absoluteTimeout = absoluteTimeout;
this.withObservable = withObservable;
this.scheduler = scheduler;
}
call(subscriber, source) {
return source.subscribe(new TimeoutWithSubscriber(subscriber, this.absoluteTimeout, this.waitFor, this.withObservable, this.scheduler));
}
}
/**
* We need this JSDoc comment for affecting ESDoc.
* @ignore
* @extends {Ignored}
*/
class TimeoutWithSubscriber extends OuterSubscriber {
constructor(destination, absoluteTimeout, waitFor, withObservable, scheduler) {
super(destination);
this.absoluteTimeout = absoluteTimeout;
this.waitFor = waitFor;
this.withObservable = withObservable;
this.scheduler = scheduler;
this.action = null;
this.scheduleTimeout();
}
static dispatchTimeout(subscriber) {
const { withObservable } = subscriber;
subscriber._unsubscribeAndRecycle();
subscriber.add(subscribeToResult(subscriber, withObservable));
}
scheduleTimeout() {
const { action } = this;
if (action) {
// Recycle the action if we've already scheduled one. All the production
// Scheduler Actions mutate their state/delay time and return themeselves.
// VirtualActions are immutable, so they create and return a clone. In this
// case, we need to set the action reference to the most recent VirtualAction,
// to ensure that's the one we clone from next time.
this.action = action.schedule(this, this.waitFor);
}
else {
this.add(this.action = this.scheduler.schedule(TimeoutWithSubscriber.dispatchTimeout, this.waitFor, this));
}
}
_next(value) {
if (!this.absoluteTimeout) {
this.scheduleTimeout();
}
super._next(value);
}
_unsubscribe() {
this.action = null;
this.scheduler = null;
this.withObservable = null;
}
}
//# sourceMappingURL=timeoutWith.js.map