leveldb

1 files changed, 181 insertions, 42 deletions

diff --git a/datastore-leveldb/src/main.cpp b/datastore-leveldb/src/main.cpp
index 4fd95df..9f8142e 100644
--- a/datastore-leveldb/src/main.cpp
+++ b/datastore-leveldb/src/main.cpp
@@ -8,13 +8,59 @@ extern "C" {
 #include <sys/sendfile.h>
 #include <sys/stat.h>
 #include <fcntl.h>
+#include <magic.h>
 }
 
+#include <algorithm>
 #include <iostream>
+#include <iomanip>
+#include <string>
+#include <sstream>
+#include <map>
+#include <forward_list>
+#include <functional>
+
+#include <boost/regex.hpp>
 
 #include "leveldb/db.h"
 #include "leveldb/comparator.h"
 
+std::forward_list<std::pair<boost::regex,std::function<void(const boost::cmatch&, struct http_request*, const int)>>> web_handler;
+
+static std::map<std::string,leveldb::DB*> dbs;
+
+static magic_t magic_cookie;
+
+
+bool sensor_name_is_sane(std::string& name) {
+    for (auto it = name.begin(); it != name.end(); ++it) {
+        if (not (*it >= '0' and *it <= '9' or
+                 *it >= 'A' and *it <= 'Z' or
+                 *it >= 'a' and *it <= 'z')) {
+            return false;
+        }
+    }
+    return true;
+}
+
+
+leveldb::DB *getDB(std::string& name) {
+    if (not sensor_name_is_sane(name)) {
+        return nullptr;
+    }
+    if (dbs.find(name) == dbs.end()) {
+        leveldb::DB *db;
+        leveldb::Options options;
+        options.create_if_missing = true;
+        leveldb::Status status = leveldb::DB::Open(options, "/tmp/testdb."+name, &db);
+        if (not status.ok()) {
+            return nullptr;
+        }
+        dbs[name] = db;
+    }
+    return dbs.at(name);
+}
+
 // see http_parser.h
 // char DELETE = 0;
 // char GET = 1;
