Skip to content

Commit

Permalink
feat(bufferTime): add maxBufferSize optional argument
Browse files Browse the repository at this point in the history
  • Loading branch information
figueredo committed Mar 30, 2016
1 parent cc9f6f2 commit 1fb6aae
Show file tree
Hide file tree
Showing 2 changed files with 145 additions and 49 deletions.
81 changes: 69 additions & 12 deletions spec/operators/bufferTime-spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ describe('Observable.prototype.bufferTime', () => {
z: []
};

const result = e1.bufferTime(t, null, rxTestScheduler);
const result = e1.bufferTime(t, null, Number.POSITIVE_INFINITY, rxTestScheduler);

expectObservable(result).toBe(expected, values);
expectSubscriptions(e1.subscriptions).toBe(subs);
Expand All @@ -34,11 +34,47 @@ describe('Observable.prototype.bufferTime', () => {
z: []
};

const result = e1.bufferTime(t, null, rxTestScheduler);
const result = e1.bufferTime(t, null, Number.POSITIVE_INFINITY, rxTestScheduler);

expectObservable(result).toBe(expected, values);
});

it('should emit buffers at intervals or when the buffer is full', () => {
const e1 = hot('---a---b---c---d---e---f---g-----|');
const subs = '^ !';
const t = time( '----------|');
const expected = '-------w-------x-------y---------(z|)';
const values = {
w: ['a', 'b'],
x: ['c', 'd'],
y: ['e', 'f'],
z: ['g']
};

const result = e1.bufferTime(t, null, 2, rxTestScheduler);

expectObservable(result).toBe(expected, values);
expectSubscriptions(e1.subscriptions).toBe(subs);
});

it('should emit buffers at intervals or when the buffer is full test 2', () => {
const e1 = hot('---a---b---c---d---e---f---g-----|');
const subs = '^ !';
const t = time( '----------|');
const expected = '----------w--------x---------y---(z|)';
const values = {
w: ['a', 'b'],
x: ['c', 'd', 'e'],
y: ['f', 'g'],
z: []
};

const result = e1.bufferTime(t, null, 3, rxTestScheduler);

expectObservable(result).toBe(expected, values);
expectSubscriptions(e1.subscriptions).toBe(subs);
});

