<commit_before>// take :: Int -> { readable, writable }
// take function takes an int n and
// returns a transform stream
// that takes the first n values
// from the input stream.
//

import { ReadableStream, WritableStream } from "@pipes/core/streams";

export default function take( count ) {
  let
    readable, writable,

    // Cleanup fn for writable
    close,
    closeFn = controller => () =>
      !controller._closeRequested && controller.close(),

    // Function that writes chunks to readable
    write,
    writeFn = controller =>
      chunk => {
        if ( count-- > 0 ) {
          return controller.enqueue( chunk );
        }

        close();
        writable._writer.close();
      },

    // Error handler for writable
    error,
    errorFn = controller => controller.error.bind( controller );

  // Readable
  readable = new ReadableStream({
    start( controller ) {
      // Init pass and err function
      write = writeFn( controller );
      error = errorFn( controller );
      close = closeFn( controller );
    }
  });

  // Writable
  writable = new WritableStream({
    write,
    close,
    error
  });

  return {
    readable,
    writable
  };
}

// Browserify compat
if ( typeof module !== "undefined" )
  module.exports = take;
<commit_msg>Added flow types to take<commit_after>// @flow

// take :: Int -> { readable, writable }
// take function takes an int n and
// returns a transform stream
// that takes the first n values
// from the input stream.
//

import type {
  ReadableWritable,
  ReadableStreamController

} from "@pipes/core/streams";

import { ReadableStream, WritableStream } from "@pipes/core/streams";

export default function take( count: number ): ReadableWritable {
  let
    readable: ReadableStream,
    writable: WritableStream,

    // Cleanup fn for writable
    close: ()=>void,
    closeFn = (controller: ReadableStreamController) =>
      () => { !controller._closeRequested && controller.close() },

    // Function that writes chunks to readable
    write: (mixed)=>?mixed,
    writeFn = (controller: ReadableStreamController) =>
      chunk => {
        if ( count-- > 0 ) {
          return controller.enqueue( chunk );
        }

        close();
        writable._writer.close();
      },

    // Error handler for writable
    error: ()=>void,
    errorFn = (controller: ReadableStreamController) =>
      controller.error.bind( controller );

  // Readable
  readable = new ReadableStream({
    start( controller: ReadableStreamController ) {
      // Init pass and err function
      write = writeFn( controller );
      error = errorFn( controller );
      close = closeFn( controller );
    }
  });

  // Writable
  writable = new WritableStream({
    write,
    close,
    error
  });

  return {
    readable,
    writable
  };
}

// FIXME: Internal flow.js resolution problem workaround
export const _take = take;

// Browserify compat
if ( typeof module !== "undefined" )
  // $FlowFixMe
  module.exports = take;

{'fraction_non_alphanumeric': 0.11692205196535643, 'fraction_numerical': 0.0006662225183211193, 'mean_word_length': 2.4959254947613503, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 20, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '855136', 'n_tokens_mistral': 864, 'n_tokens_neox': 835, 'n_words': 259}