Mosquitto

http://mosquitto.org/

Serveur MQTT écrit en C et C++

Installation rapide et simple

sur Raspberry Pi
wget https://bitbucket.org/pypa/setuptools/raw/bootstrap/ez_setup.py -O - | python sudo python get-pip.py

sur MacOS X
brew install mosquitto

sur Ubuntu
http://mosquitto.org/2013/01/mosquitto-debian-repository/

Lancement du broker
Lancez le broker Mosquitto sur l'hôte (10.0.1.3). Remarque : le broker est parfois lancé via initd (vérifiez avec ps -ax | grep mosquitto )

Sur Linux mosquitto

Sur MacOS /usr/local/sbin/mosquitto

Publication en CLI
Publication depuis le Raspberry Pi http://mosquitto.org/man/mosquitto_pub-1.html BROKER=test.mosquitto.org BROKER=localhost mosquitto_pub -h $BROKER -d -t arduino/temp -m "100"

Souscription en CLI
Souscription depuis l'hôte http://mosquitto.org/man/mosquitto_sub-1.html BROKER=test.mosquitto.org BROKER=localhost mosquitto_sub -h $BROKER -d -t arduino/temp

Souscription aux topics systèmes mosquitto_sub -h $BROKER -v -t \$SYS/#

Publication en Python
Publication depuis le Raspberry Pi (et un Arduino) http://mosquitto.org/documentation/python/ BROKER=10.0.1.3 python publisharduino.py $BROKER


 * 1) !/usr/bin/python
 * 2) simple app to read string from serial port (arduino board)
 * 3) and publish via MQTT
 * 4) uses the Python MQTT client from the Mosquitto project
 * 5) http://mosquitto.org
 * 6) initially Andy Piper http://andypiper.co.uk
 * 7) 2011/09/15
 * 1) initially Andy Piper http://andypiper.co.uk
 * 2) 2011/09/15
 * 1) 2011/09/15

import serial import mosquitto import os import sys

broker=sys.argv[1] port = 1883

serialdev = '/dev/ttyACM0'


 * 1) MQTT callbacks

def on_connect(mosq, obj, rc): if rc == 0: print("Connected successfully.") else: raise Exception

def on_disconnect(mosq, obj, rc): print("Disconnected successfully.")

def on_publish(mosq, obj, mid): print("Message "+str(mid)+" published.")

def on_subscribe(mosq, obj, mid, qos_list): print("Subscribe with mid "+str(mid)+" received.")

def on_unsubscribe(mosq, obj, mid): print("Unsubscribe with mid "+str(mid)+" received.")

def on_message(mosq, obj, msg): print("Message received on topic "+msg.topic+" with QoS "+str(msg.qos)+" and payload "+msg.payload)

def cleanup: print "Ending and cleaning up" ser.close mqttc.disconnect
 * 1) called on exit
 * 2) close serial, disconnect MQTT

try: print "Connecting... ", serialdev #connect to serial port ser = serial.Serial(serialdev, 9600, timeout=20) except: print "Failed to connect serial" #unable to continue with no serial input raise SystemExit

try: ser.flushInput #create an mqtt client mypid = os.getpid client_uniq = "arduino_pub_"+str(mypid) mqttc = mosquitto.Mosquitto(client_uniq)

#attach MQTT callbacks mqttc.on_connect = on_connect mqttc.on_disconnect = on_disconnect mqttc.on_publish = on_publish mqttc.on_subscribe = on_subscribe mqttc.on_unsubscribe = on_unsubscribe #mqttc.on_message = on_message

#connect to broker mqttc.connect(broker, port, 60)

#remain connected to broker #read data from serial and publish while mqttc.loop == 0: line = ser.readline #split line as it contains V,temp list = line.split(",") #second list element is temp temp = list[0].rstrip print("Temp is "+temp) mqttc.publish("arduino/temp", temp) pass

except (IndexError): print "No data received within serial timeout period" cleanup except (KeyboardInterrupt): print "Interrupt received" cleanup except (RuntimeError): print "uh-oh! time to die" cleanup
 * 1) handle list index error (i.e. assume no data received)
 * 1) handle app closure

