periodicallyEmittingBuffer

index.js

const { rxObserver } = require('api/v0.3');
const { zip, timer, from, Observable } = require('rxjs');
const { take } = require('rxjs/operators');

const TIMEOUT = 1;
const TAKE = 6;

const alphabet$ = from('abcdefghijklmnopqrstuvwxyz');

alphabet$.pipe(
  take(TAKE)
)
  .subscribe(rxObserver());

alphabet$.pipe(
  periodicallyEmittingBuffer(TIMEOUT),
  take(TAKE)
)
  .subscribe(rxObserver());

function periodicallyEmittingBuffer(timeout) {
  return (upstream) =>  {
    const buffer = [];
    return new Observable(s => {
      // Buffer incoming items
      upstream.subscribe({
        next: (item) => buffer.push(item)
      });
      const maybeEmit = () => {
        if (buffer.length) {
          s.next(buffer.shift());
        }        
      }
      // Emit right away if something already buffered
      maybeEmit();
      const interval = setInterval(maybeEmit, timeout);
      return () => clearInterval(interval);
    })
  }
}