First version of mqtt support. (#357)

* Cmake, readme, config changes to add libmosquitto.

* First version of mqtt client support.
Adds
- ENABLE_MQTT cmake option. Defaults to On but turns off automatically
  if libmosquitto not found.
- basic mqtt client support using libmosquitto
  - topics generated are:
    vzlogger/<channel-id>/uuid
    vzlogger/<channel-id>/raw
    vzlogger/<channel-id>/agg
- added config options for
  - host
  - port (currently no tls/sll support!)
  - user
  - password
  - topic (prefix used instead of vzlogger in above example)
  - (some more, see etc/vzlogger.conf)
- agg values get's preferred instead of raw. Using config option
  rawAndAgg this can be changed.

* adding libmosquitto-dev to travis-ci

* tell libmosquitto to be thread safe

* Add tls/cert support for MQTT

* Add id with "unparse" name of channel as well to the topic.
This should be to get a better understanding of what the channel
actually is. E.g. will contains obis ids.

* added reconnect handling if con. refused at startup
This should allow startups where the mqtt server will be available later

* Added a lock to protect chMap incase of multiple meters.

* announce uuid only if not empty

.travis.yml

@@ -3,8 +3,8 @@ dist: trusty
 
 language: cpp
 
-compiler: 
-    - clang 
+compiler:
+    - clang
     - gcc
 
 env:
@@ -36,11 +36,11 @@ before_install:
   - sudo apt-get install -y libcurl4-openssl-dev openssl libmicrohttpd-dev uuid-dev uuid-runtime libunistring-dev
 # -- get libsml --
   - git clone https://github.com/volkszaehler/libsml.git # or github.com/TheCount/libsml.git # or https://github.com/dailab/libsml.git
-  - cd libsml 
+  - cd libsml
   - # git checkout develop # only dev branch seems to work
   - make
 # -- install libsml --
-  - sudo cp sml/lib/libsml.* /usr/lib/. 
+  - sudo cp sml/lib/libsml.* /usr/lib/.
   - sudo cp -R sml/include/* /usr/include/.
   - sudo cp sml.pc /usr/lib/pkgconfig/.
   - cd ..
@@ -63,5 +63,10 @@ before_install:
   - sudo make install
   - cd ..
   - sudo apt-get install -y lcov
+
+# -- install libmosquitto-dev
+  - sudo add-apt-repository ppa:mosquitto-dev/mosquitto-ppa -y
+  - sudo apt-get update -qq
+  - sudo apt-get install -qq libmosquitto-dev
 script:
   - cmake . -DMETEREXEC_ROOTACCESS=OFF -DENABLE_OCR=ON -DSML_HOME=/usr/local/src/libsml/sml && make && make test

CMakeLists.txt

@@ -6,13 +6,13 @@ if(POLICY CMP0011)
 endif(POLICY CMP0011)
 
 set(VZLOGGER_MAJOR_VERSION 0)
-set(VZLOGGER_MINOR_VERSION 6)
-set(VZLOGGER_SUB_VERSION   1)
+set(VZLOGGER_MINOR_VERSION 7)
+set(VZLOGGER_SUB_VERSION   0)
 set(VERSION_SEP "-")
 set(VZLOGGER_SHORT_VERSION "${VZLOGGER_MAJOR_VERSION}${VERSION_SEP}${VZLOGGER_MINOR_VERSION}")
 set(VZLOGGER_VERSION "${VZLOGGER_SHORT_VERSION}${VERSION_SEP}${VZLOGGER_SUB_VERSION}")
 set(VZLOGGER_RPM_VERSION "${VZLOGGER_MAJOR_VERSION}.${VZLOGGER_MINOR_VERSION}.${VZLOGGER_SUB_VERSION}")
-set(VERSION_DATE "2012-02-20")
+set(VERSION_DATE "2018-08-04")
 set(PACKAGE_STRING "${PROJECT_NAME} ${VZLOGGER_RPM_VERSION}")
 set(VERSION "${VZLOGGER_RPM_VERSION}")
 set(CMAKE_VERBOSE_MAKEFILE 1)
@@ -55,6 +55,9 @@ OPTION(ENABLE_OMS
 OPTION(ENABLE_LOCAL
   "enable support for local HTTPd (def=yes)]"
   On)
+OPTION(ENABLE_MQTT
+  "enable MQTT client support (def=yes)"
+  On)
 OPTION(WITH_READER
   "compile reader to for testing your meters (def=yes)])"
   On)
@@ -107,6 +110,21 @@ if(ENABLE_LOCAL)
   endif(MICROHTTPD_FOUND)
 endif(ENABLE_LOCAL)
 
+if(ENABLE_MQTT)
+  find_library(MQTT_LIBRARY mosquitto)
+  find_path(MQTT_INCLUDE_DIR mosquitto.h)
+  message( STATUS "search for libmosquitto returned ${MQTT_LIBRARY} and ${MQTT_INCLUDE_DIR}")
+  if(MQTT_LIBRARY AND MQTT_INCLUDE_DIR)
+    message( STATUS "libmosquitto found at ${MQTT_LIBRARY}")
+    include_directories(${MQTT_INCLUDE_DIR})
+  else()
+    set(ENABLE_MQTT OFF)
+    message( WARNING "libmosquitto not found. Disabled ENABLE_MQTT. Consider installing libmosquitto-dev package.")
+  endif(MQTT_LIBRARY AND MQTT_INCLUDE_DIR)
+else()
+  message( STATUS "MQTT support disabled. If wanted use ENABLE_MQTT=On")
+endif(ENABLE_MQTT)
+
 if( ENABLE_OCR OR ENABLE_OCR_TESSERACT )
 	include(FindLeptonica)
 	if (NOT LEPTONICA_FOUND)
@@ -179,7 +197,7 @@ if("${CMAKE_CXX_COMPILER_ID}" STREQUAL "GNU")
     # require at least gcc 4.8
     execute_process(COMMAND ${CMAKE_CXX_COMPILER} -dumpversion
        OUTPUT_VARIABLE GOOGLETEST_GCC_COMPILER_VERSION)
-# on travis-ci the CMAKE_CXX_COMPILER_VERSION is empty!    message(WARNING "Using compiler: ${CMAKE_CXX_COMPILER_VERSION} ${GOOGLETEST_GCC_COMPILER_VERSION}") 
+# on travis-ci the CMAKE_CXX_COMPILER_VERSION is empty!    message(WARNING "Using compiler: ${CMAKE_CXX_COMPILER_VERSION} ${GOOGLETEST_GCC_COMPILER_VERSION}")
     if (GOOGLETEST_GCC_COMPILER_VERSION VERSION_LESS 4.8)
         message(WARNING "Disabled googlemock/-test tests due to GCC version < 4.8!")
 	set(ENABLE_GOOGLEMOCK FALSE)
@@ -213,6 +231,9 @@ endif(SML_FOUND)
 if(MICROHTTPD_FOUND)
 message("             microhttpd: -L${MICROHTTPD_LIBRARY} -I${MICROHTTPD_INCLUDE_DIR}")
 endif(MICROHTTPD_FOUND)
+if(ENABLE_MQTT)
+  message("             mqtt: -L${MQTT_LIBRARY} -I${MQTT_INCLUDE_DIR}")
+endif(ENABLE_MQTT)
 if(METEREXEC_ROOTACCESS)
   message("             MeterExec: root privileges")
 endif(METEREXEC_ROOTACCESS)

README.md

@@ -31,11 +31,14 @@ Ubuntu 18.04LTS (bionic) needs an additional:
    sudo apt-get install libunistring-dev
 (this might be needed on others now as well as we link unconditionally against libunistring)
 
+If you want to use MQTT support:
+  sudo apt-get install libmosquitto-dev
+
 Then run the installation:
 
     wget --no-check-certificate https://raw.github.com/volkszaehler/vzlogger/master/install.sh
     sudo bash install.sh
-    
+
 Mailing List
 -------------
 If you have questions, contact the volkszaehler mailing lists:

config.hpp.in

@@ -4,6 +4,9 @@
 /* Local interface */
 #cmakedefine LOCAL_SUPPORT 1
 
+/* mqtt support */
+#cmakedefine ENABLE_MQTT 1
+
 /* Name of package */
 #define PACKAGE "vzlogger"
 

etc/vzlogger.conf

@@ -35,6 +35,25 @@
         }
     ],
 
