Skip to content

Commit

Permalink
feat(bufferTime): add higher-order lettable version of bufferTime ope…
Browse files Browse the repository at this point in the history 
…rator
  • Loading branch information
@benlesh
benlesh committed Jul 13, 2017
1 parent 0ae2ed5 commit 0377ca6
Show file tree
Hide file tree
Showing 3 changed files with 240 additions and 160 deletions.
162 changes: 2 additions & 160 deletions src/operator/bufferTime.ts
View file
Original file line number Diff line number Diff line change
@@ -1,11 +1,8 @@
import { IScheduler } from '../Scheduler';
import { Action } from '../scheduler/Action';
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 { bufferTime as higherOrder } from '../operators';

/* tslint:disable:max-line-length */
export function bufferTime<T>(this: Observable<T>, bufferTimeSpan: number, scheduler?: IScheduler): Observable<T[]>;
Expand Down Expand Up @@ -75,160 +72,5 @@ export function bufferTime<T>(this: Observable<T>, bufferTimeSpan: number): Obse
maxBufferSize = arguments[2];
}

return this.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: IScheduler) {
}

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;
}

type CreationState<T> = {
bufferTimeSpan: number;
bufferCreationInterval: number,
subscriber: BufferTimeSubscriber<T>;
scheduler: IScheduler;
};

/**
* 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: IScheduler) {
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: CreationState<T> = { bufferTimeSpan, bufferCreationInterval, subscriber: this, scheduler };
this.add(context.closeAction = scheduler.schedule(dispatchBufferClose, bufferTimeSpan, closeState));
this.add(scheduler.schedule(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();
}

protected _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: Action<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);
}
}

interface DispatchArg<T> {
subscriber: BufferTimeSubscriber<T>;
context: Context<T>;
}

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

function dispatchBufferClose<T>(arg: DispatchArg<T>) {
const { subscriber, context } = arg;
subscriber.closeContext(context);
return higherOrder(bufferTimeSpan, bufferCreationInterval, maxBufferSize, scheduler)(this);
}
Loading

0 comments on commit 0377ca6

Please sign in to comment.