{"version":3,"file":"bufferTime.js","sourceRoot":"","sources":["../../src/operators/bufferTime.ts"],"names":[],"mappings":";;;;;;AAGA,sBAAsB,oBAAoB,CAAC,CAAA;AAE3C,2BAA2B,eAAe,CAAC,CAAA;AAE3C,4BAA4B,qBAAqB,CAAC,CAAA;AAOlD,mCAAmC;AAEnC;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;GA0CG;AACH,oBAA8B,cAAsB;IAClD,IAAI,MAAM,GAAW,SAAS,CAAC,MAAM,CAAC;IAEtC,IAAI,SAAS,GAAe,aAAK,CAAC;IAClC,EAAE,CAAC,CAAC,yBAAW,CAAC,SAAS,CAAC,SAAS,CAAC,MAAM,GAAG,CAAC,CAAC,CAAC,CAAC,CAAC,CAAC;QACjD,SAAS,GAAG,SAAS,CAAC,SAAS,CAAC,MAAM,GAAG,CAAC,CAAC,CAAC;QAC5C,MAAM,EAAE,CAAC;IACX,CAAC;IAED,IAAI,sBAAsB,GAAW,IAAI,CAAC;IAC1C,EAAE,CAAC,CAAC,MAAM,IAAI,CAAC,CAAC,CAAC,CAAC;QAChB,sBAAsB,GAAG,SAAS,CAAC,CAAC,CAAC,CAAC;IACxC,CAAC;IAED,IAAI,aAAa,GAAW,MAAM,CAAC,iBAAiB,CAAC;IACrD,EAAE,CAAC,CAAC,MAAM,IAAI,CAAC,CAAC,CAAC,CAAC;QAChB,aAAa,GAAG,SAAS,CAAC,CAAC,CAAC,CAAC;IAC/B,CAAC;IAED,MAAM,CAAC,oCAAoC,MAAqB;QAC9D,MAAM,CAAC,MAAM,CAAC,IAAI,CAAC,IAAI,kBAAkB,CAAI,cAAc,EAAE,sBAAsB,EAAE,aAAa,EAAE,SAAS,CAAC,CAAC,CAAC;IAClH,CAAC,CAAC;AACJ,CAAC;AAtBe,kBAAU,aAsBzB,CAAA;AAED;IACE,4BAAoB,cAAsB,EACtB,sBAA8B,EAC9B,aAAqB,EACrB,SAAqB;QAHrB,mBAAc,GAAd,cAAc,CAAQ;QACtB,2BAAsB,GAAtB,sBAAsB,CAAQ;QAC9B,kBAAa,GAAb,aAAa,CAAQ;QACrB,cAAS,GAAT,SAAS,CAAY;IACzC,CAAC;IAED,iCAAI,GAAJ,UAAK,UAA2B,EAAE,MAAW;QAC3C,MAAM,CAAC,MAAM,CAAC,SAAS,CAAC,IAAI,oBAAoB,CAC9C,UAAU,EAAE,IAAI,CAAC,cAAc,EAAE,IAAI,CAAC,sBAAsB,EAAE,IAAI,CAAC,aAAa,EAAE,IAAI,CAAC,SAAS,CACjG,CAAC,CAAC;IACL,CAAC;IACH,yBAAC;AAAD,CAAC,AAZD,IAYC;AAED;IAAA;QACE,WAAM,GAAQ,EAAE,CAAC;IAEnB,CAAC;IAAD,cAAC;AAAD,CAAC,AAHD,IAGC;AASD;;;;GAIG;AACH;IAAsC,wCAAa;IAIjD,8BAAY,WAA4B,EACpB,cAAsB,EACtB,sBAA8B,EAC9B,aAAqB,EACrB,SAAqB;QACvC,kBAAM,WAAW,CAAC,CAAC;QAJD,mBAAc,GAAd,cAAc,CAAQ;QACtB,2BAAsB,GAAtB,sBAAsB,CAAQ;QAC9B,kBAAa,GAAb,aAAa,CAAQ;QACrB,cAAS,GAAT,SAAS,CAAY;QAPjC,aAAQ,GAAsB,EAAE,CAAC;QASvC,IAAM,OAAO,GAAG,IAAI,CAAC,WAAW,EAAE,CAAC;QACnC,IAAI,CAAC,YAAY,GAAG,sBAAsB,IAAI,IAAI,IAAI,sBAAsB,GAAG,CAAC,CAAC;QACjF,EAAE,CAAC,CAAC,IAAI,CAAC,YAAY,CAAC,CAAC,CAAC;YACtB,IAAM,iBAAiB,GAAG,EAAE,UAAU,EAAE,IAAI,EAAE,gBAAO,EAAE,8BAAc,EAAE,CAAC;YACxE,IAAI,CAAC,GAAG,CAAC,OAAO,CAAC,WAAW,GAAG,SAAS,CAAC,QAAQ,CAAC,0BAA0B,EAAE,cAAc,EAAE,iBAAiB,CAAC,CAAC,CAAC;QACpH,CAAC;QAAC,IAAI,CAAC,CAAC;YACN,IAAM,UAAU,GAAG,EAAE,UAAU,EAAE,IAAI,EAAE,gBAAO,EAAE,CAAC;YACjD,IAAM,aAAa,GAAqB,EAAE,8BAAc,EAAE,8CAAsB,EAAE,UAAU,EAAE,IAAI,EAAE,oBAAS,EAAE,CAAC;YAChH,IAAI,CAAC,GAAG,CAAC,OAAO,CAAC,WAAW,GAAG,SAAS,CAAC,QAAQ,CAAC,mBAAmB,EAAE,cAAc,EAAE,UAAU,CAAC,CAAC,CAAC;YACpG,IAAI,CAAC,GAAG,CAAC,SAAS,CAAC,QAAQ,CAAC,sBAAsB,EAAE,sBAAsB,EAAE,aAAa,CAAC,CAAC,CAAC;QAC9F,CAAC;IACH,CAAC;IAES,oCAAK,GAAf,UAAgB,KAAQ;QACtB,IAAM,QAAQ,GAAG,IAAI,CAAC,QAAQ,CAAC;QAC/B,IAAM,GAAG,GAAG,QAAQ,CAAC,MAAM,CAAC;QAC5B,IAAI,mBAA+B,CAAC;QACpC,GAAG,CAAC,CAAC,IAAI,CAAC,GAAG,CAAC,EAAE,CAAC,GAAG,GAAG,EAAE,CAAC,EAAE,EAAE,CAAC;YAC7B,IAAM,OAAO,GAAG,QAAQ,CAAC,CAAC,CAAC,CAAC;YAC5B,IAAM,MAAM,GAAG,OAAO,CAAC,MAAM,CAAC;YAC9B,MAAM,CAAC,IAAI,CAAC,KAAK,CAAC,CAAC;YACnB,EAAE,CAAC,CAAC,MAAM,CAAC,MAAM,IAAI,IAAI,CAAC,aAAa,CAAC,CAAC,CAAC;gBACxC,mBAAmB,GAAG,OAAO,CAAC;YAChC,CAAC;QACH,CAAC;QAED,EAAE,CAAC,CAAC,mBAAmB,CAAC,CAAC,CAAC;YACxB,IAAI,CAAC,YAAY,CAAC,mBAAmB,CAAC,CAAC;QACzC,CAAC;IACH,CAAC;IAES,qCAAM,GAAhB,UAAiB,GAAQ;QACvB,IAAI,CAAC,QAAQ,CAAC,MAAM,GAAG,CAAC,CAAC;QACzB,gBAAK,CAAC,MAAM,YAAC,GAAG,CAAC,CAAC;IACpB,CAAC;IAES,wCAAS,GAAnB;QACE,IAAA,SAAsC,EAA9B,sBAAQ,EAAE,4BAAW,CAAU;QACvC,OAAO,QAAQ,CAAC,MAAM,GAAG,CAAC,EAAE,CAAC;YAC3B,IAAM,OAAO,GAAG,QAAQ,CAAC,KAAK,EAAE,CAAC;YACjC,WAAW,CAAC,IAAI,CAAC,OAAO,CAAC,MAAM,CAAC,CAAC;QACnC,CAAC;QACD,gBAAK,CAAC,SAAS,WAAE,CAAC;IACpB,CAAC;IAED,oCAAoC,CAAC,2CAAY,GAAZ;QACnC,IAAI,CAAC,QAAQ,GAAG,IAAI,CAAC;IACvB,CAAC;IAES,2CAAY,GAAtB,UAAuB,OAAmB;QACxC,IAAI,CAAC,YAAY,CAAC,OAAO,CAAC,CAAC;QAC3B,IAAM,WAAW,GAAG,OAAO,CAAC,WAAW,CAAC;QACxC,WAAW,CAAC,WAAW,EAAE,CAAC;QAC1B,IAAI,CAAC,MAAM,CAAC,WAAW,CAAC,CAAC;QAEzB,EAAE,CAAC,CAAC,CAAC,IAAI,CAAC,MAAM,IAAI,IAAI,CAAC,YAAY,CAAC,CAAC,CAAC;YACtC,OAAO,GAAG,IAAI,CAAC,WAAW,EAAE,CAAC;YAC7B,IAAM,cAAc,GAAG,IAAI,CAAC,cAAc,CAAC;YAC3C,IAAM,iBAAiB,GAAG,EAAE,UAAU,EAAE,IAAI,EAAE,gBAAO,EAAE,8BAAc,EAAE,CAAC;YACxE,IAAI,CAAC,GAAG,CAAC,OAAO,CAAC,WAAW,GAAG,IAAI,CAAC,SAAS,CAAC,QAAQ,CAAC,0BAA0B,EAAE,cAAc,EAAE,iBAAiB,CAAC,CAAC,CAAC;QACzH,CAAC;IACH,CAAC;IAED,0CAAW,GAAX;QACE,IAAM,OAAO,GAAe,IAAI,OAAO,EAAK,CAAC;QAC7C,IAAI,CAAC,QAAQ,CAAC,IAAI,CAAC,OAAO,CAAC,CAAC;QAC5B,MAAM,CAAC,OAAO,CAAC;IACjB,CAAC;IAED,2CAAY,GAAZ,UAAa,OAAmB;QAC9B,IAAI,CAAC,WAAW,CAAC,IAAI,CAAC,OAAO,CAAC,MAAM,CAAC,CAAC;QACtC,IAAM,QAAQ,GAAG,IAAI,CAAC,QAAQ,CAAC;QAE/B,IAAM,WAAW,GAAG,QAAQ,GAAG,QAAQ,CAAC,OAAO,CAAC,OAAO,CAAC,GAAG,CAAC,CAAC,CAAC;QAC9D,EAAE,CAAC,CAAC,WAAW,IAAI,CAAC,CAAC,CAAC,CAAC;YACrB,QAAQ,CAAC,MAAM,CAAC,QAAQ,CAAC,OAAO,CAAC,OAAO,CAAC,EAAE,CAAC,CAAC,CAAC;QAChD,CAAC;IACH,CAAC;IACH,2BAAC;AAAD,CAAC,AAxFD,CAAsC,uBAAU,GAwF/C;AAED,oCAAuD,KAAU;IAC/D,IAAM,UAAU,GAA8B,KAAK,CAAC,UAAU,CAAC;IAE/D,IAAM,WAAW,GAAG,KAAK,CAAC,OAAO,CAAC;IAClC,EAAE,CAAC,CAAC,WAAW,CAAC,CAAC,CAAC;QAChB,UAAU,CAAC,YAAY,CAAC,WAAW,CAAC,CAAC;IACvC,CAAC;IAED,EAAE,CAAC,CAAC,CAAC,UAAU,CAAC,MAAM,CAAC,CAAC,CAAC;QACvB,KAAK,CAAC,OAAO,GAAG,UAAU,CAAC,WAAW,EAAE,CAAC;QACzC,KAAK,CAAC,OAAO,CAAC,WAAW,GAAG,IAAI,CAAC,QAAQ,CAAC,KAAK,EAAE,KAAK,CAAC,cAAc,CAAC,CAAC;IACzE,CAAC;AACH,CAAC;AAOD,gCAAmE,KAAuB;IAChF,yDAAsB,EAAE,qCAAc,EAAE,6BAAU,EAAE,2BAAS,CAAW;IAChF,IAAM,OAAO,GAAG,UAAU,CAAC,WAAW,EAAE,CAAC;IACzC,IAAM,MAAM,GAA6B,IAAI,CAAC;IAC9C,EAAE,CAAC,CAAC,CAAC,UAAU,CAAC,MAAM,CAAC,CAAC,CAAC;QACvB,UAAU,CAAC,GAAG,CAAC,OAAO,CAAC,WAAW,GAAG,SAAS,CAAC,QAAQ,CAAiB,mBAAmB,EAAE,cAAc,EAAE,EAAE,sBAAU,EAAE,gBAAO,EAAE,CAAC,CAAC,CAAC;QACvI,MAAM,CAAC,QAAQ,CAAC,KAAK,EAAE,sBAAsB,CAAC,CAAC;IACjD,CAAC;AACH,CAAC;AAED,6BAAgC,GAAmB;IACzC,+BAAU,EAAE,qBAAO,CAAS;IACpC,UAAU,CAAC,YAAY,CAAC,OAAO,CAAC,CAAC;AACnC,CAAC","sourcesContent":["import { IScheduler } from '../Scheduler';\nimport { Action } from '../scheduler/Action';\nimport { Operator } from '../Operator';\nimport { async } from '../scheduler/async';\nimport { Observable } from '../Observable';\nimport { Subscriber } from '../Subscriber';\nimport { Subscription } from '../Subscription';\nimport { isScheduler } from '../util/isScheduler';\nimport { OperatorFunction } from '../interfaces';\n\n/* tslint:disable:max-line-length */\nexport function bufferTime<T>(bufferTimeSpan: number, scheduler?: IScheduler): OperatorFunction<T, T[]>;\nexport function bufferTime<T>(bufferTimeSpan: number, bufferCreationInterval: number, scheduler?: IScheduler): OperatorFunction<T, T[]>;\nexport function bufferTime<T>(bufferTimeSpan: number, bufferCreationInterval: number, maxBufferSize: number, scheduler?: IScheduler): OperatorFunction<T, T[]>;\n/* tslint:enable:max-line-length */\n\n/**\n * Buffers the source Observable values for a specific time period.\n *\n * <span class=\"informal\">Collects values from the past as an array, and emits\n * those arrays periodically in time.</span>\n *\n * <img src=\"./img/bufferTime.png\" width=\"100%\">\n *\n * Buffers values from the source for a specific time duration `bufferTimeSpan`.\n * Unless the optional argument `bufferCreationInterval` is given, it emits and\n * resets the buffer every `bufferTimeSpan` milliseconds. If\n * `bufferCreationInterval` is given, this operator opens the buffer every\n * `bufferCreationInterval` milliseconds and closes (emits and resets) the\n * buffer every `bufferTimeSpan` milliseconds. When the optional argument\n * `maxBufferSize` is specified, the buffer will be closed either after\n * `bufferTimeSpan` milliseconds or when it contains `maxBufferSize` elements.\n *\n * @example <caption>Every second, emit an array of the recent click events</caption>\n * var clicks = Rx.Observable.fromEvent(document, 'click');\n * var buffered = clicks.bufferTime(1000);\n * buffered.subscribe(x => console.log(x));\n *\n * @example <caption>Every 5 seconds, emit the click events from the next 2 seconds</caption>\n * var clicks = Rx.Observable.fromEvent(document, 'click');\n * var buffered = clicks.bufferTime(2000, 5000);\n * buffered.subscribe(x => console.log(x));\n *\n * @see {@link buffer}\n * @see {@link bufferCount}\n * @see {@link bufferToggle}\n * @see {@link bufferWhen}\n * @see {@link windowTime}\n *\n * @param {number} bufferTimeSpan The amount of time to fill each buffer array.\n * @param {number} [bufferCreationInterval] The interval at which to start new\n * buffers.\n * @param {number} [maxBufferSize] The maximum buffer size.\n * @param {Scheduler} [scheduler=async] The scheduler on which to schedule the\n * intervals that determine buffer boundaries.\n * @return {Observable<T[]>} An observable of arrays of buffered values.\n * @method bufferTime\n * @owner Observable\n */\nexport function bufferTime<T>(bufferTimeSpan: number): OperatorFunction<T, T[]> {\n  let length: number = arguments.length;\n\n  let scheduler: IScheduler = async;\n  if (isScheduler(arguments[arguments.length - 1])) {\n    scheduler = arguments[arguments.length - 1];\n    length--;\n  }\n\n  let bufferCreationInterval: number = null;\n  if (length >= 2) {\n    bufferCreationInterval = arguments[1];\n  }\n\n  let maxBufferSize: number = Number.POSITIVE_INFINITY;\n  if (length >= 3) {\n    maxBufferSize = arguments[2];\n  }\n\n  return function bufferTimeOperatorFunction(source: Observable<T>) {\n    return source.lift(new BufferTimeOperator<T>(bufferTimeSpan, bufferCreationInterval, maxBufferSize, scheduler));\n  };\n}\n\nclass BufferTimeOperator<T> implements Operator<T, T[]> {\n  constructor(private bufferTimeSpan: number,\n              private bufferCreationInterval: number,\n              private maxBufferSize: number,\n              private scheduler: IScheduler) {\n  }\n\n  call(subscriber: Subscriber<T[]>, source: any): any {\n    return source.subscribe(new BufferTimeSubscriber(\n      subscriber, this.bufferTimeSpan, this.bufferCreationInterval, this.maxBufferSize, this.scheduler\n    ));\n  }\n}\n\nclass Context<T> {\n  buffer: T[] = [];\n  closeAction: Subscription;\n}\n\ntype CreationState<T> = {\n  bufferTimeSpan: number;\n  bufferCreationInterval: number,\n  subscriber: BufferTimeSubscriber<T>;\n  scheduler: IScheduler;\n};\n\n/**\n * We need this JSDoc comment for affecting ESDoc.\n * @ignore\n * @extends {Ignored}\n */\nclass BufferTimeSubscriber<T> extends Subscriber<T> {\n  private contexts: Array<Context<T>> = [];\n  private timespanOnly: boolean;\n\n  constructor(destination: Subscriber<T[]>,\n              private bufferTimeSpan: number,\n              private bufferCreationInterval: number,\n              private maxBufferSize: number,\n              private scheduler: IScheduler) {\n    super(destination);\n    const context = this.openContext();\n    this.timespanOnly = bufferCreationInterval == null || bufferCreationInterval < 0;\n    if (this.timespanOnly) {\n      const timeSpanOnlyState = { subscriber: this, context, bufferTimeSpan };\n      this.add(context.closeAction = scheduler.schedule(dispatchBufferTimeSpanOnly, bufferTimeSpan, timeSpanOnlyState));\n    } else {\n      const closeState = { subscriber: this, context };\n      const creationState: CreationState<T> = { bufferTimeSpan, bufferCreationInterval, subscriber: this, scheduler };\n      this.add(context.closeAction = scheduler.schedule(dispatchBufferClose, bufferTimeSpan, closeState));\n      this.add(scheduler.schedule(dispatchBufferCreation, bufferCreationInterval, creationState));\n    }\n  }\n\n  protected _next(value: T) {\n    const contexts = this.contexts;\n    const len = contexts.length;\n    let filledBufferContext: Context<T>;\n    for (let i = 0; i < len; i++) {\n      const context = contexts[i];\n      const buffer = context.buffer;\n      buffer.push(value);\n      if (buffer.length == this.maxBufferSize) {\n        filledBufferContext = context;\n      }\n    }\n\n    if (filledBufferContext) {\n      this.onBufferFull(filledBufferContext);\n    }\n  }\n\n  protected _error(err: any) {\n    this.contexts.length = 0;\n    super._error(err);\n  }\n\n  protected _complete() {\n    const { contexts, destination } = this;\n    while (contexts.length > 0) {\n      const context = contexts.shift();\n      destination.next(context.buffer);\n    }\n    super._complete();\n  }\n\n  /** @deprecated internal use only */ _unsubscribe() {\n    this.contexts = null;\n  }\n\n  protected onBufferFull(context: Context<T>) {\n    this.closeContext(context);\n    const closeAction = context.closeAction;\n    closeAction.unsubscribe();\n    this.remove(closeAction);\n\n    if (!this.closed && this.timespanOnly) {\n      context = this.openContext();\n      const bufferTimeSpan = this.bufferTimeSpan;\n      const timeSpanOnlyState = { subscriber: this, context, bufferTimeSpan };\n      this.add(context.closeAction = this.scheduler.schedule(dispatchBufferTimeSpanOnly, bufferTimeSpan, timeSpanOnlyState));\n    }\n  }\n\n  openContext(): Context<T> {\n    const context: Context<T> = new Context<T>();\n    this.contexts.push(context);\n    return context;\n  }\n\n  closeContext(context: Context<T>) {\n    this.destination.next(context.buffer);\n    const contexts = this.contexts;\n\n    const spliceIndex = contexts ? contexts.indexOf(context) : -1;\n    if (spliceIndex >= 0) {\n      contexts.splice(contexts.indexOf(context), 1);\n    }\n  }\n}\n\nfunction dispatchBufferTimeSpanOnly(this: Action<any>, state: any) {\n  const subscriber: BufferTimeSubscriber<any> = state.subscriber;\n\n  const prevContext = state.context;\n  if (prevContext) {\n    subscriber.closeContext(prevContext);\n  }\n\n  if (!subscriber.closed) {\n    state.context = subscriber.openContext();\n    state.context.closeAction = this.schedule(state, state.bufferTimeSpan);\n  }\n}\n\ninterface DispatchArg<T> {\n  subscriber: BufferTimeSubscriber<T>;\n  context: Context<T>;\n}\n\nfunction dispatchBufferCreation<T>(this: Action<CreationState<T>>, state: CreationState<T>) {\n  const { bufferCreationInterval, bufferTimeSpan, subscriber, scheduler } = state;\n  const context = subscriber.openContext();\n  const action = <Action<CreationState<T>>>this;\n  if (!subscriber.closed) {\n    subscriber.add(context.closeAction = scheduler.schedule<DispatchArg<T>>(dispatchBufferClose, bufferTimeSpan, { subscriber, context }));\n    action.schedule(state, bufferCreationInterval);\n  }\n}\n\nfunction dispatchBufferClose<T>(arg: DispatchArg<T>) {\n  const { subscriber, context } = arg;\n  subscriber.closeContext(context);\n}\n"]}