o Fully functional reception from MQTT into Cassandra.

3 files changed, 250 insertions, 93 deletions

diff --git a/CMakeLists.txt b/CMakeLists.txt
index 9c379c5..74197cb 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -3,7 +3,7 @@ project(mqtt_cassandra_bridge)
 
 find_package(Boost COMPONENTS regex system program_options REQUIRED)
 
-add_executable(mqtt_cassandra_bridge main.cpp)
+add_executable(mqtt_cassandra_bridge main.cpp cassandra_support.h)
 target_link_libraries(mqtt_cassandra_bridge PUBLIC ${Boost_LIBRARIES})
 target_compile_options(mqtt_cassandra_bridge PUBLIC "-std=c++14")
 
diff --git a/cassandra_support.h b/cassandra_support.h
new file mode 100644
index 0000000..167f69b
--- /dev/null
+++ b/cassandra_support.h
@@ -0,0 +1,151 @@
+#ifndef TRYGVIS_CASSANDRA_SUPPORT_H
+#define TRYGVIS_CASSANDRA_SUPPORT_H
+
+#include <stddef.h>
+#include <string>
+#include <cassandra.h>
+#include <stdexcept>
+
+namespace trygvis {
+namespace cassandra_support {
+
+using namespace std;
+
+class cassandra_error : runtime_error {
+public:
+    cassandra_error(const string &context, CassError error) : runtime_error("Cassandra error: context=" + context + ", error=" + to_string((int)error)) {
+    }
+};
+
+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) {
+    CassStatement *statement = cass_statement_new(query.c_str(), 0);
+
+    CassFuture *future = cass_session_execute(session, statement);
+    cass_future_wait(future);
+
+    CassError rc = cass_future_error_code(future);
+
+    cass_future_free(future);
+    cass_statement_free(statement);
+
+    return rc;
+}
+
+void assert_ok(const string &context, CassError &err) {
+    if (err == CASS_OK) {
+        return;
+    }
+
+    throw cassandra_error(context, err);
+}
+
+class cassandra_tuple {
+public:
+    cassandra_tuple(size_t item_count) {
+        tuple = cass_tuple_new(item_count);
+    }
+
+    ~cassandra_tuple() {
+        cass_tuple_free(tuple);
+    }
+
+    void set(size_t i, cass_int32_t value) {
+        cass_tuple_set_int32(tuple, i, value);
+    }
+
+    CassTuple *tuple;
+};
+
+class cassandra_collection {
+public:
+    cassandra_collection(CassCollectionType type, size_t item_count) {
+        collection = cass_collection_new(type, item_count);
+    }
+
+    ~cassandra_collection() {
+        cass_collection_free(collection);
+    }
+
+    void append_tuple(cassandra_tuple &&tuple) {
+        cass_collection_append_tuple(collection, tuple.tuple);
+    }
+
+    CassCollection *collection;
+};
+
+CassError wait_for_future(CassFuture *future) {
+    cass_future_wait(future);
+
+    CassError rc = cass_future_error_code(future);
+
+    cass_future_free(future);
+
+    return rc;
+};
+
+class cassandra_statement {
+public:
+    cassandra_statement(string q, size_t argument_count) {
+        statement = cass_statement_new(q.c_str(), argument_count);
+    };
+
+    ~cassandra_statement() {
+        cass_statement_free(statement);
+    }
+
+    void bind(size_t i, const string &value) {
+        auto err = cass_statement_bind_string(statement, i, value.c_str());
+        assert_ok("cass_statement_bind_string", err);
+    }
+
+    void bind(size_t i, const char *value) {
+        auto err = cass_statement_bind_string(statement, i, value);
+        assert_ok("cass_statement_bind_string", err);
+    }
+
+    void bind(size_t i, const cass_int64_t value) {
+        auto err = cass_statement_bind_int64(statement, i, value);
+        assert_ok("cass_statement_bind_int64", err);
+    }
+
+    void bind(size_t i, const CassCollection *value) {
+        auto err = cass_statement_bind_collection(statement, i, value);
+        assert_ok("cass_statement_bind_collection", err);
+    }
+
+    void bind(size_t i, const cassandra_collection &value) {
+        auto err = cass_statement_bind_collection(statement, i, value.collection);
+        assert_ok("cass_statement_bind_collection", err);
+    }
+
+    CassStatement *statement;
+};
+
+class cassandra_session {
+public:
+    cassandra_session() {
+        session = cass_session_new();
+    }
+
+    ~cassandra_session() {
+        cass_session_free(session);
+    }
+
+    cassandra_session(const cassandra_session &) = delete;
+
+    cassandra_session &operator=(const cassandra_session &) = delete;
+
+    CassSession *session;
+};
+
+}
+}
+
+#endif
diff --git a/main.cpp b/main.cpp
index 0cd5622..744f697 100644
--- a/main.cpp
+++ b/main.cpp
@@ -1,5 +1,5 @@
+#include "cassandra_support.h"
 #include "mosquittopp.h"
