Брокер MQTT/Клиент

Драйвер позволяет подключатся к MQTT брокерам, а также самому выступать в роли брокера

Текущий релиз
8.1.0
Разработчик
bluefox, Apollon77, Matthias Kleine
Лицензия
MIT

MQTT Broker и клиент

MQTT (Message Queue Telemetry Transport) это легковесный протокол, применяемый для общения между устройствами (M2M — machine-to-machine). Он использует модель издатель-подписчик (publish/subscribe) для передачи сообщений поверх протокола TCP/IP. Центральная часть протокола это MQTT-сервер или брокер, который имеет доступ к издателю и подписчику. Этот протокол предельно примитивен: с коротким заголовком, без контроля целостности (поэтому передача реализована поверх TCP), не накладывает никаких ограничения на структуру, кодирование или схему данных. Единственное требование к данным в каждом пакете – они должны сопровождаться идентификатором информационного канала. Этот идентификатор в спецификации называется Topic Name.

Протокол MQTT требует обязательного наличия брокера данных. Это центральная идея технологии. Все устройства посылают данные только брокеру и принимают данные тоже только от него. Получив пакет, брокер рассылает его всем устройствам в сети согласно их подписке. Чтобы устройство что-то получило от брокера оно должно подписаться на топик. Топики возникают динамически по факту подписки или по факту прихода пакета с данным топиком. От подписки на топик можно отказаться. Таким образом топики представляют собой удобный механизм организации связей разных видов: один ко многим, многие к одному и многие ко многим.

Важные моменты:

  • устройства сами устанавливают связь с брокером, они могут находится за NAT и не иметь статических IP-адресов,
  • можно применить протокол SSL для шифрования трафика,
  • MQTT брокеры позволяют подключаться к ним через протокол WebSocket на 80-й порт,
  • разные брокеры могут соединяться между собой подписываясь на сообщения друг у друга.

Установка

Установка осуществляется на вкладке Драйвера странички администрирования системы. В группе драйверов Сетевые находим строчку с названием MQTT Adapter и нажимаем кнопку со значком плюса в этой строке справа.

На экране появится всплывающее окно установки драйвера, в конце установки оно автоматически закроется.

Если все прошло удачно, на вкладке Настройка драйверов появится строка mqtt.0 с установленным экземпляром драйвера.

Настройка

Как писалось выше, протокол MQTT подразумевает наличие брокера и клиентов. Сервер ioBroker может выступать как в роли брокера, так и в роли клиента. Настройка для указания режима работы - тип (сервер/брокер или клиент/подписчик) Рассмотрим каждый вариант.

Работа ioBroker в качестве MQTT-брокера

Основные настройки, если предполагается использование режима сервер/брокер, приведены на картинке:

  • Use WebSockets - если есть надобность использовать WEB-сокеты для соединения, необходимо установить эту опцию, при этом TCP-server будет работать параллельно с WebSocket-сервером,
  • Порт -порт для установки соединения по TCP (значение по-умолчанию 1883), сервер WebSocket (см. опция выше) запускается по порту +1 (по-умолчанию 1884),
  • SSL - данная опция используется, если необходимо шифровать трафик (TCP или WebSocket), при этом необходимо указать сертификаты - просто выбрать из списка заранее установленные (указываются в системных настройках, см. описание драйвера администрирования системы),
  • настройки аутентификации (имя пользователя и пароль) - указываются, если необходима аутентификация конкретного пользователя, данная настройка всегда используется вместе с опцией SSL-шифрования (чтобы не передавать пароли в открытом виде через незащищенное соединение).
  • Маска для собственных значений - шаблон (или несколько через запятую) для фильтрации переменных ioBroker, которые будут отправляться клиентам, можно использовать специальные символы чтобы указать группу сообщения (к примеру, memRSS, mqtt.0 - могут передаваться все переменные состояния памяти всех драйверов и все переменные экземпляра драйвера mqtt.0)
  • Отсылать только изменения - отправка данных клиенту будет произведена только в случае изменения переменной (если состояние просто обновилось - значение не поменялось, клиенту сообщение не будет отправлено), от клиента будет принято любое сообщение, даже если значение не изменилось,
  • Выдавать собственные значения при старте - для каждого успешного соединения с клиентом ему будут переданы все известные состояния (определяются маской состояний) - для того, чтобы сообщить клиенту текущее состояние ioBroker,
  • Публиковать состояния при подписке - сразу после подписки клиенту будет отправлено значение переменной, на которую он подписан (при первом старте или рестарте клиент получит значения переменных, на которые он подписан, можно использовать для инициализации переменных),
  • Префикс для всех значений - если указано значение, то оно будет добавляться как префикс к каждому отправленному топику, к примеру, если указать iobroker/, то все отправленные топики примерно следующего содержания: iobroker/mqtt/0/connected,
  • Вывод лога для каждого изменения - в лог-файле будет отображаться отладочная информация по каждому изменению,
  • Посылать не только команды, но и состояния (ack=true) - если опция не активна, то клиенту будут отправляться только переменные/команды с ack=false, если флаг установлен, то будут переданы переменные не зависимо от состояния ack (false или true),
  • Максимальная длина имени топика - максимальное кол-во символов для описания топика, включая служебные.

