amqp mem issues

gistfile1.js

var amqp = require('amqp'),
    sys = require('sys');

var mqhost = process.argv[2];
var exchange_name = process.argv[3];

function expensive_op(str) {
  return str;
}
var listen = function(f) {
  var parent = this;
  this.mq = amqp.createConnection({host:mqhost});
  this.f = f;
  var reconnect_in = function(t) { setTimeout(function() {this.listen(this.f);}, t); };
  this.mq.addListener('close', function (e) { reconnect_in(1000); });
  this.mq.addListener('ready', function () {
    var mq = this;
    var exchange = mq.exchange(exchange_name, {type: 'topic'});
    var q = mq.queue('node-test-' + process.pid, { autoDelete: true });
    q.bind(exchange, "#");
  
    q.subscribeRaw(function (m) {
      m._payload = '';
      m.addListener('data', function (d) { if(d != null) this._payload += d; });
      m.addListener('end', function () {
        var str = this._payload;
        m.acknowledge();
        f(m, function() { return expensive_op(str); });
      });
    });
  });
};

var routes = {};

listen(function (m,d) {
  if(m.routingKey in routes) {
    routes[m.routingKey].write_f(d());
  }
});