title: Using csv-parse with highlandjs
body: <p>I would like to do a bit of parsing on csv files to convert them to JSON and extract data out of them. I'm using highland as a stream processing library. I am creating an array of csv parsing streams using</p>
  
  <pre><code>import { readdir as readdirCb, createReadStream } from 'fs';
  import { promisify } from 'util';
  import _ from 'highland';
  import parse from 'csv-parse';
  
  const readdir = promisify(readdirCb);
  
  const LOGS_DIR = './logs';
  const options = '-maxdepth 1'; 
  
  async function main() {
    const files = await readdir(LOGS_DIR)
     const stream = _(files)
          .map(filename =&gt; createReadStream(`${LOGS_DIR}/${filename}`))
              .map(parse)
  
  }
  
  main();
  </code></pre>
  
  <p>I have tried to use <code>stream</code> like:</p>
  
  <pre><code> const stream = _(files)
          .map(filename =&gt; createReadStream(`${LOGS_DIR}/${filename}`))
          .map(parse)
          .each(stream =&gt; {
               stream.on('parseable', () =&gt; {
                   let record
                   while (record = stream.read()) { console.log(record) }
               })
          })
  </code></pre>
  
  <p>This does not produce any records. I am not sure as how to proceed and receive the JSON for each row for each CSV file.</p>
  
  <p>EDIT:</p>
  
  <p>Writing a function like this works for an individual file:</p>
  
  <pre><code>import parse from 'csv-parse';
  import transform from 'stream-transform';
  import { createReadStream } from 'fs';
  
  
  export default function retrieveApplicationIds(filename) {
    console.log('Parsing file', filename);
    return createReadStream(filename).pipe(parser).pipe(getApplicationId).pipe(recordUniqueId);
  }
  </code></pre>
  
  <p>Edit 2:</p>
  
  <p>I have tried using the concat streams approach:</p>
  
  <pre><code>const LOGS_DIR = './logs';
  
  function concatStreams(streamArray, streamCounter = streamArray.length) { 
    streamArray.reduce((mergedStream, stream) =&gt; {
      // pipe each stream of the array into the merged stream
     // prevent the automated 'end' event from firing
      mergedStream = stream.pipe(mergedStream, { end: false });
      // rewrite the 'end' event handler
      // Every time one of the stream ends, the counter is decremented.
      // Once the counter reaches 0, the mergedstream can emit its 'end' event.
      stream.once('end', () =&gt; --streamCounter === 0 &amp;&amp; mergedStream.emit('end'));
      return mergedStream;
    }, new PassThrough());
  }
  
  async function main() {
    const files = await readdir(LOGS_DIR)
    const streams = files.map(parseFile);
    const combinedStream = concatStreams(streams);
    combinedStream.pipe(process.stdout); 
  }
  
  main();
  </code></pre>
  
  <p>When I use this, I get the error: </p>
  
  <blockquote>
    <p>(node:1050) MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 11 unpipe listeners added to [Transformer]. Use emitter.setMaxListeners() to increase limit</p>
  </blockquote>