В качестве примера рассмотрим обмен данными между клиентом на базе платы arduino и брокером - экземпляр драйвера mqtt.0 системы ioBroker.

  • клиент - плата для разработки arduino UNO + ethernet shield на базе чипа W5100,
  • для работы с платой ethernet используется стандартная библиотека, для работы с MQTT библиотека Pubsubclient,
  • датчик AM2302 (температура и влажность) подключен на pin_8 для опроса используется библиотека с DHTlib с ресурса github.com,
  • светодиод led_green подключен на pin_9, управление в дискретном режиме вкл/откл,
  • брокер - система ioBroker драйвер mqtt.

Формат топиков обмена данными:

  • example1/send_interval - клиент подписан на изменение интервала отправки показаний температуры и влажности (значение int в секундах)
  • example1/temp - клиент публикует с заданным интервалом значение температуры с датчика DHT22 (тип float),
  • example1/hum -клиент публикует с заданным интервалом значение влажности с датчика DHT22 (тип float),
  • example1/led -клиент подписан на изменение состояния светодиода (тип text on/off или 0/1 или true/false).

Настройки драйвера будут следующие:

Подключение по TCP (WebSocket не нужен), порт по-умолчанию 1883. Клиент внутри локальной сети, поэтому шифровать трафик и проводить аутентификацию пользователя нет необходимости. Отсылать будем только изменения, так как клиент подписан на интервал отправки показаний и состояние светодиода, получать информацию только об обновлении (без изменения значения) переменной нет смысла. Публиковать состояния при подписке - отметим эту опцию, так как при первом подключении (или подключении после обрыва соединения) клиента он должен знать состояния переменных, на которые он подписан (какой текущий интервал отправки и должен ли быть включен светодиод). Настройку отсылать переменные с ack=true или false тоже стоит отметить, так как переменную (на которую подписан клиент) может изменить любой драйвер/скрипт/VIS и все изменения надо отправлять клиенту. Полный код для платы arduino будет выглядеть так:

//Подключаем библиотеки
#include 
#include 
#include  //https://github.com/knolleary/pubsubclient
#include           //https://github.com/RobTillaart/Arduino/tree/master/libraries/DHTlib
//Настройки сети
byte mac[] = {  0xAB, 0xBC, 0xCD, 0xDE, 0xEF, 0x31 };
byte ip[] = { 192, 168, 69, 31 }; //IP-адрес платы arduino
byte mqttserver[] = { 192, 168, 69, 51 }; //IP-адрес сервера ioBroker
EthernetClient ethClient;
void callback(char* topic, byte* payload, unsigned int length);
PubSubClient client(mqttserver, 1883, callback, ethClient);
//Глобальные переменные
#define LED_pin 9
unsigned int send_interval = 10; //интервал отправки показаний на сервер по-умолчанию 10 секунд
unsigned long last_time = 0; //текущее время для таймера
dht DHT;
#define DHT22_PIN 8
char buff[20];
//Функция обработки входящих соединений - прием данных по подписке
void callback(char* topic, byte* payload, unsigned int length) {
  Serial.println ("");
  Serial.println ("-------");
  Serial.println ("New callback of MQTT-broker");
  //преобразуем тему(topic) и значение (payload) в строку
  payload[length] = '\0';
  String strTopic = String(topic);
  String strPayload = String((char*)payload);
  //Исследуем что "прилетело" от сервера по подписке:
  //Изменение интервала опроса
  if (strTopic == "example1/send_interval") {
    int tmp = strPayload.toInt();
    if (tmp == 0) {
      send_interval = 10;
    } else {
      send_interval = strPayload.toInt();
    }
  }
  //Управление светодиодом
  if (strTopic == "example1/led") {
    if (strPayload == "off" || strPayload == "0" || strPayload == "false") digitalWrite(LED_pin, LOW);
    if (strPayload == "on" || strPayload == "1" || strPayload == "true") digitalWrite(LED_pin, HIGH);
  }  
  Serial.print (strTopic);
  Serial.print (" ");
  Serial.println (strPayload);
  Serial.println ("-------");
  Serial.println ("");  
}
void setup() {
  Serial.begin(9600);
  Serial.println("Start...");
  //стартуем сетевое подключение
  Ethernet.begin(mac, ip);
  Serial.print("IP: ");
  Serial.println(Ethernet.localIP());
  //Инициализируем порты ввода-вывода, прописываем начальные значения
  pinMode(LED_pin, OUTPUT);
  digitalWrite(LED_pin, LOW); //при светодиод отключен
}
void loop() {
  //Если соединение MQTT неактивно, то пытаемся установить его и опубликовать/подписаться
  if (!client.connected()) {
    Serial.print("Connect to MQTT-boker...  ");
    //Подключаемся и публикуемся/подписываемся
    if (client.connect("example1")) {
      Serial.println("success");
      //Значение с датчиков
      if (DHT.read22(DHT22_PIN)==DHTLIB_OK) {
        dtostrf(DHT.humidity, 5, 2, buff);
        client.publish("example1/hum", buff);
        dtostrf(DHT.temperature, 5, 2, buff);
        client.publish("example1/temp", buff);
      }
      //Подписываемся на интервал опроса
      client.subscribe("example1/send_interval");
      //Подписываемся на переменную управления светодиодом
      client.subscribe("example1/led");
    } else {
      //Если не подключились, ждем 10 секунд и пытаемся снова
      Serial.print("Failed, rc=");
      Serial.print(client.state());
      Serial.println(" try again in 10 seconds");
      delay (10000);
    }
  //Если соединение активно, то отправляем данные на сервер с заданным интервалом времени
  } else {
    if(millis() > (last_time + send_interval*1000)) {
      last_time = millis();
      if (DHT.read22(DHT22_PIN)==DHTLIB_OK) {
        dtostrf(DHT.humidity, 5, 2, buff);
        client.publish("example1/hum", buff);
        dtostrf(DHT.temperature, 5, 2, buff);
        client.publish("example1/temp", buff);
      }
    }
  }
  //Проверка входящих соединений по подписке
  client.loop();
}

