o More Cassandra wrappers.

o Using more futures.

1 files changed, 44 insertions, 38 deletions

diff --git a/sm-http-server.cpp b/sm-http-server.cpp
index 7ac4e67..b1cd37e 100644
--- a/sm-http-server.cpp
+++ b/sm-http-server.cpp
@@ -1,11 +1,6 @@
 #include "cassandra_support.h"
 #include "http_support.h"
-#include <nghttp2/asio_http2_server.h>
-#include <iostream>
-#include <string>
 #include <boost/program_options.hpp>
-#include <boost/algorithm/string.hpp>
-#include <boost/algorithm/string/split.hpp>
 #include <cxxabi.h>
 
 using namespace std;
@@ -53,7 +48,7 @@ cass_int32_t read_value_int32(const CassRow *row, const size_t index) {
 }
 
 void handle_device_get(const request &req, const response &res, string device) {
-    if(!current_cassandra_session) {
+    if (!current_cassandra_session) {
         header_map headers;
         headers.emplace("content-type", text_plain);
         res.write_head(503, headers);
@@ -65,19 +60,30 @@ void handle_device_get(const request &req, const response &res, string device) {
 
     cassandra_statement stmt("SELECT device, timestamp, sensors FROM sm_by_day WHERE device=? AND day IN ?", 2);
     stmt.bind(0, device);
-    vector<string> days = {"2015-07-10", "2015-07-11", "2015-07-12", "2015-07-13", "2015-07-14", "2015-07-15", "2015-07-16"};
+    vector<string> days = {"2015-07-10", "2015-07-11", "2015-07-12", "2015-07-13", "2015-07-14", "2015-07-15",
+                           "2015-07-16"};
     stmt.bind(1, std::move(days));
 
-    auto f = cass_session_execute(current_cassandra_session->session, stmt.statement);
-    handle_future(f, [&](auto future) {
+    current_cassandra_session->execute(std::move(stmt), [&](cassandra_future& future) {
+        const cassandra_result result = future.result();
+        auto x = result.underlying();
+        if (!future.ok()) {
+            header_map headers;
+            headers.emplace("content-type", text_plain);
+            res.write_head(500, headers);
+
+            stringstream buf;
+            buf << "Bad shit: " << future.error_message() << "\r\n";
+            res.end(buf.str());
+            return;
+        }
+
         header_map headers;
         headers.emplace("content-type", application_json);
         res.write_head(200, headers);
 
-        const CassResult *result = cass_future_get_result(future);
-        size_t count = cass_result_row_count(result);
-        cout << "row count: " << count << endl;
-        CassIterator *rows = cass_iterator_from_result(result);
+        size_t count = cass_result_row_count(x);
+        CassIterator *rows = cass_iterator_from_result(x);
 
         stringstream buf;
         buf << "[";
@@ -88,8 +94,7 @@ void handle_device_get(const request &req, const response &res, string device) {
 
             string d = read_string(row, 0);
             auto timestamp = read_value_int64(row, 1);
-//            auto sensors = read_string(row, 1);
-            int value = -1;
+//            auto sensors = read_list(row, 1);
 
             if (!first) {
                 buf << ",";
@@ -101,18 +106,9 @@ void handle_device_get(const request &req, const response &res, string device) {
 
         buf << endl << "]" << endl;
 
-        cass_result_free(result);
         cass_iterator_free(rows);
 
         res.end(buf.str() + "\r\n");
-    }, [&](auto future, auto err) {
-        header_map headers;
-        headers.emplace("content-type", text_plain);
-        res.write_head(500, headers);
-
-        stringstream buf;
-        buf << "Bad shit: " << error_message(future) << "\r\n";
-        res.end(buf.str());
     });
 }
 
@@ -120,7 +116,7 @@ using namespace __cxxabiv1;
 
 std::string util_demangle(std::string to_demangle) {
     int status = 0;
-    char * buff = __cxxabiv1::__cxa_demangle(to_demangle.c_str(), NULL, NULL, &status);
+    char *buff = __cxxabiv1::__cxa_demangle(to_demangle.c_str(), NULL, NULL, &status);
     std::string demangled = buff;
     std::free(buff);
     return demangled;
@@ -139,6 +135,15 @@ void internal_server_error(const response &res, const string &msg) {
     res.end(s);
 }
 
+void on_logging_from_cassandra(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;
+}
+
 int main(int argc, const char *const argv[]) {
     string cassandra_cluster;
     po::options_description all("Options");
@@ -169,30 +174,31 @@ int main(int argc, const char *const argv[]) {
         return EXIT_FAILURE;
     }
 
-    CassFuture *connect_future = nullptr;
-    CassCluster *cluster = cass_cluster_new();
-    auto session = make_unique<cassandra_session>();
+    cass_log_set_level(CASS_LOG_INFO);
+    cass_log_set_callback(on_logging_from_cassandra, nullptr);
 
+    auto cluster = cass_cluster_new();
     cass_cluster_set_contact_points(cluster, cassandra_cluster.c_str());
+    cass_cluster_set_num_threads_io(cluster, 1);
 
-    connect_future = cass_session_connect(session->session, cluster);
+    current_cassandra_session = make_unique<cassandra_session>();
+    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);
-        cerr << "Could not connect to Cassandra:" << s << endl;
+        cerr << "Could not connect to Cassandra: " << s << endl;
         return EXIT_FAILURE;
     }
 
     cout << "Connected to Cassandra" << endl;
-    current_cassandra_session = std::move(session);
 
-    execute_query(current_cassandra_session->session, "USE " + keyspace_name);
+    execute_query(current_cassandra_session->underlying(), "USE " + keyspace_name);
     boost::system::error_code ec;
     http2 server;
     server.num_threads(4);
 
     server.handle("/", [](const request &req, const response &res) {
-        cerr << req.method() << " " << req.uri().path << endl;
+        cout << req.method() << " " << req.uri().path << endl;
 
         vector<string> paths;
         auto &path = req.uri().path;
@@ -218,9 +224,9 @@ int main(int argc, const char *const argv[]) {
                 res.write_head(404);
                 res.end("Not found :(\r\n");
             }
-        } catch (const exception& ex) {
+        } catch (const exception &ex) {
             internal_server_error(res, ex.what());
-        } catch (const string& ex) {
+        } catch (const string &ex) {
             internal_server_error(res, ex);
         } catch (...) {
             auto type = util_demangle(__cxa_current_exception_type()->name());
@@ -228,11 +234,11 @@ int main(int argc, const char *const argv[]) {
         }
     });
 
-    std::cerr << "Starting server" << endl;
+    cout << "Starting HTTP listener" << endl;
     if (server.listen_and_serve(ec, "127.0.0.1", "3000")) {
-        std::cerr << "error: " << ec.message() << std::endl;
+        cout << "error: " << ec.message() << endl;
     }
-    std::cerr << "woot?" << endl;
+    cout << "woot?" << endl;
 
     return EXIT_SUCCESS;
 }