Merge remote-tracking branch 'origin/master' into enhanced_device

# Conflicts:
#	src/lib/ebus/device.cpp

.github/FUNDING.yml

@@ -0,0 +1,12 @@
+# These are supported funding model platforms
+
+github: # Replace with up to 4 GitHub Sponsors-enabled usernames e.g., [user1, user2]
+patreon: # Replace with a single Patreon username
+open_collective: # Replace with a single Open Collective username
+ko_fi: # Replace with a single Ko-fi username
+tidelift: # Replace with a single Tidelift platform-name/package-name e.g., npm/babel
+community_bridge: # Replace with a single Community Bridge project-name e.g., cloud-foundry
+liberapay: # Replace with a single Liberapay username
+issuehunt: # Replace with a single IssueHunt username
+otechie: # Replace with a single Otechie username
+custom: ['https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=5Y52QSKVXLYQG']

CMakeLists.txt

@@ -53,6 +53,7 @@ check_function_exists(pthread_setname_np HAVE_PTHREAD_SETNAME_NP)
 check_function_exists(pselect HAVE_PSELECT)
 check_function_exists(ppoll HAVE_PPOLL)
 check_include_file(linux/serial.h HAVE_LINUX_SERIAL -DHAVE_LINUX_SERIAL=1)
+check_include_file(dev/usb/uftdiio.h HAVE_FREEBSD_UFTDI -DHAVE_FREEBSD_UFTDI=1)
 check_function_exists(argp_parse HAVE_ARGP)
 if(NOT HAVE_ARGP)
   find_library(LIB_ARGP argp)

ChangeLog.md

@@ -1,9 +1,41 @@
-# next version
+# 3.4 (2019-10-?)
+
+## Bug Fixes
+* fix for always enabled "--mqttchanges" option
+* fix for dynamically adjusted poll priority
+* fix for enhanced escape chars
+* fix for too frequent logging on MQTT broker communication error
+* fix for unexpected "different version available" message during update check
+* fix for detecting subsequent fields with less than 8 bits not fitting into 1 byte
+* fix for compilation and running on FreeBSD and MacOS with low latency setting for FTDI device (thanks to samm-git)
+* fix for switching to daemon mode when log file was not opened
+* fix for checking required arguments of "write" command with "-h"
+
+## Features
+* added option to set poll priority with MQTT get topic
+* added "direct" command for listening to all valid messages from the bus
+* added MQTT /list topic for retrieval of all known messages
+* added support for init scripts on non-LSB distributions (thanks to andr2000)
+* added support for logging to syslog instead of file (thanks to samm-git)
+
+
+# 3.3 (2018-12-26)
 
 ## Bug Fixes
 * fix for missing MQTT subscription after broker reconnect
 * fix for answering to first scan only in answer mode