Работа ioBroker в качестве MQTT-клиента

Чтобы экземпляр драйвера MQTT заработал как клиент/подписчик - нужно в настройках выбрать соответствующий тип. При этом набор настроек немного поменяется:

  • Настройки соединения - указывается URL и порт брокера (если необходимо шифровать трафик, то указывается SSL) - настройки для подключения к брокеру,
  • Настройки аутентификации - имя пользователя и пароль, если брокер требует аутентификацию (уместно использовать SSL-шифрование, чтобы не передавать пароль в открытом виде),
  • Patterns - маска для переменных, на которые клиент подписывается (переменные брокера), значения перечисляются через запятую, для указания множества используется символ # (решетка),
  • Маска для собственных значений - фильтр переменных, которые необходимо публиковать (переменные клиента), значения перечисляются через запятую, для указания множества используется символ * (звездочка),
  • Отсылать только изменения - клиент будет публиковать только переменные, которые изменили значение (согласно маски),
  • Выдавать собственные значения при старте - если эту опцию отметить, то будут публиковаться все состояния (согласно маски) каждый раз, когда устанавливается соединение, чтобы объявить доступные собственные переменные и их значения,
  • Префикс для всех значений - если указано значение, то оно будет добавляться как префикс к каждому публикуемому топику, к примеру, если указать client1/, то все публикуемые топики будут примерно следующего содержания: client1/javascript/0/cubietruck,
  • Вывод лога для каждого изменения - в лог-файле будет отображаться отладочная информация по каждому изменению,
  • Посылать не только команды, но и состояния (ack=true) - если данная опция не отмечена, то брокеру отправляются только переменные/команды с ack=false, если опцию отметить, то будут отправляться все данные, независимо от ack=true или ack=false,
  • Максимальная длина топика - максимальное кол-во символов для описания топика, включая служебные.

Примеры для задания маски подписки на переменные (patterns). Рассмотрим топики:

  • "Sport"
  • "Sport/Tennis"
  • "Sport/Basketball"
  • "Sport/Swimming"
  • "Sport/Tennis/Finals"
  • "Sport/Basketball/Finals"
  • "Sport/Swimming/Finals"

Если необходимо подписаться на определенное множество топиков, можно использовать символы # (решетка) или + (знак плюс).

  • "Sport/Tennis/#" (подписка только на "Sport/Tennis" и "Sport/Tennis/Finals")
  • "Sport/Tennis/+" (подписка только на  "Sport/Tennis/Finals", но не "Sport/Tennis")

Для JMS топиков, если нужно подписаться на все топики "Finals", можно использовать символы # (решетка) или + (знак плюс)

  • "Sport/#/Finals"
  • "Sport/+/Finals"

Для MQTT топиков, если нужно подписаться на все топики "Finals", можно использовать символ + (знак плюс)

  • "Sport/+/Finals"

В качестве примера рассмотрим обмен данными между двумя системами ioBroker. Есть работающая система ioBroker на плате BananaPi (IP-адрес 192.168.69.51), на ней запущен MQTT-драйвер в режиме сервер/брокер из примера выше. К серверу подключается клиент, который публикует данные с датчика DHT22 - температуру и влажность, а так же подписывается на переменные интервал передачи показаний и состояние светодиода (включить/отключить) - так же из примера выше. Вторая работающая система ioBroker на плате Cubietruck, на ней запустим MQTT-драйвер в режиме клиент/подписчик. Он подпишется на переменные температура и влажность брокера (который в свою очередь получает от другого клиента) и будет публиковать все переменные скрипта - состояние АКБ платы (только изменения). Настройки клиента будут примерно следующие:

Тип соединения - клиент/подписчик, указывается IP-адрес брокера и порт (по-умолчанию 1883). Шифрование трафика и аутентификация не нужны.

Маска для подписки (Patterns) - mqtt/0/example1/hum,mqtt/0/example1/temp - клиент подписывается только на температуру и влажность (значения через запятую без пробелов).

Маска данных для публикации - javascript.0.cubietruck.battery.* - публикуются все переменные скрипта cubietruck в группе battery драйвера javascript.0.