-#include "cassandra.h"
 #include "trygvis/sensor/io.h"
 #include <thread>
 #include <boost/lexical_cast.hpp>
@@ -9,6 +9,7 @@ using namespace std;
 using namespace std::chrono;
 using namespace trygvis::sensor;
 using namespace trygvis::sensor::io;
+using namespace trygvis::cassandra_support;
 using namespace boost;
 namespace po = boost::program_options;
 
@@ -18,6 +19,40 @@ static auto mqtt_broker_port = 1883;
 static auto queue_name = "/trygvis";
 static string keyspace_name = "soil_moisture";
 
+static unique_ptr<cassandra_session> current_cassandra_session;
+
+struct sensor_measurement {
+    int sensor;
+    int value;
+
+    sensor_measurement(int sensor, int value) : sensor(sensor), value(value) {
+    }
+
+    ~sensor_measurement() = default;
+};
+
+struct device_measurement {
+    string device;
+    long timestamp;
+    vector<sensor_measurement> sensors;
+
+    device_measurement(string &device, long timestamp, vector<sensor_measurement> &&sensors) :
+            device(device), timestamp(timestamp), sensors(std::move(sensors)) {
+    };
+
+    ~device_measurement() = default;
+
+    string str() {
+        stringstream buf;
+        buf << "device=" << device;
+        buf << ", timestamp=" << timestamp;
+        std::for_each(sensors.begin(), sensors.end(), [&](auto &sensor) {
+            buf << ", #" << sensor.sensor << "=" + sensor.value;
+        });
+        return buf.str();
+    }
+};
+
 string &&to_string(CassFuture *f) {
     const char *message;
     size_t message_length;
@@ -98,30 +133,7 @@ private:
 };
 
 void print_error(CassFuture *future) {
-    const char *message;
-    size_t message_length;
-    cass_future_error_message(future, &message, &message_length);
-    string msg(message, message_length);
-    cout << "Cassandra error: " << msg << endl;
-}
-
-CassError execute_query(CassSession *session, const string &&query) {
-    CassError rc = CASS_OK;
-    CassFuture *future = NULL;
-    CassStatement *statement = cass_statement_new(query.c_str(), 0);
-
-    future = cass_session_execute(session, statement);
-    cass_future_wait(future);
-
-    rc = cass_future_error_code(future);
-    if (rc != CASS_OK) {
-        print_error(future);
-    }
-
-    cass_future_free(future);
-    cass_statement_free(statement);
-
-    return rc;
+    cout << "Cassandra error: " << error_message(future) << endl;
 }
 
 /*
@@ -133,84 +145,67 @@ CassError execute_query(CassSession *session, const string &&query) {
       PRIMARY KEY ((device, day), timestamp)
     )
  */
+CassError insert_into_sm_by_day(CassSession *session, device_measurement &&measurement) {
+    cassandra_statement q("INSERT INTO sm_by_day(device, day, timestamp, sensors) VALUES (?, ?, ?, ?);", 4);
 
-CassError insert_into_sm_by_day(CassSession *session, string payload) {
-    CassError rc = CASS_OK;
-    CassStatement *statement = nullptr;
-    CassFuture *future = nullptr;
-    auto query = "INSERT INTO sm_by_day(device, day, timestamp, sensors) VALUES (?, ?, ?, ?);";
-
-    statement = cass_statement_new(query, 4);
-
-    auto device = "aa:bb:cc:dd:ee:ff";
+    q.bind(0, measurement.device);
 
     std::time_t t = std::time(NULL);
     char day[100];
-    std::strftime(day, sizeof(day), "%Y-%M-%D", std::localtime(&t));
-    cass_statement_bind_string(statement, 0, device);
-    cass_statement_bind_string(statement, 1, day);
+    std::strftime(day, sizeof(day), "%Y-%m-%d", std::localtime(&t));
+    q.bind(1, day);
 
     auto timestamp = std::time(NULL);
-    cass_statement_bind_int64(statement, 2, timestamp);
-
-    future = cass_session_execute(session, statement);
-    cass_future_wait(future);
-
-    rc = cass_future_error_code(future);
-    if (rc != CASS_OK) {
-        print_error(future);
-    }
+    q.bind(2, timestamp);
+
+    cassandra_collection sensors(CASS_COLLECTION_TYPE_LIST, measurement.sensors.size());
+    for_each(measurement.sensors.cbegin(), measurement.sensors.cend(), [&](auto sensor) {
+        cassandra_tuple tuple(2);
+        tuple.set(0, sensor.sensor);
+        tuple.set(1, sensor.value);
+        sensors.append_tuple(std::move(tuple));
+    });
 
-    cass_future_free(future);
-    cass_statement_free(statement);
+    q.bind(3, sensors);
 
-    return rc;
+    return wait_for_future(cass_session_execute(session, q.statement));
 }
 
-struct sensor_measurement {
-    int sensor;
-    int value;
-
-    sensor_measurement(int sensor, int value) : sensor(sensor), value(value) {
+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;
     }
 
-    ~sensor_measurement() = default;
-};
-
-struct device_measurement {
-    string device;
-    long timestamp;
-    vector<sensor_measurement> sensors;
-
-    device_measurement(string &device, long timestamp, vector<sensor_measurement> &&sensors) :
-            device(device), timestamp(timestamp), sensors(std::move(sensors)) {
-    };
-
-    ~device_measurement() = default;
-
-    string str() {
-        stringstream buf;
-        buf << "device=" << device;
-        buf << ", timestamp=" << timestamp;
-        std::for_each(sensors.begin(), sensors.end(), [&](auto &sensor) {
-            buf << ", #" << sensor.sensor << "=" + sensor.value;
-        });
-        return buf.str();
-    }
-};
+    return make_optional(f(a));
+}
 
 template<typename Target, typename Source>