Souscription en Python
Souscription depuis l'hôte http://mosquitto.org/documentation/python/ BROKER=10.0.1.3 python subscribearduino.py $BROKER


 * 1) !/usr/bin/python
 * 2) simple app to subscribe MQTT topic
 * 3) uses the Python MQTT client from the Mosquitto project
 * 4) http://mosquitto.org
 * 5) initially Andy Piper http://andypiper.co.uk
 * 6) 2011/09/15
 * 1) initially Andy Piper http://andypiper.co.uk
 * 2) 2011/09/15
 * 1) 2011/09/15

import os import mosquitto import sys

broker=sys.argv[1] port = 1883
 * 1) broker = "10.0.1.3"

serialdev = '/dev/ttyACM0'


 * 1) MQTT callbacks

def on_connect(mosq, obj, rc): if rc == 0: print("Connected successfully.") else: raise Exception

def on_disconnect(mosq, obj, rc): print("Disconnected successfully.")

def on_publish(mosq, obj, mid): print("Message "+str(mid)+" published.")

def on_subscribe(mosq, obj, mid, qos_list): print("Subscribe with mid "+str(mid)+" received.")

def on_unsubscribe(mosq, obj, mid): print("Unsubscribe with mid "+str(mid)+" received.")

def on_message(mosq, obj, msg): print("Message received on topic "+msg.topic+" with QoS "+str(msg.qos)+" and payload "+msg.payload)

def cleanup: print "Ending and cleaning up" mqttc.disconnect
 * 1) called on exit
 * 2) disconnect MQTT

try: #create an mqtt client mypid = os.getpid client_uniq = "arduino_pub_"+str(mypid) mqttc = mosquitto.Mosquitto(client_uniq)

#attach MQTT callbacks mqttc.on_connect = on_connect mqttc.on_disconnect = on_disconnect mqttc.on_publish = on_publish mqttc.on_subscribe = on_subscribe mqttc.on_unsubscribe = on_unsubscribe mqttc.on_message = on_message

#connect to broker mqttc.connect(broker, port, 60)

mqttc.subscribe("arduino/temp")

#remain connected to broker while mqttc.loop == 0: pass

except (KeyboardInterrupt): print "Interrupt received" cleanup except (RuntimeError): print "uh-oh! time to die" cleanup
 * 1) handle app closure

Souscription avec Node.js
npm install mqtt BROKER=10.0.1.3 node subscribe.js $BROKER 1883 arduino/temp


 * 1) !/usr/bin/env node

var mqtt = require('mqtt');

var argv = process.argv;

for (var i = 2; i <= 4; i++) { if(!argv[i]) process.exit(-1); }

var port = argv[3] , host = argv[2] , topic = argv[4];

var c = mqtt.createClient(port, host);

c.on('connect', function {

c.on('message', function(topic, message) {   console.log(topic + ' ' + message);  });

c.subscribe(topic); });

Simple Souscripteur avec une interface Web npm install mqtt npm install express BROKER=10.0.1.3 node webmqttsub.js $BROKER 1883 arduino/#

node webmqttsub.js $BROKER 1883 bbc/#

curl http://localhost:3000/* curl http://localhost:3000/arduino/ curl http://localhost:3000/arduino/temp

// Simple Gateway between MQTT et HTTP (REST)

// Author: Didier Donsez, 2013 // TODO add PUT, POST (publish), DELETE method

var mqtt = require('mqtt');

var argv = process.argv;

for (var i = 2; i <= 4; i++) { if(!argv[i]) process.exit(-1); }

var brokerport = argv[3] , brokerhost = argv[2] , subscribedtopicpattern = argv[4];

var topics = new Object;

var c = mqtt.createClient(brokerport, brokerhost);

c.on('connect', function {

c.on('message', function(topic, message) {   console.log("receive "+ topic + ' ' + message);      topics[topic]=message;  });

console.log("subscribe to " + subscribedtopicpattern); c.subscribe(subscribedtopicpattern); });

