--- /dev/null
+import { Operator } from '../Operator';
+import { async } from '../scheduler/async';
+import { Observable } from '../Observable';
+import { Subscriber } from '../Subscriber';
+import { Subscription } from '../Subscription';
+import { isScheduler } from '../util/isScheduler';
+import { OperatorFunction, SchedulerAction, SchedulerLike } from '../types';
+
+/* tslint:disable:max-line-length */
+export function bufferTime<T>(bufferTimeSpan: number, scheduler?: SchedulerLike): OperatorFunction<T, T[]>;
+export function bufferTime<T>(bufferTimeSpan: number, bufferCreationInterval: number | null | undefined, scheduler?: SchedulerLike): OperatorFunction<T, T[]>;
+export function bufferTime<T>(bufferTimeSpan: number, bufferCreationInterval: number | null | undefined, maxBufferSize: number, scheduler?: SchedulerLike): OperatorFunction<T, T[]>;
+/* tslint:enable:max-line-length */
+
+/**
+ * Buffers the source Observable values for a specific time period.
+ *
+ * <span class="informal">Collects values from the past as an array, and emits
+ * those arrays periodically in time.</span>
+ *
+ * 
+ *
+ * Buffers values from the source for a specific time duration `bufferTimeSpan`.
+ * Unless the optional argument `bufferCreationInterval` is given, it emits and
+ * resets the buffer every `bufferTimeSpan` milliseconds. If
+ * `bufferCreationInterval` is given, this operator opens the buffer every
+ * `bufferCreationInterval` milliseconds and closes (emits and resets) the
+ * buffer every `bufferTimeSpan` milliseconds. When the optional argument
+ * `maxBufferSize` is specified, the buffer will be closed either after
+ * `bufferTimeSpan` milliseconds or when it contains `maxBufferSize` elements.
+ *
+ * ## Examples
+ *
+ * Every second, emit an array of the recent click events
+ *
+ * ```ts
+ * import { fromEvent } from 'rxjs';
+ * import { bufferTime } from 'rxjs/operators';
+ *
+ * const clicks = fromEvent(document, 'click');
+ * const buffered = clicks.pipe(bufferTime(1000));
+ * buffered.subscribe(x => console.log(x));
+ * ```
+ *
+ * Every 5 seconds, emit the click events from the next 2 seconds
+ *
+ * ```ts
+ * import { fromEvent } from 'rxjs';
+ * import { bufferTime } from 'rxjs/operators';
+ *
+ * const clicks = fromEvent(document, 'click');
+ * const buffered = clicks.pipe(bufferTime(2000, 5000));
+ * buffered.subscribe(x => console.log(x));
+ * ```
+ *
+ * @see {@link buffer}
+ * @see {@link bufferCount}
+ * @see {@link bufferToggle}
+ * @see {@link bufferWhen}
+ * @see {@link windowTime}
+ *
+ * @param {number} bufferTimeSpan The amount of time to fill each buffer array.
+ * @param {number} [bufferCreationInterval] The interval at which to start new
+ * buffers.
+ * @param {number} [maxBufferSize] The maximum buffer size.
+ * @param {SchedulerLike} [scheduler=async] The scheduler on which to schedule the
+ * intervals that determine buffer boundaries.
+ * @return {Observable<T[]>} An observable of arrays of buffered values.
+ * @method bufferTime
+ * @owner Observable
+ */
+export function bufferTime<T>(bufferTimeSpan: number): OperatorFunction<T, T[]> {
+ let length: number = arguments.length;
+
+ let scheduler: SchedulerLike = async;
+ if (isScheduler(arguments[arguments.length - 1])) {
+ scheduler = arguments[arguments.length - 1];
+ length--;
+ }
+
+ let bufferCreationInterval: number = null;
+ if (length >= 2) {
+ bufferCreationInterval = arguments[1];
+ }
+
+ let maxBufferSize: number = Number.POSITIVE_INFINITY;
+ if (length >= 3) {
+ maxBufferSize = arguments[2];
+ }
+
+ return function bufferTimeOperatorFunction(source: Observable<T>) {
+ return source.lift(new BufferTimeOperator<T>(bufferTimeSpan, bufferCreationInterval, maxBufferSize, scheduler));
+ };
+}
+
+class BufferTimeOperator<T> implements Operator<T, T[]> {
+ constructor(private bufferTimeSpan: number,
+ private bufferCreationInterval: number,
+ private maxBufferSize: number,
+ private scheduler: SchedulerLike) {
+ }
+
+ call(subscriber: Subscriber<T[]>, source: any): any {
+ return source.subscribe(new BufferTimeSubscriber(
+ subscriber, this.bufferTimeSpan, this.bufferCreationInterval, this.maxBufferSize, this.scheduler
+ ));
+ }
+}
+
+class Context<T> {
+ buffer: T[] = [];
+ closeAction: Subscription;
+}
+
+interface DispatchCreateArg<T> {
+ bufferTimeSpan: number;
+ bufferCreationInterval: number;
+ subscriber: BufferTimeSubscriber<T>;
+ scheduler: SchedulerLike;
+}
+
+interface DispatchCloseArg<T> {
+ subscriber: BufferTimeSubscriber<T>;
+ context: Context<T>;
+}
+
+/**
+ * We need this JSDoc comment for affecting ESDoc.
+ * @ignore
+ * @extends {Ignored}
+ */
+class BufferTimeSubscriber<T> extends Subscriber<T> {
+ private contexts: Array<Context<T>> = [];
+ private timespanOnly: boolean;
+
+ constructor(destination: Subscriber<T[]>,
+ private bufferTimeSpan: number,
+ private bufferCreationInterval: number,
+ private maxBufferSize: number,
+ private scheduler: SchedulerLike) {
+ super(destination);
+ const context = this.openContext();
+ this.timespanOnly = bufferCreationInterval == null || bufferCreationInterval < 0;
+ if (this.timespanOnly) {
+ const timeSpanOnlyState = { subscriber: this, context, bufferTimeSpan };
+ this.add(context.closeAction = scheduler.schedule(dispatchBufferTimeSpanOnly, bufferTimeSpan, timeSpanOnlyState));
+ } else {
+ const closeState = { subscriber: this, context };
+ const creationState: DispatchCreateArg<T> = { bufferTimeSpan, bufferCreationInterval, subscriber: this, scheduler };
+ this.add(context.closeAction = scheduler.schedule<DispatchCloseArg<T>>(dispatchBufferClose, bufferTimeSpan, closeState));
+ this.add(scheduler.schedule<DispatchCreateArg<T>>(dispatchBufferCreation, bufferCreationInterval, creationState));
+ }
+ }
+
+ protected _next(value: T) {
+ const contexts = this.contexts;
+ const len = contexts.length;
+ let filledBufferContext: Context<T>;
+ for (let i = 0; i < len; i++) {
+ const context = contexts[i];
+ const buffer = context.buffer;
+ buffer.push(value);
+ if (buffer.length == this.maxBufferSize) {
+ filledBufferContext = context;
+ }
+ }
+
+ if (filledBufferContext) {
+ this.onBufferFull(filledBufferContext);
+ }
+ }
+
+ protected _error(err: any) {
+ this.contexts.length = 0;
+ super._error(err);
+ }
+
+ protected _complete() {
+ const { contexts, destination } = this;
+ while (contexts.length > 0) {
+ const context = contexts.shift();
+ destination.next(context.buffer);
+ }
+ super._complete();
+ }
+
+ /** @deprecated This is an internal implementation detail, do not use. */
+ _unsubscribe() {
+ this.contexts = null;
+ }
+
+ protected onBufferFull(context: Context<T>) {
+ this.closeContext(context);
+ const closeAction = context.closeAction;
+ closeAction.unsubscribe();
+ this.remove(closeAction);
+
+ if (!this.closed && this.timespanOnly) {
+ context = this.openContext();
+ const bufferTimeSpan = this.bufferTimeSpan;
+ const timeSpanOnlyState = { subscriber: this, context, bufferTimeSpan };
+ this.add(context.closeAction = this.scheduler.schedule(dispatchBufferTimeSpanOnly, bufferTimeSpan, timeSpanOnlyState));
+ }
+ }
+
+ openContext(): Context<T> {
+ const context: Context<T> = new Context<T>();
+ this.contexts.push(context);
+ return context;
+ }
+
+ closeContext(context: Context<T>) {
+ this.destination.next(context.buffer);
+ const contexts = this.contexts;
+
+ const spliceIndex = contexts ? contexts.indexOf(context) : -1;
+ if (spliceIndex >= 0) {
+ contexts.splice(contexts.indexOf(context), 1);
+ }
+ }
+}
+
+function dispatchBufferTimeSpanOnly(this: SchedulerAction<any>, state: any) {
+ const subscriber: BufferTimeSubscriber<any> = state.subscriber;
+
+ const prevContext = state.context;
+ if (prevContext) {
+ subscriber.closeContext(prevContext);
+ }
+
+ if (!subscriber.closed) {
+ state.context = subscriber.openContext();
+ state.context.closeAction = this.schedule(state, state.bufferTimeSpan);
+ }
+}
+
+function dispatchBufferCreation<T>(this: SchedulerAction<DispatchCreateArg<T>>, state: DispatchCreateArg<T>) {
+ const { bufferCreationInterval, bufferTimeSpan, subscriber, scheduler } = state;
+ const context = subscriber.openContext();
+ const action = <SchedulerAction<DispatchCreateArg<T>>>this;
+ if (!subscriber.closed) {
+ subscriber.add(context.closeAction = scheduler.schedule<DispatchCloseArg<T>>(dispatchBufferClose, bufferTimeSpan, { subscriber, context }));
+ action.schedule(state, bufferCreationInterval);
+ }
+}
+
+function dispatchBufferClose<T>(arg: DispatchCloseArg<T>) {
+ const { subscriber, context } = arg;
+ subscriber.closeContext(context);
+}
projects / dotfiles / .git / blobdiff