-boost::optional<Target> lexical_cast_optional(boost::optional<Source> &a) {
+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>
+//boost::optional<Target> flat_map(boost::optional<Source> &a, std::function<boost::optional<Target>(Source)> f) {
+//    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>(a);
+        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);
@@ -220,7 +215,6 @@ void on_message(const struct mosquitto_message *message) {
 
     KeyDictionary dict;
     auto sample_buffer = make_shared<VectorSampleOutputStream>();
-//    auto parser = open_sample_stream_parser(sample_buffer, dict);
     auto input = make_shared<KeyValueSampleStreamParser>(sample_buffer, dict);
 
     mutable_buffers_1 buffer = boost::asio::buffer(message->payload, (std::size_t) message->payloadlen);
@@ -249,7 +243,7 @@ void on_message(const struct mosquitto_message *message) {
 
         auto device = deviceO.get();
 
-        auto timestamp = lexical_cast_optional<long>(timestampS);
+        auto timestamp = flat_map(timestampS, l_c<long>);
         if (!timestamp) {
             cout << "Invalid value for 'timestamp'" << endl;
         }
@@ -257,17 +251,15 @@ void on_message(const struct mosquitto_message *message) {
         vector<sensor_measurement> sensors;
 
         for (int i = 0; i < 10; i++) {
-            auto sensorS = sample.at(dict.indexOf("sensor" + to_string(i)));
-            auto valueS = sample.at(dict.indexOf("value" + to_string(i)));
+            auto valueS = sample.at(dict.indexOf("sensor" + to_string(i)));
 
-            auto sensor = lexical_cast_optional<int>(sensorS);
-            auto value = lexical_cast_optional<int>(valueS);
+            auto value = flat_map(valueS, l_c<int>);
 
-            if (!sensor || !value) {
+            if (!value) {
                 continue;
             }
 
-            sensors.emplace_back(sensor.get(), value.get());
+            sensors.emplace_back(i, value.get());
         }
 
         if (sensors.size() == 0) {
@@ -277,6 +269,16 @@ void on_message(const struct mosquitto_message *message) {
         device_measurement measurement(device, timestamp.get(), std::move(sensors));
 
         cout << "Measurement: " << measurement.str() << endl;
+
+        if (current_cassandra_session) {
+            auto rc = insert_into_sm_by_day(current_cassandra_session->session, std::move(measurement));
+
+            cout << "rc=" << rc << endl;
+
+            assert_ok("wait_for_future", rc);
+        } else {
+            cout << "Not connected to Cassandra" << endl;
+        }
     });
 }
 
@@ -316,20 +318,22 @@ int main(int argc, const char **argv) {
     mqtt_client mqtt_client(on_message);
     CassFuture *connect_future = nullptr;
     CassCluster *cluster = cass_cluster_new();
-    CassSession *session = cass_session_new();
+    auto session = make_unique<cassandra_session>();
 
     cass_cluster_set_contact_points(cluster, cassandra_cluster.c_str());
 
-    connect_future = cass_session_connect(session, cluster);
+    connect_future = cass_session_connect(session->session, cluster);
 
     if (cass_future_error_code(connect_future) != CASS_OK) {
         string s = to_string(connect_future);
         cerr << "Could not connect to Cassandra:" << s << endl;
+        return EXIT_FAILURE;
     }
 
     cout << "Connected to Cassandra" << endl;
+    current_cassandra_session = std::move(session);
 
-    execute_query(session, "USE " + keyspace_name);
+    execute_query(current_cassandra_session->session, "USE " + keyspace_name);
 
     should_run = true;
     while (should_run) {
@@ -337,5 +341,7 @@ int main(int argc, const char **argv) {
         std::this_thread::sleep_for(60s);
     }
 
+    current_cassandra_session.release();
+
     return 0;
 }