var express = require('express'); var app = express; app.use(function(req, res, next){   var path=req.path;    console.log('req.path=' + path);    if(path=="/*"){	res.set('Content-Type', 'application/json');	res.send(JSON.stringify(topics));     } else {         var topic=path.substring(1);	 if (topics.hasOwnProperty(topic))  {	     if(req.get('Content-Type')=='application/json'){		 res.set('Content-Type', 'application/json');		 res.send( '{"'+topic+'","'+topics[topic]+'"}' );	     } else { 		 res.set('Content-Type', 'text/plain');		 res.send(topics[topic]);	     }	 } else {	     res.send(404, 'Sorry, we cannot find topic '+ topic);	 }     } }); app.listen(3000);

Publication avec Node.js

 * 1) !/usr/bin/env node

var mqtt = require('mqtt');

var argv = process.argv;

for (var i = 2; i <= 5; i++) { if(!argv[i]) process.exit(-1); }

var port = argv[3] , host = argv[2] , topic = argv[4]; , message = argv[5];

var c = mqtt.createClient(port, host);

c.on('connect', function {

c.on('subscribe', function(packet) {     console.log("subscribe" + ' ' + packet);  });

c.on('close', function {     console.log("close");  });

c.on('disconnect', function(packet) {     console.log("disconnect" + ' ' + packet);  });

c.on('error', function(e) {    console.log("error" + ' ' + packet);  });

c.publish(topic, message);

c.end;

});

npm install mqtt BROKER=10.0.1.3 node publish.js $BROKER 1883 arduino/temp "99999999"

Bridge HTTP REST <--> MQTT
BROKER=test.mosquitto.org mosquitto_sub -h test.mosquitto.org -t "arduino/#" -v

BROKER=test.mosquitto.org mosquitto_pub -h test.mosquitto.org -t "arduino/temp" -m "246"

BRIDGE=http://test-mosquitto.herokuapp.com curl $BRIDGE/arduino/temp curl -X PUT --data-binary "247" $BRIDGE/arduino/temp curl -X POST --data-binary "248" $BRIDGE/arduino/temp

...

curl -X DELETE $BRIDGE/arduino/temp

Node-RED with Mosquitto
TODO with Arduino, Firmata, MQTT

Start the subscription BROKER=test.mosquitto.org mosquitto_sub -h $BROKER -d -t smartgrid/uk

BROKER=m2m.eclipse.org mosquitto_sub -h $BROKER -d -t /smartgrid/uk/demand

Install Terminal Notifier (MacOS X 10.8) sudo gem install terminal-notifier

node red.js flow_mqttoutput.json

Browse http://127.0.0.1:1880

Import (Ctrl-I) the following flow then deploy. The flow gets UK smartgrid data from http://www.nationalgrid.com/ngrealtime/realtime/systemdata.aspx every minute, then parses it into a message then publishes the message on the smartgrid/uk topic of the test.mosquitto.org public server. The flow is inspired from the Node-RED documentation.

