Throttle any Effection stream using a fixed size buffer

buffer.ts

/**
 * Takes a buffer limit, and returns a stream combinator that converts a stream into stream
 * limits the number of in flight items to that limit when it is subscribed to. Use with pipe:
 * ```ts
 * let doubleclicks = pipe(events, buffer(200), filter(isDoubleClick));
 * ```
 * Or as a standalone:
 * let buffered = buffer(200)(events);
 * 
 * No buffer is actually allocated until the resulting stream is subscribed to.
 */
export function buffer(limit: number): <T,TClose>(stream: Stream<T,TClose>) => Stream<T,TClose> {
  return (source) => function* () {
    let buffer: T[] = [];
    let { port, stream } = createChannel<T, TClose>();
    yield* spawn(function*() {
      let subscription = yield* source;
      while (true) {
        let next = yield* subscription.next();
        if (next.done) {
          yield* port.close(next.value);
          break;
        } else if (buffer.length < limit) {
          buffer.unshift(next.value);          
          yield* port.send(buffer.pop() as T);        
        } else {
          // buffer overflow. What to do?
        }                                         
      }
    });      
    return yield* stream;                                  
  });
}