LevelUP Read Stream Transform Memory Stress Test

levelup-stream-memory-test.js

'use strict';

var Transform = require('stream').Transform;
var inherits = require('util').inherits;
var crypto = require('crypto');

var levelup = require('levelup');
var async = require('async');

var maxNumber = 1500000;
var maxNumberBuffer = new Buffer(new Array(4));
maxNumberBuffer.writeUInt32BE(maxNumber);
var minNumber = 0;
var minNumberBuffer = new Buffer(new Array(4));
minNumberBuffer.writeUInt32BE(minNumber);

function StreamTransform(options) {
  Transform.call(this, {
    objectMode: true
  });
  this._count = 0;
}

inherits(StreamTransform, Transform);

StreamTransform.prototype._transform = function(chunk, encoding, callback) {
  var self = this;
  this._count += 1;
  var obj = {
    count: this._count
  };
  self.push(obj);
  callback();
};

function doStream() {

  var db = levelup('./testdb', {
    fillCache: false,
    cacheSize: 1024,
    maxOpenFiles: 200,
    blockSize: 4096 * 10
  });

  // async.timesLimit(maxNumber, 100, function(n, next) {
  //   var key = new Buffer(new Array(4));
  //   key.writeUInt32BE(n);
  //   var value = crypto.randomBytes(4058);
  //   console.log(n);
  //   db.put(key, value, next);
  // }, function(err, results) {
  //   if (err) {
  //     throw err;
  //   }
  //   console.log('Prep data done');

  // });

  var streamTransform = new StreamTransform();

  var stream = db.createReadStream({
    gte: minNumberBuffer,
    lte: maxNumberBuffer,
    valueEncoding: 'binary',
    keyEncoding: 'binary',
    fillCache: false
  }).pipe(streamTransform);

  var count = 0;
  stream.on('data', function(data) {
    //console.log(data);
  });

  stream.on('error', function(err) {
    console.error(err);
    console.error(err.stack);
  });

  stream.on('end', function() {
    console.log('End!');
    db.close();
    setTimeout(function() {
      doStream();
    }, 2000);
  });

}

doStream();