diff --git a/.github/FUNDING.yml b/.github/FUNDING.yml new file mode 100644 index 00000000..c9ca4361 --- /dev/null +++ b/.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'] diff --git a/CMakeLists.txt b/CMakeLists.txt index aa3dc83f..deb5ce96 100755 --- a/CMakeLists.txt +++ b/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) diff --git a/ChangeLog.md b/ChangeLog.md old mode 100755 new mode 100644 index 13aff986..f16c783a --- a/ChangeLog.md +++ b/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 diff --git a/README.md b/README.md index e512678e..a9c1d1e0 100755 --- a/README.md +++ b/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 diff --git a/VERSION b/VERSION index a3ec5a4b..f30101c0 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -3.2 +3.3 \ No newline at end of file diff --git a/config.h.cmake b/config.h.cmake index 5e88633c..b194d390 100755 --- a/config.h.cmake +++ b/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 diff --git a/configure.ac b/configure.ac index 08bed21d..ed7eb853 100755 --- a/configure.ac +++ b/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]) diff --git a/contrib/archlinux/PKGBUILD b/contrib/archlinux/PKGBUILD index 812b0c10..867baf3f 100644 --- a/contrib/archlinux/PKGBUILD +++ b/contrib/archlinux/PKGBUILD @@ -2,7 +2,7 @@ # Contributor: Milan Knizek # 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') diff --git a/contrib/archlinux/PKGBUILD.git b/contrib/archlinux/PKGBUILD.git index 56e334fc..cba8bebb 100644 --- a/contrib/archlinux/PKGBUILD.git +++ b/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') diff --git a/contrib/archlinux/systemd/ebusd.service b/contrib/archlinux/systemd/ebusd.service index e0d44006..5bb9f4d2 100644 --- a/contrib/archlinux/systemd/ebusd.service +++ b/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 diff --git a/contrib/config/README.md b/contrib/config/README.md new file mode 100644 index 00000000..3d6e48a7 --- /dev/null +++ b/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. + diff --git a/contrib/config/index.php b/contrib/config/index.php new file mode 100644 index 00000000..b318a033 --- /dev/null +++ b/contrib/config/index.php @@ -0,0 +1,61 @@ + diff --git a/contrib/debian/init.d/ebusd b/contrib/debian/init.d/ebusd index 356fd1f0..db77359a 100755 --- a/contrib/debian/init.d/ebusd +++ b/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 diff --git a/contrib/debian/systemd/ebusd.service b/contrib/debian/systemd/ebusd.service index ca93a24e..1398bdcb 100644 --- a/contrib/debian/systemd/ebusd.service +++ b/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 diff --git a/contrib/docker/Dockerfile b/contrib/docker/Dockerfile index 2289a9b0..ad939472 100755 --- a/contrib/docker/Dockerfile +++ b/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"] diff --git a/contrib/docker/release/Dockerfile b/contrib/docker/release/Dockerfile index b6e1caa1..fe10ab69 100755 --- a/contrib/docker/release/Dockerfile +++ b/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}" diff --git a/contrib/docker/runtime/Dockerfile b/contrib/docker/runtime/Dockerfile deleted file mode 100755 index 9c6d6d31..00000000 --- a/contrib/docker/runtime/Dockerfile +++ /dev/null @@ -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"] diff --git a/contrib/docker/runtime/docker-entrypoint.sh b/contrib/docker/runtime/docker-entrypoint.sh deleted file mode 100755 index 142a9a26..00000000 --- a/contrib/docker/runtime/docker-entrypoint.sh +++ /dev/null @@ -1,8 +0,0 @@ -#!/bin/bash -set -e - -if [ "${1#-}" != "$1" ]; then - set -- ebusd "$@" -fi - -exec "$@" \ No newline at end of file diff --git a/contrib/docker/runtime/ebusd-3.2_amd64_mqtt1.deb b/contrib/docker/runtime/ebusd-3.2_amd64_mqtt1.deb deleted file mode 100644 index 63db5d54..00000000 Binary files a/contrib/docker/runtime/ebusd-3.2_amd64_mqtt1.deb and /dev/null differ diff --git a/contrib/etc/logrotate.d/ebusd b/contrib/etc/logrotate.d/ebusd old mode 100755 new mode 100644 index 630a56e5..80aa6d1f --- a/contrib/etc/logrotate.d/ebusd +++ b/contrib/etc/logrotate.d/ebusd @@ -8,5 +8,5 @@ daily postrotate /usr/bin/killall -HUP ebusd - endscript + endscript } diff --git a/contrib/updatecheck/README.md b/contrib/updatecheck/README.md new file mode 100644 index 00000000..8fbaa92b --- /dev/null +++ b/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. + diff --git a/contrib/updatecheck/calcversions.sh b/contrib/updatecheck/calcversions.sh index da502b97..c688342e 100755 --- a/contrib/updatecheck/calcversions.sh +++ b/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 diff --git a/contrib/updatecheck/index.php b/contrib/updatecheck/index.php index 9a0105a6..6a4f092e 100644 --- a/contrib/updatecheck/index.php +++ b/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'); diff --git a/contrib/updatecheck/prepend.inc b/contrib/updatecheck/prepend.inc index c4f2c144..bb93f14c 100644 --- a/contrib/updatecheck/prepend.inc +++ b/contrib/updatecheck/prepend.inc @@ -1,21 +1,26 @@ 1) { - $versions[$val[0]] = explode(',', $val[1]); - } - }; - array_walk($v, $func); +function readVersions($old=false) { + global $versions; + $versions = array(); + $v = @file_get_contents(($old?'old':'').'versions.txt'); + if ($v) { + $v = explode("\n", $v); + $func = function($val, $k) { + global $versions; + $val = explode('=', $val); + 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; diff --git a/src/ebusd/bushandler.cpp b/src/ebusd/bushandler.cpp index 78fd29c1..664a9ffb 100755 --- a/src/ebusd/bushandler.cpp +++ b/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 (message) { - *output << ": " << message->getCircuit() << " " << message->getName(); + if (!isDirectMode) { + *output << " = " << m_count; + 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"; - } else { - bool first = true; - for (const auto& it : m_grabbedMessages) { - if (it.second.dump(unknown, m_messages, first, decode, output)) { - first = false; - } + if (!isDirectMode) { + *output << "grab disabled"; } + 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; } } diff --git a/src/ebusd/bushandler.h b/src/ebusd/bushandler.h index e66a5544..da93e88c 100755 --- a/src/ebusd/bushandler.h +++ b/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. diff --git a/src/ebusd/main.cpp b/src/ebusd/main.cpp index 52a2ca23..1da71025 100755 --- a/src/ebusd/main.cpp +++ b/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), - errorDescription.c_str()); + logError(lf_main, "error reading config files from %s: %s, last error: %s", opt.configPath, + 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 } diff --git a/src/ebusd/mainloop.cpp b/src/ebusd/mainloop.cpp index 927e16cc..9229278c 100755 --- a/src/ebusd/mainloop.cpp +++ b/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); - if (!listening) { + ClientMode mode = netMessage->getMode(&since); + 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 = token[0]; + escaped = 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& 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& 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& args, const string levels, argPos++; } - if ((hex && (newDefinition || dstAddress != SYN || !circuit.empty() || args.size() < argPos + 1)) - || (newDefinition && (hex || !circuit.empty() || args.size() < argPos + 1 || args.size() > argPos + 2))) { + if ((args.size() < argPos + 1) + || (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& 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& args, const string levels, return RESULT_OK; } -result_t MainLoop::executeHex(const vector& args, ostringstream* ostream) { - size_t argPos = 1; +result_t MainLoop::parseHexAndSend(const vector& args, size_t& argPos, bool isDirectMode, + ostringstream* ostream) { symbol_t srcAddress = SYN; if (args.size() > argPos && args[argPos] == "-s") { argPos++; if (argPos >= args.size()) { argPos = 0; // print usage - } else { - 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; + 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; argPos++; } if (args.size() < argPos + 1 || (args.size() > argPos && args[argPos][0] == '-')) { argPos = 0; // print usage + return RESULT_OK; } - if (argPos > 0) { - MasterSymbolString master; - result_t ret = parseHexMaster(args, argPos, srcAddress, &master); - 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)); + MasterSymbolString master; + result_t ret = parseHexMaster(args, argPos, srcAddress, &master); + argPos = args.size(); // mark as successfully parsed + if (ret != RESULT_OK) { 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& 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& 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& args, const string& levels, return RESULT_OK; } -result_t MainLoop::executeListen(const vector& args, bool* listening, ostringstream* ostream) { +result_t MainLoop::executeListen(const vector& 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& 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" diff --git a/src/ebusd/mainloop.h b/src/ebusd/mainloop.h index 0ccd0099..f07f9866 100755 --- a/src/ebusd/mainloop.h +++ b/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& 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& 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& 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& 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& args, bool* listening, ostringstream* ostream); + result_t executeListen(const vector& args, ClientMode* mode, ostringstream* ostream); /** * Execute the state command. diff --git a/src/ebusd/mqtthandler.cpp b/src/ebusd/mqtthandler.cpp old mode 100755 new mode 100644 index 84b882d7..986e6e46 --- a/src/ebusd/mqtthandler.cpp +++ b/src/ebusd/mqtthandler.cpp @@ -22,6 +22,7 @@ #include "ebusd/mqtthandler.h" #include +#include #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 "_]", 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 g_topicStrs; static vector 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* 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(obj); + if (!handler || !message || !handler->isRunning()) { + return; + } + string topic(message->topic); + string data(message->payloadlen > 0 ? reinterpret_cast(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 (mosquitto_lib_init() != MOSQ_ERR_SUCCESS) { - logOtherError("mqtt", "unable to initialize"); - } else { + if (check(mosquitto_lib_init(), "unable to initialize")) { 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 - | MOSQ_LOG_NOTICE | MOSQ_LOG_INFO, MOSQ_LOG_STDERR);*/ +#if (LIBMOSQUITTO_VERSION_NUMBER >= 1004001) + check(mosquitto_threaded_set(m_mosquitto, true), "threaded_set"); + check(mosquitto_opts_set(m_mosquitto, MOSQ_OPT_PROTOCOL_VERSION, reinterpret_cast(&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(obj); - if (!handler || !message || !handler->isRunning()) { - return; - } - string topic(message->topic); - string data(message->payloadlen > 0 ? reinterpret_cast(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); - mosquitto_subscribe(m_mosquitto, nullptr, m_subscribeTopic.c_str(), 0); + check(mosquitto_subscribe(m_mosquitto, nullptr, m_subscribeTopic.c_str(), 0), "subscribe"); } } @@ -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"; - if (!isWrite && direction != "get") { + bool isWrite = direction == "set"; + 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; - for (idx = 0; idx < g_topicStrs.size()+1; idx++) { + bool finalField = false; + 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 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) { - if (m_mosquitto) { - int ret; +bool MqttHandler::handleTraffic(bool allowReconnect) { + if (!m_mosquitto) { + 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_initialConnectFailed) { + if (!m_connected && ret == MOSQ_ERR_NO_CONN && allowReconnect) { + 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) { - logOtherError("mqtt", "unable to connect (invalid parameters), retrying"); - } - if (ret == MOSQ_ERR_SUCCESS) { - m_initialConnectFailed = false; - } - } else { - ret = mosquitto_reconnect(m_mosquitto); + if (ret == MOSQ_ERR_INVAL) { + logOtherError("mqtt", "unable to connect (invalid parameters), retrying"); + } + if (ret == MOSQ_ERR_SUCCESS) { + m_initialConnectFailed = false; } - } - 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()); - mosquitto_publish(m_mosquitto, nullptr, topic.c_str(), (uint32_t)data.size(), - reinterpret_cast(data.c_str()), 0, g_retain || retain); + const char* topicStr = topic.c_str(); + const char* dataStr = data.c_str(); + 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(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 diff --git a/src/ebusd/mqtthandler.h b/src/ebusd/mqtthandler.h index 84fea954..12f70cf5 100644 --- a/src/ebusd/mqtthandler.h +++ b/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 diff --git a/src/ebusd/network.cpp b/src/ebusd/network.cpp index bc1c6123..b53831d5 100755 --- a/src/ebusd/network.cpp +++ b/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(); diff --git a/src/ebusd/network.h b/src/ebusd/network.h index 2b961fe1..fdcabc05 100755 --- a/src/ebusd/network.h +++ b/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. - * @param listenSince set to the start time from which to add updates (inclusive). - * @return whether the client is in listening mode. + * Return the client mode. + * @param listenSince set listening to the specified start time from which to add updates (inclusive). + * @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. */ - bool m_listening; + /** the client mode. */ + 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; /** diff --git a/src/lib/ebus/data.cpp b/src/lib/ebus/data.cpp old mode 100755 new mode 100644 index 3f47def2..d86e882c --- a/src/lib/ebus/data.cpp +++ b/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 - || (after && firstBit + (m_dataType->getBitCount() % 8) >= 8); + bool ret = (m_dataType->getBitCount() % 8) == 0 + || (firstBit == previousFirstBit) || (after && firstBit + (m_dataType->getBitCount() % 8) >= 8); + // std::cout<<(after?"after,":"before,")<<"prev="<(previousFirstBit)<<",first=" + // <(firstBit)<<",length="<(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) { diff --git a/src/lib/ebus/data.h b/src/lib/ebus/data.h index b6256feb..4d9eb34c 100755 --- a/src/lib/ebus/data.h +++ b/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; diff --git a/src/lib/ebus/datatype.cpp b/src/lib/ebus/datatype.cpp index 59b5cab2..2f2d24ec 100755 --- a/src/lib/ebus/datatype.cpp +++ b/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 { diff --git a/src/lib/ebus/device.cpp b/src/lib/ebus/device.cpp index d3f29ce9..9732e05d 100755 --- a/src/lib/ebus/device.cpp +++ b/src/lib/ebus/device.cpp @@ -30,6 +30,9 @@ #ifdef HAVE_LINUX_SERIAL # include #endif +#ifdef HAVE_FREEBSD_UFTDI +# include +#endif #include #ifdef HAVE_PPOLL # include @@ -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, ¶m) == 0) { + ioctl(m_fd, UFTDIIOC_RESET_IO, ¶m); + param = 1; + ioctl(m_fd, UFTDIIOC_SET_LATENCY, ¶m); + } +#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); diff --git a/src/lib/ebus/device.h b/src/lib/ebus/device.h index 050c1e1d..6d8e0b15 100755 --- a/src/lib/ebus/device.h +++ b/src/lib/ebus/device.h @@ -22,6 +22,7 @@ #include #include #include +#include #include #include #include diff --git a/src/lib/ebus/filereader.cpp b/src/lib/ebus/filereader.cpp index 200a0170..7daab6df 100755 --- a/src/lib/ebus/filereader.cpp +++ b/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; diff --git a/src/lib/ebus/message.cpp b/src/lib/ebus/message.cpp old mode 100755 new mode 100644 index 5bb97027..008cb078 --- a/src/lib/ebus/message.cpp +++ b/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& 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* 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& messages, const MasterSymbolString* sameIdExtAs, const bool onlyAvailable = true) { @@ -1287,6 +1304,7 @@ Message* getFirstAvailable(const vector& 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& 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 >::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 >::const_iterator it; + Message* message; if (withPassive) { - it = m_messagesByKey.find(key); - if (it != m_messagesByKey.end()) { - Message* message = getFirstAvailable(it->second, &master, onlyAvailable); + message = getFirstAvailableFromIterator(m_messagesByKey.find(key), &master, onlyAvailable); + if (message) { + 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 - if (it != m_messagesByKey.end()) { - Message* message = getFirstAvailable(it->second, &master, onlyAvailable); - if (message) { - return message; - } + // try again with special value for active read + message = getFirstAvailableFromIterator( + m_messagesByKey.find(key | (isWriteDest ? ID_SOURCE_ACTIVE_READ_MASTER : ID_SOURCE_ACTIVE_READ)), + &master, onlyAvailable); + if (message) { + return message; } } if (withWrite) { - it = m_messagesByKey.find(key | ID_SOURCE_ACTIVE_WRITE); // try again with special value for active write - if (it != m_messagesByKey.end()) { - Message* message = getFirstAvailable(it->second, &master, onlyAvailable); - if (message) { - return message; - } + // try again with special value for active write + message = getFirstAvailableFromIterator( + m_messagesByKey.find(key | (isWriteDest ? ID_SOURCE_ACTIVE_WRITE_MASTER : ID_SOURCE_ACTIVE_WRITE)), + &master, onlyAvailable); + 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; } diff --git a/src/lib/ebus/message.h b/src/lib/ebus/message.h old mode 100755 new mode 100644 index daf7fd80..0f2a2d13 --- a/src/lib/ebus/message.h +++ b/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. */ - unsigned int m_pollCount; + /** the polling order of this message (roughly number of polls * priority). */ + 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* 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 >::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. diff --git a/src/lib/ebus/test/test_data.cpp b/src/lib/ebus/test/test_data.cpp index 3ca8fd9a..13b812dc 100755 --- a/src/lib/ebus/test/test_data.cpp +++ b/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 diff --git a/src/lib/ebus/test/test_message.cpp b/src/lib/ebus/test/test_message.cpp old mode 100755 new mode 100644 index 16d54bc0..aea4f361 --- a/src/lib/ebus/test/test_message.cpp +++ b/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; diff --git a/src/lib/utils/log.cpp b/src/lib/utils/log.cpp index ecf26021..350cda61 100755 --- a/src/lib/utils/log.cpp +++ b/src/lib/utils/log.cpp @@ -27,12 +27,13 @@ #include #include #include +#include #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(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(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(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) { - if (s_logFile == nullptr) { +void logWrite(const char* facility, const LogLevel level, const char* message, va_list ap) { + 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", - 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, level, buf); - fflush(s_logFile); + if (s_useSyslog) { + syslog(s_syslogLevels[level], "[%s %s] %s", facility, s_levelNames[level], buf); + } else { + struct timespec ts; + 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); } diff --git a/src/lib/utils/log.h b/src/lib/utils/log.h index 0cd41ae8..4a6e264a 100755 --- a/src/lib/utils/log.h +++ b/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); diff --git a/src/lib/utils/tcpsocket.h b/src/lib/utils/tcpsocket.h index 4d0dcb1b..fd82b225 100755 --- a/src/lib/utils/tcpsocket.h +++ b/src/lib/utils/tcpsocket.h @@ -21,6 +21,7 @@ #include #include +#include #include #include #include