-* fix don't add transfer latency to receive timeout when acting as SYN generator
+* fix dont add transfer latency to receive timeout when acting as SYN generator
+* fix for bit combinations during write to SymbolString
+* fix for MQTT handling after broker disconnect
+* fix for NaN in JSON
+* fix for deadlock in libmosquitto
+
+## Features
+* wait for being online before starting and automatically restart after 30 seconds
+* added "--mqttchanges" option to only publish changed messages and changed to publish all messages by default
+* added "--mqttclientid" to set own client ID instead of using the default
+* added support for single quotes to all commands
+* added "--mqttlog" and "--mqttversion" options
 
 ## Breaking Changes
 * added support for enhanced network protocol mode for recent [ebusd-esp firmware](https://github.com/john30/ebusd-esp/) that allows the arbitration to be done directly by the Wemos

README.md

@@ -38,7 +38,7 @@ Installation
 ------------
 
 Either pick the [latest release package](https://github.com/john30/ebusd/releases/latest)
-suitable for your system or build it yourself.
+suitable for your system, use the Debian repository as [described here](https://github.com/john30/ebusd-debian/blob/master/README.md), or build it yourself.
 
 Building ebusd from the source requires the following packages and/or features:
  * autoconf (>=2.63) + automake (>=1.11) or cmake

VERSION

@@ -1 +1 @@
-3.2
+3.3

config.h.cmake

@@ -16,6 +16,9 @@
 /* Defined if linux/serial.h is available. */
 #cmakedefine HAVE_LINUX_SERIAL
 
+/* Defined if dev/usb/uftdiio.h is available. */
+#cmakedefine HAVE_FREEBSD_UFTDI
+
 /* Defined if pthread_setname_np is available. */
 #cmakedefine HAVE_PTHREAD_SETNAME_NP
 

configure.ac

@@ -36,6 +36,7 @@ AC_SUBST(EXTRA_LIBS)
 AC_CHECK_FUNC([pselect], [AC_DEFINE(HAVE_PSELECT, [1], [Defined if pselect() is available.])])
 AC_CHECK_FUNC([ppoll], [AC_DEFINE(HAVE_PPOLL, [1], [Defined if ppoll() is available.])])
 AC_CHECK_HEADER([linux/serial.h], [AC_DEFINE(HAVE_LINUX_SERIAL, [1], [Defined if linux/serial.h is available.])])
+AC_CHECK_HEADER([dev/usb/uftdiio.h], [AC_DEFINE(HAVE_FREEBSD_UFTDI, [1], [Defined if dev/usb/uftdiio.h is available.])])
 
 AC_ARG_ENABLE(coverage, AS_HELP_STRING([--enable-coverage], [enable code coverage tracking]), [CXXFLAGS+=" -coverage -O0"], [])
 AC_ARG_WITH(contrib, AS_HELP_STRING([--without-contrib], [disable inclusion of contributed sources]), [], [with_contrib=yes])

contrib/archlinux/PKGBUILD

@@ -2,7 +2,7 @@
 # Contributor: Milan Knizek <knizek volny cz>
 # Usage: makepkg
 pkgname=ebusd
-pkgver=3.2
+pkgver=3.3
 pkgrel=1
 pkgdesc="ebusd, the daemon for communication with eBUS heating systems."
 arch=('i686' 'x86_64' 'armv6h' 'armv7h' 'aarch64')
@@ -37,4 +37,4 @@ package() {
   install -m 0644 contrib/archlinux/conf.d/ebusd          "${pkgdir}/etc/conf.d/ebusd"
 }
 # update md5sums: updpkgsums
-md5sums=('413886187f5e93138674cbea916ad1fb')
+md5sums=('f65252d59ccce4c244d3c4c5597a6b8e')

contrib/archlinux/PKGBUILD.git

@@ -3,7 +3,7 @@
 # Usage: makepkg -p PKGBUILD.git
 pkgname=ebusd-git
 _gitname=ebusd
-pkgver=3.2
+pkgver=3.3
 pkgrel=1
 pkgdesc="ebusd, the daemon for communication with eBUS heating systems."
 arch=('i686' 'x86_64' 'armv6h' 'armv7h' 'aarch64')

contrib/archlinux/systemd/ebusd.service

@@ -6,6 +6,7 @@ ConditionPathExists=/var/log
 [Service]
 Type=forking
 Restart=always
+RestartSec=30
 PIDFile=/run/ebusd.pid
 EnvironmentFile=-/etc/conf.d/ebusd
 ExecStart=/usr/bin/ebusd $EBUSD_OPTS

contrib/config/README.md

@@ -0,0 +1,8 @@
+
+ebusd.eu config webservice
+==========================
+
+This is the code running on the ebusd.eu config webservice that allows ebusd to download the needed CSV configuration files instead of having to install the ebusd-configuration package.
+
+It is enabled by default in ebusd due to the default value of the `--configpath=http://ebusd.eu/config/` commandline option.
+

contrib/config/index.php

@@ -0,0 +1,61 @@
+<?php
+$agent = $_SERVER['HTTP_USER_AGENT'];
+if (substr($agent, 0, 5)!=='ebusd') {
+  header('HTTP/1.1 404 Not Found');
+  exit;
+}
+$p=@$_REQUEST['p'];
+$e=@$_REQUEST['t'];
+$a=@$_REQUEST['a'];
+if ($p && $p[0]==='/') {
+  $p=substr($p, 1);
+}
+if ($p && $p[strlen($p)-1]==='/') {
+  $p=substr($p, 0, strlen($p)-1);
+}
+if ($p && $p[0]!=='/' && strpos($p, '..')===FALSE && is_file($p)) {
+  header('Content-Type: text/comma-separated-values');
+  header('Cache-Control: Private');
+  $t=@filemtime($p);
+  if ($t) {
+    header('Last-Modified: '.gmdate('D, d M Y H:i:s', $t).' GMT');
+  }
+  ob_start();
+  echo @file_get_contents($p);
+  header('Content-Length: '.ob_get_length());
+  ob_end_flush();
+  exit;
+}
+if (!$e) {
+  header('HTTP/1.1 404 Not Found');
+  exit;
+}
+$e=".$e";
+if (!$p) {
+  $p='.';
+} else if (strpos($p, '/')!==FALSE || strpos($p, '\\')!==FALSE || strpos($p, '..')!==FALSE) {
+  header('HTTP/1.1 400 Bad Request');
+  exit;
+}
+$prefix=$a ? $a==='-' ? $a : substr(dechex(0x100|hexdec($a)), 1).'.' : NULL;
+$dir=@opendir($p);
+if ($dir===FALSE) {
+  header('HTTP/1.1 404 Not Found');
+  exit;
+}
+header('Content-Type: text/comma-separated-values');
+header('Cache-Control: Private');
+ob_start();
+while(($f=@readdir())!==FALSE) {
+  if (!$f || !is_file($p.'/'.$f) || substr($f, -strlen($e))!==$e) {
+    continue;
+  }
+  if ($f!=='_templates'.$e && $prefix && ($prefix==='-' ? strpos($f, '.') === 2 : substr($f, 0, strlen($prefix))!==$prefix)) {
+    continue;
+  }
+  echo "$f\n";
+}
+@closedir($dir);
+header('Content-Length: '.ob_get_length());
+ob_end_flush();
+?>

contrib/debian/init.d/ebusd

@@ -13,7 +13,19 @@ DAEMON=/usr/bin/ebusd
 PIDFILE_PREFIX=/var/run/ebusd
 PIDFILE_SUFFIX=.pid
 
-. /lib/lsb/init-functions
+if test -f /lib/lsb/init-functions; then
+	. /lib/lsb/init-functions
+else
+	. /etc/init.d/functions
+
+	log_daemon_msg()  { logger "$@"; }
+	log_end_msg()     { [ $1 -eq 0 ] && RES=OK; logger ${RES:=FAIL}; }
+	log_failure_msg() { logger "FAIL $@"; }
+	log_warning_msg() { logger "WARN $@"; }
+	status_of_proc () { status $3; }
+
+fi
+
 [ -r /etc/default/ebusd ] && . /etc/default/ebusd
 
 if [ ! -x $DAEMON ]; then

contrib/debian/systemd/ebusd.service

@@ -6,6 +6,7 @@ ConditionPathExists=/var/log
 [Service]
 Type=forking
 Restart=always
+RestartSec=30
 PIDFile=/var/run/ebusd.pid
 EnvironmentFile=-/etc/default/ebusd
 ExecStart=/usr/bin/ebusd $EBUSD_OPTS

contrib/docker/Dockerfile

@@ -1,4 +1,4 @@
-FROM debian:jessie
+FROM debian:stretch
 
 RUN apt-get update && apt-get install -y \
     logrotate libmosquitto1 libstdc++6 libc6 libgcc1 \
@@ -12,7 +12,7 @@ RUN apt-get update && apt-get install -y \
 
 LABEL maintainer "ebusd@ebusd.eu"
 
-ENV EBUSD_VERSION 3.2
+ENV EBUSD_VERSION 3.3
 ENV EBUSD_ARCH amd64
 
 LABEL version "${EBUSD_VERSION}-${EBUSD_ARCH}-devel"
@@ -30,6 +30,6 @@ RUN rm -rf /build
 
 EXPOSE 8888
 
-COPY runtime/docker-entrypoint.sh /
+COPY release/docker-entrypoint.sh /
 ENTRYPOINT ["/docker-entrypoint.sh"]
 CMD ["-f", "--scanconfig"]

contrib/docker/release/Dockerfile

@@ -7,7 +7,7 @@ RUN apt-get update && apt-get install -y \
 
 LABEL maintainer "ebusd@ebusd.eu"
 
-ENV EBUSD_VERSION 3.2
+ENV EBUSD_VERSION 3.3
 ENV EBUSD_ARCH amd64
 
 LABEL version "${EBUSD_VERSION}-${EBUSD_ARCH}"

contrib/docker/runtime/Dockerfile

@@ -1,25 +0,0 @@
-FROM debian:stretch-slim
-
-RUN apt-get update && apt-get install -y \
-    logrotate libmosquitto1 libstdc++6 libc6 libgcc1 \
-    curl \
-    && rm -rf /var/lib/apt/lists/*
-
-LABEL maintainer "ebusd@ebusd.eu"
-
-ENV EBUSD_VERSION 3.2
-ENV EBUSD_ARCH amd64
-
-LABEL version "${EBUSD_VERSION}-${EBUSD_ARCH}-mqtt1"
-
-COPY ebusd-${EBUSD_VERSION}_${EBUSD_ARCH}_mqtt1.deb ebusd.deb
-
-RUN dpkg -i ebusd.deb
-
-RUN rm ebusd.deb
-
-EXPOSE 8888
-
-COPY docker-entrypoint.sh /
-ENTRYPOINT ["/docker-entrypoint.sh"]
-CMD ["-f", "--scanconfig"]

contrib/docker/runtime/docker-entrypoint.sh

@@ -1,8 +0,0 @@
-#!/bin/bash
-set -e
-
-if [ "${1#-}" != "$1" ]; then
-	set -- ebusd "$@"
-fi
-
-exec "$@"

contrib/etc/logrotate.d/ebusd

@@ -8,5 +8,5 @@
 	daily
 	postrotate
 		/usr/bin/killall -HUP ebusd
-    endscript
+	endscript
 }

contrib/updatecheck/README.md

@@ -0,0 +1,8 @@
+
+ebusd.eu update webservice
+==========================
+
+This is the code running on the ebusd.eu update webservice that allows ebusd to check its own version and the configuration files for availability of new versions.
+
+It is enabled by default in ebusd due to the default value of the `--updatecheck=on` commandline option.
+

contrib/updatecheck/calcversions.sh

@@ -2,5 +2,7 @@
 version=`head -n 1 ../../VERSION`
 revision=`git describe --always`
 echo "ebusd=${version},${revision}" > versions.txt
+echo "ebusd=${version},${revision}" > oldversions.txt
 files=`find config/ -type f -or -type l`
 ../../src/lib/ebus/test/test_filereader $files|sed -e 's#^config/##' -e 's#^\([^ ]*\) #\1=#' -e 's# #,#g'|sort >> versions.txt
+./oldtest_filereader $files|sed -e 's#^config/##' -e 's#^\([^ ]*\) #\1=#' -e 's# #,#g'|sort >> oldversions.txt

contrib/updatecheck/index.php

@@ -5,7 +5,7 @@ if (substr($agent, 0, 5)==='ebusd') {
   $r = @file_get_contents('php://input');
   header('Content-Type: text/plain');
   $r = @json_decode($r, true);
-  echo checkUpdate(@$r['v'], @$r['r'], @$r['l']);
+  echo checkUpdate(@$r['v'], @$r['r'], @$r['a'], @$r['l']);
   exit;
 }
 header('HTTP/1.1 404 Not Found');

contrib/updatecheck/prepend.inc

@@ -1,21 +1,26 @@
 <?php
-$v = @file_get_contents('versions.txt');
 $versions = array();
-if ($v) {
+function readVersions($old=false) {
-  $v = explode("\n", $v);
+  global $versions;
-  $func = function($val, $k) {
+  $versions = array();
-    global $versions;
+  $v = @file_get_contents(($old?'old':'').'versions.txt');
-    $val = explode('=', $val);
+  if ($v) {
-    if (count($val)>1) {
+    $v = explode("\n", $v);
-      $versions[$val[0]] = explode(',', $val[1]);
+    $func = function($val, $k) {
-    }
+      global $versions;
-  };
+      $val = explode('=', $val);
-  array_walk($v, $func);
+      if (count($val)>1) {
+        $versions[$val[0]] = explode(',', $val[1]);
+      }
+    };
+    array_walk($v, $func);
+  }
 }
-function checkUpdate($ebusdVersion, $ebusdRelease, $loadedFiles) {
+function checkUpdate($ebusdVersion, $ebusdRelease, $architecture, $loadedFiles) {
   if (!$ebusdVersion) {
     return 'invalid request';
   }
+  readVersions($architecture && substr($architecture, 0, 3)!=='arm' && ($ebusdVersion<'3.3' || ($ebusdVersion==='3.3' && ($ebusdRelease==='v3.3' || (strtok($ebusdRelease, '-')==='v3.3') && strtok('-')<18))));
   global $versions;
   $ret = 'unknown';
   if ($ebusdVersion==$versions['ebusd'][0] && $ebusdRelease==$versions['ebusd'][1]) {
@@ -39,7 +44,7 @@ function checkUpdate($ebusdVersion, $ebusdRelease, $loadedFiles) {
       }
     }
   }
-  if (strlen($configs)>100) {
+  if (strlen($configs)>1000) {
     return $ret.($newerAvailable?', '.$newerAvailable.' newer':', different').' configuration files available';
   }
   return $ret.$configs;

src/ebusd/bushandler.cpp

@@ -190,6 +190,7 @@ bool ActiveBusRequest::notify(result_t result, const SlaveSymbolString& slave) {
 
 
 void GrabbedMessage::setLastData(const MasterSymbolString& master, const SlaveSymbolString& slave) {
+  time(&m_lastTime);
   m_lastMaster = master;
   m_lastSlave = slave;
   m_count++;
@@ -249,7 +250,8 @@ bool decodeType(const DataType* type, const SymbolString& input, size_t length,
   return !first;
 }
 
-bool GrabbedMessage::dump(bool unknown, MessageMap* messages, bool first, bool decode, ostringstream* output) const {
+bool GrabbedMessage::dump(bool unknown, MessageMap* messages, bool first, bool decode, ostringstream* output,
+    bool isDirectMode) const {
   Message* message = messages->find(m_lastMaster);
   if (unknown && message) {
     return false;
@@ -260,11 +262,13 @@ bool GrabbedMessage::dump(bool unknown, MessageMap* messages, bool first, bool d
   symbol_t dstAddress = m_lastMaster[1];
   *output << m_lastMaster.getStr();
   if (dstAddress != BROADCAST && !isMaster(dstAddress)) {
-    *output << " / " << m_lastSlave.getStr();
+    *output << (isDirectMode ? " " : " / ") << m_lastSlave.getStr();
   }
-  *output << " = " << m_count;
+  if (!isDirectMode) {
-  if (message) {
+    *output << " = " << m_count;
-    *output << ": " << message->getCircuit() << " " << message->getName();
+    if (message) {
+      *output << ": " << message->getCircuit() << " " << message->getName();
+    }
   }
   if (decode) {
     DataTypeList *types = DataTypeList::getInstance();
@@ -1585,16 +1589,26 @@ bool BusHandler::enableGrab(bool enable) {
   return true;
 }
 
-void BusHandler::formatGrabResult(bool unknown, bool decode, ostringstream* output) const {
+void BusHandler::formatGrabResult(bool unknown, bool decode, ostringstream* output, bool isDirectMode,
+    time_t since, time_t until) const {
   if (!m_grabMessages) {
-    *output << "grab disabled";
+    if (!isDirectMode) {
-  } else {
+      *output << "grab disabled";
-    bool first = true;
-    for (const auto& it : m_grabbedMessages) {
-      if (it.second.dump(unknown, m_messages, first, decode, output)) {
-        first = false;
-      }
     }
+    return;
+  }
+  bool first = true;
+  for (const auto& it : m_grabbedMessages) {
+    if ((since > 0 && it.second.getLastTime() < since)
+    || (until > 0 && it.second.getLastTime() >= until)) {
+      continue;
+    }
+    if (it.second.dump(unknown, m_messages, first, decode, output, isDirectMode)) {
+      first = false;
+    }
+  }
+  if (isDirectMode && !first) {
+    *output << endl;
   }
 }
 

src/ebusd/bushandler.h

@@ -296,7 +296,7 @@ class GrabbedMessage {
   /**
    * Construct a new instance.
    */
-  GrabbedMessage() : m_count(0) {}
+  GrabbedMessage() : m_lastTime(0), m_count(0) {}
 
   /**
    * Copy constructor.
@@ -314,6 +314,12 @@ class GrabbedMessage {
    */
   void setLastData(const MasterSymbolString& master, const SlaveSymbolString& slave);
 
+  /**
+   * Get the last received time.
+   * @return the last received time.
+   */
+  time_t getLastTime() const { return m_lastTime; }
+
   /**
    * Get the last @a MasterSymbolString.
    * @return the last @a MasterSymbolString.
@@ -327,12 +333,17 @@ class GrabbedMessage {
    * @param first whether this is the first message to be added to the output.
    * @param decode whether to add decoding hints.
    * @param output the @a ostringstream to format the messages to.
+   * @param isDirectMode true for direct mode, false for grab command.
    * @return whether the message was added to the output.
    */
-  bool dump(bool unknown, MessageMap* messages, bool first, bool decode, ostringstream* output) const;
+  bool dump(bool unknown, MessageMap* messages, bool first, bool decode, ostringstream* output,
+      bool isDirectMode = false) const;
 
 
  private:
+  /** the last received time. */
+  time_t m_lastTime;
+
   /** the last @a MasterSymbolString. */
   MasterSymbolString m_lastMaster;
 
@@ -516,13 +527,23 @@ class BusHandler : public WaitThread {
    */
   bool enableGrab(bool enable = true);
 
+  /**
+   * Return whether grabbing unknown messages is enabled.
+   * @return whether grabbing unknown messages is enabled.
+   */
+  bool isGrabEnabled() { return m_grabMessages; }
+
   /**
    * Format the grabbed messages to the @a ostringstream.
    * @param unknown whether to dump only unknown messages.
    * @param decode whether to add decoding hints.
    * @param output the @a ostringstream to format the messages to.
+   * @param isDirectMode true for direct mode, false for grab command.
+   * @param since the start time from which to add received messages (inclusive), or 0 for all.
+   * @param until the end time to which to add received messages (exclusive), or 0 for all.
    */
-  void formatGrabResult(bool unknown, bool decode, ostringstream* output) const;
+  void formatGrabResult(bool unknown, bool decode, ostringstream* output, bool isDirectMode = false,
+      time_t since = 0, time_t until = 0) const;
 
   /**
    * Return true when a signal on the bus is available.

src/ebusd/main.cpp

@@ -223,7 +223,7 @@ static const struct argp_option argpoptions[] = {
   {"updatecheck",    O_UPDCHK, "MODE",     0, "Set automatic update check to MODE (on|off) [on]", 0 },
 
   {nullptr,          0,        nullptr,    0, "Log options:", 5 },
-  {"logfile",        'l',      "FILE",     0, "Write log to FILE (only for daemon) [" PACKAGE_LOGFILE "]", 0 },
+  {"logfile",        'l',      "FILE",     0, "Write log to FILE (only for daemon, empty string for using syslog) [" PACKAGE_LOGFILE "]", 0 },
   {"log",            O_LOG, "AREAS LEVEL", 0, "Only write log for matching AREA(S) below or equal to LEVEL"
       " (alternative to --logareas/--logevel, may be used multiple times) [all notice]", 0 },
   {"logareas",       O_LOGARE, "AREAS",    0, "Only write log for matching AREA(S): main|network|bus|update|all"
@@ -487,7 +487,7 @@ error_t parse_opt(int key, char *arg, struct argp_state *state) {
 
   // Log options:
   case 'l':  // --logfile=/var/log/ebusd.log
-    if (arg == nullptr || arg[0] == 0 || strcmp("/", arg) == 0) {
+    if (arg == nullptr || strcmp("/", arg) == 0) {
       argp_error(state, "invalid logfile");
       return EINVAL;
     }
@@ -599,7 +599,7 @@ void daemonize() {
   pid_t pid = fork();
 
   if (pid < 0) {
-    logError(lf_main, "fork() failed");
+    logError(lf_main, "fork() failed, exiting");
     exit(EXIT_FAILURE);
   }
 
@@ -612,14 +612,14 @@ void daemonize() {
   // Create a new SID for the child process and
   // detach the process from the parent (normally a shell)
   if (setsid() < 0) {
-    logError(lf_main, "setsid() failed");
+    logError(lf_main, "setsid() failed, exiting");
     exit(EXIT_FAILURE);
   }
 
   // Change the current working directory. This prevents the current
   // directory from being locked; hence not being able to remove it.
   if (chdir("/tmp") < 0) {
-    logError(lf_main, "daemon chdir() failed");
+    logError(lf_main, "daemon chdir() failed, exiting");
     exit(EXIT_FAILURE);
   }
 
@@ -642,7 +642,7 @@ void daemonize() {
     }
   }
   if (pidFile == nullptr) {
-    logError(lf_main, "can't open pidfile: %s", opt.pidFile);
+    logError(lf_main, "can't open pidfile: %s, exiting", opt.pidFile);
     exit(EXIT_FAILURE);
   }
 
@@ -701,7 +701,7 @@ void signalHandler(int sig) {
   switch (sig) {
   case SIGHUP:
     logNotice(lf_main, "SIGHUP received");
-    if (!opt.foreground) {
+    if (!opt.foreground && opt.logFile && opt.logFile[0] != 0) { // for log file rotation
       closeLogFile();
       setLogFile(opt.logFile);
     }
@@ -988,8 +988,8 @@ result_t loadConfigFiles(MessageMap* messages, bool verbose, bool denyRecursive)
   if (result == RESULT_OK) {
     logInfo(lf_main, "read config files");
   } else {
-    logError(lf_main, "error reading config files: %s, last error: %s", getResultCode(result),
+    logError(lf_main, "error reading config files from %s: %s, last error: %s", opt.configPath,
-        errorDescription.c_str());
+        getResultCode(result), errorDescription.c_str());
   }
   messages->unlock();
   return RESULT_OK;
@@ -1293,7 +1293,10 @@ int main(int argc, char* argv[]) {
   }
 
   if (!opt.foreground) {
-    setLogFile(opt.logFile);
+    if (!setLogFile(opt.logFile)) {
+      logError(lf_main, "unable to open log file %s", opt.logFile);
+      return EINVAL;
+    }
     daemonize();  // make me daemon
   }
 

src/ebusd/mainloop.cpp

@@ -371,22 +371,24 @@ void MainLoop::run() {
       continue;
     }
     if (m_shutdown) {
-      netMessage->setResult("ERR: shutdown", "", false, now, true);
+      netMessage->setResult("ERR: shutdown", "", cm_normal, now, true);
       break;
     }
     string request = netMessage->getRequest();
     string user = netMessage->getUser();
-    bool listening = netMessage->isListening(&since);
+    ClientMode mode = netMessage->getMode(&since);
-    if (!listening) {
+    if (!netMessage->isListeningMode()) {
       since = now;
     }
     ostringstream ostream;
     bool connected = true;
     if (request.length() > 0) {
       logDebug(lf_main, ">>> %s", request.c_str());
-      result_t result = decodeMessage(request, netMessage->isHttp(), &connected, &listening, &user, &reload, &ostream);
+      result_t result = decodeMessage(request, netMessage->isHttp(), &connected, &mode, &user, &reload, &ostream);
       if (!netMessage->isHttp() && (ostream.tellp() == 0 || result != RESULT_OK)) {
-        ostream.str("");
+        if (mode != cm_direct) {
+          ostream.str("");
+        }
         ostream << getResultCode(result);
       }
       if (ostream.tellp() > 100) {
@@ -397,10 +399,10 @@ void MainLoop::run() {
       if (ostream.tellp() == 0) {
         ostream << "\n";  // only for HTTP
       } else if (!netMessage->isHttp()) {
-        ostream << "\n\n";
+        ostream << (mode == cm_direct ? "\n" : "\n\n");
       }
     }
-    if (listening) {
+    if (mode == cm_listen) {
       string levels = getUserLevels(user);
       messages.clear();
       m_messages->findAll("", "", levels, false, true, true, true, true, true, since, now, true, &messages);
@@ -409,9 +411,13 @@ void MainLoop::run() {
         message->decodeLastData(false, nullptr, -1, 0, &ostream);
         ostream << endl;
       }
+    } else if (mode == cm_direct) {
+      if (m_busHandler->isGrabEnabled()) {
+        m_busHandler->formatGrabResult(false, false, &ostream, true, since, now);
+      }
     }
     // send result to client
-    netMessage->setResult(ostream.str(), user, listening, now, !connected);
+    netMessage->setResult(ostream.str(), user, mode, now, !connected);
   }
 }
 
@@ -456,7 +462,7 @@ void MainLoop::notifyDeviceData(symbol_t symbol, bool received) {
   }
 }
 
-result_t MainLoop::decodeMessage(const string &data, bool isHttp, bool* connected, bool* listening,
+result_t MainLoop::decodeMessage(const string &data, bool isHttp, bool* connected, ClientMode* mode,
     string* user, bool* reload, ostringstream* ostream) {
   string token, previous;
   istringstream stream(data);
@@ -476,11 +482,11 @@ result_t MainLoop::decodeMessage(const string &data, bool isHttp, bool* connecte
       } else if (token.length() == 0) {  // allow multiple space chars for a single delimiter
         continue;
       } else if (token[0] == '"' || token[0] == '\'') {
+        escaped = token[0];
         token.erase(0, 1);
-        if (token.length() > 0 && token[token.length()-1] == token[0]) {
+        if (token.length() > 0 && token[token.length()-1] == escaped) {
           token.erase(token.length() - 1, 1);
-        } else {
+          escaped = 0;
-          escaped = token[0];
         }
       }
     }
@@ -506,21 +512,25 @@ result_t MainLoop::decodeMessage(const string &data, bool isHttp, bool* connecte
     return RESULT_OK;
   }
 
-  if (args.size() == 0) {
+  string cmd = args.size() > 0 ? args[0] : "";
+  transform(cmd.begin(), cmd.end(), cmd.begin(), ::toupper);
+  if (cmd == "?" || cmd == "H" || cmd == "HELP") {
+    // found "HELP CMD"
+    cmd = args.size() > 1 ? args[1] : "";
+    transform(cmd.begin(), cmd.end(), cmd.begin(), ::toupper);
+    args.clear();  // empty args is used as command help indicator
+  }
+  if (*mode == cm_direct) {
+    return executeDirect(args, mode, ostream);
+  }
+  if (cmd.empty() && args.size() == 0) {
     return executeHelp(ostream);
   }
-  string cmd = args[0];
-  transform(cmd.begin(), cmd.end(), cmd.begin(), ::toupper);
   if (args.size() == 2) {
     string arg = args[1];
     if (arg == "?" || arg == "-?" || arg == "--help") {
       // found "CMD HELP"
       args.clear();  // empty args is used as command help indicator
-    } else if (cmd == "?" || cmd == "H" || cmd == "HELP") {
-      // found "HELP CMD"
-      cmd = args[1];
-      transform(cmd.begin(), cmd.end(), cmd.begin(), ::toupper);
-      args.clear();  // empty args is used as command help indicator
     }
   }
   if (cmd == "AUTH" || cmd == "A") {
@@ -543,7 +553,10 @@ result_t MainLoop::decodeMessage(const string &data, bool isHttp, bool* connecte
     return executeFind(args, getUserLevels(*user), ostream);
   }
   if (cmd == "L" || cmd == "LISTEN") {
-    return executeListen(args, listening, ostream);
+    return executeListen(args, mode, ostream);
+  }
+  if (cmd == "DIRECT") {
+    return executeDirect(args, mode, ostream);
   }
   if (cmd == "S" || cmd == "STATE") {
     return executeState(args, ostream);
@@ -748,7 +761,8 @@ result_t MainLoop::executeRead(const vector<string>& args, const string& levels,
     *ostream <<
         "usage: read [-f] [-m SECONDS] [-s QQ] [-d ZZ] [-c CIRCUIT] [-p PRIO] [-v|-V] [-n|-N] [-i VALUE[;VALUE]*]"
         " NAME [FIELD[.N]]\n"
-        "  or:  read [-f] [-m SECONDS] [-s QQ] [-d ZZ] [-v|-V] [-n|-N] [-i VALUE[;VALUE]*] -def DEFINITION\n"
+        "  or:  read [-f] [-m SECONDS] [-s QQ] [-d ZZ] [-v|-V] [-n|-N] [-i VALUE[;VALUE]*] -def DEFINITION "
+        "(only if enabled)\n"
         "  or:  read [-f] [-m SECONDS] [-s QQ] [-c CIRCUIT] -h ZZPBSBNN[DD]*\n"
         " Read value(s) or hex message.\n"
         "  -f           force reading from the bus (same as '-m 0')\n"
@@ -765,7 +779,7 @@ result_t MainLoop::executeRead(const vector<string>& args, const string& levels,
         "  NAME         NAME of the message to send\n"
         "  FIELD        only retrieve the field named FIELD\n"
         "  N            only retrieve the N'th field named FIELD (0-based)\n"
-        "  -def         read with explicit message definition:\n"
+        "  -def         read with explicit message definition (only if enabled):\n"
         "    DEFINITION message definition to use instead of known definition\n"
         "  -h           send hex read message (or answer from cache):\n"
         "    ZZ         destination address\n"
@@ -987,14 +1001,16 @@ result_t MainLoop::executeWrite(const vector<string>& args, const string levels,
     argPos++;
   }
 
-  if ((hex && (newDefinition || dstAddress != SYN || !circuit.empty() || args.size() < argPos + 1))
+  if ((args.size() < argPos + 1)
-  || (newDefinition && (hex || !circuit.empty() || args.size() < argPos + 1 || args.size() > argPos + 2))) {
+  || (hex && (newDefinition || dstAddress != SYN))
+  || (newDefinition && (hex || !circuit.empty() || args.size() > argPos + 2))
+  || (!newDefinition && !hex && (circuit.empty() || args.size() > argPos + 2))) {
     argPos = 0;  // print usage
   }
 
-  if (argPos == 0 || (!newDefinition && (circuit.empty() || (args.size() != argPos+2 && args.size() != argPos+1)))) {
+  if (argPos == 0) {
     *ostream << "usage: write [-s QQ] [-d ZZ] -c CIRCUIT NAME [VALUE[;VALUE]*]\n"
-        "  or:  write [-s QQ] [-d ZZ] -def DEFINITION [VALUE[;VALUE]*]\n"
+        "  or:  write [-s QQ] [-d ZZ] -def DEFINITION [VALUE[;VALUE]*] (only if enabled)\n"
         "  or:  write [-s QQ] [-c CIRCUIT] -h ZZPBSBNN[DD]*\n"
         " Write value(s) or hex message.\n"
         "  -s QQ        override source address QQ\n"
@@ -1002,7 +1018,7 @@ result_t MainLoop::executeWrite(const vector<string>& args, const string levels,
         "  -c CIRCUIT   CIRCUIT of the message to send\n"
         "  NAME         NAME of the message to send\n"
         "  VALUE        a single field VALUE\n"
-        "  -def         write with explicit message definition:\n"
+        "  -def         write with explicit message definition (only if enabled):\n"
         "    DEFINITION message definition to use instead of known definition\n"
         "  -h           send hex write message:\n"
         "    ZZ         destination address\n"
@@ -1133,56 +1149,114 @@ result_t MainLoop::executeWrite(const vector<string>& args, const string levels,
   return RESULT_OK;
 }
 
-result_t MainLoop::executeHex(const vector<string>& args, ostringstream* ostream) {
+result_t MainLoop::parseHexAndSend(const vector<string>& args, size_t& argPos, bool isDirectMode,
-  size_t argPos = 1;
+    ostringstream* ostream) {
   symbol_t srcAddress = SYN;
   if (args.size() > argPos && args[argPos] == "-s") {
     argPos++;
     if (argPos >= args.size()) {
       argPos = 0;  // print usage
-    } else {
+      return RESULT_OK;
-      result_t ret;
-      symbol_t address = (symbol_t)parseInt(args[argPos].c_str(), 16, 0, 0xff, &ret);
-      if (ret != RESULT_OK || !isValidAddress(address, false) || !isMaster(address)) {
-        return RESULT_ERR_INVALID_ADDR;
-      }
-      srcAddress = address == m_address ? SYN : address;
     }
+    result_t ret;
+    symbol_t address = (symbol_t)parseInt(args[argPos].c_str(), 16, 0, 0xff, &ret);
+    if (ret != RESULT_OK || !isValidAddress(address, false) || !isMaster(address)) {
+      return RESULT_ERR_INVALID_ADDR;
+    }
+    srcAddress = address == m_address ? SYN : address;
     argPos++;
   }
   if (args.size() < argPos + 1 || (args.size() > argPos && args[argPos][0] == '-')) {
     argPos = 0;  // print usage
+    return RESULT_OK;
   }
 
-  if (argPos > 0) {
+  MasterSymbolString master;
-    MasterSymbolString master;
+  result_t ret = parseHexMaster(args, argPos, srcAddress, &master);
-    result_t ret = parseHexMaster(args, argPos, srcAddress, &master);
+  argPos = args.size();  // mark as successfully parsed
-    if (ret != RESULT_OK) {
+  if (ret != RESULT_OK) {
-      return ret;
-    }
-    logNotice(lf_main, "hex cmd: %s", master.getStr().c_str());
-
-    // send message
-    SlaveSymbolString slave;
-    ret = m_busHandler->sendAndWait(master, &slave);
-
-    if (ret == RESULT_OK) {
-      if (master[1] == BROADCAST) {
-        *ostream << "done broadcast";
-        return RESULT_OK;
-      }
-      if (isMaster(master[1])) {
-        return RESULT_OK;
-      }
-      *ostream << slave.getStr();
-      return RESULT_OK;
-    }
-    logError(lf_main, "hex: %s", getResultCode(ret));
     return ret;
   }
+  logNotice(lf_main, isDirectMode ? "direct cmd: %s" : "hex cmd: %s", master.getStr().c_str());
 
+  // send message
+  SlaveSymbolString slave;
+  ret = m_busHandler->sendAndWait(master, &slave);
+
+  if (ret == RESULT_OK) {
+    if (master[1] == BROADCAST) {
+      *ostream << "done broadcast";
+      return RESULT_OK;
+    }
+    if (isMaster(master[1])) {
+      *ostream << "done";
+      return RESULT_OK;
+    }
+    *ostream << slave.getStr();
+    return RESULT_OK;
+  }
+  logError(lf_main, isDirectMode ? "direct: %s" : "hex: %s", getResultCode(ret));
+  return ret;
+}
+
+result_t MainLoop::executeHex(const vector<string>& args, ostringstream* ostream) {
+  size_t argPos = 1;
+  result_t ret = parseHexAndSend(args, argPos, false, ostream);
+  if (argPos == args.size()) {
+    return ret;
+  }
   *ostream << "usage: hex [-s QQ] ZZPBSBNN[DD]*\n"
-              " Send arbitrary data in hex (only if enabled).\n"
+              " Send arbitrary data in hex.\n"
+              "  -s QQ  override source address QQ\n"
+              "  ZZ     destination address\n"
+              "  PB SB  primary/secondary command byte\n"
+              "  NN     number of following data bytes\n"
+              "  DD     data byte(s) to send";
+  return RESULT_OK;
+}
+
+result_t MainLoop::executeDirect(const vector<string>& args, ClientMode* mode, ostringstream* ostream) {
+  if (*mode != cm_direct) {
+    if (args.size() == 1) {
+      *mode = cm_direct;
+      m_busHandler->enableGrab(true);  // needed for listening to all messages
+      *ostream << "direct mode started";
+      return RESULT_OK;
+    }
+    *ostream << "usage: direct\n"
+                " Enter direct mode.";
+    return RESULT_OK;
+  }
+  if (args.size() > 0) {
+    string firstArg = args[0];
+    if (firstArg == "stop") {
+      *mode = cm_normal;
+      *ostream << "direct mode stopped";
+      return RESULT_OK;
+    }
+    if (firstArg != "") {
+      for (size_t argPos = 0; argPos < args.size(); argPos++) {
+        if (argPos > 0) {
+          *ostream << " ";
+        }
+        *ostream << args[argPos];
+      }
+      *ostream << ":";
+      if (!m_enableHex) {
+        *ostream << "ERR: command not enabled";
+        return RESULT_OK;
+      }
+      size_t argPos = 0;
+      result_t ret = parseHexAndSend(args, argPos, true, ostream);
+      if (ret == RESULT_OK && argPos != args.size()) {
+        ret = RESULT_ERR_INVALID_ARG;
+      }
+      return ret;
+    }
+  }
+  *ostream << "usage: [-s QQ] ZZPBSBNN[DD]*\n"
+              "  or: stop\n"
+              " Send arbitrary data in hex (only if enabled) or stop direct mode.\n"
               "  -s QQ  override source address QQ\n"
               "  ZZ     destination address\n"
               "  PB SB  primary/secondary command byte\n"
@@ -1405,13 +1479,13 @@ result_t MainLoop::executeFind(const vector<string>& args, const string& levels,
   return RESULT_OK;
 }
 
-result_t MainLoop::executeListen(const vector<string>& args, bool* listening, ostringstream* ostream) {
+result_t MainLoop::executeListen(const vector<string>& args, ClientMode* mode, ostringstream* ostream) {
   if (args.size() == 1) {
-    if (*listening) {
+    if (*mode == cm_listen) {
       *ostream << "listen continued";
       return RESULT_OK;
     }
-    *listening = true;
+    *mode = cm_listen;
     *ostream << "listen started";
     return RESULT_OK;
   }
@@ -1421,7 +1495,7 @@ result_t MainLoop::executeListen(const vector<string>& args, bool* listening, os
                 " Listen for updates or stop it.";
     return RESULT_OK;
   }
-  *listening = false;
+  *mode = cm_normal;
   *ostream << "listen stopped";
   return RESULT_OK;
 }
@@ -1781,17 +1855,18 @@ result_t MainLoop::executeHelp(ostringstream* ostream) {
   *ostream << "usage:\n"
       " read|r    Read value(s):         read [-f] [-m SECONDS] [-s QQ] [-d ZZ] [-c CIRCUIT] [-p PRIO] [-v|-V] [-n|-N]"
       " [-i VALUE[;VALUE]*] NAME [FIELD[.N]]\n"
-      "           Read by new defintion: read [-f] [-m SECONDS] [-s QQ] [-d ZZ] [-v|-V] [-n|-N]"
+      "           Read by new defintion: read [-f] [-m SECONDS] [-s QQ] [-d ZZ] [-v|-V] [-n|-N] (if enabled)"
       " [-i VALUE[;VALUE]*] -def DEFINITION\n"
       "           Read hex message:      read [-f] [-m SECONDS] [-s QQ] [-c CIRCUIT] -h ZZPBSBNN[DD]*\n"
       " write|w   Write value(s):        write [-s QQ] [-d ZZ] -c CIRCUIT NAME [VALUE[;VALUE]*]\n"
-      "           Write by new def.:     write [-s QQ] [-d ZZ] -def DEFINITION [VALUE[;VALUE]*]\n"
+      "           Write by new def.:     write [-s QQ] [-d ZZ] -def DEFINITION [VALUE[;VALUE]*] (if enabled)\n"
       "           Write hex message:     write [-s QQ] [-c CIRCUIT] -h ZZPBSBNN[DD]*\n"
       " auth|a    Authenticate user:     auth USER SECRET\n"
-      " hex       Send hex data:         hex [-s QQ] ZZPBSBNN[DD]*\n"
+      " hex       Send hex data:         hex [-s QQ] ZZPBSBNN[DD]* (if enabled)\n"
       " find|f    Find message(s):       find [-v|-V] [-r] [-w] [-p] [-a] [-d] [-h] [-i ID] [-f] [-F COL[,COL]*] [-e]"
       " [-c CIRCUIT] [-l LEVEL] [NAME]\n"
       " listen|l  Listen for updates:    listen [stop]\n"
+      " direct    Enter direct mode\n"
       " state|s   Report bus state\n"
       " info|i    Report information about the daemon, the configuration, and seen devices.\n"
       " grab|g    Grab messages:         grab [stop]\n"

src/ebusd/mainloop.h

@@ -144,13 +144,13 @@ class MainLoop : public Thread, DeviceListener {
    * @param data the data string to decode (may be empty).
    * @param connected set to false when the client connection shall be closed.
    * @param isHttp true for HTTP message.
-   * @param listening set to true when the client is in listening mode.
+   * @param mode set to the new client mode.
    * @param user set to the new user name when changed by authentication.
    * @param reload set to true when the configuration files were reloaded.
    * @param ostream the @a ostringstream to format the result string to.
    * @return the result code.
    */
-  result_t decodeMessage(const string& data, bool isHttp, bool* connected, bool* listening,
+  result_t decodeMessage(const string& data, bool isHttp, bool* connected, ClientMode* mode,
       string* user, bool* reload, ostringstream* ostream);
 
   /**
@@ -198,6 +198,17 @@ class MainLoop : public Thread, DeviceListener {
    */
   result_t executeWrite(const vector<string>& args, const string levels, ostringstream* ostream);
 
+  /**
+   * Parse a hex or direct command and send it on the bus.
+   * @param args the arguments passed to the command.
+   * @param argPos the position in the arguments to the first parameter, will be set to other than args.size() on
+   * invalid input.
+   * @param isDirectMode true for direct mode, false for hex command.
+   * @param ostream the @a ostringstream to format the result string to.
+   * @return the result code.
+   */
+  result_t parseHexAndSend(const vector<string>& args, size_t& argPos, bool isDirectMode, ostringstream* ostream);
+
   /**
    * Execute the hex command.
    * @param args the arguments passed to the command (starting with the command itself), or empty for help.
@@ -206,6 +217,15 @@ class MainLoop : public Thread, DeviceListener {
    */
   result_t executeHex(const vector<string>& args, ostringstream* ostream);
 
+  /**
+   * Execute the direct command.
+   * @param args the arguments passed to the command (starting with the command itself), or empty for help.
+   * @param mode set to the new client mode.
+   * @param ostream the @a ostringstream to format the result string to.
+   * @return the result code.
+   */
+  result_t executeDirect(const vector<string>& args, ClientMode* mode, ostringstream* ostream);
+
   /**
    * Execute the find command.
    * @param args the arguments passed to the command (starting with the command itself), or empty for help.
@@ -218,11 +238,11 @@ class MainLoop : public Thread, DeviceListener {
   /**
    * Execute the listen command.
    * @param args the arguments passed to the command (starting with the command itself), or empty for help.
-   * @param listening set to true when the client is in listening mode.
+   * @param mode set to the new client mode.
    * @param ostream the @a ostringstream to format the result string to.
    * @return the result code.
    */
-  result_t executeListen(const vector<string>& args, bool* listening, ostringstream* ostream);
+  result_t executeListen(const vector<string>& args, ClientMode* mode, ostringstream* ostream);
 
   /**
    * Execute the state command.

src/ebusd/mqtthandler.cpp

@@ -22,6 +22,7 @@
 
 #include "ebusd/mqtthandler.h"
 #include <csignal>
+#include <deque>
 #include "lib/utils/log.h"
 
 namespace ebusd {
@@ -36,7 +37,9 @@ using std::dec;
 #define O_TOPI (O_PASS+1)
 #define O_RETA (O_TOPI+1)
 #define O_JSON (O_RETA+1)
-#define O_IGIN (O_JSON+1)
+#define O_LOGL (O_JSON+1)
+#define O_VERS (O_LOGL+1)
+#define O_IGIN (O_VERS+1)
 #define O_CHGS (O_IGIN+1)
 #define O_CAFI (O_CHGS+1)
 #define O_CERT (O_CAFI+1)
@@ -48,7 +51,7 @@ static const struct argp_option g_mqtt_argp_options[] = {
   {nullptr,       0,      nullptr,       0, "MQTT options:", 1 },
   {"mqtthost",    O_HOST, "HOST",        0, "Connect to MQTT broker on HOST [localhost]", 0 },
   {"mqttport",    O_PORT, "PORT",        0, "Connect to MQTT broker on PORT (usually 1883), 0 to disable [0]", 0 },
-  {"mqttclientid",O_CLID, "ID",          0, "Set client ID for connection to MQTT broker [" PACKAGE_NAME "_"
+  {"mqttclientid", O_CLID, "ID",         0, "Set client ID for connection to MQTT broker [" PACKAGE_NAME "_"
    PACKAGE_VERSION "_<pid>]", 0 },
   {"mqttuser",    O_USER, "USER",        0, "Connect as USER to MQTT broker (no default)", 0 },
   {"mqttpass",    O_PASS, "PASSWORD",    0, "Use PASSWORD when connecting to MQTT broker (no default)", 0 },
@@ -56,6 +59,12 @@ static const struct argp_option g_mqtt_argp_options[] = {
    "Use MQTT TOPIC (prefix before /%circuit/%name or complete format) [ebusd]", 0 },
   {"mqttretain",  O_RETA, nullptr,       0, "Retain all topics instead of only selected global ones", 0 },
   {"mqttjson",    O_JSON, nullptr,       0, "Publish in JSON format instead of strings", 0 },
+#if (LIBMOSQUITTO_VERSION_NUMBER >= 1003001)
+  {"mqttlog",     O_LOGL, nullptr,       0, "Log library events", 0 },
+#endif
+#if (LIBMOSQUITTO_VERSION_NUMBER >= 1004001)
+  {"mqttversion", O_VERS, "VERSION",     0, "Use protocol VERSION [3.1]", 0 },
+#endif
   {"mqttignoreinvalid", O_IGIN, nullptr, 0,
    "Ignore invalid parameters during init (e.g. for DNS not resolvable yet)", 0 },
   {"mqttchanges", O_CHGS, nullptr,       0, "Whether to only publish changed messages instead of all received", 0 },
@@ -81,8 +90,14 @@ static vector<string> g_topicStrs;
 static vector<string> g_topicFields;
 static bool g_retain = false;             //!< whether to retail all topics
 static OutputFormat g_publishFormat = 0;  //!< the OutputFormat for publishing messages
+#if (LIBMOSQUITTO_VERSION_NUMBER >= 1003001)
+static bool g_logFromLib = false;         //!< log library events
+#endif
+#if (LIBMOSQUITTO_VERSION_NUMBER >= 1004001)
+static int g_version = MQTT_PROTOCOL_V31;  //!< protocol version to use
+#endif
 static bool g_ignoreInvalidParams = false;  //!< ignore invalid parameters during init
-static bool g_onlyChanges = true;         //!< whether to only publish changed messages instead of all received
+static bool g_onlyChanges = false;        //!< whether to only publish changed messages instead of all received
 
 #if (LIBMOSQUITTO_MAJOR >= 1)
 static const char* g_cafile = nullptr;    //!< CA file for TLS
@@ -162,6 +177,22 @@ static error_t mqtt_parse_opt(int key, char *arg, struct argp_state *state) {
     g_publishFormat |= OF_JSON|OF_NAMES;
     break;
 
+#if (LIBMOSQUITTO_VERSION_NUMBER >= 1003001)
+  case O_LOGL:
+    g_logFromLib = true;
+    break;
+#endif
+
+#if (LIBMOSQUITTO_VERSION_NUMBER >= 1004001)
+  case O_VERS:  // --mqttversion=3.1.1
+    if (arg == nullptr || arg[0] == 0 || (strcmp(arg, "3.1") != 0 && strcmp(arg, "3.1.1") != 0)) {
+      argp_error(state, "invalid mqttversion");
+      return EINVAL;
+    }
+    g_version = strcmp(arg, "3.1.1") == 0 ? MQTT_PROTOCOL_V311 : MQTT_PROTOCOL_V31;
+    break;
+#endif
+
   case O_IGIN:
     g_ignoreInvalidParams = true;
     break;
@@ -228,15 +259,37 @@ const struct argp_child* mqtthandler_getargs() {
   return &g_mqtt_argp_child;
 }
 
+bool check(int code, const char* method) {
+  if (code == MOSQ_ERR_SUCCESS) {
+    return true;
+  }
+  if (code == MOSQ_ERR_ERRNO) {
+    char* error = strerror(errno);
+    logOtherError("mqtt", "%s: errno %d=%s", method, errno, error);
+    return false;
+  }
+#if (LIBMOSQUITTO_VERSION_NUMBER >= 1003001)
+  const char* msg = mosquitto_strerror(code);
+  logOtherError("mqtt", "%s: %s", method, msg);
+#else
+  logOtherError("mqtt", "%s: error code %d", method, code);
+#endif
+  return false;
+}
+
 bool mqtthandler_register(UserInfo* userInfo, BusHandler* busHandler, MessageMap* messages,
     list<DataHandler*>* handlers) {
   if (g_port > 0) {
     int major = -1;
-    mosquitto_lib_version(&major, nullptr, nullptr);
+    int minor = -1;
+    int revision = -1;
+    mosquitto_lib_version(&major, &minor, &revision);
     if (major != LIBMOSQUITTO_MAJOR) {
       logOtherError("mqtt", "invalid mosquitto version %d instead of %d", major, LIBMOSQUITTO_MAJOR);
       return false;
     }
+    logOtherInfo("mqtt", "mosquitto version %d.%d.%d (compiled with %d.%d.%d)", major, minor, revision,
+      LIBMOSQUITTO_MAJOR, LIBMOSQUITTO_MINOR, LIBMOSQUITTO_REVISION);
     handlers->push_back(new MqttHandler(userInfo, busHandler, messages));
   }
   return true;
@@ -331,10 +384,49 @@ void on_connect(
   }
 }
 
+#if (LIBMOSQUITTO_VERSION_NUMBER >= 1003001)
+void on_log(struct mosquitto *mosq, void *obj, int level, const char* msg) {
+  switch (level) {
+  case MOSQ_LOG_DEBUG:
+    logOtherDebug("mqtt", "log %s", msg);
+    break;
+  case MOSQ_LOG_INFO:
+    logOtherInfo("mqtt", "log %s", msg);
+    break;
+  case MOSQ_LOG_NOTICE:
+    logOtherNotice("mqtt", "log %s", msg);
+    break;
+  case MOSQ_LOG_WARNING:
+    logOtherNotice("mqtt", "log warning %s", msg);
+    break;
+  case MOSQ_LOG_ERR:
+    logOtherError("mqtt", "log %s", msg);
+    break;
+  default:
+    logOtherError("mqtt", "log other %s", msg);
+    break;
+  }
+}
+#endif
+
+void on_message(
+#if (LIBMOSQUITTO_MAJOR >= 1)
+  struct mosquitto *mosq,
+#endif
+  void *obj, const struct mosquitto_message *message) {
+  MqttHandler* handler = reinterpret_cast<MqttHandler*>(obj);
+  if (!handler || !message || !handler->isRunning()) {
+    return;
+  }
+  string topic(message->topic);
+  string data(message->payloadlen > 0 ? reinterpret_cast<char*>(message->payload) : "");
+  handler->notifyTopic(topic, data);
+}
+
 
 MqttHandler::MqttHandler(UserInfo* userInfo, BusHandler* busHandler, MessageMap* messages)
   : DataSink(userInfo, "mqtt"), DataSource(busHandler), WaitThread(), m_messages(messages), m_connected(false),
-    m_initialConnectFailed(false), m_lastUpdateCheckResult(".") {
+    m_initialConnectFailed(false), m_lastUpdateCheckResult("."), m_lastErrorLogTime(0) {
   m_publishByField = false;
   m_mosquitto = nullptr;
   if (g_topicFields.empty()) {
@@ -359,10 +451,7 @@ MqttHandler::MqttHandler(UserInfo* userInfo, BusHandler* busHandler, MessageMap*
   }
   m_globalTopic = getTopic(nullptr, "global/");
   m_subscribeTopic = getTopic(nullptr, "#");
-  m_mosquitto = nullptr;
+  if (check(mosquitto_lib_init(), "unable to initialize")) {
-  if (mosquitto_lib_init() != MOSQ_ERR_SUCCESS) {
-    logOtherError("mqtt", "unable to initialize");
-  } else {
     signal(SIGPIPE, SIG_IGN);  // needed before libmosquitto v. 1.1.3
     ostringstream clientId;
     if (g_clientId) {
@@ -380,8 +469,11 @@ MqttHandler::MqttHandler(UserInfo* userInfo, BusHandler* busHandler, MessageMap*
     }
   }
   if (m_mosquitto) {
-    /*mosquitto_log_init(m_mosquitto, MOSQ_LOG_DEBUG | MOSQ_LOG_ERR | MOSQ_LOG_WARNING
+#if (LIBMOSQUITTO_VERSION_NUMBER >= 1004001)
-                | MOSQ_LOG_NOTICE | MOSQ_LOG_INFO, MOSQ_LOG_STDERR);*/
+    check(mosquitto_threaded_set(m_mosquitto, true), "threaded_set");
+    check(mosquitto_opts_set(m_mosquitto, MOSQ_OPT_PROTOCOL_VERSION, reinterpret_cast<void*>(&g_version)),
+       "opts_set protocol version");
+#endif
     if (g_username || g_password) {
       if (!g_username) {
         g_username = PACKAGE;
@@ -409,8 +501,14 @@ MqttHandler::MqttHandler(UserInfo* userInfo, BusHandler* busHandler, MessageMap*
         logOtherError("mqtt", "unable to set TLS: %d", ret);
       }
     }
+#endif
+#if (LIBMOSQUITTO_VERSION_NUMBER >= 1003001)
+    if (g_logFromLib) {
+      mosquitto_log_callback_set(m_mosquitto, on_log);
+    }
 #endif
     mosquitto_connect_callback_set(m_mosquitto, on_connect);
+    mosquitto_message_callback_set(m_mosquitto, on_message);
     int ret;
 #if (LIBMOSQUITTO_MAJOR >= 1)
     ret = mosquitto_connect(m_mosquitto, g_host, g_port, 60);
@@ -421,16 +519,9 @@ MqttHandler::MqttHandler(UserInfo* userInfo, BusHandler* busHandler, MessageMap*
       logOtherError("mqtt", "unable to connect (invalid parameters)");
       mosquitto_destroy(m_mosquitto);
       m_mosquitto = nullptr;
-    } else if (ret != MOSQ_ERR_SUCCESS) {
+    } else if (!check(ret, "unable to connect, retrying")) {
       m_connected = false;
       m_initialConnectFailed = g_ignoreInvalidParams;
-      char* error;
-      if (ret != MOSQ_ERR_ERRNO) {
-        error = (char*)("unkown mosquitto error");
-      } else {
-        error = strerror(errno);
-      }
-      logOtherError("mqtt", "unable to connect, retrying: %s", error);
     } else {
       m_connected = true;  // assume success until connect_callback says otherwise
       logOtherDebug("mqtt", "connection requested");
@@ -453,27 +544,12 @@ void MqttHandler::start() {
   }
 }
 
-void on_message(
-#if (LIBMOSQUITTO_MAJOR >= 1)
-  struct mosquitto *mosq,
-#endif
-  void *obj, const struct mosquitto_message *message) {
-  MqttHandler* handler = reinterpret_cast<MqttHandler*>(obj);
-  if (!handler || !message || !handler->isRunning()) {
-    return;
-  }
-  string topic(message->topic);
-  string data(message->payloadlen > 0 ? reinterpret_cast<char*>(message->payload) : "");
-  handler->notifyTopic(topic, data);
-}
-
 void MqttHandler::notifyConnected() {
   if (m_mosquitto && isRunning()) {
     const string sep = (g_publishFormat & OF_JSON) ? "\"" : "";
     publishTopic(m_globalTopic+"version", sep + (PACKAGE_STRING "." REVISION) + sep, true);
     publishTopic(m_globalTopic+"running", "true", true);
-    mosquitto_message_callback_set(m_mosquitto, on_message);
+    check(mosquitto_subscribe(m_mosquitto, nullptr, m_subscribeTopic.c_str(), 0), "subscribe");
-    mosquitto_subscribe(m_mosquitto, nullptr, m_subscribeTopic.c_str(), 0);
   }
 }
 
@@ -483,33 +559,43 @@ void MqttHandler::notifyTopic(const string& topic, const string& data) {
     return;
   }
   string direction = topic.substr(pos+1);
-  bool isWrite = false;
   if (direction.empty()) {
     return;
   }
-  isWrite = direction == "set";
+  bool isWrite = direction == "set";
-  if (!isWrite && direction != "get") {
+  bool isList = !isWrite && direction == "list";
+  if (!isWrite && !isList && direction != "get") {
     return;
   }
 
-  logOtherDebug("mqtt", "received topic %s", topic.c_str(), data.c_str());
+  logOtherDebug("mqtt", "received topic %s with data %s", topic.c_str(), data.c_str());
   string remain = topic.substr(0, pos);
   size_t last = 0;
   string circuit, name;
-  size_t idx;
+  bool finalField = false;
-  for (idx = 0; idx < g_topicStrs.size()+1; idx++) {
+  for (size_t idx = 0; idx < g_topicStrs.size()+1 && !finalField; idx++) {
     string field;
     string chk;
     if (idx < g_topicStrs.size()) {
       chk = g_topicStrs[idx];
       pos = remain.find(chk, last);
       if (pos == string::npos) {
-        return;
+        if (!isList) {
+          return;
+        }
+        if (idx == 0 && remain+"/" == chk) {  // check for only first prefix, e.g. "ebusd/"
+          break;
+        }
+        pos = remain.size();
+        finalField = true;
       }
     } else if (idx-1 < g_topicFields.size()) {
       pos = remain.size();
     } else if (last < remain.size()) {
-      return;
+      if (!isList) {
+        return;
+      }
+      break;
     } else {
       break;
     }
@@ -521,7 +607,10 @@ void MqttHandler::notifyTopic(const string& topic, const string& data) {
       }
     } else {
       if (field.empty()) {
-        return;
+        if (!isList) {
+          return;
+        }
+        continue;
       }
       string fieldName = g_topicFields[idx-1];
       if (fieldName == "circuit") {
@@ -535,10 +624,25 @@ void MqttHandler::notifyTopic(const string& topic, const string& data) {
       }
     }
   }
+  if (isList) {
+    logOtherInfo("mqtt", "received list topic for %s %s", circuit.c_str(), name.c_str());
+    deque<Message*> messages;
+    m_messages->findAll(circuit, name, m_levels, true, true, true, true, true, true, 0, 0, false, &messages);
+    bool onlyWithData = !data.empty();
+    for (const auto message : messages) {
+      time_t lastup = message->getLastUpdateTime();
+      if (onlyWithData && lastup == 0) {
+        continue;
+      }
+      ostringstream ostream;
+      publishMessage(message, &ostream, true);
+    }
+    return;
+  }
   if (name.empty()) {
     return;
   }
-  logOtherInfo("mqtt", "received topic for %s %s", circuit.c_str(), name.c_str());
+  logOtherInfo("mqtt", "received %s topic for %s %s", direction.c_str(), circuit.c_str(), name.c_str());
   Message* message = m_messages->find(circuit, name, m_levels, isWrite);
   if (message == nullptr) {
     message = m_messages->find(circuit, name, m_levels, isWrite, true);
@@ -548,7 +652,25 @@ void MqttHandler::notifyTopic(const string& topic, const string& data) {
     return;
   }
   if (!message->isPassive()) {
-    result_t result = m_busHandler->readFromBus(message, data);
+    string useData = data;
+    if (!isWrite && !data.empty()) {
+      size_t pos = useData.find_last_of('?');
+      if (pos != string::npos && pos > 0 && useData[pos-1] != UI_FIELD_SEPARATOR) {
+        pos = string::npos;
+      }
+      if (pos != string::npos) {
+        string args = useData.substr(pos + 1);
+        useData = useData.substr(0, pos > 0 ? pos - 1 : pos);
+        if (!args.empty()) {
+          result_t ret = RESULT_OK;
+          size_t pollPriority = (size_t)parseInt(args.c_str(), 10, 1, 9, &ret);
+          if (ret == RESULT_OK && pollPriority > 0 && message->setPollPriority(pollPriority)) {
+            m_messages->addPollMessage(false, message);
+          }
+        }
+      }
+    }
+    result_t result = m_busHandler->readFromBus(message, useData);
     if (result != RESULT_OK) {
       logOtherError("mqtt", "%s %s %s: %s", isWrite?"write":"read", circuit.c_str(), name.c_str(),
           getResultCode(result));
@@ -580,7 +702,7 @@ void MqttHandler::run() {
   bool allowReconnect = false;
   while (isRunning()) {
     bool wasConnected = m_connected;
-    handleTraffic(allowReconnect);
+    bool needsWait = handleTraffic(allowReconnect);
     bool reconnected = !wasConnected && m_connected;
     allowReconnect = false;
     time(&now);
@@ -606,12 +728,12 @@ void MqttHandler::run() {
         lastSignal = now;
         if (!signal || reconnected) {
           signal = true;
-          publishTopic(signalTopic, "true");
+          publishTopic(signalTopic, "true", true);
         }
       } else {
         if (signal || reconnected) {
           signal = false;
-          publishTopic(signalTopic, "false");
+          publishTopic(signalTopic, "false", true);
         }
       }
     }
@@ -639,52 +761,60 @@ void MqttHandler::run() {
         m_updatedMessages.clear();
       }
     }
-    if (!m_connected && !Wait(5)) {
+    if ((!m_connected && !Wait(5)) || (needsWait && !Wait(1))) {
       break;
     }
   }
+  publishTopic(signalTopic, "false", true);
 }
 
-void MqttHandler::handleTraffic(bool allowReconnect) {
+bool MqttHandler::handleTraffic(bool allowReconnect) {
-  if (m_mosquitto) {
+  if (!m_mosquitto) {
-    int ret;
+    return false;
+  }
+  int ret;
 #if (LIBMOSQUITTO_MAJOR >= 1)
-    ret = mosquitto_loop(m_mosquitto, -1, 1);  // waits up to 1 second for network traffic
+  ret = mosquitto_loop(m_mosquitto, -1, 1);  // waits up to 1 second for network traffic
 #else
-    ret = mosquitto_loop(m_mosquitto, -1);  // waits up to 1 second for network traffic
+  ret = mosquitto_loop(m_mosquitto, -1);  // waits up to 1 second for network traffic
 #endif
-    if (!m_connected && ret == MOSQ_ERR_NO_CONN && allowReconnect) {
+  if (!m_connected && ret == MOSQ_ERR_NO_CONN && allowReconnect) {
-      if (m_initialConnectFailed) {
+    if (m_initialConnectFailed) {
 #if (LIBMOSQUITTO_MAJOR >= 1)
-        ret = mosquitto_connect(m_mosquitto, g_host, g_port, 60);
+      ret = mosquitto_connect(m_mosquitto, g_host, g_port, 60);
 #else
-        ret = mosquitto_connect(m_mosquitto, g_host, g_port, 60, true);
+      ret = mosquitto_connect(m_mosquitto, g_host, g_port, 60, true);
 #endif
-        if (ret == MOSQ_ERR_INVAL) {
+      if (ret == MOSQ_ERR_INVAL) {
-          logOtherError("mqtt", "unable to connect (invalid parameters), retrying");
+        logOtherError("mqtt", "unable to connect (invalid parameters), retrying");
-        }
+      }
-        if (ret == MOSQ_ERR_SUCCESS) {
+      if (ret == MOSQ_ERR_SUCCESS) {
-          m_initialConnectFailed = false;
+        m_initialConnectFailed = false;
-        }
-      } else {
-        ret = mosquitto_reconnect(m_mosquitto);
       }
-    }
-    if (!m_connected && ret == MOSQ_ERR_SUCCESS) {
-      m_connected = true;
-      logOtherNotice("mqtt", "connection re-established");
-    }
-    if (!m_connected || ret == MOSQ_ERR_SUCCESS) {
-      return;
-    }
-    if (ret == MOSQ_ERR_NO_CONN || ret == MOSQ_ERR_CONN_LOST || ret == MOSQ_ERR_CONN_REFUSED) {
-      logOtherError("mqtt", "communication error: %s", ret == MOSQ_ERR_NO_CONN ? "not connected"
-                    : (ret == MOSQ_ERR_CONN_LOST ? "connection lost" : "connection refused"));
-      m_connected = false;
     } else {
-      logOtherError("mqtt", "communication error: %d", ret);
+      ret = mosquitto_reconnect(m_mosquitto);
     }
   }
+  if (!m_connected && ret == MOSQ_ERR_SUCCESS) {
+    m_connected = true;
+    logOtherNotice("mqtt", "connection re-established");
+  }
+  if (!m_connected || ret == MOSQ_ERR_SUCCESS) {
+    return false;
+  }
+  if (ret == MOSQ_ERR_NO_CONN || ret == MOSQ_ERR_CONN_LOST || ret == MOSQ_ERR_CONN_REFUSED) {
+    logOtherError("mqtt", "communication error: %s", ret == MOSQ_ERR_NO_CONN ? "not connected"
+                  : (ret == MOSQ_ERR_CONN_LOST ? "connection lost" : "connection refused"));
+    m_connected = false;
+  } else {
+    time_t now;
+    time(&now);
+    if (now > m_lastErrorLogTime + 10) {  // log at most every 10 seconds
+      m_lastErrorLogTime = now;
+      check(ret, "communication error");
+    }
+  }
+  return true;
 }
 
 string MqttHandler::getTopic(const Message* message, const string& suffix, const string& fieldName) {
@@ -708,10 +838,15 @@ string MqttHandler::getTopic(const Message* message, const string& suffix, const
   return ret.str();
 }
 
-void MqttHandler::publishMessage(const Message* message, ostringstream* updates) {
+void MqttHandler::publishMessage(const Message* message, ostringstream* updates, bool includeWithoutData) {
   OutputFormat outputFormat = g_publishFormat;
   bool json = outputFormat & OF_JSON;
+  bool noData = includeWithoutData && message->getLastUpdateTime() == 0;
   if (!m_publishByField) {
+    if (noData) {
+      publishEmptyTopic(getTopic(message));  // alternatively: , json ? "null" : "");
+      return;
+    }
     if (json) {
       *updates << "{";
     }
@@ -732,6 +867,10 @@ void MqttHandler::publishMessage(const Message* message, ostringstream* updates)
   }
   for (size_t index = 0; index < message->getFieldCount(); index++) {
     string name = message->getFieldName(index);
+    if (noData) {
+      publishEmptyTopic(getTopic(message, "", name));  // alternatively: , json ? "null" : "");
+      continue;
+    }
     result_t result = message->decodeLastData(false, nullptr, index, outputFormat, updates);
     if (result != RESULT_OK) {
       logOtherError("mqtt", "decode %s %s %s: %s", message->getCircuit().c_str(), message->getName().c_str(),
@@ -745,9 +884,18 @@ void MqttHandler::publishMessage(const Message* message, ostringstream* updates)
 }
 
 void MqttHandler::publishTopic(const string& topic, const string& data, bool retain) {
-  logOtherDebug("mqtt", "publish %s %s", topic.c_str(), data.c_str());
+  const char* topicStr = topic.c_str();
-  mosquitto_publish(m_mosquitto, nullptr, topic.c_str(), (uint32_t)data.size(),
+  const char* dataStr = data.c_str();
-      reinterpret_cast<const uint8_t*>(data.c_str()), 0, g_retain || retain);
+  const size_t len = strlen(dataStr);
+  logOtherDebug("mqtt", "publish %s %s", topicStr, dataStr);
+  check(mosquitto_publish(m_mosquitto, nullptr, topicStr, (uint32_t)len,
+      reinterpret_cast<const uint8_t*>(dataStr), 0, g_retain || retain), "publish");
+}
+
+void MqttHandler::publishEmptyTopic(const string& topic) {
+  const char* topicStr = topic.c_str();
+  logOtherDebug("mqtt", "publish empty %s", topicStr);
+  check(mosquitto_publish(m_mosquitto, nullptr, topicStr, 0, nullptr, 0, g_retain), "publish empty");
 }
 
 }  // namespace ebusd

src/ebusd/mqtthandler.h

@@ -100,8 +100,9 @@ class MqttHandler : public DataSink, public DataSource, public WaitThread {
   /**
    * Called regularly to handle MQTT traffic.
    * @param allowReconnect true when reconnecting to the broker is allowed.
+   * @return true on error for waiting a bit until next call, or false otherwise.
    */
-  void handleTraffic(bool allowReconnect);
+  bool handleTraffic(bool allowReconnect);
 
   /**
    * Build the MQTT topic string for the @a Message.
@@ -116,8 +117,9 @@ class MqttHandler : public DataSink, public DataSource, public WaitThread {
    * Prepare a @a Message and publish as topic.
    * @param message the @a Message to publish.
    * @param updates the @a ostringstream for preparation.
+   * @param includeWithoutData whether to publish messages without data as well.
    */
-  void publishMessage(const Message* message, ostringstream* updates);
+  void publishMessage(const Message* message, ostringstream* updates, bool includeWithoutData = false);
 
   /**
    * Publish a topic update to MQTT.
@@ -127,6 +129,12 @@ class MqttHandler : public DataSink, public DataSource, public WaitThread {
    */
   void publishTopic(const string& topic, const string& data, bool retain = false);
 
+  /**
+   * Publish a topic update to MQTT without any data.
+   * @param topic the topic string.
+   */
+  void publishEmptyTopic(const string& topic);
+
   /** the @a MessageMap instance. */
   MessageMap* m_messages;
 
@@ -150,6 +158,9 @@ class MqttHandler : public DataSink, public DataSource, public WaitThread {
 
   /** the last update check result. */
   string m_lastUpdateCheckResult;
+
+  /** the last system time when a communication error was logged. */
+  time_t m_lastErrorLogTime;
 };
 
 }  // namespace ebusd

src/ebusd/network.cpp

@@ -65,7 +65,7 @@ bool NetMessage::add(const char* request) {
     }
     return true;
   }
-  return m_request.length() == 0 && m_listening;
+  return m_request.length() == 0 && isListeningMode();
 }
 
 
@@ -144,7 +144,7 @@ void Connection::run() {
 #endif
     }
 
-    if (newData || message.isListening()) {
+    if (newData || message.isListeningMode()) {
       char data[256];
 
       if (!m_socket->isValid()) {
@@ -208,7 +208,7 @@ Network::~Network() {
   stop();
   NetMessage* netMsg;
   while ((netMsg = m_netQueue->pop()) != nullptr) {
-    netMsg->setResult("ERR: shutdown", "", false, 0, true);
+    netMsg->setResult("ERR: shutdown", "", cm_normal, 0, true);
   }
   while (!m_connections.empty()) {
     Connection* connection = m_connections.back();

src/ebusd/network.h

@@ -37,6 +37,13 @@ namespace ebusd {
 /** Forward declaration for @a Connection. */
 class Connection;
 
+/** the possible client modes. */
+enum ClientMode {
+  cm_normal,  //!< normal mode
+  cm_listen,  //!< listening mode
+  cm_direct,  //!< direct mode
+};
+
 /**
  * Class for data/message transfer between @a Connection and @a MainLoop.
  */
@@ -47,7 +54,7 @@ class NetMessage {
    * @param isHttp whether this is a HTTP message.
    */
   explicit NetMessage(bool isHttp)
-    : m_isHttp(isHttp), m_resultSet(false), m_disconnect(false), m_listening(false), m_listenSince(0) {
+    : m_isHttp(isHttp), m_resultSet(false), m_disconnect(false), m_mode(cm_normal), m_listenSince(0) {
     pthread_mutex_init(&m_mutex, nullptr);
     pthread_cond_init(&m_cond, nullptr);
   }
@@ -117,16 +124,16 @@ class NetMessage {
    * Set the result string and notify the waiting thread.
    * @param result the result string.
    * @param user the new user name.
-   * @param listening whether the client is in listening mode.
+   * @param mode the new client mode.
    * @param listenUntil the end time to which to updates were added (exclusive).
    * @param disconnect true when the client shall be disconnected.
    */
-  void setResult(const string& result, const string& user, bool listening, time_t listenUntil, bool disconnect) {
+  void setResult(const string& result, const string& user, ClientMode mode, time_t listenUntil, bool disconnect) {
     pthread_mutex_lock(&m_mutex);
     m_result = result;
     m_user = user;
     m_disconnect = disconnect;
-    m_listening = listening;
+    m_mode = mode;
     m_listenSince = listenUntil;
     m_resultSet = true;
     pthread_cond_signal(&m_cond);
@@ -134,17 +141,23 @@ class NetMessage {
   }
 
   /**
-   * Return whether the client is in listening mode.
+   * Return the client mode.
-   * @param listenSince set to the start time from which to add updates (inclusive).
+   * @param listenSince set listening to the specified start time from which to add updates (inclusive).
-   * @return whether the client is in listening mode.
+   * @return the client mode.
    */
-  bool isListening(time_t* listenSince = nullptr) {
+  ClientMode getMode(time_t* listenSince = nullptr) {
     if (listenSince) {
       *listenSince = m_listenSince;
     }
-    return m_listening;
+    return m_mode;
   }
 
+  /**
+   * Return whether this instance is in one of the listening modes.
+   * @return whether this instance is in one of the listening modes.
+   */
+  bool isListeningMode() { return m_mode == cm_listen || m_mode == cm_direct; }
+
   /**
    * Return whether the client shall be disconnected.
    * @return true when the client shall be disconnected.
@@ -177,8 +190,8 @@ class NetMessage {
   /** condition variable for exclusive lock. */
   pthread_cond_t m_cond;
 
-  /** whether the client is in listening mode. */
+  /** the client mode. */
-  bool m_listening;
+  ClientMode m_mode;
 
   /** start timestamp of listening update. */
   time_t m_listenSince;
@@ -284,7 +297,7 @@ class Network : public Thread {
   /** @a Notify object for shutdown procedure. */
   Notify m_notify;
 
-  /** true if this instance is listening */
+  /** true if this instance is listening. */
   bool m_listening;
 
   /**

src/lib/ebus/data.cpp

@@ -647,8 +647,11 @@ size_t SingleDataField::getLength(PartType partType, size_t maxLength) const {
   return remainder ? maxLength : m_length;
 }
 
-bool SingleDataField::hasFullByteOffset(bool after) const {
+bool SingleDataField::hasFullByteOffset(bool after, int16_t& previousFirstBit) const {
   if (m_length > 1) {
+    if (after) {
+      previousFirstBit = -1;
+    }
     return true;
   }
   int16_t firstBit;
@@ -658,8 +661,15 @@ bool SingleDataField::hasFullByteOffset(bool after) const {
   } else {
     firstBit = 0;
   }
-  return (m_dataType->getBitCount() % 8) == 0
+  bool ret = (m_dataType->getBitCount() % 8) == 0
-  || (after && firstBit + (m_dataType->getBitCount() % 8) >= 8);
+  || (firstBit == previousFirstBit) || (after && firstBit + (m_dataType->getBitCount() % 8) >= 8);
+  // std::cout<<(after?"after,":"before,")<<"prev="<<static_cast<unsigned>(previousFirstBit)<<",first="
+  // <<static_cast<unsigned>(firstBit)<<",length="<<static_cast<unsigned>(m_dataType->getBitCount())
+  // <<" => "<<(ret?"true":"false")<<"\n";
+  if (after) {
+    previousFirstBit = ret ? -1 : firstBit;
+  }
+  return ret;
 }
 
 size_t SingleDataField::getCount(PartType partType, const char* fieldName) const {
@@ -915,9 +925,10 @@ const DataFieldSet* DataFieldSet::clone() const {
 size_t DataFieldSet::getLength(PartType partType, size_t maxLength) const {
   size_t length = 0;
   bool previousFullByteOffset[] = { true, true, true, true };
+  int16_t previousFirstBit[] = { -1, -1, -1, -1 };
   for (const auto field : m_fields) {
     if (field->getPartType() == partType) {
-      if (!previousFullByteOffset[partType] && !field->hasFullByteOffset(false)) {
+      if (!previousFullByteOffset[partType] && !field->hasFullByteOffset(false, previousFirstBit[partType])) {
         length--;
       }
       size_t fieldLength = field->getLength(partType, maxLength);
@@ -928,7 +939,7 @@ size_t DataFieldSet::getLength(PartType partType, size_t maxLength) const {
       }
       length = length + fieldLength;
 
-      previousFullByteOffset[partType] = field->hasFullByteOffset(true);
+      previousFullByteOffset[partType] = field->hasFullByteOffset(true, previousFirstBit[partType]);
     }
   }
 
@@ -1010,12 +1021,13 @@ void DataFieldSet::dump(bool prependFieldSeparator, bool asJson, ostream* output
 result_t DataFieldSet::read(const SymbolString& data, size_t offset,
     const char* fieldName, ssize_t fieldIndex, unsigned int* output) const {
   bool previousFullByteOffset = true, found = false, findFieldIndex = fieldIndex >= 0;
+  int16_t previousFirstBit = -1;
   PartType partType = data.isMaster() ? pt_masterData : pt_slaveData;
   for (const auto field : m_fields) {
     if (field->getPartType() != partType) {
       continue;
     }
-    if (!previousFullByteOffset && !field->hasFullByteOffset(false)) {
+    if (!previousFullByteOffset && !field->hasFullByteOffset(false, previousFirstBit)) {
       offset--;
     }
     result_t result = field->read(data, offset, fieldName, fieldIndex, output);
@@ -1023,7 +1035,7 @@ result_t DataFieldSet::read(const SymbolString& data, size_t offset,
       return result;
     }
     offset += field->getLength(partType, data.getDataSize()-offset);
-    previousFullByteOffset = field->hasFullByteOffset(true);
+    previousFullByteOffset = field->hasFullByteOffset(true, previousFirstBit);
     if (result != RESULT_EMPTY) {
       found = true;
     }
@@ -1049,6 +1061,7 @@ result_t DataFieldSet::read(const SymbolString& data, size_t offset,
     bool leadingSeparator, const char* fieldName, ssize_t fieldIndex,
     OutputFormat outputFormat, ssize_t outputIndex, ostream* output) const {
   bool previousFullByteOffset = true, found = false, findFieldIndex = fieldIndex >= 0;
+  int16_t previousFirstBit = -1;
   if (outputIndex < 0 && (!m_uniqueNames || ((outputFormat & OF_JSON) && !(outputFormat & OF_NAMES)))) {
     outputIndex = 0;
   }
@@ -1060,7 +1073,7 @@ result_t DataFieldSet::read(const SymbolString& data, size_t offset,
       }
       continue;
     }
-    if (!previousFullByteOffset && !field->hasFullByteOffset(false)) {
+    if (!previousFullByteOffset && !field->hasFullByteOffset(false, previousFirstBit)) {
       offset--;
     }
     result_t result = field->read(data, offset, leadingSeparator, fieldName, fieldIndex,
@@ -1069,7 +1082,7 @@ result_t DataFieldSet::read(const SymbolString& data, size_t offset,
       return result;
     }
     offset += field->getLength(partType, data.getDataSize()-offset);
-    previousFullByteOffset = field->hasFullByteOffset(true);
+    previousFullByteOffset = field->hasFullByteOffset(true, previousFirstBit);
     if (result != RESULT_EMPTY) {
       found = true;
       leadingSeparator = true;
@@ -1099,12 +1112,13 @@ result_t DataFieldSet::write(char separator, size_t offset, istringstream* input
   string token;
   PartType partType = data->isMaster() ? pt_masterData : pt_slaveData;
   bool previousFullByteOffset = true;
+  int16_t previousFirstBit = -1;
   size_t baseOffset = offset;
   for (const auto field : m_fields) {
     if (field->getPartType() != partType) {
       continue;
     }
-    if (!previousFullByteOffset && !field->hasFullByteOffset(false)) {
+    if (!previousFullByteOffset && !field->hasFullByteOffset(false, previousFirstBit)) {
       offset--;
     }
     result_t result;
@@ -1124,7 +1138,7 @@ result_t DataFieldSet::write(char separator, size_t offset, istringstream* input
       return result;
     }
     offset += fieldLength;
-    previousFullByteOffset = field->hasFullByteOffset(true);
+    previousFullByteOffset = field->hasFullByteOffset(true, previousFirstBit);
   }
 
   if (usedLength != nullptr) {

src/lib/ebus/data.h

@@ -401,10 +401,13 @@ class SingleDataField : public DataField {
   /**
    * Get whether this field uses a full byte offset.
    * @param after @p true to check after consuming the bits, @p false to check before.
+   * @param previousFirstBit the index to the first bit of the previous field, or -1
+   * if the previous field used a full byte offset. Will be updated during the call
+   * when after was true.
    * @return @p true if this field uses a full byte offset, @p false if this field
    * only consumes a part of a byte and a subsequent field may re-use the same offset.
    */
-  bool hasFullByteOffset(bool after) const;
+  bool hasFullByteOffset(bool after, int16_t& previousFirstBit) const;
 
   // @copydoc
   size_t getCount(PartType partType = pt_any, const char* fieldName = nullptr) const override;

src/lib/ebus/datatype.cpp

@@ -696,7 +696,7 @@ result_t NumberDataType::readSymbols(size_t offset, size_t length, const SymbolS
         }
       }
 #endif
-      if (isnan(val)) {
+      if (val != val) {  // !isnan(val)
         if (outputFormat & OF_JSON) {
           *output << "null";
         } else {

src/lib/ebus/device.cpp

@@ -30,6 +30,9 @@
 #ifdef HAVE_LINUX_SERIAL
 #  include <linux/serial.h>
 #endif
+#ifdef HAVE_FREEBSD_UFTDI
+#  include <dev/usb/uftdiio.h>
+#endif
 #include <errno.h>
 #ifdef HAVE_PPOLL
 #  include <poll.h>
@@ -392,7 +395,7 @@ result_t SerialDevice::open() {
   struct termios newSettings;
 
   // open file descriptor
-  m_fd = ::open(m_name, O_RDWR | O_NOCTTY);
+  m_fd = ::open(m_name, O_RDWR | O_NOCTTY | O_NDELAY);
 
   if (m_fd < 0) {
     return RESULT_ERR_NOTFOUND;
@@ -415,13 +418,24 @@ result_t SerialDevice::open() {
   }
 #endif
 
+#ifdef HAVE_FREEBSD_UFTDI
+  int param = 0;
+  // flush tx/rx and set low latency on uftdi device
+  if (ioctl(m_fd, UFTDIIOC_GET_LATENCY, &param) == 0) {
+    ioctl(m_fd, UFTDIIOC_RESET_IO, &param);
+    param = 1;
+    ioctl(m_fd, UFTDIIOC_SET_LATENCY, &param);
+  }
+#endif
+
   // save current settings
   tcgetattr(m_fd, &m_oldSettings);
 
   // create new settings
   memset(&newSettings, 0, sizeof(newSettings));
 
-  newSettings.c_cflag |= ((m_enhancedProto ? B115200 : B2400) | CS8 | CLOCAL | CREAD);
+  cfsetspeed(&newSettings, m_enhancedProto ? B115200 : B2400);
+  newSettings.c_cflag |= (CS8 | CLOCAL | CREAD);
   newSettings.c_lflag &= ~(ICANON | ECHO | ECHOE | ISIG);  // non-canonical mode
   newSettings.c_iflag |= IGNPAR;  // ignore parity errors
   newSettings.c_oflag &= ~OPOST;
@@ -434,7 +448,10 @@ result_t SerialDevice::open() {
   tcflush(m_fd, TCIFLUSH);
 
   // activate new settings of serial device
-  tcsetattr(m_fd, TCSAFLUSH, &newSettings);
+  if (tcsetattr(m_fd, TCSAFLUSH, &newSettings)) {
+    close();
+    return RESULT_ERR_DEVICE;
+  }
 
   // set serial device into blocking mode
   fcntl(m_fd, F_SETFL, fcntl(m_fd, F_GETFL) & ~O_NONBLOCK);

src/lib/ebus/device.h

@@ -22,6 +22,7 @@
 #include <unistd.h>
 #include <termios.h>
 #include <arpa/inet.h>
+#include <netinet/in.h>
 #include <netdb.h>
 #include <iostream>
 #include <fstream>

src/lib/ebus/filereader.cpp

@@ -120,7 +120,7 @@ void FileReader::tolower(string* str) {
 
 static size_t hashFunction(const string& str) {
   size_t hash = 0;
-  for (char c : str) {
+  for (unsigned char c : str) {
     hash = (31 * hash) ^ c;
   }
   return hash;

src/lib/ebus/message.cpp

@@ -45,12 +45,24 @@ using std::endl;
 /** the bit mask for the ID length and combined ID bytes in the message key. */
 #define ID_LENGTH_AND_IDS_MASK ((7LL << (8 * 7 + 5)) | 0xffffffffLL)
 
-/** the bits in the @a ID_SOURCE_MASK for arbitrary source and active read message. */
+/** the bits in the @a ID_SOURCE_MASK for arbitrary source and active write message. */
 #define ID_SOURCE_ACTIVE_WRITE (0x1fLL << (8 * 7))
 
-/** the bits in the @a ID_SOURCE_MASK for arbitrary source and active write message. */
+/** the bits in the @a ID_SOURCE_MASK for arbitrary source and active read message. */
 #define ID_SOURCE_ACTIVE_READ (0x1eLL << (8 * 7))
 
+/**
+ * the bits in the @a ID_SOURCE_MASK for arbitrary source and active write message
+ * to a master (same value as ID_SOURCE_ACTIVE_WRITE for now).
+ */
+#define ID_SOURCE_ACTIVE_WRITE_MASTER (0x1fLL << (8 * 7))
+
+/**
+ * the bits in the @a ID_SOURCE_MASK for arbitrary source and active read message
+ * to a master (same value as ID_SOURCE_ACTIVE_WRITE for now).
+ */
+#define ID_SOURCE_ACTIVE_READ_MASTER (0x1eLL << (8 * 7))
+
 /** special value for invalid message key. */
 #define INVALID_KEY 0xffffffffffffffffLL
 
@@ -79,6 +91,9 @@ static const char* defaultMessageFieldMap[] = {  // access level not included in
     "*name", "part", "type", "divisor/values", "unit", "comment",
 };
 
+/** the m_pollOrder of the last polled message. */
+static unsigned int g_lastPollOrder = 0;
+
 extern DataFieldTemplates* getTemplates(const string& filename);
 
 extern result_t loadDefinitionsFromConfigPath(FileReader* reader, const string& filename, bool verbose,
@@ -99,7 +114,7 @@ Message::Message(const string& circuit, const string& level, const string& name,
       m_data(data), m_deleteData(deleteData),
       m_pollPriority(pollPriority),
       m_usedByCondition(false), m_isScanMessage(false), m_condition(condition),
-      m_lastUpdateTime(0), m_lastChangeTime(0), m_pollCount(0), m_lastPollTime(0) {
+      m_lastUpdateTime(0), m_lastChangeTime(0), m_pollOrder(0), m_lastPollTime(0) {
   if (circuit == "scan") {
     setScanMessage();
     m_pollPriority = 0;
@@ -116,7 +131,7 @@ Message::Message(const string& circuit, const string& level, const string& name,
       m_data(data), m_deleteData(deleteData),
       m_pollPriority(0),
       m_usedByCondition(false), m_isScanMessage(true), m_condition(nullptr),
-      m_lastUpdateTime(0), m_lastChangeTime(0), m_pollCount(0), m_lastPollTime(0) {
+      m_lastUpdateTime(0), m_lastChangeTime(0), m_pollOrder(0), m_lastPollTime(0) {
 }
 
 
@@ -158,7 +173,8 @@ uint64_t Message::createKey(const vector<symbol_t>& id, bool isWrite, bool isPas
   if (isPassive) {
     key |= (uint64_t)getMasterNumber(srcAddress) << (8 * 7);  // 0..25
   } else {
-    key |= (isWrite ? 0x1fLL : 0x1eLL) << (8 * 7);  // special values for active
+    key |= isMaster(dstAddress) ? (isWrite ? ID_SOURCE_ACTIVE_WRITE_MASTER : ID_SOURCE_ACTIVE_READ_MASTER)
+           : (isWrite ? ID_SOURCE_ACTIVE_WRITE : ID_SOURCE_ACTIVE_READ);  // special values for active
   }
   key |= (uint64_t)dstAddress << (8 * 6);
   int exp = 5;
@@ -199,7 +215,7 @@ uint64_t Message::createKey(const MasterSymbolString& master, size_t maxIdLength
 
 uint64_t Message::createKey(symbol_t pb, symbol_t sb, bool broadcast) {
   uint64_t key = 0;
-  key |= (broadcast ? 0x1fLL : 0x1eLL) << (8 * 7);  // special values for active
+  key |= broadcast ? ID_SOURCE_ACTIVE_WRITE : ID_SOURCE_ACTIVE_READ;  // special values for active
   key |= (uint64_t)(broadcast ? BROADCAST : SYN) << (8 * 6);
   key |= (uint64_t)pb << (8 * 5);
   key |= (uint64_t)sb << (8 * 4);
@@ -596,9 +612,29 @@ bool Message::setPollPriority(size_t priority) {
   }
   bool ret = m_pollPriority == 0 && usePriority > 0;
   m_pollPriority = usePriority;
+  if (ret || m_pollOrder > g_lastPollOrder+(unsigned int)m_pollPriority) {
+    // ensure a later increased or newly set priority does not prefer that message before all others
+    m_pollOrder = g_lastPollOrder+(unsigned int)m_pollPriority;
+  }
   return ret;
 }
 
+bool Message::isLessPollWeight(const Message* other) const {
+  if (m_pollOrder > other->m_pollOrder) {
+    return true;
+  }
+  if (m_pollOrder < other->m_pollOrder) {
+    return false;
+  }
+  if (m_pollPriority > other->m_pollPriority) {
+    return true;
+  }
+  if (m_pollPriority < other->m_pollPriority) {
+    return false;
+  }
+  return m_lastPollTime > other->m_lastPollTime;
+}
+
 void Message::setUsedByCondition() {
   if (m_usedByCondition) {
     return;
@@ -786,26 +822,6 @@ result_t Message::decodeLastDataNumField(const char* fieldName, ssize_t fieldInd
   return result;
 }
 
-bool Message::isLessPollWeight(const Message* other) const {
-  size_t tprio = m_pollPriority;
-  size_t oprio = other->m_pollPriority;
-  size_t tw = tprio * m_pollCount;
-  size_t ow = oprio * other->m_pollCount;
-  if (tw > ow) {
-    return true;
-  }
-  if (tw < ow) {
-    return false;
-  }
-  if (tprio > oprio) {
-    return true;
-  }
-  if (tprio < oprio) {
-    return false;
-  }
-  return m_lastPollTime > other->m_lastPollTime;
-}
-
 void Message::dumpHeader(const vector<string>* fieldNames, ostream* output) {
   bool first = true;
   if (fieldNames == nullptr) {
@@ -1267,6 +1283,7 @@ void ChainedMessage::dumpField(const string& fieldName, bool withConditions, ost
  * @param sameIdExtAs the optional @a MasterSymbolString to check for having the same ID.
  * @param onlyAvailable true to include only available messages (default true), false to also include messages that
  * are currently not available (e.g. due to unresolved or false conditions).
+ * @return the first available @a Message from the list.
  */
 Message* getFirstAvailable(const vector<Message*>& messages, const MasterSymbolString* sameIdExtAs,
     const bool onlyAvailable = true) {
@@ -1287,6 +1304,7 @@ Message* getFirstAvailable(const vector<Message*>& messages, const MasterSymbolS
  * @param sameIdExtAs the optional @a Message to check for having the same ID.
  * @param onlyAvailable true to include only available messages (default true), false to also include messages that
  * are currently not available (e.g. due to unresolved or false conditions).
+ * @return the first available @a Message from the list.
  */
 Message* getFirstAvailable(const vector<Message*>& messages, const Message* sameIdExtAs = nullptr,
     const bool onlyAvailable = true) {
@@ -2566,6 +2584,14 @@ void MessageMap::findAll(const string& circuit, const string& name, const string
   }
 }
 
+Message* MessageMap::getFirstAvailableFromIterator(const map<uint64_t, vector<Message*> >::const_iterator& it,
+    const MasterSymbolString* sameIdExtAs, bool onlyAvailable) const {
+  if (it != m_messagesByKey.end()) {
+    return getFirstAvailable(it->second, sameIdExtAs, onlyAvailable);
+  }
+  return nullptr;
+}
+
 Message* MessageMap::find(const MasterSymbolString& master, bool anyDestination,
   bool withRead, bool withWrite, bool withPassive, bool onlyAvailable) const {
   if (anyDestination && master.size() >= 5 && master[4] == 0 && master[2] == 0x07 && master[3] == 0x04) {
@@ -2576,6 +2602,7 @@ Message* MessageMap::find(const MasterSymbolString& master, bool anyDestination,
   if (baseKey == INVALID_KEY) {
     return nullptr;
   }
+  bool isWriteDest = isMaster(master[1]) || master[1] == BROADCAST;
   size_t maxIdLength = Message::getKeyLength(baseKey);
   for (size_t idLength = maxIdLength; true; idLength--) {
     uint64_t key = baseKey;
@@ -2591,44 +2618,39 @@ Message* MessageMap::find(const MasterSymbolString& master, bool anyDestination,
         }
       }
     }
-    map<uint64_t, vector<Message*> >::const_iterator it;
+    Message* message;
     if (withPassive) {
-      it = m_messagesByKey.find(key);
+      message = getFirstAvailableFromIterator(m_messagesByKey.find(key), &master, onlyAvailable);
-      if (it != m_messagesByKey.end()) {
+      if (message) {
-        Message* message = getFirstAvailable(it->second, &master, onlyAvailable);
+        return message;
+      }
+    }
+    if ((key & ID_SOURCE_MASK) != 0) {
+      key &= ~ID_SOURCE_MASK;
+      if (withPassive) {
+        // try again without specific source master
+        message = getFirstAvailableFromIterator(m_messagesByKey.find(key), &master, onlyAvailable);
         if (message) {
           return message;
         }
       }
-      if ((key & ID_SOURCE_MASK) != 0) {
-        key &= ~ID_SOURCE_MASK;
-        it = m_messagesByKey.find(key & ~ID_SOURCE_MASK);  // try again without specific source master
-        if (it != m_messagesByKey.end()) {
-          Message* message = getFirstAvailable(it->second, &master, onlyAvailable);
-          if (message) {
-            return message;
-          }
-        }
-      }
-    } else {
-      key &= ~ID_SOURCE_MASK;
     }
     if (withRead) {
-      it = m_messagesByKey.find(key | ID_SOURCE_ACTIVE_READ);  // try again with special value for active read
+      // try again with special value for active read
-      if (it != m_messagesByKey.end()) {
+      message = getFirstAvailableFromIterator(
-        Message* message = getFirstAvailable(it->second, &master, onlyAvailable);
+        m_messagesByKey.find(key | (isWriteDest ? ID_SOURCE_ACTIVE_READ_MASTER : ID_SOURCE_ACTIVE_READ)),
-        if (message) {
+        &master, onlyAvailable);
-          return message;
+      if (message) {
-        }
+        return message;
       }
     }
     if (withWrite) {
-      it = m_messagesByKey.find(key | ID_SOURCE_ACTIVE_WRITE);  // try again with special value for active write
+      // try again with special value for active write
-      if (it != m_messagesByKey.end()) {
+      message = getFirstAvailableFromIterator(
-        Message* message = getFirstAvailable(it->second, &master, onlyAvailable);
+        m_messagesByKey.find(key | (isWriteDest ? ID_SOURCE_ACTIVE_WRITE_MASTER : ID_SOURCE_ACTIVE_WRITE)),
-        if (message) {
+        &master, onlyAvailable);
-          return message;
+      if (message) {
-        }
+        return message;
       }
     }
     if (idLength == 0) {
@@ -2751,11 +2773,16 @@ Message* MessageMap::getNextPoll() {
   if (m_pollMessages.empty()) {
     return nullptr;
   }
+  lock();
   Message* ret = m_pollMessages.top();
   m_pollMessages.pop();
-  ret->m_pollCount++;
+  if (ret->m_pollOrder > g_lastPollOrder) {
+    g_lastPollOrder = ret->m_pollOrder;
+  }
+  ret->m_pollOrder += (unsigned int)ret->m_pollPriority;
   time(&(ret->m_lastPollTime));
   m_pollMessages.push(ret);  // re-insert at new position
+  unlock();
   return ret;
 }
 

src/lib/ebus/message.h

@@ -642,8 +642,8 @@ class Message : public AttributedItem {
   /** the system time when the message content was last changed, 0 for never. */
   time_t m_lastChangeTime;
 
-  /** the number of times this messages was already polled for. */
+  /** the polling order of this message (roughly number of polls * priority). */
-  unsigned int m_pollCount;
+  unsigned int m_pollOrder;
 
   /** the system time when this message was last polled for, 0 for never. */
   time_t m_lastPollTime;
@@ -1403,13 +1403,24 @@ class MessageMap : public MappedFileReader {
    * address), or 0 to ignore.
    * @param until the end time to which to add updates (exclusive, also removes messages with unset destination
    * address), or 0 to ignore.
-   * @changedSince true to use the last change time for the since/until range, false to use the last seen time.
+   * @param changedSince true to use the last change time for the since/until range, false to use the last seen time.
    * @param messages the @a deque to which to add the found @a Message instances.
    */
   void findAll(const string& circuit, const string& name, const string& levels,
     bool completeMatch, bool withRead, bool withWrite, bool withPassive, bool includeEmptyLevel, bool onlyAvailable,
     time_t since, time_t until, bool changedSince, deque<Message*>* messages) const;
 
+  /**
+   * Get the first available @a Message from the first map iterator entry.
+   * @param it the map iterator with list of @a Message instances to check.
+   * @param sameIdExtAs the optional @a MasterSymbolString to check for having the same ID.
+   * @param onlyAvailable true to include only available messages (default true), false to also include messages that
+   * are currently not available (e.g. due to unresolved or false conditions).
+   * @return the first available @a Message from the first map iterator entry.
+   */
+  Message* getFirstAvailableFromIterator(const map<uint64_t, vector<Message*> >::const_iterator& it,
+    const MasterSymbolString* sameIdExtAs, bool onlyAvailable) const;
+
   /**
    * Find the @a Message instance for the specified master data.
    * @param master the @a MasterSymbolString for identifying the @a Message.

src/lib/ebus/test/test_data.cpp

@@ -228,6 +228,9 @@ int main() {
     {"x,,ttq", "23:45", "10fe0700015f",   "00", ""},
     {"x,,ttq", "24:00", "10fe07000160",   "00", ""},
     {"x,,ttq", "-:-",   "10fe07000100",   "00", ""},
+    {"x,,ttq,,,,,,ttq", "23:00;05:45", "10fe0700025c17", "00", ""},
+    {"x,,ttq,,,,,,bi7,,,,,,ttq", "23:00;0;05:45", "10fe0700025c17", "00", ""},
+    {"x,,ttq,,,,,,bi7,,,,,,ttq", "23:00;1;05:45", "10fe070002dc17", "00", ""},
     {"x,,ttq", "",      "10fe07000161",   "00", "rw"},
     {"x,,ttq,2", "",    "",               "",   "c"},
     {"x,,ttq,,,,y,,bi7", "23:45;0",  "10fe0700015f", "00", ""},  // bit combination

src/lib/ebus/test/test_message.cpp

@@ -173,7 +173,11 @@ int main() {
     {"w,,x,,,,,,b0,,BI0:1,,,,b1,,BI1:1,,,,b2,,BI2:6,,,,c0,,BI0:1,,,,c1,,BI1:1,,,,c2,,BI2:1", "1;1;1;0;0;0", "ff08b509030e0700", "00", "di" },
     {"w,,x,,,,,,b0,,BI0:1,,,,b1,,BI1:1,,,,b2,,BI2:6,,,,c0,,BI0:1,,,,c1,,BI1:1,,,,c2,,BI2:1", "1;0;0;0;0;1", "ff08b509030e0104", "00", "di" },
     {"w,,x,,,,,,b0,,BI0:1,,,,b1,,BI1:1,,,,b2,,BI2:6,,,,c0,,BI0:1,,,,c1,,BI1:1,,,,c2,,BI2:1", "0;0;1;0;1;1", "ff08b509030e0406", "00", "di" },
-    {"*r,cir*cuit#level,na*me,com*ment,ff,75,b509,0d", "", "", "", "" },
+    {"r,470,ccTimer.Monday,,,15,B515,0002,,,IGN:1,,,,from,,TTM", "", "", "", "M"},
+    {"w,470,ccTimer.Monday,,,10,B515,0002,from,,TTM", "", "", "", "kM*"},
+    {"", "19:00", "3115b515020002", "080272", "kd"},
+    {"", "19:00", "3110b51503000272", "00", "kd"},
+    {"*r,cir*cuit#level,na*me,com*ment,ff,75,b509,0d", "", "", "", ""},
     {"r,CIRCUIT,NAME,COMMENT,,,,0100,field,,UCH", "r,cirCIRCUITcuit,naNAMEme,comCOMMENTment,ff,75,b509,0d0100,field,s,UCH,,,: field=42", "ff75b509030d0100", "012a", "DN"},
     {"r,CIRCUIT,NAME,COMMENT,,,,0100,field,,UCH",
         // "\"naNAMEme\": {r,cirCIRCUITcuit,naNAMEme,comCOMMENTment,ff,75,b509,0d0100,field,s,UCH,,,: field=42"
@@ -420,7 +424,7 @@ int main() {
       result = message->decodeLastData(false, nullptr, -1,
           (decodeVerbose?OF_NAMES|OF_UNITS|OF_COMMENTS:0)|(decodeJson?OF_NAMES|OF_JSON:0), &output);
       if (result != RESULT_OK) {
-        cout << "  \"" << check[2] << "\" / \"" << check[3] << "\": decode error: "
+        cout << "  \"" << check[2] << "\" / \"" << check[3] << "\": decode error " << (message->isWrite() ? "write: " : "read: ")
             << getResultCode(result) << endl;
         error = true;
         continue;

src/lib/utils/log.cpp

@@ -27,12 +27,13 @@
 #include <sys/time.h>
 #include <stdarg.h>
 #include <string.h>
+#include <syslog.h>
 #include "lib/utils/clock.h"
 
 namespace ebusd {
 
 /** the name of each @a LogFacility. */
-static const char *facilityNames[] = {
+static const char *s_facilityNames[] = {
   "main",
   "network",
   "bus",
@@ -43,7 +44,7 @@ static const char *facilityNames[] = {
 };
 
 /** the name of each @a LogLevel. */
-static const char* levelNames[] = {
+static const char* s_levelNames[] = {
   "none",
   "error",
   "notice",
@@ -52,19 +53,32 @@ static const char* levelNames[] = {
   nullptr
 };
 
+/** the syslog level of each @a LogLevel. */
+static const int s_syslogLevels[] = {
+  LOG_INFO,
+  LOG_ERR,
+  LOG_NOTICE,
+  LOG_INFO,
+  LOG_DEBUG,
+  0
+};
+
 /** the current log level by log facility. */
 static LogLevel s_facilityLogLevel[] = { ll_notice, ll_notice, ll_notice, ll_notice, ll_notice, };
 
-/** the current log FILE. */
+/** the current log FILE, or nullptr if closed or syslog is used. */
 static FILE* s_logFile = stdout;
 
+/** whether to log to syslog. */
+static bool s_useSyslog = false;
+
 LogFacility parseLogFacility(const char* facility) {
   if (!facility) {
     return lf_COUNT;
   }
   char *input = strdup(facility);
   char *opt = reinterpret_cast<char*>(input), *value = nullptr;
-  int val = getsubopt(&opt, (char *const *)facilityNames, &value);
+  int val = getsubopt(&opt, (char *const *)s_facilityNames, &value);
   if (val < 0 || val >= lf_COUNT || value || *opt) {
     free(input);
     return lf_COUNT;
@@ -78,7 +92,7 @@ int parseLogFacilities(const char* facilities) {
   char *opt = reinterpret_cast<char*>(input), *value = nullptr;
   int newFacilites = 0;
   while (*opt) {
-    int val = getsubopt(&opt, (char *const *)facilityNames, &value);
+    int val = getsubopt(&opt, (char *const *)s_facilityNames, &value);
     if (val < 0 || val > lf_COUNT || value) {
       free(input);
       return -1;
@@ -99,7 +113,7 @@ LogLevel parseLogLevel(const char* level) {
   }
   char *input = strdup(level);
   char *opt = reinterpret_cast<char*>(input), *value = nullptr;
-  int val = getsubopt(&opt, (char *const *)levelNames, &value);
+  int val = getsubopt(&opt, (char *const *)s_levelNames, &value);
   if (val < 0 || val >= ll_COUNT || value || *opt) {
     free(input);
     return ll_COUNT;
@@ -109,11 +123,11 @@ LogLevel parseLogLevel(const char* level) {
 }
 
 const char* getLogFacilityStr(LogFacility facility) {
-  return facilityNames[facility];
+  return s_facilityNames[facility];
 }
 
 const char* getLogLevelStr(LogLevel level) {
-  return levelNames[level];
+  return s_levelNames[level];
 }
 
 bool setFacilitiesLogLevel(int facilities, LogLevel level) {
@@ -132,6 +146,12 @@ LogLevel getFacilityLogLevel(LogFacility facility) {
 }
 
 bool setLogFile(const char* filename) {
+  if (filename[0] == 0) {
+    closeLogFile();
+    openlog("ebusd", LOG_NDELAY|LOG_PID, LOG_USER);
+    s_useSyslog = true;
+    return true;
+  }
   FILE* newFile = fopen(filename, "a");
   if (newFile == nullptr) {
     return false;
@@ -148,27 +168,35 @@ void closeLogFile() {
     }
     s_logFile = nullptr;
   }
+  if (s_useSyslog) {
+    closelog();
+    s_useSyslog = false;
+  }
 }
 
 bool needsLog(const LogFacility facility, const LogLevel level) {
   return s_facilityLogLevel[facility] >= level;
 }
 
-void logWrite(const char* facility, const char* level, const char* message, va_list ap) {
+void logWrite(const char* facility, const LogLevel level, const char* message, va_list ap) {
-  if (s_logFile == nullptr) {
+  if (s_logFile == nullptr && !s_useSyslog) {
     return;
   }
-  struct timespec ts;
-  struct tm td;
-  clockGettime(&ts);
-  localtime_r(&ts.tv_sec, &td);
   char* buf;
   if (vasprintf(&buf, message, ap) >= 0 && buf) {
-    fprintf(s_logFile, "%04d-%02d-%02d %02d:%02d:%02d.%03ld [%s %s] %s\n",
+    if (s_useSyslog) {
-      td.tm_year+1900, td.tm_mon+1, td.tm_mday,
+      syslog(s_syslogLevels[level], "[%s %s] %s", facility, s_levelNames[level], buf);
-      td.tm_hour, td.tm_min, td.tm_sec, ts.tv_nsec/1000000,
+    } else {
-      facility, level, buf);
+      struct timespec ts;
-    fflush(s_logFile);
+      struct tm td;
+      clockGettime(&ts);
+      localtime_r(&ts.tv_sec, &td);
+      fprintf(s_logFile, "%04d-%02d-%02d %02d:%02d:%02d.%03ld [%s %s] %s\n",
+        td.tm_year+1900, td.tm_mon+1, td.tm_mday,
+        td.tm_hour, td.tm_min, td.tm_sec, ts.tv_nsec/1000000,
+        facility, s_levelNames[level], buf);
+      fflush(s_logFile);
+    }
   }
   if (buf) {
     free(buf);
@@ -178,14 +206,14 @@ void logWrite(const char* facility, const char* level, const char* message, va_l
 void logWrite(const LogFacility facility, const LogLevel level, const char* message, ...) {
   va_list ap;
   va_start(ap, message);
-  logWrite(facilityNames[facility], levelNames[level], message, ap);
+  logWrite(s_facilityNames[facility], level, message, ap);
   va_end(ap);
 }
 
 void logWrite(const char* facility, const LogLevel level, const char* message, ...) {
   va_list ap;
   va_start(ap, message);
-  logWrite(facility, levelNames[level], message, ap);
+  logWrite(facility, level, message, ap);
   va_end(ap);
 }
 

src/lib/utils/log.h

@@ -98,7 +98,7 @@ LogLevel getFacilityLogLevel(LogFacility facility);
 
 /**
  * Set the log file to use.
- * @param filename the name of the log file to use.
+ * @param filename the name of the log file to use, or the empty string for syslog.
  * @return true on success, false on error.
  */
 bool setLogFile(const char* filename);

src/lib/utils/tcpsocket.h

@@ -21,6 +21,7 @@
 
 #include <unistd.h>
 #include <sys/socket.h>
+#include <netinet/in.h>
 #include <sys/time.h>
 #include <stdint.h>
 #include <string>