Отсылать только изменения - отправляем переменные состояния АКБ (нет смысла отправлять, если значение не изменилось). Выдавать собственные значения при старте - при старте драйвера, клиент сразу опубликует все переменные согласно маске - даже если они нулевые или пустые, чтобы создать переменные в брокере. Посылать данные с ack=false - переменные работы АКБ обновляются драйвером javascript, поэтому они всегда ack=false.

Результат работы на стороне клиента (данные температуры и влажности другого клиента - см. пример выше):

Результат работы со стороны брокера (данные состояния АКБ клиента):

Применение - шлюз протоколов MQTT - ModBus RTU

Драйвер MQTT можно использовать как шлюз различных протоколов, чтобы подключить новые устройства в систему ioBroker или любую другую. Универсальной базой для разработки подобных решений являются платы arduino. В сети много примеров, библиотек и наработок. Огромное сообщество работает с этими контроллерами и в систему интегрированы множество устройств и оборудования. Для примера, рассмотрим распространенный промышленный протокол ModBus. В системе ioBroker имеется драйвер для работы с ним - версии ModBus TCP (по сети ethernet). Множество датчиков, контроллеров и исполнительных устройств работают физически по сети RS-485/232 и протоколу ModBus RTU. Чтобы интегрировать их можно применить шлюз MQTT - ModBus RTU на базе платформы arduino.

Рассмотрим пример.

Имеется датчик температуры и влажности (для теста на базе платы arduino pro mini и сенсора DHT22),  который выдает данные по ModBUS RTU:

  • Порт UART (можно с помощью микросхемы MAX485 преобразовать в интерфейс RS-485) работает на скорости 9600 с параметрами 8E1 (1 start bit, 8 data bits, 1 Even parity bit, 1 stop bit)
  • Адрес ModBus - 10
  • температура - адрес 0 значение, умноженное на 10 (read function 3)
  • влажность - адрес 1 значение, умноженное на 10 (read function 3)
  • PWM LED - адрес 2 значение 0...1023 для проверки функции записи (write function 6)

Схема соединения:

by Fritzing

Код для контроллера arduino pro mini получается следующий:

#include  //https://github.com/RobTillaart/Arduino/tree/master/libraries/DHTlib
#include  //https://code.google.com/archive/p/simple-modbus/
#include  //https://github.com/PaulStoffregen/MsTimer2
//регистры modbus
enum {
  TEMP,
  HUM,
  PWM,
  TEST,
  HOLDING_REGS_SIZE
};
#define ID_MODBUS 10 //адрес modbus slave устройства
unsigned int holdingRegs[HOLDING_REGS_SIZE]; //массив регистров modbus
//датчик температуры и влажности DHT22
dht DHT;
#define DHT22_PIN 2
#define LED 9 //светодиод подключен на PWM-пин 9
void setup() {
  //конфигурируем modbus
  modbus_configure(&Serial, 9600, SERIAL_8E1, ID_MODBUS, 0, HOLDING_REGS_SIZE, holdingRegs);
  holdingRegs[TEST] = -157; //для теста отрицательных значений
  //инициализируем таймер 2 секунды обновления данных в регистрах температуры и влажности
  MsTimer2::set(2000, read_sensors);
  MsTimer2::start(); //запускаем таймер
  pinMode(LED, OUTPUT); //инициализация порта светодиода
}
//функция, запускаемая по таймеру каждые 2 секунды
void read_sensors() {
  if (DHT.read22(DHT22_PIN)==DHTLIB_OK) { //если данные с датчика DHT22 удалось прочитать
    //записываем в регистр влажности целочисленное значение
    holdingRegs[HUM] = 10 * DHT.humidity;
    //записываем в регистр температуры целочисленное значение
    holdingRegs[TEMP]= 10 * DHT.temperature;
  } else {
    //если не удалось прочитать данные с датчика DHT22, записываем в регистры нули
    holdingRegs[HUM] = 0;
    holdingRegs[TEMP] = 0;
  }
}
void loop() {
  modbus_update(); //обновляем данные modbus
  //данные из регистра управления светодиодом передаем в ШИМ (битовый сдвиг на 2 разряда)
  analogWrite(LED, holdingRegs[PWM]>>2);
}
  • порт (выбрать из списка к какому порту подключен serial платы ардуино);
  • скорость и прочие параметры - 9600 8E1;
  • slave id: 10, чтение: функция №3 read holding registers, начальный адрес: 0, число регистров: 3,
  • slave id: 10, запись: функция №6 write single register, начальный адрес: 2,

Ответ в программе при чтении должен быть примерно следующий:

Ответ в программе при записи:

Теперь настроем сам шлюз и подключим его в iobroker Шлюз будет на базе платформы arduino MEGA 2560 с ethernet shield - клиент MQTT, брокер - экземпляр драйвера mqtt.0 системы ioBroker. Выбор именно MEGA 2560 обусловлен тем, что на этой плате более одного UART-порта, соответственно нулевой Serial0 (pin_0 (RX) и зшт_1 (TX)) или просто Serial - используем для вывода отладочных сообщений, а Serial1 (pin_19 (RX) и pin_18 (TX)) - для работы с slave-устройством по ModBus.

  • клиент - плата для разработки arduino MEGA 2560 + ethernet shield на базе чипа W5100;
  • для работы с платой ethernet используется стандартная библиотека, для работы с MQTT библиотека Pubsubclient;
  • для опроса по modbus используется библиотека SimpleModbus версии master;
  • опрос по порту UART (просто соединить RX порт master, TX порт slave и соответственно TX порт master, RX порт slave), порт управления передачей не используется (он для RS-485);
  • параметры порта: скорость 9600, 8Е1;
  • адрес slave-устройства 10, функция чтения №3 (read holding registers), функция записи №6 (write single register);
  • брокер - система ioBroker драйвер mqtt.

Формат топиков обмена данными:

  • modbusgateway/send_interval - клиент подписан на изменение интервала отправки показаний температуры и влажности (значение int в секундах)
  • modbusgateway/temp - клиент публикует с заданным интервалом значение температуры с датчика DHT22 (тип float),
  • modbusgateway/hum -клиент публикует с заданным интервалом значение влажности с датчика DHT22 (тип float),
  • modbusgateway/led -клиент подписан на изменение состояния светодиода (ШИМ управление, значения 0...1024).

Схема соединений получится примерно такая:

By Fritzing[/caption] Для теста slave-устройство запитаем от master-устройства. Master в свою очередь, будет работать от USB-порта, по которому ведется отладка (Serial0). Настройки драйвера будут следующие:

Подключение по TCP (WebSocket не нужен), порт по-умолчанию 1883. Клиент внутри локальной сети, поэтому шифровать трафик и проводить аутентификацию пользователя нет необходимости. Отсылать будем только изменения, так как клиент подписан на интервал отправки показаний и состояние светодиода, получать информацию только об обновлении (без изменения значения) переменной нет смысла. Публиковать состояния при подписке - отметим эту опцию, так как при первом подключении (или подключении после обрыва соединения) клиента он должен знать состояния переменных, на которые он подписан (какой текущий интервал отправки и должен ли быть включен светодиод). Настройку отсылать переменные с ack=true или false тоже стоит отметить, так как переменную (на которую подписан клиент) может изменить любой драйвер/скрипт/VIS и все изменения надо отправлять клиенту.

Полный код для платы arduino MEGA 2560 будет выглядеть так:

//Подключаем библиотеки
#include 
#include 
#include        //https://github.com/knolleary/pubsubclient
#include  //https://code.google.com/archive/p/simple-modbus/
//Настройки сети
byte mac[] = {  0xAB, 0xBC, 0xCD, 0xDE, 0xEF, 0x31 };
byte ip[] = { 192, 168, 69, 31 }; //IP-адрес платы arduino
byte mqttserver[] = { 192, 168, 69, 51 }; //IP-адрес сервера ioBroker
//Объекты/переменные/функции ethernet и MQTT
EthernetClient ethClient;
void callback(char* topic, byte* payload, unsigned int length);
PubSubClient client(mqttserver, 1883, callback, ethClient);
//Переменные временного интервала и буфер для отправки данных по MQTT
unsigned int send_interval = 10; //интервал отправки показаний на сервер по-умолчанию 10 секунд
unsigned long last_time = 0; //текущее время для таймера
char buff[20];
//Параметры порта Serial1 (19 (RX) and 18 (TX))
#define baud 9600       //скорость порта
#define timeout 1000    //интервал времени ожидания ответа /(мс)
#define polling 200     //интервал опроса (мс)
#define retry_count 10  //кол-во повторов при неудачном опросе
#define TxEnablePin 0   //пин управления передачей для RS485 (в UART не используется = 0)
// Общая сумма доступной памяти на master устройстве для хранения данных
// Из слейва запрашиваем 4 регистра, в массиве regs должно быть не меньше 4х ячеек
#define TOTAL_NO_OF_REGISTERS 4
//Массив регистров для работы с modbus (хранение, чтение, запись)
unsigned int regs[TOTAL_NO_OF_REGISTERS];
//Определение пакетов mmodbus
enum {
  TEMP,
  HUM,
  PWM,
  TEST,
  TOTAL_NO_OF_PACKETS //всегда последней записью
};
//Создание массива пакетов modbus
Packet packets[TOTAL_NO_OF_PACKETS];

//Функция обработки входящих соединений - прием данных по подписке
void callback(char* topic, byte* payload, unsigned int length) {
  Serial.println ("");
  Serial.println ("-------");
  Serial.println ("New callback of MQTT-broker");
  //преобразуем тему(topic) и значение (payload) в строку
  payload[length] = '\0';
  String strTopic = String(topic);
  String strPayload = String((char*)payload);
  //Исследуем что "прилетело" от сервера по подписке:
  //Изменение интервала опроса
  if (strTopic == "modbusgateway/send_interval") {
    int tmp = strPayload.toInt();
    if (tmp == 0) {
      send_interval = 10;
    } else {
      send_interval = strPayload.toInt();
    }
  }
  //Управление светодиодом значение int от 0 до 1023
  if (strTopic == "modbusgateway/led") {
    int tmp = strPayload.toInt();
    if (tmp >= 0 && tmp <=1023) {
      regs[2] = tmp;
    }
  }  
  Serial.print (strTopic);
  Serial.print (" ");
  Serial.println (strPayload);
  Serial.println ("-------");
  Serial.println ("");  
}

