testMosca.js

var Q = require("q");
var persistence = require("persistence");
var mosca = require("mosca");
var mqtt = require("mqtt");
var chai = require("chai");

var should = chai.should();
var Promise = Q.Promise;

function promiseServer(){
    var moscaOpts = {
        backend: {},
        logger: {},
        stats: true,
        port:null,
        persistence: {
            factory: persistence.Memory
        },
        http:{
            port:3000,
            static:false,
            bundle:false
        },
        onlyHttp:true
    };

    return new Promise(function(resolve, reject){
        var mqttServer = new mosca.Server(moscaOpts);
        mqttServer.on("ready", function(){
           resolve(mqttServer);
        });
    });
}

function promiseClient(){
    return new Promise(function(resolve, reject){
        var mqttClient = mqtt.connect("ws://localhost:3000");
        mqttClient.on("connect", function(){
            resolve(mqttClient);
        });
    });
}

function promiseSubscription(client, topic){
    return new Promise(function(resolve, reject){
        client.subscribe(topic, {qos:2}, function(err, granted){
            //Array.prototype.slice.call(arguments).should.deep.equal([null, [{qos:2,topic:"Hello"}]]); //THIS IS THE CORRECT TEST
            Array.prototype.slice.call(arguments).should.deep.equal([null, [{qos:1,topic:"Hello"}]]); //THIS INCORRECT TEST ACTUALLY PASSES
            resolve(arguments);
        });
    });
}

function promisePublication(client,topic,message){
    return new Promise(function(resolve, reject){
        client.publish(topic, message, {qos:2, retain:true}, function(){ //THIS LINE IS CALLED
            resolve(arguments); //THIS LINE IS NEVER CALLED
        });
        setTimeout(resolve, 0); //WORKAROUND FOR MISSING CALLBACK
        //setTimeout(resolve, 500); //DELAYING RESOLUTION BY 500 milliseconds causes test to fail altogether
    });
}

function promiseReceipt(client){
    return new Promise(function(resolve, reject){
       client.on("message", function(topic, message, packet){
           resolve([topic, message.toString()]);
       });
    });
}

var testTopic = "Hello",
    testMessage = "World";

var testPromise = Promise.resolve()
    .then(function(){ //launch server
      return promiseServer();
    })
    .then(function(server) { //create receiver
        return promiseClient();
    })
    .then(function(receiver) { //subscribe receiver
        return promiseSubscription(receiver, testTopic)
            .then(function(subscription) { //create sender
                return promiseClient();
            }).then(function(sender) { // publish sender
                return promisePublication(sender, testTopic, testMessage);
            }).then(function(){
                return promiseReceipt(receiver); //expect receipt
            });
    })
    .timeout(10000)
    .then(function(message){
        console.log("Message Received: " + message);
    })
    .catch(function(err){
        console.log("Chain failed" + err);
    });