o Adding a MQTT consumer that stores the parsed record in Cassandra.

9 files changed, 565 insertions, 90 deletions

diff --git a/.gitmodules b/.gitmodules
deleted file mode 100644
index e69de29..0000000
--- a/.gitmodules
+++ /dev/null
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 75645c0..f63f023 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -6,6 +6,7 @@ find_package(Boost COMPONENTS system program_options unit_test_framework REQUIRE
 include(ExternalProject)
 
 set(SHARED_COMPILE_OPTIONS "-std=c++14")
+set(SHARED_SOURCES misc_support.h)
 
 # Cassandra
 set(CPP_DRIVER ${CMAKE_CURRENT_BINARY_DIR}/cpp-driver)
@@ -13,17 +14,17 @@ ExternalProject_Add(cpp-driver
   URL https://github.com/datastax/cpp-driver/archive/2.1.0-beta.tar.gz
   URL_MD5 d3cfde8731acc2f0f51ef9caf41068dc
   PREFIX ${CPP_DRIVER}
-  CMAKE_ARGS -DCMAKE_INSTALL_PREFIX:PATH=<INSTALL_DIR>
+  CMAKE_ARGS -DCMAKE_INSTALL_PREFIX:PATH=<INSTALL_DIR> -DCMAKE_C_COMPILER:FILE=${CMAKE_C_COMPILER} -DCMAKE_CXX_COMPILER:FILE=${CMAKE_CXX_COMPILER}
 )
 
 # ble-toys
 set(BLE_TOYS ${CMAKE_CURRENT_BINARY_DIR}/ble-toys)
 ExternalProject_Add(ble-toys
   GIT_REPOSITORY https://trygvis.io/git/2015/02/ble-toys.git
-  GIT_TAG 650fb016ce36cfda2e8073764196655ee6a50567
+  GIT_TAG f6493150c1a7172bcd8c9cc1790829285f707ee9
   GIT_SUBMODULES json
   PREFIX ${BLE_TOYS}
-  CMAKE_ARGS -DCMAKE_INSTALL_PREFIX:PATH=<INSTALL_DIR>
+  CMAKE_ARGS -DCMAKE_INSTALL_PREFIX:PATH=<INSTALL_DIR> -DCMAKE_C_COMPILER:FILE=${CMAKE_C_COMPILER} -DCMAKE_CXX_COMPILER:FILE=${CMAKE_CXX_COMPILER}
 )
 
 # nghttp2
@@ -45,7 +46,7 @@ ExternalProject_Add(nghttp2
 # TODO: proper discovery
 
 # sm-mqtt-consumer
-add_executable(sm-mqtt-consumer sm-mqtt-consumer.cpp cassandra_support.h)
+add_executable(sm-mqtt-consumer sm-mqtt-consumer.cpp cassandra_support.h mqtt_support.h ${SHARED_SOURCES})
 add_dependencies(sm-mqtt-consumer cpp-driver)
 ## Boost
 target_link_libraries(sm-mqtt-consumer PRIVATE ${Boost_LIBRARIES})
@@ -60,8 +61,24 @@ add_dependencies(sm-mqtt-consumer ble-toys)
 target_include_directories(sm-mqtt-consumer PRIVATE ${BLE_TOYS}/include)
 target_link_libraries(sm-mqtt-consumer PRIVATE ${BLE_TOYS}/lib/trygvis/libtrygvis-sensor.a)
 
+# raw-mqtt-consumer
+add_executable(raw-mqtt-consumer raw-mqtt-consumer.cpp cassandra_support.h mqtt_support.h ${SHARED_SOURCES})
+add_dependencies(raw-mqtt-consumer cpp-driver)
+## Boost
+target_link_libraries(raw-mqtt-consumer PRIVATE ${Boost_LIBRARIES})
+## Cassandra
+target_include_directories(raw-mqtt-consumer PRIVATE ${CPP_DRIVER}/include)
+target_link_libraries(raw-mqtt-consumer PRIVATE ${CPP_DRIVER}/lib/libcassandra.so)
+## Mosquitto
+target_compile_options(raw-mqtt-consumer PUBLIC ${SHARED_COMPILE_OPTIONS})
+target_link_libraries(raw-mqtt-consumer PRIVATE mosquitto mosquittopp)
+## Ble toys
+add_dependencies(raw-mqtt-consumer ble-toys)
+target_include_directories(raw-mqtt-consumer PRIVATE ${BLE_TOYS}/include)
+target_link_libraries(raw-mqtt-consumer PRIVATE ${BLE_TOYS}/lib/trygvis/libtrygvis-sensor.a)
+
 # sm-http-server
-add_executable(sm-http-server sm-http-server.cpp cassandra_support.h http_support.h)
+add_executable(sm-http-server sm-http-server.cpp cassandra_support.h http_support.h ${SHARED_SOURCES})
 target_compile_options(sm-http-server PUBLIC ${SHARED_COMPILE_OPTIONS})
 ## Boost
 target_link_libraries(sm-http-server PRIVATE ${Boost_LIBRARIES})
@@ -76,6 +93,7 @@ target_link_libraries(sm-http-server PRIVATE ${NGHTTP2}/lib/libnghttp2_asio.a ${
 ## Misc
 target_link_libraries(sm-http-server PRIVATE ssl crypto pthread)
 
+# Testing
 enable_testing()
 add_executable(http-tests http-tests.cpp http_support.h)
 target_compile_options(http-tests PUBLIC ${SHARED_COMPILE_OPTIONS})
diff --git a/README.md b/README.md
index 91681e4..5d39e63 100644
--- a/README.md
+++ b/README.md
@@ -1,4 +1,4 @@
-# Schema
+# Soil Moisture Schema
 
     CREATE TABLE sm_by_day (
         device text,
@@ -11,6 +11,17 @@
         )
     );
 
+# Raw Schema
+
+    CREATE TABLE raw_record (
+        day text,
+        timestamp timestamp,
+        records list<frozen<tuple<text, text>>>,
+        PRIMARY KEY (
+            (day), 
+            timestamp
+        )
+    );
 
 # Create Schema
 
diff --git a/cassandra_support.h b/cassandra_support.h
index 7b50296..3f49be2 100644
--- a/cassandra_support.h
+++ b/cassandra_support.h
@@ -1,14 +1,18 @@
 #ifndef TRYGVIS_CASSANDRA_SUPPORT_H
 #define TRYGVIS_CASSANDRA_SUPPORT_H
 
-#include <stddef.h>
+#include <cassandra.h>
+
 #include <string>
+#include <cassert>
 #include <algorithm>
-#include <cassandra.h>
 #include <stdexcept>
 #include <functional>
 #include <vector>
 #include <iostream>
+#include <sstream>
+#include <iomanip>
+#include <mutex>
 
 namespace trygvis {
 namespace cassandra_support {
@@ -17,7 +21,7 @@ using namespace std;
 
 class cassandra_error : public runtime_error {
 public:
-    cassandra_error(const string &context, string &&error) : runtime_error(
+    cassandra_error(const string &context, const string &error) : runtime_error(
             "Cassandra error: context=" + context + ", error=" + error) {
     }
 
@@ -26,14 +30,7 @@ public:
     }
 };
 
-string error_message(CassFuture *future) {
-    const char *message;
-    size_t message_length;
-    cass_future_error_message(future, &message, &message_length);
-    return string(message, message_length);
-}
-
-static CassError execute_query(CassSession *session, const string &&query) {
+static CassError execute_query(CassSession *session, const string query) {
     CassStatement *statement = cass_statement_new(query.c_str(), 0);
 
     CassFuture *future = cass_session_execute(session, statement);
@@ -82,7 +79,7 @@ protected:
     cassandra_wrapper &operator=(const cassandra_wrapper &) = delete;
 
     virtual ~cassandra_wrapper() {
-        underlying_ = (Underlying *)0xaabbccdd;
+        underlying_ = reinterpret_cast<Underlying *>(0xaabbccdd);
     }
 
     Underlying *underlying_;
@@ -107,18 +104,155 @@ public:
     cassandra_result &operator=(const cassandra_result &) = delete;
 };
 
-static const CassResult *r;
+class cassandra_future2 {
+public:
+    // TODO: the values shouldn't be the future, but a specific result object instead
+    typedef void(callback_t)(cassandra_future2 &);
+
+    static
+    cassandra_future2 *wrap(CassFuture *future) {
+        return new cassandra_future2(future);
+    }
+
+private:
+    typedef std::lock_guard<std::mutex> guard;
+
+    callback_t *callback_ = nullptr;
+    CassFuture *future;
+    bool has_callback = false;
+    bool has_data = false;
+
+    std::mutex mutex;
+
+    cassandra_future2(CassFuture *future) : future(future), callback_(default_callback) {
+        cout << "cassandra_future2: this=" << std::hex << std::setw(8) << this <<
+        ", future=" << std::hex << std::setw(8) << future << endl;
+        cass_future_set_callback(future, callback, this);
+    }
+
+    static void default_callback(cassandra_future2 &) {
+        cout << "default_callback" << endl;
+    }
+
+    ~cassandra_future2() {
+        cout << "~cassandra_future2: this=" << std::hex << std::setw(8) << this << endl;
+//        cout << std::hex << std::setw(8) << "freeing future: " << underlying_ << endl;
+//        cass_future_free(underlying_);
+    }
+
+    cassandra_future2(const cassandra_future2 &) = delete;
+
+    cassandra_future2 &operator=(const cassandra_future2 &) = delete;
+
+public:
+    void then(callback_t callback_) {
+        bool do_delete = false;
+
+        {
+            guard lock(mutex);
+
+            assert(callback_ != nullptr);
+            this->callback_ = callback_;
+            has_callback = true;
+
+            if (has_data) {
+                cout << "Had early data" << endl;
+                this->callback(future);
+
+                cout << "freeing future: " << std::hex << std::setw(8) << future << endl;
+                cass_future_free(future);
+                do_delete = true;
+            }
+        }
+
+        if (do_delete) {
+            delete this;
+        }
+    }
+
+    bool fetched = false;
+
+    cassandra_result result() {
+        if (fetched) {
+            throw cassandra_error("cassandra_result::result()", "Already fetched");
+        }
+        fetched = true;
+        const CassResult *x = cass_future_get_result(future);
+        size_t count = cass_result_row_count(x);
+        return std::move(cassandra_result(x));
+    }
+
+    static
+    string error_message(CassFuture * future) {
+        const char *message;
+        size_t message_length;
+        cass_future_error_message(future, &message, &message_length);
+        return string(message, message_length);
+    }
+
+    string error_message() {
+        return error_message(future);
+    }
+
+    CassError error_code() const {
+        return cass_future_error_code(future);
+    }
 
-class cassandra_future;
+    bool operator!() const {
+        return !ok();
+    }
 
-typedef std::function<void(cassandra_future &)> callback_type;
+    operator bool() const {
+        return ok();
+    }
 
-struct tmp {
-    callback_type cb;
+    bool ok() const {
+        return error_code() == CASS_OK;
+    }
+
+private:
+
+    static void callback(CassFuture *f, void *data) {
+        cassandra_future2 *c_f = static_cast<cassandra_future2 *>(data);
+        c_f->callback(f);
+    }
+
+    void callback(CassFuture *future) {
+        bool do_delete = false;
+
+        {
+            guard lock(mutex);
+
+            cout << "cassandra_future::callback, error=" << cassandra_future2::error_message(future) << endl;
+            cout << "cassandra_future::callback, this=" << std::hex << std::setw(8) << (this) << endl;
+            CassError rc = cass_future_error_code(future);
+
+            has_data = true;
+            if (has_callback) {
+                cout << "had callback already" << endl;
+                callback_(*this);
+
+                cout << "freeing future: " << std::hex << std::setw(8) << future << endl;
+                cass_future_free(future);
+
+                do_delete = true;
+            }
+        }
+
+        if (do_delete) {
+            delete this;
+        }
+    }
 };
 
 class cassandra_future : public cassandra_wrapper<CassFuture> {
 public:
+    typedef std::function<void(cassandra_future &)> callback_type;
+
+    struct tmp {
+        cassandra_future::callback_type cb;
+    };
+
     cassandra_future(CassFuture *future) : cassandra_wrapper(future) {
     }
 
@@ -140,11 +274,13 @@ public:
         return std::move(cassandra_result(x));
     }
 
+    static
+    string error_message(CassFuture *future) {
+        return cassandra_future2::error_message(future);
+    }
+
     string error_message() {
-        const char *message;
-        size_t message_length;
-        cass_future_error_message(underlying(), &message, &message_length);
-        return string(message, message_length);
+        return error_message(underlying());
     }
 
     CassError error_code() const {
@@ -185,10 +321,14 @@ public:
 
     cassandra_tuple &operator=(const cassandra_tuple &) = delete;
 
-    void set(size_t i, cass_int32_t value) {
+    void set(size_t i, const cass_int32_t value) {
         cass_tuple_set_int32(tuple, i, value);
     }
 
+    void set(size_t i, const string &s) {
+        cass_tuple_set_string_n(tuple, i, s.c_str(), s.length());
+    }
+
     CassTuple *tuple;
 };
 
@@ -317,10 +457,19 @@ public:
 
     cassandra_session &operator=(const cassandra_session &) = delete;
 
-    void execute(cassandra_statement &&stmt, callback_type cb_) {
-        auto statement = cass_statement_new("SELECT device, timestamp, sensors FROM sm_by_day", 0);
+    auto connect(CassCluster *cluster) {
+        return cass_session_connect(session, cluster);
+    }
+
+    [[deprecated]]
+    void execute(cassandra_statement &&stmt, cassandra_future::callback_type cb_) {
+        auto future = cass_session_execute(session, stmt.underlying());
+        cass_future_set_callback(future, cassandra_future::callback, new cassandra_future::tmp{cb_});
+    }
+
+    cassandra_future2 *execute2(cassandra_statement &&stmt) {
         auto future = cass_session_execute(session, stmt.underlying());
-        cass_future_set_callback(future, cassandra_future::callback, new tmp{cb_});
+        return cassandra_future2::wrap(future);
     }
 
     operator CassSession *() const {
@@ -335,6 +484,28 @@ private:
     CassSession *session;
 };
 
+class cassandra_logging {
+public:
+    cassandra_logging(CassLogLevel log_level = CASS_LOG_DEBUG) {
+        cass_log_set_level(log_level);
+        cass_log_set_callback(on_log, this);
+    }
+
+    ~cassandra_logging() {
+    }
+
+private:
+    static
+    void on_log(const CassLogMessage *message, void *data) {
+        stringstream buf;
+        buf << message->time_ms << " " << cass_log_level_string(message->severity) << " " <<
+        message->file << ":" << message->function << ":" <<
+        message->line << ":" << message->message;
+
+        cout << "CASSANDRA: " << buf.str() << endl;
+    }
+};
+
 }
 }
 
diff --git a/misc_support.h b/misc_support.h
new file mode 100644
index 0000000..b00f6a8
--- /dev/null
+++ b/misc_support.h
@@ -0,0 +1,49 @@
+#ifndef SOIL_MOISTURE_MISC_SUPPORT_H
+#define SOIL_MOISTURE_MISC_SUPPORT_H
+
+#include <boost/lexical_cast.hpp>
+#include <boost/program_options.hpp>
+#include <boost/optional.hpp>
+#include <string>
+
+namespace trygvis {
+namespace misc_support {
+
+using namespace boost;
+using namespace std;
+namespace po = boost::program_options;
+
+template<typename Target, typename Source>
+static
+boost::optional<Target> map(boost::optional<Source> &a, std::function<Target(Source)> f) {
+    if (!a.is_initialized()) {
+        return boost::none;
+    }
+
+    return make_optional(f(a));
+}
+
+template<typename Target, typename Source>
+static
+boost::optional<Target> flat_map(boost::optional<Source> &a, boost::optional<Target> (&f)(Source)) {
+    if (!a.is_initialized()) {
+        return boost::none;
+    }
+
+    return f(a.get());
+}
+
+template<typename Target, typename Source = string>
+static
+boost::optional<Target> l_c(const Source source) {
+    try {
+        return boost::lexical_cast<Target>(source);
+    } catch (bad_lexical_cast &e) {
+        return boost::none;
+    }
+};
+
+}
+}
+
+#endif
diff --git a/mqtt_support.h b/mqtt_support.h
new file mode 100644
index 0000000..9a9f7fe
--- /dev/null
+++ b/mqtt_support.h
@@ -0,0 +1,25 @@
+#ifndef TRYGVIS_MQTT_SUPPORT_H
+#define TRYGVIS_MQTT_SUPPORT_H
+
+#include "mosquittopp.h"
+
+namespace trygvis {
+namespace mqtt_support {
+
+using namespace mosqpp;
+
+class mqtt_lib {
+public:
+    mqtt_lib() {
+        mosquitto_lib_init();
+    }
+
+    ~mqtt_lib() {
+        mosquitto_lib_cleanup();
+    }
+};
+
+}
+}
+
+#endif
diff --git a/raw-mqtt-consumer.cpp b/raw-mqtt-consumer.cpp
new file mode 100644
index 0000000..421d18c
--- /dev/null
+++ b/raw-mqtt-consumer.cpp
@@ -0,0 +1,239 @@
+#include "cassandra_support.h"
+#include "mqtt_support.h"
+#include <thread>
+#include <boost/lexical_cast.hpp>
+#include <boost/program_options.hpp>
+#include <trygvis/sensor.h>
+#include <trygvis/sensor/io.h>
+
+namespace raw_mqtt_consumer {
+
+using namespace std;
+using namespace std::chrono;
+using namespace trygvis::cassandra_support;
+using namespace trygvis::mqtt_support;
+using namespace trygvis::sensor;
+using namespace trygvis::sensor::io;
+using namespace boost;
+namespace po = boost::program_options;
+
+static bool should_run;
+static string mqtt_host;
+static int mqtt_port;
+static string mqtt_topic;
+static string keyspace_name = "soil_moisture";
+
+static unique_ptr<cassandra_session> current_cassandra_session;
+
+struct measurement {
+    KeyDictionary dict;
+};
+
+class raw_mqtt_client : private mosqpp::mosquittopp {
+public:
+    typedef std::function<void(const struct mosquitto_message *)> callback_t;
+
+    raw_mqtt_client(callback_t on_message_) : mosquittopp(),
+                                              on_message_(on_message_) {
+        cout << "Connecting to " << mqtt_host << ":" << mqtt_port << endl;
+        loop_start();
+        connect_async(mqtt_host.c_str(), mqtt_port, 10);
+    }
+
+    ~raw_mqtt_client() {
+        loop_stop(true);
+        disconnect();
+    }
+
+private:
+    callback_t on_message_;
+    bool subscribed = false;
+
+    void on_connect(int rc) override {
+        cout << "Connected to MQTT broker, rc=" << rc << endl;
+//        should_run = false;
+        int qos = 0;
+        if (!subscribed) {
+            subscribed = true;
+            cout << "Subscribing..." << endl;
+            subscribe(nullptr, mqtt_topic.c_str(), qos);
+        }
+    }
+
+    void on_disconnect(int rc) override {
+        subscribed = false;
+
+        cout << "Oops, disconnected, rc=" << rc << endl;
+    }
+
+    void on_publish(int mid) override {
+    }
+
+    void on_message(const struct mosquitto_message *message) override {
+        string payload((const char *) message->payload, (size_t) message->payloadlen);
+        on_message_(message);
+    }
+
+    void on_subscribe(int mid, int qos_count, const int *granted_qos) override {
+        cout << "Subscribed" << endl;
+    }
+
+    void on_unsubscribe(int mid) override {
+        cout << "Oops, unsubscribed" << endl;
+    }
+
+    void on_log(int level, const char *str) override {
+        cout << "MQTT: " << level << ":" << str << endl;
+    }
+
+    void on_error() override {
+        cout << "Oops, error" << endl;
+    }
+};
+
+cassandra_future2 *insert_into_raw(unique_ptr<cassandra_session> &session, const SampleRecord &record) {
+    cassandra_statement q("INSERT INTO raw_record(day, timestamp, records) VALUES (?, ?, ?);", 3);
+
+    auto system_now = system_clock::now();
+
+    char day[100];
+    std::time_t t = system_clock::to_time_t(system_now);
+    std::strftime(day, sizeof(day), "%Y-%m-%d", std::localtime(&t));
+    q.bind(0, day);
+    cout << "day=" << day << endl;
+
+    auto now_ms = std::chrono::time_point_cast<std::chrono::milliseconds>(system_now);
+    long timestamp = now_ms.time_since_epoch().count();
+    q.bind(1, timestamp);
+    cout << "timestamp=" << timestamp << endl;
+
+    auto buf = make_shared<stringstream>();
+    auto output = trygvis::sensor::io::open_sample_output_stream(buf, record.dict, sample_format_type::KEY_VALUE);
+    output->write(record);
+
+    cassandra_collection c(CASS_COLLECTION_TYPE_LIST, record.dict.size());
+    for_each(record.dict.begin(), record.dict.end(), [&](const SampleKey *key) {
+        cassandra_tuple tuple(2);
+        tuple.set(0, key->name);
+        o<string> value = record.at(key);
+        if (value) {
+            tuple.set(1, value.get());
+        }
+        c.append(std::move(tuple));
+    });
+
+    q.bind(2, c);
+
+    return session->execute2(std::move(q));
+}
+
+void on_message(const struct mosquitto_message *message) {
+    string payload((const char *) message->payload, (size_t) message->payloadlen);
+    cout << "storing message: " << endl;
+    cout << payload << endl;
+    cout << "----------------------------------------" << endl;
+
+    KeyDictionary dict;
+    auto sample_buffer = make_shared<VectorSampleOutputStream>();
+    auto input = make_shared<KeyValueSampleStreamParser>(sample_buffer, dict);
+
+    mutable_buffers_1 buffer = boost::asio::buffer(message->payload, (std::size_t) message->payloadlen);
+
+    input->process(buffer);
+    input->finish();
+
+    cout << "sample_buffer->samples: " << sample_buffer->samples.size() << endl;
+
+    if (current_cassandra_session) {
+        std::for_each(sample_buffer->samples.cbegin(), sample_buffer->samples.cend(), [&](auto &sample) {
+            cout << "Sample: " << sample.to_string() << endl;
+
+            insert_into_raw(current_cassandra_session, sample)->then([&](cassandra_future2 &f) {
+                if (f)
+                    cout << "Success!" << endl;
+                else {
+                    cout << "Error: " << f.error_message() << endl;
+                }
+            });
+
+            cout << "sample insert scheduled" << endl;
+        });
+    } else {
+        cout << "Not connected to Cassandra" << endl;
+    }
+}
+
+int main(int argc, const char **argv) {
+    mqtt_lib mqtt_lib;
+    cassandra_logging cassandra_logging;
+
+    string cassandra_cluster;
+    po::options_description all("Options");
+    all.add_options()("cassandra-cluster", po::value<string>(&cassandra_cluster)->default_value("127.0.0.1"));
+    all.add_options()("mqtt-host", po::value<>(&mqtt_host)->default_value("trygvis.io"));
+    all.add_options()("mqtt-port", po::value<>(&mqtt_port)->default_value(1883));
+    all.add_options()("mqtt-topic", po::value<>(&mqtt_topic)->required());
+
+    po::variables_map vm;
+    try {
+        auto parsed = po::parse_command_line(argc, argv, all);
+        po::store(parsed, vm);
+        po::notify(vm);
+        auto unrecognized = po::collect_unrecognized(parsed.options, po::include_positional);
+
+        if (vm.count("help")) {
+            cerr << all << "\n";
+            return EXIT_FAILURE;
+        }
+
+        if (unrecognized.size()) {
+            cerr << "Unrecognized option: " << unrecognized.at(0) << "\n";
+            return EXIT_FAILURE;
+        }
+
+    } catch (po::required_option &e) {
+        cerr << "Missing required option: " << e.get_option_name() << endl;
+        cerr << all << endl;
+    } catch (po::unknown_option &e) {
+        cerr << e.what() << endl;
+        return EXIT_FAILURE;
+    }
+
+    CassCluster *cluster = cass_cluster_new();
+    auto session = make_unique<cassandra_session>();
+
+    cout << "Connecting to Cassandra at " << cassandra_cluster << endl;
+    cass_cluster_set_contact_points(cluster, cassandra_cluster.c_str());
+
+    auto connect_future = session->connect(cluster);
+
+    if (cass_future_error_code(connect_future) != CASS_OK) {
+        string s = cassandra_future::error_message(connect_future);
+        cerr << "Could not connect to Cassandra: " << s << endl;
+        return EXIT_FAILURE;
+    }
+
+    execute_query(session->underlying(), "USE " + keyspace_name);
+
+    cout << "Connected to Cassandra" << endl;
+    current_cassandra_session = std::move(session);
+
+    raw_mqtt_client mqtt_client(on_message);
+
+    should_run = true;
+
+    while (should_run) {
+        cout << "sleeping.." << endl;
+        std::this_thread::sleep_for(60s);
+    }
+
+    current_cassandra_session.release();
+
+    return 0;
+}
+
+}
+
+int main(int argc, const char **argv) {
+    return raw_mqtt_consumer::main(argc, argv);
+}
diff --git a/sm-http-server.cpp b/sm-http-server.cpp
index b1cd37e..6b3169d 100644
--- a/sm-http-server.cpp
+++ b/sm-http-server.cpp
@@ -185,7 +185,7 @@ int main(int argc, const char *const argv[]) {
     auto connect_future = cass_session_connect(current_cassandra_session->underlying(), cluster);
 
     if (cass_future_error_code(connect_future) != CASS_OK) {
-        string s = error_message(connect_future);
+        string s = cassandra_future::error_message(connect_future);
         cerr << "Could not connect to Cassandra: " << s << endl;
         return EXIT_FAILURE;
     }
diff --git a/sm-mqtt-consumer.cpp b/sm-mqtt-consumer.cpp
index e1f6801..b91b663 100644
--- a/sm-mqtt-consumer.cpp
+++ b/sm-mqtt-consumer.cpp
@@ -1,15 +1,19 @@
 #include "cassandra_support.h"
-#include "mosquittopp.h"
-#include "trygvis/sensor/io.h"
+#include "mqtt_support.h"
+#include "misc_support.h"
 #include <thread>
-#include <boost/lexical_cast.hpp>
 #include <boost/program_options.hpp>
+#include <trygvis/sensor/io.h>
+
+namespace sm_mqtt_consumer {
 
 using namespace std;
 using namespace std::chrono;
 using namespace trygvis::sensor;
 using namespace trygvis::sensor::io;
 using namespace trygvis::cassandra_support;
+using namespace trygvis::mqtt_support;
+using namespace trygvis::misc_support;
 using namespace boost;
 namespace po = boost::program_options;
 
@@ -53,29 +57,12 @@ struct device_measurement {
     }
 };
 
-string &&to_string(CassFuture *f) {
-    const char *message;
-    size_t message_length;
-    cass_future_error_message(f, &message, &message_length);
-
-    return std::move(string(message, message_length));
-}
-
-class mqtt_lib {
-public:
-    mqtt_lib() {
-        mosquitto_lib_init();
-    }
-
-    ~mqtt_lib() {
-        mosquitto_lib_cleanup();
-    }
-};
-
 class mqtt_client : private mosqpp::mosquittopp {
 public:
-    mqtt_client(std::function<void(const struct mosquitto_message *)> on_message_) : mosquittopp(),
-                                                                                     on_message_(on_message_) {
+    typedef std::function<void(const struct mosquitto_message *)> callback_t;
+
+    mqtt_client(callback_t on_message_) : mosquittopp(),
+                                          on_message_(on_message_) {
         cout << "Connecting to " << mqtt_broker_host << ":" << mqtt_broker_port << endl;
         loop_start();
         connect_async(mqtt_broker_host.c_str(), mqtt_broker_port, 10);
@@ -87,7 +74,7 @@ public:
     }
 
 private:
-    std::function<void(const struct mosquitto_message *)> on_message_;
+    callback_t on_message_;
     bool subscribed = false;
 
     void on_connect(int rc) override {
@@ -132,10 +119,6 @@ private:
     }
 };
 
-void print_error(CassFuture *future) {
-    cout << "Cassandra error: " << error_message(future) << endl;
-}
-
 auto insert_into_sm_by_day(CassSession *session, device_measurement &&measurement) {
     cassandra_statement q("INSERT INTO sm_by_day(device, day, timestamp, sensors) VALUES (?, ?, ?, ?);", 4);
 
@@ -161,33 +144,6 @@ auto insert_into_sm_by_day(CassSession *session, device_measurement &&measuremen
     return cass_session_execute(session, q);
 }
 
-template<typename Target, typename Source>
-boost::optional<Target> map(boost::optional<Source> &a, std::function<Target(Source)> f) {
-    if (!a.is_initialized()) {
-        return boost::none;
-    }
-
-    return make_optional(f(a));
-}
-
-template<typename Target, typename Source>
-boost::optional<Target> flat_map(boost::optional<Source> &a, boost::optional<Target> (&f)(Source)) {
-    if (!a.is_initialized()) {
-        return boost::none;
-    }
-
-    return f(a.get());
-}
-
-template<typename Target, typename Source = string>
-boost::optional<Target> l_c(const Source source) {
-    try {
-        return boost::lexical_cast<Target>(source);
-    } catch (bad_lexical_cast &e) {
-        return boost::none;
-    }
-};
-
 void on_message(const struct mosquitto_message *message) {
     string payload((const char *) message->payload, (size_t) message->payloadlen);
     cout << "storing message: " << endl;
@@ -255,7 +211,7 @@ void on_message(const struct mosquitto_message *message) {
             handle_future(insert_into_sm_by_day((CassSession*) current_cassandra_session.get(), std::move(measurement)), [&](auto future) {
                 cout << "Success!" << endl;
             }, [&](auto future, auto err) {
-                cout << "Failure: " << error_message(future) << endl;
+                cout << "Failure: " << cassandra_future::error_message(future) << endl;
             });
         } else {
             cout << "Not connected to Cassandra" << endl;
@@ -306,7 +262,7 @@ int main(int argc, const char **argv) {
     connect_future = cass_session_connect((CassSession*) session.get(), cluster);
 
     if (cass_future_error_code(connect_future) != CASS_OK) {
-        string s = to_string(connect_future);
+        string s = cassandra_future::error_message(connect_future);
         cerr << "Could not connect to Cassandra:" << s << endl;
         return EXIT_FAILURE;
     }
@@ -326,3 +282,9 @@ int main(int argc, const char **argv) {
 
     return 0;
 }
+
+}
+
+int main(int argc, const char **argv) {
+    return sm_mqtt_consumer::main(argc, argv);
+}