+    // mqtt client support (if ENABLE_MQTT set at cmake generation)
+    "mqtt": {
+        "enabled": false,  // enable mqtt client. needs host and port as well
+        "host": "test.mosquitto.org", // mqtt server addr
+        "port": 1883, // 1883 for unencrypted, 8883 enc, 8884 enc cert needed,
+        "cafile": "", // optional file with server CA
+        "capath": "", // optional path for server CAs. see mosquitto.conf. Specify only cafile or capath
+        "certfile": "", // optional file for your client certificate (e.g. client.crt)
+        "keyfile": "", // optional path for your client certficate private key (e.g. client.key)
+        "keypass": "", // optional password for your private key
+        "keepalive": 30, // optional keepalive in seconds.
+        "topic": "vzlogger/data", // optional topic dont use $ at start and no / at end
+        "user": "", // optional user name for the mqtt server
+        "pass": "", // optional password for the mqtt server
+        "retain": false, // optional use retain message flag
+        "rawAndAgg": false, // optional publish raw values even if agg mode is used
+    },
+
+
     // Meter configuration
     "meters": [
         {

include/Buffer.hpp

@@ -67,6 +67,7 @@ class Buffer {
 	inline void have_newValues() { _newValues =  true; }
 
 	inline void set_aggmode(Buffer::aggmode m) {_aggmode=m;}
+	inline aggmode get_aggmode() const { return _aggmode; }
 
 	private:
 	Buffer(const Buffer &); // don't allow copy constructor

include/Channel.hpp

@@ -68,7 +68,7 @@ class Channel {
 
 	bool running() const { return _thread_running; }
 
-	const char* name()                  { return _name.c_str(); }
+	const char* name() const            { return _name.c_str(); }
 	std::list<Option> &options()        { return _options; }
 
 	ReadingIdentifier::Ptr identifier() {

include/mqtt.hpp

@@ -0,0 +1,73 @@
+/*
+ * Author: Matthias Behr, mbehr (a) mcbehr dot de
+ * (c) 2018
+ * */
+
+#ifndef __mqtt_hpp_
+#define __mqtt_hpp_
+
+#include <string>
+#include <mutex>
+#include <unordered_map>
+#include <vector>
+#include "Channel.hpp"
+#include "Reading.hpp"
+
+struct mosquitto; // forward decl. to avoid pulling the header here
+
+class MqttClient
+{
+  public:
+	MqttClient(struct json_object *option);
+	MqttClient() = delete;					 // no default constr.
+	MqttClient(const MqttClient &) = delete; // no copy constr.
+	~MqttClient();
+	bool isConfigured() const;
+
+	void publish(Channel::Ptr ch, Reading &rds, bool aggregate = false); // thread safe, non blocking
+  protected:
+	friend void *mqtt_client_thread(void *);
+	void connect_callback(struct mosquitto *mosq, int result);
+	void disconnect_callback(struct mosquitto *mosq, int result);
+	void message_callback(struct mosquitto *mosq, const struct mosquitto_message *msg);
+
+	bool _enabled;
+	std::string _host;
+	int _port = 0;
+	int _keepalive = 10;
+	std::string _user;
+	std::string _pwd;
+	std::string _cafile;
+	std::string _capath;
+	std::string _certfile;
+	std::string _keyfile;
+	std::string _keypass;
+	bool _retain = false;
+	bool _rawAndAgg = false;
+	std::string _topic;
+
+	bool _isConnected  = false;
+
+	struct mosquitto *_mcs = nullptr; // mosquitto client session data
+
+	struct ChannelEntry
+	{
+		bool _announced = false;
+		bool _sendRaw = true;
+		bool _sendAgg = true;
+		std::string _fullTopicRaw;
+		std::string _fullTopicAgg;
+		std::string _announceName;
+		std::vector<std::pair<std::string, std::string>> _announceValues;
+		void generateNames(const std::string &prefix, Channel &ch);
+	};
+	std::mutex _chMapMutex;
+	std::unordered_map<std::string, ChannelEntry> _chMap;
+};
+
+extern MqttClient *mqttClient;
+
+void *mqtt_client_thread(void *arg);
+void end_mqtt_client_thread(); // notifies the thread to stop. does not wait for the thread
+
+#endif

src/CMakeLists.txt

@@ -16,18 +16,24 @@ else(LOCAL_SUPPORT)
   set(local_srcs "")
 endif(LOCAL_SUPPORT)
 
+if(ENABLE_MQTT)
+  set(mqtt_srcs mqtt.cpp)
+else(ENABLE_MQTT)
+  set(mqtt_srcs "")
+endif(ENABLE_MQTT)
+
 configure_file("${CMAKE_SOURCE_DIR}/src/gitSha1.cpp.in" "${CMAKE_BINARY_DIR}/gitSha1.cpp" @ONLY)
 
 set(vzlogger_srcs
-  vzlogger.cpp 
+  vzlogger.cpp
   ltqnorm.cpp
   Meter.cpp
   ${CMAKE_BINARY_DIR}/gitSha1.cpp
   CurlSessionProvider.cpp
   PushData.cpp ../include/PushData.hpp
 )
 
-set(libvz_srcs 
+set(libvz_srcs
   Channel.cpp
   Config_Options.cpp
   threads.cpp
@@ -37,6 +43,7 @@ set(libvz_srcs
   Reading.cpp
   exception.cpp
   ${local_srcs}
+  ${mqtt_srcs}
   MeterMap.cpp
   )
 
@@ -58,6 +65,10 @@ if(LOCAL_SUPPORT)
 target_link_libraries(vzlogger ${MICROHTTPD_LIBRARY})
 endif(LOCAL_SUPPORT)
 
+if(ENABLE_MQTT)
+target_link_libraries(vzlogger ${MQTT_LIBRARY})
+endif(ENABLE_MQTT)
+
 target_link_libraries(vzlogger ${LIBGCRYPT})
 target_link_libraries(vzlogger pthread m ${LIBUUID})
 target_link_libraries(vzlogger dl)
@@ -72,8 +83,8 @@ if( TARGET )
 endif( TARGET )
 target_link_libraries(vzlogger ${CURL_STATIC_LIBRARIES} ${CURL_LIBRARIES} unistring ${GNUTLS_LIBRARIES} ${OPENSSL_LIBRARIES} )
 
-# add programs to the install target 
-INSTALL(PROGRAMS 
+# add programs to the install target
+INSTALL(PROGRAMS
   ${CMAKE_CURRENT_BINARY_DIR}/vzlogger
   DESTINATION bin)
 install(TARGETS vz

src/Config_Options.cpp

@@ -28,10 +28,13 @@
 #include <ctype.h>
 #include <regex>
 
+#include "config.hpp"
 #include <Config_Options.hpp>
 #include "Channel.hpp"
 #include <VZException.hpp>
-
+#ifdef ENABLE_MQTT
+#include "mqtt.hpp"
+#endif
 
 static const char *option_type_str[] = { "null", "boolean", "double", "int", "object", "array", "string" };
 
@@ -184,6 +187,22 @@ void Config_Options::config_parse(
 				} else
 					print(log_error, "Ignoring push entry due to empty array or duplicate section", "push");
 			}
+#ifdef ENABLE_MQTT
+			else if ((strcmp(key, "mqtt") == 0) && type == json_type_object ) {
+				if (!mqttClient)
+				{
+					mqttClient = new MqttClient(value);
+					if (!mqttClient->isConfigured())
+					{
+						delete mqttClient;
+						mqttClient = 0;
+						print(log_debug, "mqtt client not configured. stopped.", "mqtt");
+					}
+				}
+				else
+					print(log_error, "Ignoring mqtt entry due to empty array or duplicate section", "mqtt");
+			}
+#endif
 			else {
 				print(log_alert, "Ignoring invalid field or type: %s=%s (%s)",
 							NULL, key, json_object_get_string(value), option_type_str[type]);