diff options
Diffstat (limited to 'node_modules/rxjs/src/internal/operators/buffer.ts')
| -rw-r--r-- | node_modules/rxjs/src/internal/operators/buffer.ts | 81 |
1 files changed, 0 insertions, 81 deletions
diff --git a/node_modules/rxjs/src/internal/operators/buffer.ts b/node_modules/rxjs/src/internal/operators/buffer.ts deleted file mode 100644 index 2ca2fde..0000000 --- a/node_modules/rxjs/src/internal/operators/buffer.ts +++ /dev/null @@ -1,81 +0,0 @@ -import { OperatorFunction, ObservableInput } from '../types'; -import { operate } from '../util/lift'; -import { noop } from '../util/noop'; -import { createOperatorSubscriber } from './OperatorSubscriber'; -import { innerFrom } from '../observable/innerFrom'; - -/** - * Buffers the source Observable values until `closingNotifier` emits. - * - * <span class="informal">Collects values from the past as an array, and emits - * that array only when another Observable emits.</span> - * - *  - * - * Buffers the incoming Observable values until the given `closingNotifier` - * `ObservableInput` (that internally gets converted to an Observable) - * emits a value, at which point it emits the buffer on the output - * Observable and starts a new buffer internally, awaiting the next time - * `closingNotifier` emits. - * - * ## Example - * - * On every click, emit array of most recent interval events - * - * ```ts - * import { fromEvent, interval, buffer } from 'rxjs'; - * - * const clicks = fromEvent(document, 'click'); - * const intervalEvents = interval(1000); - * const buffered = intervalEvents.pipe(buffer(clicks)); - * buffered.subscribe(x => console.log(x)); - * ``` - * - * @see {@link bufferCount} - * @see {@link bufferTime} - * @see {@link bufferToggle} - * @see {@link bufferWhen} - * @see {@link window} - * - * @param closingNotifier An `ObservableInput` that signals the - * buffer to be emitted on the output Observable. - * @return A function that returns an Observable of buffers, which are arrays - * of values. - */ -export function buffer<T>(closingNotifier: ObservableInput<any>): OperatorFunction<T, T[]> { - return operate((source, subscriber) => { - // The current buffered values. - let currentBuffer: T[] = []; - - // Subscribe to our source. - source.subscribe( - createOperatorSubscriber( - subscriber, - (value) => currentBuffer.push(value), - () => { - subscriber.next(currentBuffer); - subscriber.complete(); - } - ) - ); - - // Subscribe to the closing notifier. - innerFrom(closingNotifier).subscribe( - createOperatorSubscriber( - subscriber, - () => { - // Start a new buffer and emit the previous one. - const b = currentBuffer; - currentBuffer = []; - subscriber.next(b); - }, - noop - ) - ); - - return () => { - // Ensure buffered values are released on finalization. - currentBuffer = null!; - }; - }); -} |
