MQTT – DECODE X NET

以前、他のブログでテストしたMQTTですが、再勉強も兼ねてまた違ったやり方でテストしてみました。

MQTT/RabbitMQ http://bitlife.me/archives/tag/mq
MSMQ/PowerShell http://crossframe.iiv.jp/20140503733/

最近ますます注目度が上がっている感がある、MQTT(MQ Telemetry Transport)ですが、今回はラズパイ上でnode.js(http://node-arm.herokuapp.com/)を使いました。

参考 : MQTTとJavaScript　http://ledsun.hatenablog.com/entry/2014/08/13/141908

インストール

dpkg -i node_latest_armhf.deb
npm install mqtt

バージョン : npm: 2.11.2, node: 0.12.6

broker.js, publisher.js, subscriber.js という三つのプログラムを実行します。
brokerを媒介して、publisherからsubscriberにメッセージが送信されます。publisher, subscriberは複数実行して、違うメッセージを送ることができます。

broker.js

var mqtt = require('mqtt');
var util = require('util');

mqtt.createServer(function(client) {
  var self = this;
  if (!self.clients) self.clients = {};
  client.on('connect', function(packet) {
    client.connack({returnCode: 0});
    client.id = packet.clientId;
    self.clients[client.id] = client;
    console.log("connect ---");
    console.log(util.inspect(packet, false, null));
  });
  client.on('publish', function(packet) {
    console.log("publish ---");
    console.log(util.inspect(packet, false, null));
    for (var k in self.clients) {
      self.clients[k].publish({topic: packet.topic, payload: packet.payload})
;
    }
  });
  client.on('subscribe', function(packet) {
    var granted = [];
    console.log("subscribe ---");
    console.log(util.inspect(packet, false, null));
    for (var i = 0; i < packet.subscriptions.length; i++) {
      granted.push(packet.subscriptions[i].qos);
    }
    client.suback({granted: granted, messageId: packet.messageId});
  });
  client.on('pingreq', function(packet) {
    console.log("pingrep ---");
    console.log(util.inspect(packet, false, null));
    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();
    console.log('error ---');
  });
}).listen(25252);

var mqtt = require('mqtt');

var util = require('util');

mqtt.createServer(function(client) {

var self = this;

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

client.on('connect', function(packet) {

client.connack({returnCode: 0});

client.id = packet.clientId;

self.clients[client.id] = client;

console.log("connect ---");

console.log(util.inspect(packet, false, null));

});

client.on('publish', function(packet) {

console.log("publish ---");

console.log(util.inspect(packet, false, null));

for (var k in self.clients) {

self.clients[k].publish({topic: packet.topic, payload: packet.payload})

;

}

});

client.on('subscribe', function(packet) {

var granted = [];

console.log("subscribe ---");

console.log(util.inspect(packet, false, null));

for (var i = 0; i < packet.subscriptions.length; i++) {

granted.push(packet.subscriptions[i].qos);

}

client.suback({granted: granted, messageId: packet.messageId});

});

client.on('pingreq', function(packet) {

console.log("pingrep ---");

console.log(util.inspect(packet, false, null));

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();

console.log('error ---');

});

}).listen(25252);

publisher.js

var mqtt = require('mqtt');
var client = mqtt.connect({host: 'localhost', port: 25252,
  username: 'admin',
  password: 'pass'
});
var cnt = 0;
setInterval(function() {
  client.publish('msg1', cnt + "");
  client.publish('msg2', cnt + 100 + "");
  cnt ++;
}, 10000);

var mqtt = require('mqtt');

var client = mqtt.connect({host: 'localhost', port: 25252,

username: 'admin',

password: 'pass'

});

var cnt = 0;

setInterval(function() {

client.publish('msg1', cnt + "");

client.publish('msg2', cnt + 100 + "");

cnt ++;

}, 10000);

subscriber.js

var mqtt = require('mqtt'),
  client = mqtt
  .connect({host:'192.168.11.111', port:25252,
    username: 'admin',
    password: 'pass'
  })
  .on('connect', function() {
    console.log('connected');
  })
  .subscribe('msg1', function(err, granted) {
    console.log('subscribe', granted);
  })
  .on('message', function(msg, p1) {
    console.log(p1);
  })

var mqtt = require('mqtt'),

client = mqtt

.connect({host:'192.168.11.111', port:25252,

username: 'admin',

password: 'pass'

})

.on('connect', function() {

console.log('connected');

})

.subscribe('msg1', function(err, granted) {

console.log('subscribe', granted);

})

.on('message', function(msg, p1) {

console.log(p1);

})

broker, publisher, subscriberの順で動かします。publisherから二つのメッセージを送っていますが、複数のsubscriberに別のメッセージを送ろうと意図しましたが、できませんでした。クライアントのIDとトピックが紐づかないからです。このような用途はまた違ったやり方をするのでしょうか・・
下記は、subscriberとbrokerの実行結果です。publisherは何も表示しません。

subscriberが接続したときのbrokder

subscriberがメッセージを受信したときのbroker

また、Ubuntu14.04からsubscriberで接続する確認をしました。

インストール

apt-get install nodejs npm
update-alternatives –install /usr/bin/node node /usr/bin/nodejs 10

バージョン : npm: 1.3.10, node: 0.10.25

またテストをしてみたいと思っています。