Besser MQTT einsetzen für Spark Core und ESP8266?
-
Hi Bluefox
> Mach bitte Flag AN. Dann wirst du auch Zustände und nicht nur Befehle bekommen.das wars ! Danke Dir. Das Flag hatte ich irgendwie völlig ignoriert :oops:
Gruss Marco
-
-> ich nutze eineNode-red mqtt Input Node auf Port 1883 (dürfte dann der Server sein)
- Und parallel dazu mqtt-ioBroker-Adapter als Server auf einem Anderem port?
-> und hier wird es kurios. der mqtt Adapter ist ebenfalls aktiv auf Port 1883, was ja nicht sein kann.
dazu unten mehr. `
Sorry aber zwei Server können nicht auf einem Port laufen. (mindestens wenn beide TCP oder UDP sind).Der erste gewinnt, zweite soll dann die Fehler schreiben.
Kannst du einen Stoppen und das zum laufen bringen?
-
Sorry aber zwei Server können nicht auf einem Port laufen. (mindestens wenn beide TCP oder UDP sind).
Der erste gewinnt, zweite soll dann die Fehler schreiben.
Kannst du einen Stoppen und das zum laufen bringen? `
Ja, das ist bekannt. Lässt der tvp/ip Stack nicht zu.
Da beide eine Verbindung angezeigt haben und Daten rausgeschmissen habe, war meine Vermutung, dass die mqtt Node den mqtt Adapter benötigt und nicht selbst auf Port1883 lauscht.
Die Vermutung wurde auch daher gestützt, dass die Node-red mqtt Input Node nur grün wurde, wenn auf dem Port auch der mqtt Adapter aktiv ist.
Wie dem auch sei…
Habe die Node-red Node nun auf Port 8883 geändert. Ist rot, da keine Gegenstelle.
Der mqtt Adapter ist aktiv. Schmeisst aber nur Dezimalzahlen raus...
-
Ich kann natürlich einbauen, dass wenn zwei Dezimalzahlen (oder mehr) da sind, dann wird es als Buffer interpretiert….
if (typeof packet.payload == 'object') packet.payload = packet.payload.toString('utf8'); if (typeof packet.payload == 'string' && packet.payload.match(/(\d)+,\s?(\d)+/)) { var parts = packet.payload.split(','); var str = ''; for (var p = 0; p < parts.length; p++) { str += String.fromCharCode(parseInt(parts.trim(), 10)); } packet.payload = str; }Kannst du das ausprobieren?
620_bildschirmfoto_2017-03-15_um_16.19.01.png
620_bildschirmfoto_2017-03-15_um_16.19.14.png -
Ich kann natürlich einbauen, dass wenn zwei Dezimalzahlen (oder mehr) da sind, dann wird es als Buffer interpretiert….
Kannst du das ausprobieren? `
Guten Morgen,
Ich habe den Code in main.js unter opt/iobroker/node_modules/iobroker.mqtt eingesetzt.
Anstelle von:
// if (typeof packet.payload == 'object') packet.payload = packet.payload.toString(); // if (typeof packet.payload == 'object') packet.payload = packet.payload.toString('utf8');mqtt.Adapter in ioBroker neu gestartet. Log:
! ` > mqtt-0 2015-05-17 08:18:25 info Client [iPhone 6] connected
mqtt-0 2015-05-17 08:18:25 info Client [iPhone 6] closed
mqtt-0 2015-05-17 08:18:24 info Client [iPhone 6] disconnected
mqtt-0 2015-05-17 08:13:20 info Client [iPhone 6] subscribes on "mqtt.0.owntracks.ruhr.#" with regex /^mqtt.0.owntracks.ruhr..*/
mqtt-0 2015-05-17 08:13:20 info Client [iPhone 6] subscribes on "owntracks.ruhr.#" with regex /^owntracks.ruhr..*/
mqtt-0 2015-05-17 08:13:20 info Client [iPhone 6] subscribes on "mqtt.0.owntracks.+.+" with regex /mqtt.0.owntracks.[^.].[^.].$/
mqtt-0 2015-05-17 08:13:20 info Client [iPhone 6] subscribes on "owntracks.+.+" with regex /owntracks.[^.].[^.].$/
mqtt-0 2015-05-17 08:13:20 info Client [iPhone 6] connected
mqtt-0 2015-05-17 08:13:03 info Starting MQTT authenticated server on port 1883
mqtt-0 2015-05-17 08:13:03 info starting. Version 0.2.7 in /opt/iobroker/node_modules/iobroker.mqtt
mqtt-0 2015-05-17 08:12:58 info terminating
mqtt-0 2015-05-17 08:12:58 info Client [iPhone 6] closed
mqtt-0 2015-05-17 08:12:42 info Client [iPhone 6] subscribes on "mqtt.0.owntracks.ruhr.#" with regex /^mqtt.0.owntracks.ruhr..*/ `
Es werden weiterhin nur Dezimalzahlen ausgegeben.
123,34,116,115,116,34,58,34,49,52,51,49,53,52,51,50,55,48,34,44,34,97,99,99,...entspricht:
{"tst":"1431543270","accWenn ich mir tst ansehe, ergibt dies:
13.05.2015 20:54:30 (also ein altes Datum, wo owntracks noch funktioniert hat).
Schickt Owntracks eine alte Meldung, die noch quittiert werden muss?
Die Dezimalzahlen sehe ich in Node-red in folgender Konstellation:
-
ioBroker Input mit Topic mqtt.0.owntracks.ruhr, Payload = Value.
-
Verbunden mit einem Debug Node.
In dem Moment, bei dem ich in Owntracks auf "Publish Now" drücke, kommt im Debug Node die Zahlenkolonne an.
Warum mit einem alten Timestamp?
In Historie von mqtt.0.owntracks.ruhr sehe ich die Dezimalzahlen ebenfalls als aktuellen Eintrag.
Gruß,
Michael
-
-
In dem Moment, bei dem ich in Owntracks auf "Publish Now" drücke, kommt im Debug Node die Zahlenkolonne an.
Warum mit einem alten Timestamp? `
Warum im Debug Node???? Und nicht im Log von ioBroker? -
In dem Moment, bei dem ich in Owntracks auf "Publish Now" drücke, kommt im Debug Node die Zahlenkolonne an.
Warum mit einem alten Timestamp?
Warum im Debug Node???? Und nicht im Log von ioBroker?Das Log habe ich oben unter Spoiler dabei.
Dort sind keine Meldungen. Ich wüsste auch nicht woher.
Der mqtt Adapter schreibt direkt in die Variable mqtt.0.owntracks.ruhr
Da stehen die Dezimalzahlen schon drin.
Ich hatte es mir nur in Rode-Ned angesehen, da ich dort in Debug direkt sehe, was passiert.
ioBroker Input mqtt.0.owntracks.ruhr-> Debug
-
Kannst du folgenden Kode erweitern?
adapter.log.info('Type: ' + typeof packet.payload); if (typeof packet.payload == 'object') packet.payload = packet.payload.toString('utf8'); if (typeof packet.payload == 'string' && packet.payload.match(/(\d)+,\s?(\d)+/)) { adapter.log.info('Try to convert ' + packet.payload); var parts = packet.payload.split(','); var str = ''; for (var p = 0; p < parts.length; p++) { str += String.fromCharCode(parseInt(parts.trim(), 10)); } packet.payload = str; adapter.log.info('Converted ' + packet.payload); }Und natürlich den Adapter neu starten.
-
Kannst du folgenden Kode erweitern?
adapter.log.info('Type: ' + typeof packet.payload); if (typeof packet.payload == 'object') packet.payload = packet.payload.toString('utf8'); if (typeof packet.payload == 'string' && packet.payload.match(/(\d)+,\s?(\d)+/)) { adapter.log.info('Try to convert ' + packet.payload); var parts = packet.payload.split(','); var str = ''; for (var p = 0; p < parts.length; p++) { str += String.fromCharCode(parseInt(parts.trim(), 10)); } packet.payload = str; adapter.log.info('Converted ' + packet.payload); }Und natürlich den Adapter neu starten. `
Code ausgetauscht. mqtt Adapter neu gestartet.
Mehrmals bei Owntracks "Publish now" gedrückt.
Log:
Zeit Nachricht x mqtt-0 2015-05-17 09:16:23 info Type: object mqtt-0 2015-05-17 09:15:57 info Type: object mqtt-0 2015-05-17 09:15:32 info Type: object mqtt-0 2015-05-17 09:15:30 info Client [iPhone 6] subscribes on "mqtt.0.owntracks.ruhr.#" with regex /^mqtt\.0\.owntracks\.ruhr\..*/ mqtt-0 2015-05-17 09:15:30 info Client [iPhone 6] subscribes on "owntracks.ruhr.#" with regex /^owntracks\.ruhr\..*/ mqtt-0 2015-05-17 09:15:30 info Client [iPhone 6] subscribes on "mqtt.0.owntracks.+.+" with regex /mqtt\.0\.owntracks\.[^.]*\.[^\.].*$/ mqtt-0 2015-05-17 09:15:30 info Client [iPhone 6] subscribes on "owntracks.+.+" with regex /owntracks\.[^.]*\.[^\.].*$/ mqtt-0 2015-05-17 09:15:30 info Type: object mqtt-0 2015-05-17 09:15:30 info Client [iPhone 6] connected mqtt-0 2015-05-17 09:15:12 info Client [iPhone 6] closed mqtt-0 2015-05-17 09:15:12 info Client [iPhone 6] disconnected mqtt-0 2015-05-17 09:15:05 info Client [iPhone 6] subscribes on "mqtt.0.owntracks.ruhr.#" with regex /^mqtt\.0\.owntracks\.ruhr\..*/ mqtt-0 2015-05-17 09:15:05 info Client [iPhone 6] subscribes on "owntracks.ruhr.#" with regex /^owntracks\.ruhr\..*/ mqtt-0 2015-05-17 09:15:05 info Client [iPhone 6] subscribes on "mqtt.0.owntracks.+.+" with regex /mqtt\.0\.owntracks\.[^.]*\.[^\.].*$/ mqtt-0 2015-05-17 09:15:05 info Client [iPhone 6] subscribes on "owntracks.+.+" with regex /owntracks\.[^.]*\.[^\.].*$/ mqtt-0 2015-05-17 09:15:05 info Client [iPhone 6] connected mqtt-0 2015-05-17 09:14:53 info Starting MQTT authenticated server on port 1883 mqtt-0 2015-05-17 09:14:53 info starting. Version 0.2.7 in /opt/iobroker/node_modules/iobroker.mqtt mqtt-0 2015-05-17 09:01:39 info Client [iPhone 6] closed mqtt-0 2015-05-17 09:01:39 info Client [iPhone 6] disconnected
-
Und noch mal ausprobieren.
adapter.log.info('Type: ' + typeof packet.payload); if (typeof packet.payload == 'object') packet.payload = packet.payload.toString('utf8'); adapter.log.info('Type: ' + typeof packet.payload +'; Payload: ' + packet.payload); if (typeof packet.payload == 'string' && packet.payload.match(/(\d)+,\s?(\d)+/)) { adapter.log.info('Try to convert ' + packet.payload); var parts = packet.payload.split(','); var str = ''; for (var p = 0; p < parts.length; p++) { str += String.fromCharCode(parseInt(parts.trim(), 10)); } packet.payload = str; adapter.log.info('Converted ' + packet.payload); } -
mqtt-0 2015-05-17 09:38:47 info Type: string; Payload: {"cog":-1,"batt":"100","lon":"7.92283","acc":"65","vel":-1,"vac":10,"lat":"52.3427","t":"u","tst":"1431848304","alt":43,"_type":"location","tid":"MH"} mqtt-0 2015-05-17 09:38:47 info Type: object mqtt-0 2015-05-17 09:38:41 info Client [iPhone 6] subscribes on "mqtt.0.owntracks.ruhr.#" with regex /^mqtt\.0\.owntracks\.ruhr\..*/ mqtt-0 2015-05-17 09:38:41 info Client [iPhone 6] subscribes on "owntracks.ruhr.#" with regex /^owntracks\.ruhr\..*/ mqtt-0 2015-05-17 09:38:41 info Client [iPhone 6] subscribes on "mqtt.0.owntracks.+.+" with regex /mqtt\.0\.owntracks\.[^.]*\.[^\.].*$/ mqtt-0 2015-05-17 09:38:41 info Client [iPhone 6] subscribes on "owntracks.+.+" with regex /owntracks\.[^.]*\.[^\.].*$/ mqtt-0 2015-05-17 09:38:41 info Client [iPhone 6] connected mqtt-0 2015-05-17 09:38:27 info Client [iPhone 6] closed mqtt-0 2015-05-17 09:38:27 info Client [iPhone 6] disconnected mqtt-0 2015-05-17 09:38:26 info Type: string; Payload: {"cog":-1,"batt":"100","lon":"7.92283","acc":"65","vel":-1,"vac":10,"lat":"52.2427","t":"u","tst":"1431848304","alt":43,"_type":"location","tid":"MH"} mqtt-0 2015-05-17 09:38:26 info Type: object mqtt-0 2015-05-17 09:38:24 info Type: string; Payload: {"tst":"1431848304","acc":"65","_type":"location","alt":43,"lon":"7.92283","vac":10,"vel":-1,"lat":"52.3427","cog":-1,"tid":"MH","batt":"100"} mqtt-0 2015-05-17 09:38:24 info Type: object mqtt-0 2015-05-17 09:38:20 info Client [iPhone 6] subscribes on "mqtt.0.owntracks.ruhr.#" with regex /^mqtt\.0\.owntracks\.ruhr\..*/ mqtt-0 2015-05-17 09:38:20 info Client [iPhone 6] subscribes on "owntracks.ruhr.#" with regex /^owntracks\.ruhr\..*/ mqtt-0 2015-05-17 09:38:20 info Client [iPhone 6] subscribes on "mqtt.0.owntracks.+.+" with regex /mqtt\.0\.owntracks\.[^.]*\.[^\.].*$/ mqtt-0 2015-05-17 09:38:20 info Client [iPhone 6] subscribes on "owntracks.+.+" with regex /owntracks\.[^.]*\.[^\.].*$/ mqtt-0 2015-05-17 09:38:20 info Client [iPhone 6] connected mqtt-0 2015-05-17 09:38:11 info Starting MQTT authenticated server on port 1883 mqtt-0 2015-05-17 09:38:10 info starting. Version 0.2.7 in /opt/iobroker/node_modules/iobroker.mqttIm Log steht die Zeichenkette. Im Historie stehen trotzdem die Dezimahlzahlen.
-
Ich glaube, ich habe es.
Kannst du noch die Zeile (ca 465):
if (typeof packet.payload == 'string' && packet.payload[0] == '{') { try { packet.payload = JSON.parse(packet.payload); ....auf
if (0 || typeof packet.payload == 'string' && packet.payload[0] == '{') { -
Zeile ausgetauscht. mqqt Adapter neu gestartet.
In der Variable stehen immer noch die Dezimalzahlen (ganz aktuell von 19:05).
mqtt-0 2015-05-17 19:10:17 info Client [iPhone 6] closed mqtt-0 2015-05-17 19:10:17 info Client [iPhone 6] disconnected mqtt-0 2015-05-17 19:05:16 info Type: string; Payload: {"cog":-1,"batt":"100","lon":"7.92287","acc":"65","vel":-1,"vac":10,"lat":"52.2426","t":"u","tst":"1431882313","alt":41,"_type":"location","tid":"MH"} mqtt-0 2015-05-17 19:05:16 info Type: object mqtt-0 2015-05-17 19:05:14 info Client [iPhone 6] subscribes on "mqtt.0.owntracks.ruhr.#" with regex /^mqtt\.0\.owntracks\.ruhr\..*/ mqtt-0 2015-05-17 19:05:14 info Client [iPhone 6] subscribes on "owntracks.ruhr.#" with regex /^owntracks\.ruhr\..*/ mqtt-0 2015-05-17 19:05:14 info Client [iPhone 6] subscribes on "mqtt.0.owntracks.+.+" with regex /mqtt\.0\.owntracks\.[^.]*\.[^\.].*$/ mqtt-0 2015-05-17 19:05:14 info Client [iPhone 6] subscribes on "owntracks.+.+" with regex /owntracks\.[^.]*\.[^\.].*$/ mqtt-0 2015-05-17 19:05:14 info Client [iPhone 6] connected mqtt-0 2015-05-17 19:04:56 info Starting MQTT authenticated server on port 1883 mqtt-0 2015-05-17 19:04:55 info starting. Version 0.2.7 in /opt/iobroker/node_modules/iobroker.mqtt mqtt-0 2015-05-17 19:04:53 info terminatingIm Log steht der Text ja schon so, wie wie wir ihn brauchen.
Ich hoffe nur, dass das Dezimalthema nicht von mir kommt :shock:
-
Kann nicht sein
Kannst du main.js damit komplett austauschen?
! ```
`/**-
ioBroker mqtt Adapter -
(c) 2014-2015 bluefox -
MIT License
*/
! var utils = require(__dirname + '/lib/utils'); // Get common adapter utils
var adapter = utils.adapter('mqtt');
! var mqtt = null;
! var client = null;
var servers = null;
var clients = {};
var states = {};
var objects = [];
! var messageboxRegex = new RegExp(".messagebox$");
! function decrypt(key, value) {
var result = "";
for (var i = 0; i < value.length; ++i) {
result += String.fromCharCode(key[i].charCodeAt(0) ^ value.charCodeAt(i));
}
return result;
}
! adapter.on('message', function (obj) {
if (obj) processMessage(obj);
processMessages();
});
! adapter.on('ready', function () {
var fs = require('fs');
adapter.config.pass = decrypt("Zgfr56gFe87jJOM", adapter.config.pass);
if (adapter.config.ssl && adapter.config.type == 'server') {
// Read the certificates and store it under
// "adapter/mqtt/cert/privatekey.pem" and
// "adapter/mqtt/cert/certificate.pem"
// because mqtt does not support certificates not from file
adapter.getForeignObject('system.certificates', function (err, obj) {
if (err || !obj || !obj.native || !obj.native.certificates || !obj.native.certificates[adapter.config.certPublic] || !obj.native.certificates[adapter.config.certPrivate]) {
adapter.log.error('Cannot enable secure MQTT server, because no certificates found: ' + adapter.config.certPublic + ', ' + adapter.config.certPrivate);
setTimeout(function () {
process.exit(1);
}, 500);
} else {
adapter.config.certificates = {
key: obj.native.certificates[adapter.config.certPrivate],
cert: obj.native.certificates[adapter.config.certPublic]
};
! main();
}
});
} else {
// Start
main();
}
});
! adapter.on('unload', function () {
if (client) {
client.end();
client = null;
}
! if (servers) {
// to release all resources
servers.destroy(function(){
console.log('all gone!');
});
servers = null;
}
});
! function checkPattern(patterns, id) {
for (var pattern in patterns) {
if (patterns[pattern].regex.test(id)) return patterns[pattern];
}
! return null;
}
! function id2topic(id, pattern) {
var topic;
if (pattern.substring(0, (adapter.config.prefix + adapter.namespace).length) == (adapter.config.prefix + adapter.namespace)) {
topic = adapter.config.prefix + id;
} else if (pattern.substring(0, adapter.namespace.length) == adapter.namespace) {
topic = id;
} else if (adapter.config.prefix && pattern.substring(0, adapter.config.prefix.length) == adapter.config.prefix) {
topic = adapter.config.prefix + id;//.substring(adapter.namespace.length + 1);
} else if (id.substring(0, adapter.namespace.length) == adapter.namespace) {
topic = id.substring(adapter.namespace.length + 1);
} else {
topic = id;
}
topic = topic.replace(/./g, '/');
return topic;
}
! function state2string(val) {
return (val === null) ? 'null' : (val === undefined ? 'undefined' : val.toString());
}
/*
4.7.1.2 Multi-level wildcard
! The number sign (‘#’ U+0023) is a wildcard character that matches any number of levels within a topic. The multi-level wildcard represents the parent and any number of child levels. The multi-level wildcard character MUST be specified either on its own or following a topic level separator. In either case it MUST be the last character specified in the Topic Filter [MQTT-4.7.1-2].
! Non normative comment
For example, if a Client subscribes to “sport/tennis/player1/#”, it would receive messages published using these topic names:
· “sport/tennis/player1”
· “sport/tennis/player1/ranking”
· “sport/tennis/player1/score/wimbledon”
! Non normative comment
· “sport/#” also matches the singular “sport”, since # includes the parent level.
· “#” is valid and will receive every Application Message
· “sport/tennis/#” is valid
· “sport/tennis#” is not valid
· “sport/tennis/#/ranking” is not valid
! /
! /4.7.1.3 Single level wildcard
! The plus sign (‘+’ U+002B) is a wildcard character that matches only one topic level.
! The single-level wildcard can be used at any level in the Topic Filter, including first and last levels. Where it is used it MUST occupy an entire level of the filter [MQTT-4.7.1-3]. It can be used at more than one level in the Topic Filter and can be used in conjunction with the multilevel wildcard.
! Non normative comment
For example, “sport/tennis/+” matches “sport/tennis/player1” and “sport/tennis/player2”, but not “sport/tennis/player1/ranking”. Also, because the single-level wildcard matches only a single level, “sport/+” does not match “sport” but it does match “sport/”.
! Non normative comment
· “+” is valid
· “+/tennis/#” is valid
· “sport+” is not valid
· “sport/+/player1” is valid
· “/finance” matches “+/+” and “/+”, but not “+”
/
function pattern2RegEx(pattern) {
pattern = topic2id(pattern, true);
pattern = pattern.replace(/#/g, '');
pattern = pattern.replace(/$/g, '\/r>);
pattern = pattern.replace(/^/g, '\^');
! if (pattern != '') {
if (pattern[0] == '' && pattern[pattern.length - 1] != '') pattern += '/r>;
if (pattern[0] != '' && pattern[pattern.length - 1] == '') pattern = '^' + pattern;
if (pattern[0] == '+') pattern = '^[^.]' + pattern.substring(1);
if (pattern[pattern.length - 1] == '+') pattern = pattern.substring(0, pattern.length - 1) + '[^.]/r>;
}
pattern = pattern.replace(/./g, '\.');
pattern = pattern.replace(/*/g, '.');
pattern = pattern.replace(/+/g, '[^.]');
return pattern;
}
! function topic2id(topic, dontCutNamespace) {
topic = topic.replace(///g, '.');
if (topic[0] == '.') topic = topic.substring(1);
if (topic[topic.length - 1] == '.') topic = topic.substring(0, topic.length - 1);
! // Remove own prefix if
if (adapter.config.prefix && topic.substring(0, adapter.config.prefix.length) == adapter.config.prefix) {
topic = topic.substring(adapter.config.prefix.length);
}
! if (!dontCutNamespace && topic.substring(0, adapter.namespace.length) == adapter.namespace) {
topic = topic.substring(adapter.namespace.length + 1);
}
! return topic;
}
! function send2Client(client, id, state) {
if (messageboxRegex.test(id)) return;
! if (!client._subsID ||
client._subsID[id] !== undefined) {
var topic = (client._subsID) ? client._subsID[id].pattern : adapter.config.prefix + id.replace(/./g, '/');
! if (adapter.config.debug) adapter.log.info('Send to client [' + client.id + '] "' + topic + '": ' + (state ? state2string(state.val) : 'deleted'));
client.publish({topic: topic, payload: (state ? state2string(state.val) : null)});
! } else
// Check patterns
if (client._subs) {
var pattern = checkPattern(client._subs, id);
! if (pattern) {
var topic = id2topic(id, pattern.pattern);
// Cache the value
client._subsID[id] = {
qos: pattern,
pattern: topic
};
if (adapter.config.debug) adapter.log.info('Send to client [' + client.id + '] "' + topic + '": ' + (state ? state2string(state.val) : 'deleted'));
client.publish({topic: topic, payload: (state ? state2string(state.val) : null)});
}
}
}
! // is called if a subscribed state changes
adapter.on('stateChange', function (id, state) {
! // State deleted
if (!state) {
delete states[id];
// If SEVRER
if (servers) {
for (var k in clients) {
send2Client(clients[k], id, state);
}
} else
// if CLIENT
if (client) {
var topic = id.replace(/./g, '/');
if (adapter.config.debug) adapter.log.info('Send to server "' + adapter.config.prefix + topic + '": deleted');
client.publish(adapter.config.prefix + topic, null);
}
} else
// you can use the ack flag to detect if state is desired or acknowledged
if ((adapter.config.sendAckToo || !state.ack) && !messageboxRegex.test(id)) {
var old = states[id] ? states[id].val : null;
states[id] = state;
! // If value realy changed
if (!adapter.config.onchange || old !== state.val) {
if (servers) {
for (var k in clients) {
send2Client(clients[k], id, state);
}
} else
if (client) {
var topic = id.replace(/./g, '/');
if (adapter.config.debug) adapter.log.info('Send to server "' + adapter.config.prefix + topic + '": ' + state2string(state.val));
client.publish(adapter.config.prefix + topic, state2string(state.val));
}
}
}
});
! function updateClients() {
var text = '';
if (clients) {
for (var id in clients) {
text += (text ? ',' : '') + id;
}
}
! adapter.setState('clients', {val: text, ack: true});
}
! function processMessage(obj) {
if (!obj || !obj.command) return;
switch (obj.command) {
case 'test': {
// Try to connect to mqtt broker
if (obj.callback && obj.message) {
mqtt = mqtt || require('mqtt');
var _url = 'mqtt://' + (obj.message.user ? (obj.message.user + ':' + obj.message.pass + '@') : '') + obj.message.url + (obj.message.port ? (':' + obj.message.port) : '') + '?clientId=ioBroker.' + adapter.namespace;
var _client = mqtt.connect(_url);
// Set timeout for connection
var timeout = setTimeout(function () {
_client.end();
adapter.sendTo(obj.from, obj.command, 'timeout', obj.callback);
}, 2000);
! // If connected, return success
_client.on('connect', function () {
_client.end();
clearTimeout(timeout);
adapter.sendTo(obj.from, obj.command, 'connected', obj.callback);
});
}
}
}
}
! function processMessages() {
adapter.getMessage(function (err, obj) {
if (obj) {
processMessage(obj.command, obj.message);
processMessages();
}
});
}
! function createClient(config) {
mqtt = mqtt || require('mqtt');
! var _url = ((!config.ssl) ? 'mqtt' : 'mqtts') + '://' + (config.user ? (config.user + ':' + config.pass + '@') : '') + config.url + (config.port ? (':' + config.port) : '') + '?clientId=ioBroker.' + adapter.namespace;
var __url = ((!config.ssl) ? 'mqtt' : 'mqtts') + '://' + (config.user ? (config.user + ':***************@') : '') + config.url + (config.port ? (':' + config.port) : '') + '?clientId=ioBroker.' + adapter.namespace;
adapter.log.info('Try to connect to ' + __url);
client = mqtt.connect(_url);
! // By default subscribr on all topics
if (!config.patterns) config.patterns = '#';
! if (typeof config.patterns == 'string') {
config.patterns = config.patterns.split(',');
}
! for (var i = 0; i < config.patterns.length; i++) {
config.patterns[i] = config.patterns[i].trim();
client.subscribe(config.patterns[i]);
}
! client.on('message', function (topic, message) {
if (!topic) return;
! // Ignore message if value does not changed
if (config.onchange) {
var oldValue = states[topic];
if (oldValue !== undefined && oldValue == message) {
return;
} else {
states[topic] = message;
}
}
topic = topic2id(topic);
! if (typeof message == 'object') message = message.toString();
! var f = parseFloat(message);
! if (objects.indexOf(topic) == -1) {
objects.push(topic);
// Create object if exists
adapter.getObject(topic, function (err, obj) {
if (!obj) {
adapter.getForeignObject(topic, function (err, obj) {
if (!obj) {
adapter.createState('', '', topic, {
name: topic,
write: true,
read: true,
role: 'variable',
desc: 'mqtt client variable',
type: (f.toString() == message) ? 'number' : 'string'
}, {
origin: adapter.namespace
});
}
});
}
});
}
! if (f.toString() == message) message = f;
! if (config.debug) adapter.log.info('Server publishes "' + adapter.namespace + '.' + topic + '": ' + message);
! if (typeof message == 'string' && message[0] == '{') {
try {
message = JSON.parse(message);
adapter.setState(topic, message);
return;
} catch (e) {
adapter.log.warn('Cannot parse "' + topic + '": ' + message);
}
}
! adapter.setState(topic, {val: message, ack: true});
});
! client.on('connect', function () {
adapter.log.info('Connected to ' + config.url);
if (config.publishAllOnStart) {
for (var id in states) {
if (!messageboxRegex.test(id)) {
client.publish(config.prefix + id.replace(/./g, '/'), state2string(states[id].val));
}
}
}
});
! client.on('error', function (err) {
adapter.log.error(err);
});
}
! function createServer(config) {
var createStreamServer = require('create-stream-server');
var mqtt = require('mqtt-connection');
! var cltFunction = function (client) {
! client.on('connect', function (packet) {
client.id = packet.clientId;
if (config.user) {
if (config.user != packet.username ||
config.pass != packet.password) {
adapter.log.warn('Client [' + packet.clientId + '] has invalid password(' + packet.password + ') or username(' + packet.username + ')');
client.connack({returnCode: 4});
if (clients[client.id]) delete clients[client.id];
client.stream.end();
updateClients();
return;
}
}
! adapter.log.info('Client [' + packet.clientId + '] connected');
client.connack({returnCode: 0});
clients[client.id] = client;
updateClients();
! // Send all subscribed variables to client
if (config.publishAllOnStart) {
! // Give to client 2 seconds to send subscribe
client._sendOnStart = setTimeout(function () {
client._sendOnStart = null;
// If client still connected
for (var id in states) {
send2Client(client, id, states[id]);
}
}, 2000);
}
});
! client.on('publish', function (packet) {
/for (var k in clients) {
clients[k].publish({topic: packet.topic, payload: packet.payload});
}/
var topic = topic2id(packet.topic);
! adapter.log.info('Type: ' + typeof packet.payload);
! if (typeof packet.payload == 'object') packet.payload = packet.payload.toString('utf8');
! if (typeof packet.payload == 'string' && packet.payload.match(/(\d)+,\s?(\d)+/)) {
adapter.log.info('Try to convert ' + packet.payload);
! var parts = packet.payload.split(',');
var str = '';
for (var p = 0; p < parts.length; p++) {
str += String.fromCharCode(parseInt(parts.trim(), 10));
}
packet.payload = str;
adapter.log.info('Converted ' + packet.payload);
}
! var f = parseFloat(packet.payload);
! // If state is unknown => create mqtt.X.topic
if (states[topic] === undefined && states[adapter.namespace + '.' + topic] === undefined) {
adapter.log.info('Create state ' + adapter.namespace + '.' + topic);
! adapter.setObject(topic, {
type: 'state',
common: {
name: packet.topic,
write: true,
read: true,
role: 'variable',
desc: 'mqtt variable',
type: (f.toString() == packet.payload) ? 'number' : 'string'
},
native: {
origin: client.id
}
});
topic = adapter.namespace + '.' + topic;
states[topic] = {};
}
! // Try to convert into float
if (f.toString() == packet.payload) packet.payload = f;
! /if (typeof packet.payload == 'string' && packet.payload[0] == '{') {
try {
packet.payload = JSON.parse(packet.payload);
! if (states[topic]) {
if (config.debug) adapter.log.info('Client [' + client.id + '] publishes "' + topic + '": ' + JSON.stringify(packet.payload));
! adapter.setForeignState(topic, packet.payload, function (id) {
states[id] = packet.payload;
});
} else {
if (config.debug) adapter.log.info('Client [' + client.id + '] publishes "' + adapter.namespace + '.' + topic + '": ' + JSON.stringify(packet.payload));
! adapter.setState(topic, packet.payload, function (id) {
states[id] = packet.payload;
});
}
return;
} catch (e) {
adapter.log.warn('Cannot parse "' + topic + '": ' + packet.payload);
}
}/
if (states[topic]) {
if (config.debug) adapter.log.info('Client [' + client.id + '] publishes "' + topic + '" (' + typeof packet.payload + '): ' + packet.payload);
adapter.setForeignState(topic, {val: packet.payload, ack: true}, function (id) {
states[id] = {val: packet.payload, ack: true};
});
} else {
if (config.debug) adapter.log.info('Client [' + client.id + '] publishes "' + adapter.namespace + '.' + topic + '"(' + typeof packet.payload + '): ' + packet.payload);
adapter.setState(topic, {val: packet.payload, ack: true}, function (id) {
states[id] = {val: packet.payload, ack: true};
});
}
});
! client.on('subscribe', function (packet) {
var granted = [];
if (!client._subsID) client._subsID = {};
if (!client._subs) client._subs = {};
! for (var i = 0; i < packet.subscriptions.length; i++) {
granted.push(packet.subscriptions[i].qos);
! var topic = topic2id(packet.subscriptions[i].topic);
! if (topic.indexOf('') == -1 && topic.indexOf('#') == -1 && topic.indexOf('+') == -1) {
// If state is unknown => create mqtt.X.topic
if (states[topic] === undefined && states[adapter.namespace + '.' + topic] === undefined) {
adapter.log.info('Create state ' + adapter.namespace + '.' + topic + ' for subscribe');
adapter.setObject(topic, {
type: 'state',
common: {
name: packet.subscriptions[i].topic,
write: true,
read: true,
role: 'variable',
desc: 'mqtt variable',
type: 'string',
def: null
},
native: {
origin: client.id
}
}, function (err, obj) {
states[obj.id] = {val: null, ack: true};
adapter.setState(obj.id, {val: null, ack: true});
});
topic = adapter.namespace + '.' + topic;
client._subsID[topic] = {
qos: packet.subscriptions[i].qos,
pattern: packet.subscriptions[i].topic
}
adapter.log.info('Client [' + client.id + '] subscribes on "' + topic + '"');
} else {
if (states[topic]) {
client._subsID[topic] = {
qos: packet.subscriptions[i].qos,
pattern: packet.subscriptions[i].topic
};
adapter.log.info('Client [' + client.id + '] subscribes on "' + topic + '"');
} else {
client._subsID[adapter.namespace + '.' + topic] = {
qos: packet.subscriptions[i].qos,
pattern: packet.subscriptions[i].topic
};
adapter.log.info('Client [' + client.id + '] subscribes on "' + adapter.namespace + '.' + topic + '"');
}
! }
} else {
var topic = pattern2RegEx(packet.subscriptions[i].topic);
! var pattern = packet.subscriptions[i].topic.replace(///g, '.');
if (pattern[0] == '.') pattern = pattern.substring(1);
! client._subs[packet.subscriptions[i].topic] = {
regex: new RegExp(topic),
qos: packet.subscriptions[i].qos,
pattern: pattern
};
adapter.log.info('Client [' + client.id + '] subscribes on "' + topic2id(packet.subscriptions[i].topic) + '" with regex /' + topic + '/');
! topic = adapter.namespace + '.' + topic2id(packet.subscriptions[i].topic);
client._subs[topic] = {
regex: new RegExp(pattern2RegEx(topic)),
qos: packet.subscriptions[i].qos,
pattern: pattern
};
adapter.log.info('Client [' + client.id + '] subscribes on "' + topic + '" with regex /' + pattern2RegEx(topic) + '/');
}
}
! client.suback({granted: granted, messageId: packet.messageId});
});
! client.on('pingreq', function (packet) {
adapter.log.debug('Client [' + client.id + '] pingreq');
client.pingresp();
});
! client.on('disconnect', function (packet) {
if (client._sendOnStart) {
clearTimeout(client._sendOnStart);
client._sendOnStart = null;
}
adapter.log.info('Client [' + client.id + '] disconnected');
client.stream.end();
});
! client.on('close', function (err) {
if (client._sendOnStart) {
clearTimeout(client._sendOnStart);
client._sendOnStart = null;
}
adapter.log.info('Client [' + client.id + '] closed');
delete clients[client.id];
updateClients();
});
! client.on('error', function (err) {
if (client._sendOnStart) {
clearTimeout(client._sendOnStart);
client._sendOnStart = null;
}
adapter.log.warn('[' + client.id + '] ' + err);
! if (!clients[client.id]) return;
! delete clients[client.id];
updateClients();
client.stream.end();
});
};
! var serverConfig = {};
! if (serverConfig.ssl) {
serverConfig.mqtts = 'ssl://0.0.0.0:' + (config.port || 1883);
if (serverConfig.webSocket) {
serverConfig.mqtwss = 'wss://0.0.0.0:' + ((config.port || 1883) + 1);
}
} else {
serverConfig.mqtts = 'tcp://0.0.0.0:' + (config.port || 1883);
if (serverConfig.webSocket) {
serverConfig.mqtwss = 'ws://0.0.0.0:' + ((config.port || 1883) + 1);
}
}
! var options = {
ssl: adapter.config.certificates,
emitEvents: true // default
};
! servers = createStreamServer(serverConfig, options, function (clientStream) {
cltFunction(mqtt(clientStream, {
notData: !options.emitEvents
}));
});
! // to start
servers.listen(function () {
if (config.ssl) {
adapter.log.info('Starting MQTT (Secure) ' + (config.user ? 'authenticated ' : '') + ' server on port ' + (config.port || 1883));
if (config.webSocket) {
adapter.log.info('Starting MQTT-WebSocket (Secure) ' + (config.user ? 'authenticated ' : '') + ' server on port ' + ((config.port || 1883) + 1));
}
} else {
adapter.log.info('Starting MQTT ' + (config.user ? 'authenticated ' : '') + ' server on port ' + (config.port || 1883));
if (config.webSocket) {
adapter.log.info('Starting MQTT-WebSocket ' + (config.user ? 'authenticated ' : '') + ' server on port ' + ((config.port || 1883) + 1));
}
}
});
}
! var cnt = 0;
function readStatesForPattern(pattern) {
adapter.getForeignStates(pattern, function (err, res) {
if (!err && res) {
if (!states) states = {};
! for (var id in res) {
if (!messageboxRegex.test(id)) {
states[id] = res[id];
}
}
}
cnt--;
// If all patters answered, start client or server
if (!cnt) {
if (adapter.config.type == 'client') {
createClient(adapter.config);
} else {
createServer(adapter.config);
}
}
});
}
! function main() {
// Subscribe on own variables to publish it
if (adapter.config.publish) {
var parts = adapter.config.publish.split(',');
for (var t = 0; t < parts.length; t++) {
parts[t] = parts[t].trim();
adapter.subscribeForeignStates(parts[t].trim());
cnt++;
readStatesForPattern(parts[t]);
}
} else {
// subscribe for all variables
adapter.subscribeForeignStates('');
readStatesForPattern('');
}
// If no subscription, start client or server
if (!cnt) {
if (adapter.config.type == 'client') {
createClient(adapter.config);
} else {
createServer(adapter.config);
}
}
}` [/i][/i][/i][/i][/i][/i][/i][/i][/i][/i][/i][/i][/i][/i][/i][/i][/i][/i][/i] -
-
.
funktioniert!! :!:
Was mir noch auffällt:
Wenn ich die ioBroker Variable in Node-red einlese, sieht die msg im Debug so aus (Formatierung Payload):
{ "topic": "mqtt/0/owntracks/ruhr", "payload": "{\"cog\":-1,\"batt\":\"100\",\"lon\":\"7.92283\",\"acc\":\"...Im Debug sieht der Payload so aus (der Payload aus der selben Message oben):
{"cog":-1,"batt":"100","lon":"7.92283","acc":"65","...Verstehe ich zwar nicht, warum der anders formatiert ist, da es doch der Payload der Message ist.
Die Funktion ist trotzdem zu 100% gegeben. Nach einem JSON Node kann ich ganz normal die Daten verarbeiten
-
Darf ich fragen, wie weit ihr inzwischen gekommen seid?
Ich versuche gerade, mehrere ESP8266-12 mit ioBroker MQTT zu verbinden, um sie über eine eigene Webpage über MQTT steuern zu können. Das wäre sozusagen der "Schlüssel des IoT's" für mich, da ich dann überall kleine 8266er aufbauen könnte, die verschiedene Sensoren haben und Dinge schalten, und sich zentral über eine Webseite steuern lassen.
8266 zu ioBrokerMQTT funktioniert auch schon super.
Aber Webpage zu MQTT geht leider noch nicht. Zumindest stoße ich da auf Probleme beim eingebauten MQTT-broker (frag grad im Fehler-Forum).
Aber vielleicht habt ihr ja eine tolle Lösung, wie man viele 8266er über eine Webpage steuern kann?
So dass man über eine unabhängige Webseite in Echtzeit Signale empfangen und auch Befehle senden kann?
Das wäre toll.