it('should emit buffers that have been created at intervals and close after the specified delay', () => {
const e1 = hot('---a---b---c----d----e----f----g----h----i----(k|)');
// --------------------*--------------------*---- start interval
Expand All @@ -54,7 +90,28 @@ describe('Observable.prototype.bufferTime', () => {
z: ['i', 'k']
};

const result = e1.bufferTime(t, interval, rxTestScheduler);
const result = e1.bufferTime(t, interval, Number.POSITIVE_INFINITY, rxTestScheduler);

expectObservable(result).toBe(expected, values);
});

it('should emit buffers that have been created at intervals and close after the specified delay ' +
'or when the buffer is full', () => {
const e1 = hot('---a---b---c----d----e----f----g----h----i----(k|)');
// --------------------*--------------------*---- start interval
// ---------------------| timespans
// ---------------------|
// -----|
const t = time( '---------------------|');
const interval = time( '--------------------|');
const expected = '----------------x-------------------y---------(z|)';
const values = {
x: ['a', 'b', 'c', 'd'],
y: ['e', 'f', 'g', 'h'],
z: ['i', 'k']
};

const result = e1.bufferTime(t, interval, 4, rxTestScheduler);

expectObservable(result).toBe(expected, values);
});
Expand All @@ -81,7 +138,7 @@ describe('Observable.prototype.bufferTime', () => {
f: []
};

const result = e1.bufferTime(t, interval, rxTestScheduler);
const result = e1.bufferTime(t, interval, Number.POSITIVE_INFINITY, rxTestScheduler);

expectObservable(result).toBe(expected, values);
expectSubscriptions(e1.subscriptions).toBe(e1subs);
Expand All @@ -107,7 +164,7 @@ describe('Observable.prototype.bufferTime', () => {
e: []
};

const result = e1.bufferTime(t, interval, rxTestScheduler);
const result = e1.bufferTime(t, interval, Number.POSITIVE_INFINITY, rxTestScheduler);

expectObservable(result).toBe(expected, values);
});
Expand All @@ -127,7 +184,7 @@ describe('Observable.prototype.bufferTime', () => {
a: ['2', '3', '4']
};

const result = e1.bufferTime(t, interval, rxTestScheduler);
const result = e1.bufferTime(t, interval, Number.POSITIVE_INFINITY, rxTestScheduler);

expectObservable(result, unsub).toBe(expected, values);
expectSubscriptions(e1.subscriptions).toBe(subs);
Expand All @@ -150,7 +207,7 @@ describe('Observable.prototype.bufferTime', () => {

const result = e1
.mergeMap((x: any) => Observable.of(x))
.bufferTime(t, interval, rxTestScheduler)
.bufferTime(t, interval, Number.POSITIVE_INFINITY, rxTestScheduler)
.mergeMap((x: any) => Observable.of(x));

expectObservable(result, unsub).toBe(expected, values);
Expand All @@ -164,7 +221,7 @@ describe('Observable.prototype.bufferTime', () => {
const values = { b: [] };
const t = time('----------|');

const result = e1.bufferTime(t, null, rxTestScheduler);
const result = e1.bufferTime(t, null, Number.POSITIVE_INFINITY, rxTestScheduler);

expectObservable(result).toBe(expected, values);
expectSubscriptions(e1.subscriptions).toBe(e1subs);
Expand All @@ -176,7 +233,7 @@ describe('Observable.prototype.bufferTime', () => {
const t = time( '----------|');
const expected = '----------a---------a---------a---------a----';

const result = e1.bufferTime(t, null, rxTestScheduler);
const result = e1.bufferTime(t, null, Number.POSITIVE_INFINITY, rxTestScheduler);

expectObservable(result, unsub).toBe(expected, { a: [] });
});
Expand All @@ -186,7 +243,7 @@ describe('Observable.prototype.bufferTime', () => {
const expected = '#';
const t = time('----------|');

const result = e1.bufferTime(t, null, rxTestScheduler);
const result = e1.bufferTime(t, null, Number.POSITIVE_INFINITY, rxTestScheduler);

expectObservable(result).toBe(expected, undefined, new Error('haha'));
});
Expand All @@ -200,7 +257,7 @@ describe('Observable.prototype.bufferTime', () => {
w: ['a', 'b']
};

const result = e1.bufferTime(t, null, rxTestScheduler);
const result = e1.bufferTime(t, null, Number.POSITIVE_INFINITY, rxTestScheduler);

expectObservable(result).toBe(expected, values);
expectSubscriptions(e1.subscriptions).toBe(e1subs);
Expand All @@ -222,7 +279,7 @@ describe('Observable.prototype.bufferTime', () => {
y: ['e', 'f', 'g', 'h', 'i']
};

const result = e1.bufferTime(t, interval, rxTestScheduler);
const result = e1.bufferTime(t, interval, Number.POSITIVE_INFINITY, rxTestScheduler);

expectObservable(result).toBe(expected, values);
expectSubscriptions(e1.subscriptions).toBe(e1subs);
Expand Down
113 changes: 76 additions & 37 deletions src/operator/bufferTime.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import {Operator} from '../Operator';
import {Subscriber} from '../Subscriber';
import {Subscription} from '../Subscription';
import {Observable} from '../Observable';
import {Scheduler} from '../Scheduler';
import {Action} from '../scheduler/Action';
Expand All @@ -18,7 +19,9 @@ import {async} from '../scheduler/async';
* 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.
* 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.
*
* @example <caption>Every second, emit an array of the recent click events</caption>
* var clicks = Rx.Observable.fromEvent(document, 'click');
Expand All @@ -39,6 +42,7 @@ import {async} from '../scheduler/async';
* @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 {Scheduler} [scheduler=async] The scheduler on which to schedule the
* intervals that determine buffer boundaries.
* @return {Observable<T[]>} An observable of arrays of buffered values.
Expand All @@ -47,27 +51,34 @@ import {async} from '../scheduler/async';
*/
export function bufferTime<T>(bufferTimeSpan: number,
bufferCreationInterval: number = null,
maxBufferSize: number = Number.POSITIVE_INFINITY,
scheduler: Scheduler = async): Observable<T[]> {
return this.lift(new BufferTimeOperator<T>(bufferTimeSpan, bufferCreationInterval, scheduler));
return this.lift(new BufferTimeOperator<T>(bufferTimeSpan, bufferCreationInterval, maxBufferSize, scheduler));
}

export interface BufferTimeSignature<T> {
(bufferTimeSpan: number, bufferCreationInterval?: number, scheduler?: Scheduler): Observable<T[]>;
(bufferTimeSpan: number, bufferCreationInterval?: number, maxBufferSize?: number, scheduler?: Scheduler): Observable<T[]>;
}

class BufferTimeOperator<T> implements Operator<T, T[]> {
constructor(private bufferTimeSpan: number,
private bufferCreationInterval: number,
private maxBufferSize: number,
private scheduler: Scheduler) {
}

call(subscriber: Subscriber<T[]>): Subscriber<T> {
return new BufferTimeSubscriber(
subscriber, this.bufferTimeSpan, this.bufferCreationInterval, this.scheduler
subscriber, this.bufferTimeSpan, this.bufferCreationInterval, this.maxBufferSize, this.scheduler
);
}
}

class Context<T> {
buffer: T[] = [];
closeAction: Subscription;
}

type CreationState<T> = {
bufferTimeSpan: number;
bufferCreationInterval: number,
Expand All @@ -76,87 +87,115 @@ type CreationState<T> = {
};

class BufferTimeSubscriber<T> extends Subscriber<T> {
private buffers: Array<T[]> = [];
private contexts: Array<Context<T>> = [];
private timespanOnly: boolean;

constructor(destination: Subscriber<T[]>,
private bufferTimeSpan: number,
private bufferCreationInterval: number,
private maxBufferSize: number,
private scheduler: Scheduler) {
super(destination);
const buffer = this.openBuffer();
if (bufferCreationInterval !== null && bufferCreationInterval >= 0) {
const closeState = { subscriber: this, buffer };
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: CreationState<T> = { bufferTimeSpan, bufferCreationInterval, subscriber: this, scheduler };
this.add(scheduler.schedule(dispatchBufferClose, bufferTimeSpan, closeState));
this.add(context.closeAction = scheduler.schedule(dispatchBufferClose, bufferTimeSpan, closeState));
this.add(scheduler.schedule(dispatchBufferCreation, bufferCreationInterval, creationState));
} else {
const timeSpanOnlyState = { subscriber: this, buffer, bufferTimeSpan };
this.add(scheduler.schedule(dispatchBufferTimeSpanOnly, bufferTimeSpan, timeSpanOnlyState));
}
}

protected _next(value: T) {
const buffers = this.buffers;
const len = buffers.length;
const contexts = this.contexts;
const len = contexts.length;
let filledBufferContext: Context<T>;
for (let i = 0; i < len; i++) {
buffers[i].push(value);
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.buffers.length = 0;
this.contexts.length = 0;
super._error(err);
}

protected _complete() {
const { buffers, destination } = this;
while (buffers.length > 0) {
destination.next(buffers.shift());
const { contexts, destination } = this;
while (contexts.length > 0) {
const context = contexts.shift();
destination.next(context.buffer);
}
super._complete();
}

protected _unsubscribe() {
this.buffers = null;
this.contexts = null;
}

protected onBufferFull(context: Context<T>) {
this.closeContext(context);
const closeAction = context.closeAction;
closeAction.unsubscribe();
this.remove(closeAction);

if (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));
}
}

openBuffer(): T[] {
let buffer: T[] = [];
this.buffers.push(buffer);
return buffer;
openContext(): Context<T> {
let context: Context<T> = new Context<T>();
this.contexts.push(context);
return context;
}

closeBuffer(buffer: T[]) {
this.destination.next(buffer);
const buffers = this.buffers;
buffers.splice(buffers.indexOf(buffer), 1);
closeContext(context: Context<T>) {
this.destination.next(context.buffer);
const contexts = this.contexts;
contexts.splice(contexts.indexOf(context), 1);
}
}

function dispatchBufferTimeSpanOnly(state: any) {
const subscriber: BufferTimeSubscriber<any> = state.subscriber;

const prevBuffer = state.buffer;
if (prevBuffer) {
subscriber.closeBuffer(prevBuffer);
const prevContext = state.context;
if (prevContext) {
subscriber.closeContext(prevContext);
}

state.buffer = subscriber.openBuffer();
state.context = subscriber.openContext();
if (!subscriber.isUnsubscribed) {
(<any>this).schedule(state, state.bufferTimeSpan);
state.context.closeAction = (<any>this).schedule(state, state.bufferTimeSpan);
}
}

function dispatchBufferCreation<T>(state: CreationState<T>) {
const { bufferCreationInterval, bufferTimeSpan, subscriber, scheduler } = state;
const buffer = subscriber.openBuffer();
const context = subscriber.openContext();
const action = <Action<CreationState<T>>>this;
if (!subscriber.isUnsubscribed) {
action.add(scheduler.schedule(dispatchBufferClose, bufferTimeSpan, { subscriber, buffer }));
subscriber.add(context.closeAction = scheduler.schedule(dispatchBufferClose, bufferTimeSpan, { subscriber, context }));
action.schedule(state, bufferCreationInterval);
}
}

function dispatchBufferClose({ subscriber, buffer }) {
subscriber.closeBuffer(buffer);
function dispatchBufferClose({ subscriber, context }) {
subscriber.closeContext(context);
}

0 comments on commit 1fb6aae

Please sign in to comment.