node readable stream memory issue

README.md

Run server with

$ node server.js

Run client with

$ node client.js 30000

Increase parameter and run multiple times.

When using .on data handler watch memory slowly getting bigger and bigger without reducing. When using .pipe method watch no memory issue arise.

Is this misuse of streams, misunderstanding of streams or memory leak ?

Tested on:

$ node --version
v0.12.3
$ iojs --version
v2.0.2

client.js

var net = require('net');

var size = process.argv[2] || 100000000;

var client = net.connect(9000);

client.setNoDelay(true);

var count = 0;

client.on('data', function(data) {
    console.log('data', data.length);

    count += data.length;

    if(count >= size) {
        console.log('count', count);
        client.end();
    }
});

client.on('end', function() {
    console.log('client ended');

    client.end();
});

client.write('R:' + size);

server.js

var ZeroStream = require('./zero-stream.js');

var net = require('net');

net.createServer(function(conn) {

    console.log('connected conn');

    conn.zeroStream = null;

    conn.on('data', function(data) {
        data = data.toString();

        var command = data[0];

        if(command = 'R') {
            var size = data.split(':')[1];
            console.log('generating stream of size', size);
            conn.zeroStream = new ZeroStream(size);

            conn.zeroStream.on('error', console.log.bind(console, 'zero stream error'));

// WHEN USING .pipe THERE IS NO MEMORY ISSUE
            conn.zeroStream.pipe(conn, { end: false });

// WHEN USING .on data HANDLER THERE IS MEMORY ISSUE
            // conn.zeroStream.on('data', function(data) {
            //     // data = data.toString();
            //     // process.stdout.write('.');
            //     conn.write(data);
            // });

            conn.zeroStream.on('end', function() {
                console.log('end');
                conn.write('ended', function() {
                    console.log('write end');
                    conn.end();
                });
            });
        }
    })



}).listen(9000, function() {
    console.log('listening...');
});

zero-stream.js

var stream = require('stream');
var ReadableStream = stream.Readable;

var util = require('util');

function ZeroBuf(size) {
    var buf = new Buffer(size);
    buf.fill(0);
    return buf;
}

function ZeroStream(maxSize) {
    if(!(this instanceof ZeroStream)) {
        return new ZeroStream(maxSize);
    }

    ReadableStream.call(this);

    this.pushed = 0;
    this.maxSize = maxSize;

    this._read = function(size) {
        if(this.pushed + size > this.maxSize) {
            var newSize = this.maxSize - this.pushed;
            var buf2 = ZeroBuf(newSize);
            this.push(buf2);
            this.push(null);
        } else {
            var buf = ZeroBuf(size);
            this.push(buf);
            this.pushed += buf.length;
        }
    };
}

util.inherits(ZeroStream, ReadableStream);

module.exports = ZeroStream;