void setup() {
  Serial.begin(9600);
  Serial.println("Start...");
  //стартуем сетевое подключение
  Ethernet.begin(mac, ip);
  Serial.print("IP: ");
  Serial.println(Ethernet.localIP());
  //Инициализируем все пакеты modbus
  modbus_construct(&packets[TEMP], 10, READ_HOLDING_REGISTERS, 0, 1, 0); //температура
  modbus_construct(&packets[HUM], 10, READ_HOLDING_REGISTERS, 1, 1, 1); //влажность
  modbus_construct(&packets[PWM], 10, PRESET_SINGLE_REGISTER, 2, 1, 2); //данные ШИМ для светодиода
  modbus_construct(&packets[TEST], 10, READ_HOLDING_REGISTERS, 3, 1, 3); //тест
  //Конфигурируем соединение modbus (порт serial1, скорость и пр.)
  modbus_configure(&Serial1, baud, SERIAL_8E1, timeout, polling, retry_count, TxEnablePin, packets, TOTAL_NO_OF_PACKETS, regs);
}

void loop() {
  //Обоновляем данные в регистрах modbus
  modbus_update();
  //Если соединение MQTT неактивно, то пытаемся установить его и опубликовать/подписаться
  if (!client.connected()) {
    Serial.print("Connect to MQTT-boker...  ");
    //Подключаемся и публикуемся/подписываемся
    if (client.connect("modbusgateway")) {
      Serial.println("success");
      //Значение с датчиков температуры и влажности
      dtostrf((float)regs[0]/10, 5, 1, buff);
      client.publish("modbusgateway/temp", buff);
      dtostrf((float)regs[1]/10, 5, 1, buff);
      client.publish("modbusgateway/hum", buff);
      //Подписываемся на интервал опроса
      client.subscribe("modbusgateway/send_interval");
      //Подписываемся на переменную управления светодиодом
      client.subscribe("modbusgateway/led");
    } else {
      //Если не подключились, пытаемся снова
      Serial.print("Failed, rc=");
      Serial.print(client.state());
      Serial.println(" try again");
      delay(10000);
    }
  //Если соединение активно, то отправляем данные на сервер с заданным интервалом времени
  } else {
    if(millis() > (last_time + send_interval*1000)) {
      last_time = millis();
      //Значение с датчиков температуры и влажности
      dtostrf((float)regs[0]/10, 5, 1, buff);
      client.publish("modbusgateway/temp", buff);
      dtostrf((float)regs[1]/10, 5, 1, buff);
      client.publish("modbusgateway/hum", buff);
    }
  }
  //Проверка входящих соединений по подписке
  client.loop();
}

Применение - подключение мобильных клиентов

В последнее время протокол MQTT получил большое распространение ввиду простоты, экономии трафика и хорошей проработке библиотек под разные платформы. Существует множество программ для работы с MQTT на мобильных устройствах, к примеру IoT MQTT Dashboard. С помощью этой программы можно подключиться к MQTT-брокеру в локальной сети или в сети интернет. Рассмотрим пример - в роли брокера будет выступать система ioBroker, к которой по MQTT будет подключаться клиент - приложение IoT MQTT Dashboard. В данном примере будем управлять светом с помощью контроллера MegaD-328, который подключен к ioBroker с помощью драйвера MegaD. Управляет непосредственно реле (MegaD порт P7) света в холле специальный скрипт, который подписывается на состояние порта-кнопки P0 и состояние MQTT-переменной mqtt.0.remotectrl.light.hall, которую будет публиковать мобильный клиент. При этом скрипт переключает состояние порта, привязанного к реле (порт P7), т.е. инвертирует его.

Получается, что при каждом нажатии на кнопку, электрически подключенную к порту P0 (вылавливается состояние true) и при каждой публикации переменной mqtt.0.remotectrl.light.hall значением так же true, порт P7 будет включать или выключать свет. Текст скрипта будет примерно такой:

//Управление освещением в зале с помощью кнопки порт p0 контроллера MegaD драйвер экземпляр megad.0
on({id: 'megad.0.p0_P0', change: 'any'}, function (obj) {
   if (obj.newState.val !== '' || typeof obj.newState.val !== "undefined"){
     if (obj.newState.val === true) {
       if (getState('megad.0.p7_P7').val === true) {
         setState('megad.0.p7_P7', false);
       } else {
         setState('megad.0.p7_P7', true);
       }
     }
   }
});
//Управление освещением в зале удаленно по MQTT топик "mqtt.0.remotectrl.light.hall"
on({id: 'mqtt.0.remotectrl.light.hall', change: 'any'}, function (obj) {
  if (obj.newState.val !== '' || typeof obj.newState.val !== "undefined"){
    if (obj.newState.val === true) {
      if (getState('megad.0.p7_P7').val === true) {
        setState('megad.0.p7_P7', false);
      } else {
        setState('megad.0.p7_P7', true);
      }
    }
  }
});

Применение - работа с облачными серверами

