Service discovery

client.js

var mqtt = require('mqttjs');
var mdns = require('mdns');

var topic = "my.data";

var connectedClient = null;

discoverService();
startMessageLoop();

function discoverService() {
    
    var browser = mdns.createBrowser( mdns.tcp( 'mqtt' ) );

    browser.on( 'serviceUp', function( service ) {
        console.log("service up");
        serviceUp( service );
    });

    browser.on('serviceDown', function(service) {
      console.log("service down");
      serviceDown();
    });

    browser.start();
}

function serviceUp( service ) {
    if ( !connectedClient && isUsableService( service ) ) {
        connect( service.port, service.addresses[ 0 ] )
    }
}

function serviceDown() {
    stopPublishing();
}

function connect( port, host ) {
    console.log( "connecting to service..." );

    mqtt.createClient( port, host, function( client ) {
        client.connect( {keepalive: 3000} );

        client.on('connack', function(packet) {
            if (packet.returnCode === 0) {
                connectedClient = client;
            } else {
                console.log('Connack error %d', packet.returnCode);
            }
        });

        client.on('close', function() {
            console.log("Client closed");
            stopPublishing();
        });

        client.on('error', function( e ) {
            console.log('Client error %s', e);
            stopPublishing();
        });
    });   
}

function stopPublishing() {
    connectedClient = null;
}

function startMessageLoop() {
    var messageCount = 0;

    setInterval( function () {
        if ( connectedClient ) {
            console.log( "Sending message '%d'", messageCount );
            connectedClient.publish( { topic: topic, payload: "hello world #" + messageCount++ } );
        }
    }, 200);
}

function isUsableService( service ) {
    return service.flags == 2;
}

server.js

var mqtt = require('mqttjs');
var mdns = require('mdns');

var port = 1883;

startMqttServer();

function startMqttServer() {
	mqtt.createServer( function( client ) {
		var self = this;

		if (!self.clients) self.clients = {};

		client.on('connect', function(packet) {
			client.connack({returnCode: 0});
			client.id = packet.client;
			self.clients[client.id] = client;
		});

		client.on('publish', function(packet) {
			console.log( "we got a message on topic '%s' with payload '%s'", packet.topic, packet.payload );
		});

		client.on('pingreq', function(packet) {
			client.pingresp();
		});

		client.on('disconnect', function(packet) {
			client.stream.end();
		});

		client.on('close', function(err) {
			delete self.clients[client.id];
		});

		client.on('error', function(err) {
			client.stream.end();
			util.log('error!');
		});
	}).listen( port, function() {
		console.log( "broadbasting........." );
		mdns.createAdvertisement( mdns.tcp( 'mqtt' ), port ).start();
	});	
}