From 2ca532122d60cff4dbdc7f24fbc5783bcc5ad68d Mon Sep 17 00:00:00 2001
From: Trygve Laugstøl <trygvis@inamo.no>
Date: Tue, 12 Apr 2016 20:06:47 +0200
Subject: Soil Moisture: Adding support for controlling lights. Bluetooth:
 refectorying, trying to be more c++ idiomatic and modern. SM/Diller: adding
 bluetooth to Diller bridge.

---
 apps/mqtt_support.h | 72 +++++++++++++++++++++++++++++++++++++++--------------
 1 file changed, 54 insertions(+), 18 deletions(-)

(limited to 'apps/mqtt_support.h')

diff --git a/apps/mqtt_support.h b/apps/mqtt_support.h
index 48a6c39..ab319fd 100644
--- a/apps/mqtt_support.h
+++ b/apps/mqtt_support.h
@@ -2,10 +2,10 @@
 #define TRYGVIS_MQTT_SUPPORT_H
 
 #include <mutex>
+#include <shared_mutex>
 #include <string>
 #include <exception>
 #include <cstring>
-#include <span.h>
 #include <log4cplus/logger.h>
 #include <log4cplus/loggingmacros.h>
 #include <atomic>
@@ -13,16 +13,14 @@
 #include <limits.h>
 #include <experimental/optional>
 #include "mosquitto.h"
+#include "apps.h"
 
 namespace trygvis {
 namespace mqtt_support {
 
-template<typename T>
-using o = std::experimental::optional<T>;
-
+using namespace trygvis::apps;
 using namespace std;
 using namespace log4cplus;
-using namespace gsl;
 
 static inline
 string error_to_string(int rc) {
@@ -32,6 +30,26 @@ string error_to_string(int rc) {
     return string(mosquitto_strerror(rc));
 }
 
+static
+vector<string> mqtt_tokenize_topic(string path) {
+    char **topics;
+    int topic_count;
+    int i;
+
+    mosquitto_sub_topic_tokenise(path.c_str(), &topics, &topic_count);
+
+    vector<string> res;
+    for (i = 0; i < topic_count; i++) {
+        if (topics[i] != NULL) {
+            res.emplace_back(topics[i]);
+        }
+    }
+
+    mosquitto_sub_topic_tokens_free(&topics, topic_count);
+
+    return res;
+}
+
 class mqtt_error : public std::runtime_error {
 
 public:
@@ -79,7 +97,7 @@ enum mqtt_client_personality {
 };
 
 template<mqtt_client_personality personality>
-class mqtt_client : private mqtt_lib {
+class mqtt_client : public waitable, private mqtt_lib {
     template<bool>
     struct personality_tag {
     };
@@ -101,8 +119,8 @@ class mqtt_client : private mqtt_lib {
 //    bool should_reconnect_;
 
     int unacked_messages_;
-    condition_variable cv;
-    mutex cv_mutex;
+//    condition_variable cv;
+//    mutex cv_mutex;
 
     void assert_success(const string &function, int rc) {
         if (rc != MOSQ_ERR_SUCCESS) {
@@ -113,9 +131,20 @@ class mqtt_client : private mqtt_lib {
 public:
     mqtt_client(const string &host, const int port, const int keep_alive, const o<string> &client_id,
                 const bool clean_session) :
-            host(host), port(port), connecting_(false), connected_(false), /*should_reconnect_(false),*/
-            keep_alive(keep_alive), unacked_messages_(0) {
-        mosquitto = mosquitto_new(client_id ? (*client_id).c_str() : nullptr, clean_session, this);
+        host(host), port(port), connecting_(false), connected_(false), /*should_reconnect_(false),*/
+        keep_alive(keep_alive), unacked_messages_(0) {
+        const char *id = nullptr;
+
+        if (client_id) {
+            id = client_id->c_str();
+        }
+        else {
+            if (!clean_session) {
+                throw mqtt_error("If client id is not specified, clean session must be true", MOSQ_ERR_INVAL);
+            }
+        }
+
+        mosquitto = mosquitto_new(id, clean_session, this);
         if (!mosquitto) {
             string err = strerror(errno);
             throw runtime_error("Could not initialize mosquitto instance: " + err);
@@ -162,11 +191,6 @@ private:
     }
 
 public:
-    void wait() {
-        unique_lock<mutex> lk(cv_mutex);
-        cv.wait(lk);
-    }
-
     int unacked_messages() {
         guard lock(this_mutex);
         return unacked_messages_;
@@ -235,7 +259,7 @@ private:
     void on_disconnect_wrapper(int rc) {
         guard lock(this_mutex);
 
-        LOG4CPLUS_INFO(logger, "Disconnected");
+        LOG4CPLUS_INFO(logger, "Disconnected, rc=" << error_to_string(rc));
 
         bool was_connecting = connecting_, was_connected = connected_;
         connecting_ = connected_ = false;
@@ -320,6 +344,18 @@ public:
         assert_success("mosquitto_subscribe", rc);
     }
 
+    void publish(int *mid, const string &topic, int qos, bool retain, const string &s) {
+        auto len = s.length();
+
+        auto int_max = std::numeric_limits<int>::max();
+
+        if (len > int_max) {
+            len = static_cast<decltype(len)>(int_max);
+        }
+
+        publish(mid, topic, qos, retain, static_cast<int>(len), s.c_str());
+    }
+
     void publish(int *mid, const string &topic, int qos, bool retain, int payload_len, const void *payload) {
 //        if (!connected_) {
 //            throw mqtt_error("not connected", MOSQ_ERR_NO_CONN);
@@ -329,7 +365,7 @@ public:
 
         int rc = mosquitto_publish(mosquitto, mid, topic.c_str(), payload_len, payload, qos, retain);
 
-        if(rc == MOSQ_ERR_SUCCESS) {
+        if (rc == MOSQ_ERR_SUCCESS) {
             guard lock(this_mutex);
             unacked_messages_++;
         }
-- 
cgit v1.2.3