@@ -22,69 +68,139 @@ extern "C" {
 // char POST = 3;
 // char PUT = 4;
 
+
+static inline void http_ok(int fd, const char *content_type, const char *extra_headers) {
+#define write_const(fd, text) \
+    write(fd, text, strlen(text));
+ 
+    write_const(fd, "HTTP/1.1 200 OK\r\nContent-Type: ");
+    write_const(fd, content_type);
+    write_const(fd, "\r\n");
+    if (extra_headers) write_const(fd, extra_headers);
+    write_const(fd, "\r\n");
+#undef write_const
+}
+
+
 static inline void send_file(const char *path, int fd_out) {
     int fd_in;
     struct stat stat_buf;
 
+    const char *mime_type = magic_file(magic_cookie, path);
+    if (mime_type == NULL) {
+        mime_type = "application/octet-stream";
+    }
+
+    http_ok(fd_out, mime_type, nullptr);
+
     fd_in = open(path, O_RDONLY);
     fstat(fd_in, &stat_buf);
+    std::cerr << "GET " << path << std::endl;
     sendfile(fd_out, fd_in, 0, stat_buf.st_size);
     close(fd_in);
 }
 
-static inline void http_ok(int fd, const char *content_type) {
-#define write_const(fd, text) \
-    write(fd, text, strlen(text));
- 
-    write_const(fd, "HTTP/1.1 200 OK\r\nContent-Type: ");
-    write_const(fd, content_type);
-    write_const(fd, "\r\n\r\n");
-#undef write_const
+std::string make_key(uint64_t timestamp) {
+    std::stringstream key;
+    key << "ts-";
+    key << std::setfill('0') << std::setw(20) << timestamp;
+    return key.str();
 }
 
-static inline void http_fail(int fd) {
-    write(fd, "HTTP/1.1 501 FAIL\r\n\r\n",25);
+void web_handle_api_value(const boost::cmatch &match, const struct http_request *request, const int fd) {
+    const char* reply_OK = "HTTP/1.1 200 Value received\r\n\r\n";
+    const char* reply_ERR = "HTTP/1.1 500 Internal Error\r\n\r\n";
+    
+    std::string sensor(match[1].str());
+    uint64_t timestamp = std::stoul(match[2].str());
+    std::string value(request->body);
+
+    leveldb::DB *db = getDB(sensor);
+    if (db == nullptr) {
+        write(fd, reply_ERR, strlen(reply_ERR));
+        return;
+    }
+
+    std::cout << "sensor=" << sensor << " key=" << make_key(timestamp) << std::endl;
+    db->Put(leveldb::WriteOptions(), make_key(timestamp), value);
+    write(fd, reply_OK, strlen(reply_OK));
+}
+
+void web_handle_api_range(const boost::cmatch &match, const struct http_request *request, const int fd) {
+    static const leveldb::Comparator *cmp = leveldb::BytewiseComparator();
+
+    const char* reply_OK = "HTTP/1.1 200 Value received\r\n";
+    const char* reply_ERR = "HTTP/1.1 500 Internal Error\r\n\r\n";
+    const char* content_type = "Content-Type: application/json; encoding=UTF-8\r\n";
+ 
+    std::string sensor(match[1].str());
+    uint64_t start = std::stoul(match[2].str());
+    uint64_t end = std::stoul(match[3].str());
+    std::string key_start(std::move(make_key(start)));
+    std::string key_end(std::move(make_key(end)));
+
+    leveldb::DB *db = getDB(sensor);
+    if (db == nullptr) {
+        write(fd, reply_ERR, strlen(reply_ERR));
+        return;
+    }
+        
+    http_ok(fd, "application/json; encoding=UTF-8", nullptr);
+
+    std::cout << "sensor=" << sensor << " start=" << start << " end=" << end << std::endl;
+
+    std::ostringstream out;
+    out << "{'sensor':'" << sensor << "', 'error':null, 'data':[";
+    write(fd, out.str().c_str(), out.str().size());
+
+    usleep(2000000);
+    leveldb::Iterator* it = db->NewIterator(leveldb::ReadOptions());
+    bool first = true;
+    for (it->Seek(key_start); 
+             it->Valid() && cmp->Compare(it->key(), key_end) < 0;
+             it->Next()) {
+            if (it->key().size() != 20+3) {
+                std::cerr << "invalid key" << std::endl;
+                return;
+            }
+            uint64_t timestamp = std::stoul(it->key().data()+3);
+            std::string s_timestamp = std::to_string(timestamp);
+
+            if (first) {
+                first = false;
+                write(fd, "[", 1);
+            } else {
+                write(fd, ",[", 2);
+            }
+            write(fd, s_timestamp.c_str(), s_timestamp.size());
+            write(fd, ",'", 2);
+            write(fd, it->value().data(), it->value().size());
+            write(fd, "']", 2);
+    }
+
+    delete it;
 }
 
 void handle_request(struct http_request *request, int fd) {
-#define write_const(fd, text) \
-    write(fd, text, strlen(text));
-    
-    if (request->url == NULL) { // happens only under high load
+    boost::cmatch match;
+    const char *error = "HTTP/1.1 404 Not Found\r\n\r\n";
+
+    if (request->url == NULL) { // happens only under high load, why?
         std::cerr << "url is null" << std::endl;
-        http_fail(fd);
-        close(fd);
-        return;
+        goto handled;
     }
 
-    std::string url(request->url);
-
-    if (url == "/") {
-        http_ok(fd, "text/html; charset=UTF-8");
-        send_file("index.html", fd);        
-    } else if (url.find("/api/put") == 0) {
-        http_ok(fd, "text/html; charset=UTF-8");
-        std::cout << "put" << std::endl;
-    } else if (url.find("/debug") == 0) {
-        http_ok(fd, "text/html; charset=UTF-8");
-        struct http_header *header = request->headers;
-        write_const(fd, "<pre>Headers:\n");
-        while (header != NULL) {
-            write_const(fd, header->name);
-            write_const(fd, ": ");
-            write_const(fd, header->value);
-            write_const(fd, "\n");
-            header = header->next;
-        }
-        if (request->flags & F_HREQ_KEEPALIVE) {
-             write_const(fd, "\nis keepalive.\n");
+    for (auto item = web_handler.begin(); item != web_handler.end(); ++item) {
+        if (boost::regex_match(request->url, match, (*item).first)) {
+            (*item).second(match, request, fd);
+            goto handled;
         }
-        write_const(fd, "\r\n\r\n");
-    } else {
-        http_fail(fd);
     }
+    
+    write(fd, error, strlen(error));
+
+    handled:
     close(fd);
-#undef write_const
 }
 
 static struct http_server server;
@@ -118,6 +234,29 @@ int main(int argc, char **argv) {
     on_sigint.sa_flags = 0;
     sigaction(SIGINT, &on_sigint, NULL);
 
+    // Routing
+    web_handler.push_front(std::make_pair(
+                boost::regex("/"),
+                [](const boost::cmatch &match, const struct http_request *request, const int fd){
+                    send_file("index.html", fd);
+                }));
+    web_handler.push_front(std::make_pair(
+                boost::regex("/public/(.+)"),
+                [](const boost::cmatch &match, const struct http_request *request, const int fd){
+                    /// XXX possible directory traversion
+                    std::string path("public/" + match[1].str());
+                    send_file(path.c_str(), fd);
+                }));
+    web_handler.push_front(std::make_pair(
+                boost::regex("/api/value/([a-zA-Z0-9]+)/([0-9]+)"),
+                web_handle_api_value));
+    web_handler.push_front(std::make_pair(
+                boost::regex("/api/range/([a-zA-Z0-9]+)/([0-9]+)/([0-9]+)"),
+                web_handle_api_range));
+
+    magic_cookie = magic_open(MAGIC_MIME_TYPE);
+    magic_load(magic_cookie, NULL);
+
     // start the server
     return http_server_loop(&server);
 }