flow_mqttoutput.json [{"id":"1d2be19e.6a1c1e","type":"mqtt-broker","broker":"test.mosquitto.org","port":"1883"},{"id":"5ff55c38.ef0014","type":"mqtt-broker","broker":"m2m.eclipse.org","port":"1883"},{"id":"26aba4b0.704bb4","type":"mqtt out","name":"Mosquitto Server","topic":"smartgrid/uk","broker":"5ff55c38.ef0014","x":541,"y":698,"wires":[]},{"id":"963ae491.d8ac68","type":"debug","name":"Debug MQTT","active":true,"complete":"false","x":612,"y":497,"wires":[]},{"id":"b52ffa92.d6b9c","type":"http request","name":"UK National Grid Open Data","method":"GET","url":"http://www.nationalgrid.com/ngrealtime/realtime/systemdata.aspx","x":171,"y":499,"wires":"dbe1deb4.42e0d8","778334c9.ca3c3c"},{"id":"dbe1deb4.42e0d8","type":"function","name":"Parsing Grid Data","func":"// The received message is stored in 'msg'\n// It will have at least a 'payload' property:\n//  console.log(msg.payload);\n// The 'context' object is available to store state\n// between invocations of the function\n//   context = {};\n\n    // does a simple text extract parse of the http output to provide an\n    // object containing the uk power demand, frequency and time\n    \n    console.log(msg.payload);\n    \n    if (~msg.payload.indexOf('= 3) {\n        msg.payload = {};\n        msg.payload.demand = parseInt(words[0].split(\":\")[1]);\n        msg.payload.frequency = parseFloat(words[2].split(\":\")[1]);\n        msg.payload.time = words[1].split(\">\")[1];\n        \n        // Create the true/false signal based on the frequency.\n        msg2 = {};\n        msg2.payload = (msg.payload.frequency >= 50) ? true : false;\n       \n        return [msg,msg2];\n      }\n    }\n    return null;","outputs":1,"x":312,"y":600,"wires":"26aba4b0.704bb4"},{"id":"1e20ecfd.c0c393","type":"inject","name":"Every minute","topic":"","payload":"","repeat":"0","crontab":"*/1 0-22 * * *","once":true,"x":147,"y":378,"wires":"b52ffa92.d6b9c"},{"id":"c3092d54.828b5","type":"exec","command":"terminal-notifier -title 'Electricity Demand' -message ","append":"","useSpawn":"","name":"Notify","x":732,"y":420,"wires":],[],[},{"id":"62d8096.aa27f78","type":"mqtt out","name":"Eclipse Paho Server","topic":"smartgrid/uk/demand","broker":"5ff55c38.ef0014","x":563,"y":586,"wires":[]},{"id":"778334c9.ca3c3c","type":"function","name":"Parsing Demand Only","func":"// The received message is stored in 'msg'\n// It will have at least a 'payload' property:\n//   console.log(msg.payload);\n// The 'context' object is available to store state\n// between invocations of the function\n//   context = {};\n\n    // does a simple text extract parse of the http output to provide an\n    // object containing the uk power demand, frequency and time\n    \n    console.log(msg.payload);\n    \n    if (~msg.payload.indexOf('= 1) {\n\n        msg.payload = parseInt(words[0].split(\":\")[1]);\n        \n        return msg;\n      }\n    }\n    return null;","outputs":1,"x":398,"y":437,"wires":"62d8096.aa27f78","a4521ff1.b25f98","963ae491.d8ac68","c3092d54.828b5"},{"id":"a4521ff1.b25f98","type":"mqtt out","name":"Mosquitto Server - Demand only","topic":"/smartgrid/uk/demand","broker":"1d2be19e.6a1c1e","x":553,"y":379,"wires":[]}]

From another machine, start : node red.js flow_mqttinput.json

flow_mqttinput.json [{"id":"f7611a87.089ee8","type":"mqtt-broker","broker":"test.mosquitto.org","port":"1883"},{"id":"80408ea7.7fbf7","type":"mqtt in","name":"UK Smart grid","topic":"smartgrid/uk ","broker":"f7611a87.089ee8","x":920,"y":291,"wires":"f4f3a91.f0b0c58","ca880431.3577f8","5eb0c39b.a14f3c"},{"id":"f4f3a91.f0b0c58","type":"debug","name":"","active":true,"complete":"false","x":1172,"y":296,"wires":[]},{"id":"ca880431.3577f8","type":"function","name":"Console Log","func":"// The received message is stored in 'msg'\n// It will have at least a 'payload' property:\n//  console.log(msg.payload);\n// The 'context' object is available to store state\n// between invocations of the function\n//   context = {};\n\nconsole.log(msg.payload);\nreturn msg;","outputs":1,"x":1191,"y":341,"wires":},{"id":"5eb0c39b.a14f3c","type":"notify","title":"","name":"","x":1199,"y":400,"wires":[]}]

MQTT Panel
MQTT Panel est une application (Node.js) de visualisation de données publiées sur des topics MQTT. Elle utilise JQuery.sparkline.

Telechargez https://github.com/fabaff/mqtt-panel

Term 1: mosquitto /usr/local/sbin/mosquitto

Term 2: node server.js

open index.html

Term 3: ./test-messages.py

InfluxDB
Persitance avec InfluxDB

Examples

 * Arduino-Based Sensor for Agriculture
 * SmartDollHouse