Publishing has unexpected outcome about amqp-coffee HOT 5 CLOSED

var AMQP = require('amqp-coffee');
var async = require('async');

var users = [{
    'user_id': '123'
}, {
    'user_id': '456'
}];

var getTestConnection = function (cb) {
    console.log('creating connection');
    var opts = {
        'host': 'localhost'
    };
    var conn = new AMQP(opts, function (err, c) {
        if (err) return cb(err);
        cb(null, conn);
    });
};

var createTopicExchange = function (conn, done) {
    var name = 'inbox.events';
    if (typeof name != 'string') throw new Error('suppy topic exchange name');
    var opts = {
        'exchange': name,
        'type': 'topic',
        'passive': false,
        'durable': false
    };
    var declareExchange = function (opts, exchange, cb) {
        console.log('declaring exchange');
        exchange.declare(opts, function (err, res) {
            if (err) return cb(err);
            cb(null, exchange);
        });
    };
    var fns = [
        conn.exchange.bind(conn, opts),
        declareExchange.bind(null, opts)
    ];
    async.waterfall(fns, function (err, exchange) {
        if (err) done(err);
        done(null, conn);
    });
};

var createTwoQueuesAndBindToExchange = function (users, conn, done) {
    var queuePrefix = 'user.events';
    var createDeclareBind = function (u, cb) {
        var opts = {
            'queue': [queuePrefix, u.user_id].join('.'),
            'durable': false,
            'passive': false
        };
        var fns = [
            conn.queue.bind(conn, opts),
            function declareQueue (queue, next) {
                console.log('declaring queue')
                queue.declare(opts, function (err, res) {
                    if (err) return next(err);
                    next(null, queue);
                });
            },
            function bindQueue (queue, next) {
                console.log('binding queue');
                var routing = [queuePrefix, u.user_id, '#'].join('.');
                var name = [queuePrefix, u.user_id ].join('.');
                queue.bind('inbox.events', routing, name, function (err) {
                    if (err) return next(err);
                    next(null, queue);
                });
            }
        ];
        async.waterfall(fns, cb);
    };
    async.map(users, createDeclareBind, function (err, queues) {
        if (err) done(err);
        done(null, conn);
    });
};

function sendEvents(users, conn, done) {
    var msg = {'test': 'msg'};
    msg = new Buffer(JSON.stringify(msg));
    var publishOpts = {
        'confirm': true,
        'contentType': 'application/json'
    };
    var Publish = function (user, cb) {
        var uid = user.user_id;
        var pattern = ['user.events', uid].join('.');
        console.log('publishING to pattern >', pattern);
        conn.publish('inbox.events', pattern, msg, publishOpts, function (err) {
            if (err) return cb(err);
            console.log('publishED to pattern >', pattern);
            cb();
        });
    };
    async.map(users, Publish, function (err) {
        if (err) return done(err);
        console.log('publishing complete');
        done(null, conn);
    });
};

var fns = [
    getTestConnection,
    createTopicExchange,
    createTwoQueuesAndBindToExchange.bind(null, users),
    sendEvents.bind(null, users)
];

async.waterfall(fns, function (err, result) {
    if (err) throw err;
    console.log('script complete');
});