Описанный выше пример имеет ряд недостатков. Во-первых, не всегда мобильный клиент может находиться в одной локальной сети с сервером ioBroker, во-вторых, даже если осуществить проброс портов в интернет и защитить соединение, не всегда сам сервер ioBroker может принять входящие подключения (находится за NAT, к которому нет доступа для настройки). В глобальной сети много различных сервисов, которые поддерживают MQTT - платных и бесплатных, к примеру отправка погодных данных, геолокации и пр. Некоторые сервисы могут выступать в качестве брокера протокола MQTT и их можно использовать как шлюз (мост) для вывода данных из ioBroker в глобальную сеть или для получения данных в ioBroker. В качестве примера рассмотрим работу связки:

  • сервер/брокер - сервис cloudmqtt.com (есть бесплатный тариф),
  • клиент/подписчик - система ioBroker с выходом в сеть интернет, публикует данные температуры и влажности (см. пример выше), публикует реальное состояние портов P7-P13 (реле MegaD драйвера megad.0 - управление освещением), подписка на переменные удаленного управления светом (экземпляр драйвера mqtt.0),
  • клиент/подписчик - приложение IoT MQTT Dashboard для удаленной работы - подписка на данные сенсора температуры и влажности, подписка на реальное состояние портов P7-P13 (реле MegaD драйвера megad.0), публикация переменных удаленного управления светом (экземпляр драйвера mqtt.0).

В итоге получается следующая структура:

Связка драйвер mqtt.1 (брокер) - Arduino UNO + Ethernet + DHT22 (клиент) как в примере выше с несколькими изменениями. Настройки экземпляра драйвера mqtt.1:

Код для платформы arduino:

//Подключаем библиотеки
#include 
#include 
#include  //https://github.com/knolleary/pubsubclient
#include           //https://github.com/RobTillaart/Arduino/tree/master/libraries/DHTlib
//Настройки сети
byte mac[] = {  0xAB, 0xBC, 0xCD, 0xDE, 0xEF, 0x31 };
byte ip[] = { 192, 168, 69, 31 }; //IP-адрес платы arduino
byte mqttserver[] = { 192, 168, 69, 51 }; //IP-адрес сервера ioBroker
EthernetClient ethClient;
void callback(char* topic, byte* payload, unsigned int length);
PubSubClient client(mqttserver, 1884, callback, ethClient);
//Глобальные переменные
unsigned int send_interval = 10; //интервал отправки показаний на сервер по-умолчанию 10 секунд
unsigned long last_time = 0; //текущее время для таймера
dht DHT;
#define DHT22_PIN 8
char buff[20];

