includes/writer.js
function processBuffer(flush, callback) {
    function taskCallback(err, payload) {
      if (err && !didError) {
        debug(`Queue task failed with error ${err.name}`);
        didError = true;
        q.kill();
        writer.emit('error', err);
      }
    }

    if (flush || buffer.length >= bufferSize) {
      // work through the buffer to break off bufferSize chunks
      // and feed the chunks to the queue
      do {
        // split the buffer into bufferSize chunks
        var toSend = buffer.splice(0, bufferSize);

        // and add the chunk to the queue
        debug(`Adding ${toSend.length} to the write queue.`);
        q.push({ docs: toSend }, taskCallback);
      } while (buffer.length >= bufferSize);

      // send any leftover documents to the queue
      if (flush && buffer.length > 0) {
        debug(`Adding remaining ${buffer.length} to the write queue.`);
        q.push({ docs: buffer }, taskCallback);
      }

      // wait until the queue size falls to a reasonable level
      async.until(
        // wait until the queue length drops to twice the paralellism
        // or until empty on the last write
        function() {
          // if we encountered an error, stop this until loop
          if (didError) {
            return true;
          }
          if (flush) {
            return q.idle() && q.length() === 0;
          } else {
            return q.length() <= parallelism * 2;
          }
        },

        function(cb) {
          setTimeout(cb, 20);
        },

        function() {
          if (flush && !didError) {
            writer.emit('finished', { total: written });
          }
          // callback when we're happy with the queue size
          callback();
        });
    } else {
      callback();
    }
  }
function processBuffer(flush, callback) {
    function taskCallback(err, payload) {
      if (err && !didError) {
        debug(`Queue task failed with error ${err.name}`);
        didError = true;
        q.kill();
        writer.emit('error', err);
      }
    }

    if (flush || buffer.length >= bufferSize) {
      // work through the buffer to break off bufferSize chunks
      // and feed the chunks to the queue
      do {
        // split the buffer into bufferSize chunks
        var toSend = buffer.splice(0, bufferSize);

        // and add the chunk to the queue
        debug(`Adding ${toSend.length} to the write queue.`);
        q.push({ docs: toSend }, taskCallback);
      } while (buffer.length >= bufferSize);

      // send any leftover documents to the queue
      if (flush && buffer.length > 0) {
        debug(`Adding remaining ${buffer.length} to the write queue.`);
        q.push({ docs: buffer }, taskCallback);
      }

      // wait until the queue size falls to a reasonable level
      async.until(
        // wait until the queue length drops to twice the paralellism
        // or until empty on the last write
        function() {
          // if we encountered an error, stop this until loop
          if (didError) {
            return true;
          }
          if (flush) {
            return q.idle() && q.length() === 0;
          } else {
            return q.length() <= parallelism * 2;
          }
        },

        function(cb) {
          setTimeout(cb, 20);
        },

        function() {
          if (flush && !didError) {
            writer.emit('finished', { total: written });
          }
          // callback when we're happy with the queue size
          callback();
        });
    } else {
      callback();
    }
  }
function processBuffer ( flush , callback ) { function taskCallback ( err , payload ) { if ( err && ! didError ) { debug ( ` ${ err . name } ` ) ; didError = true ; q . kill ( ) ; writer . emit ( 'error' , err ) ; } } if ( flush || buffer . length >= bufferSize ) { // work through the buffer to break off bufferSize chunks // and feed the chunks to the queue do { // split the buffer into bufferSize chunks var toSend = buffer . splice ( 0 , bufferSize ) ; // and add the chunk to the queue debug ( ` ${ toSend . length } ` ) ; q . push ( { docs : toSend } , taskCallback ) ; } while ( buffer . length >= bufferSize ) ; // send any leftover documents to the queue if ( flush && buffer . length > 0 ) { debug ( ` ${ buffer . length } ` ) ; q . push ( { docs : buffer } , taskCallback ) ; } // wait until the queue size falls to a reasonable level async . until ( // wait until the queue length drops to twice the paralellism // or until empty on the last write function ( ) { // if we encountered an error, stop this until loop if ( didError ) { return true ; } if ( flush ) { return q . idle ( ) && q . length ( ) === 0 ; } else { return q . length ( ) <= parallelism * 2 ; } } , function ( cb ) { setTimeout ( cb , 20 ) ; } , function ( ) { if ( flush && ! didError ) { writer . emit ( 'finished' , { total : written } ) ; } // callback when we're happy with the queue size callback ( ) ; } ) ; } else { callback ( ) ; } }
write the contents of the buffer to CouchDB in blocks of bufferSize
write the contents of the buffer to CouchDB in blocks of bufferSize