Transformation Operators part II

• map
• scan


• mergeMap
• concatMap
• switchMap
• exhaustMap


• buffer
• bufferWhen
• bufferCount
• bufferTime
• bufferToggle

Buffering

Streams are stateless and don’t store any data nevertheless, sometimes useful to be able to temporarily cache some events and throw a batch of them

• buffer(predicateObservable)
• bufferCount(number)
• bufferTime(periodNumber)
• bufferWhen(closingFunction: Observable)
• bufferToggle(openingObservable, closingFunction: Observable)

Buffer Operator

        const interval$ = rxjs.interval(2000)
        const bufferBy$ = rxjs.fromEvent(document, 'click')
        interval$.pipe(rxjs.operators.buffer(bufferBy$))
    

Collects values from the past as an array, and emits that array only when another Observable emits | Collect output values until provided observable emits, emit as array

BufferCount Operator

        const interval$ = rxjs.interval(1000)
        //After three values are emitted, pass on as an array of buffered values
        interval$.pipe(rxjs.operators.bufferCount(3))
    

Collect emitted values until provided number is fulfilled, emit as array

BufferTime Operator

        const interval$ = rxjs.interval(1000)
        //After two seconds, emit buffered values as an array
        interval$.pipe(rxjs.operators.bufferTime(2000))
    

Collects values from the past as an array, and emits those arrays periodically in time

BufferWhen Operator

        const interval$ = rxjs.interval(2000)
        const bufferBy$ = rxjs.fromEvent(document, 'click')
        interval$.pipe(rxjs.operators.bufferWhen(() => bufferBy$))
    

Collects values from the past as an array. When it starts collecting values, it calls a function that returns an Observable that tells when to close the buffer and restart collecting

BufferToggle Operator

        const sourceInterval$ = rxjs.interval(1000)

        const startInterval$ = rxjs.interval(5000)
        // val will be the startInterval$ value emitted
        const closingInterval$ = (val) => rxjs.interval(1000)
        //every 5s a new buffer will start, collecting emitted values for 1s then emitting buffered values
        sourceInterval$.pipe(rxjs.operators.bufferToggle(
            startInterval$,
            closingInterval$
        )
    

Collects values from the past as an array. Starts collecting only when opening emits, and calls the closingSelector function to get an Observable that tells when to close the buffer

Other Buffering Transformation Operators

• pairwise() => Emit the previous and current values as an array here


• window(predicateObservable) => It's like buffer, but emits a nested Observable instead of an array
• windowCount(number) => It's like bufferCount, but emits a nested Observable instead of an array
• windowToggle() => It's like bufferToggle, but emits a nested Observable instead of an array
• windowTime(periodNumber) => It's like bufferTime, but emits a nested Observable instead of an array
• windowWhen(predicateFunction) => It's like bufferWhen, but emits a nested Observable instead of an array