//Функция обработки входящих соединений - прием данных по подписке
void callback(char* topic, byte* payload, unsigned int length) {
  Serial.println ("");
  Serial.println ("-------");
  Serial.println ("New callback of MQTT-broker");
  //преобразуем тему(topic) и значение (payload) в строку
  payload[length] = '\0';
  String strTopic = String(topic);
  String strPayload = String((char*)payload);
  //Исследуем что "прилетело" от сервера по подписке:
  //Изменение интервала опроса
  if (strTopic == "example2/send_interval") {
    int tmp = strPayload.toInt();
    if (tmp == 0) {
      send_interval = 10;
    } else {
      send_interval = strPayload.toInt();
    }
  }
  Serial.print (strTopic);
  Serial.print (" ");
  Serial.println (strPayload);
  Serial.println ("-------");
  Serial.println ("");  
}
void setup() {
  Serial.begin(9600);
  Serial.println("Start...");
  //стартуем сетевое подключение
  Ethernet.begin(mac, ip);
  Serial.print("IP: ");
  Serial.println(Ethernet.localIP());
  //Инициализируем порты ввода-вывода, прописываем начальные значения
}
void loop() {
  //Если соединение MQTT неактивно, то пытаемся установить его и опубликовать/подписаться
  if (!client.connected()) {
    Serial.print("Connect to MQTT-boker...  ");
    //Подключаемся и публикуемся/подписываемся
    if (client.connect("example2")) {
      Serial.println("success");
      //Значение с датчиков
      if (DHT.read22(DHT22_PIN)==DHTLIB_OK) {
        dtostrf(DHT.humidity, 5, 2, buff);
        client.publish("example2/hum", buff);
        dtostrf(DHT.temperature, 5, 2, buff);
        client.publish("example2/temp", buff);
      }
      //Подписываемся на интервал опроса
      client.subscribe("example2/send_interval");
    } else {
      //Если не подключились, ждем 10 секунд и пытаемся снова
      Serial.print("Failed, rc=");
      Serial.print(client.state());
      Serial.println(" try again in 10 seconds");
      delay (10000);
    }
  //Если соединение активно, то отправляем данные на сервер с заданным интервалом времени
  } else {
    if(millis() > (last_time + send_interval*1000)) {
      last_time = millis();
      if (DHT.read22(DHT22_PIN)==DHTLIB_OK) {
        dtostrf(DHT.humidity, 5, 2, buff);
        client.publish("example2/hum", buff);
        dtostrf(DHT.temperature, 5, 2, buff);
        client.publish("example2/temp", buff);
      }
    }
  }
  //Проверка входящих соединений по подписке
  client.loop();
}
  • тип соединения - клиент/подписчик,
  • настройки соединения - указываем URL выданный в панели управления cloudmqtt.com порт будем использовать 22809, который работает с SSL,
  • в настройках аутентификации указываем имя пользователя и пароль,
  • patterns - наш клиент ioBroker будет подписан на все топики, что есть в облаке, поэтому указываем здесь символ решетка (#), можно использовать маску и подписаться выборочно,
  • маска для собственных значений - клиент будет публиковать на сервер значения температуры/влажности и состояния всех портов megaD (порты с реле P7-P13),в этом поле через запятую указываем необходимые переменные: mqtt.1.example2.hum,mqtt.1.example2.temp,megad.0.p7_P7,megad.0.p8_P8,megad.0.p9_P9,megad.0.p10_P10,megad.0.p11_P11,megad.0.p12_P12,megad.0.p13_P13,
  • отсылать только изменения - ставим галочку, публиковать будем только изменения,
  • выдавать собственные значения при старте - так же указываем для создания топиков,
  • посылать не только команды, но и состояния (ack=true) - необходимо отметить эту настройку, так как данные температуры/влажности обновляются драйвером mqtt (ack=true).

Настройки сохраняем, убеждаемся, что соединение установилось (можно на панели управления cloudmqtt.com посмотреть лог сервера). Через некоторое время появятся данные (в панели ссылка WebsocketUI):

В итоге остается только настроить мобильный клиент, к примеру IoT MQTT Dashboard. Создаем новое подключение:

Создаем топики для публикации (на примере освещения зала - порт P7 MegaD):

Создаем топики подписок (температура, влажность, освещение зала порт P7 MegaD):

В итоге dashboard может выглядеть примерно так:

После создания публикаций на мобильном устройстве, в экземпляре драйвера mqtt.0 системы ioBroker должны появиться переменные управления светом, которые следует использовать в скрипте управления освещением (см. пример выше):

Поздравляем! Теперь вы сможете управления системой ioBroker и получать от нее данные через облачный сервис!

Changelog

8.1.0 (2026-09-06)

  • (@GermanBluefox) Added the option "Interpretation of the received payload": besides the previous behavior ("Full") a received message can now be stored without unpacking a JSON into an ioBroker state object ("Numbers and booleans only"), or completely unparsed as string ("None")

8.0.1 (2026-08-28)

  • (@GermanBluefox) The Blockly block is now built from TypeScript sources in src-blockly/ into admin/blockly.js, with the translations in src-blockly/i18n/
  • (@GermanBluefox) Fixed: the Blockly block produced no code at all in newer Blockly editors. Since Blockly 10 a generator has to be registered in Blockly.JavaScript.forBlock, and the editor migrates its own generators there before it loads an adapter's block

8.0.0 (2026-08-28)

  • (@GermanBluefox) Added MQTT 5.0 support. In server mode every client is answered with the protocol level it announced, so MQTT 5 and MQTT 3.1.1 clients can share one instance. Includes Reason codes, topic aliases, the subscription options "No Local", "Retain Handling" and "Retain As Published", subscription identifiers, message expiry, will delay, maximum packet size, receive maximum, shared subscriptions and the session expiry interval. MQTT 3.1.1 clients are not affected by the new rules
  • (@GermanBluefox) Client mode: the MQTT version (3.1 / 3.1.1 / 5.0) can now be selected. If the broker refuses it, the adapter falls back to MQTT 3.1.1
  • (@GermanBluefox) Added MQTT 5 enhanced authentication with SCRAM-SHA-256 (RFC 7677) in server mode. The password is no longer sent over the connection; both sides prove that they know it. It uses the configured user name and password, needs no configuration and does not change anything for clients that log in the usual way
  • (@GermanBluefox) Fixed: a rejected login could lose its CONNACK, because the socket was torn down before the answer had left it. The client saw a closed connection instead of the reason
  • (@GermanBluefox) Replaced the unmaintained mqtt-connection package with an own connection layer on mqtt-packet, which removes a second, outdated copy of mqtt-packet from the dependency tree

7.1.3 (2026-08-28)

  • (@Tarvion) Fixed: a value received for a topic that was published from a state ID containing "#", "+" or a space (e.g. the Shelly IDs like shelly.0.SHCB-1#3494546B9BEC#1) is now written back to that state instead of creating a new state in the adapter's own namespace
  • (@GermanBluefox) Fixed: the resolution above now also corrects an instance that already carries the wrongly created mqtt.<n>.* state from an earlier version. Such a state was published at the start and claimed the topic before any message arrived. It is logged once as a leftover and can be deleted

7.1.2 (2026-07-26)

  • (@GermanBluefox) Added a "Replace dots in topic names with underscore" setting to keep MQTT topics that contain dots (e.g. Wolf heating via ism7mqtt) as a single object level (#413)
  • (@GermanBluefox) Client mode: added loop protection so a value just received from the broker is not published straight back, plus a new "Do not subscribe to own states" option (#414)

License

The MIT License (MIT)

Copyright (c) 2014-2026, bluefox dogafox@gmail.com

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.