hellcat.cpp

#include <iostream>
#include <sstream>
#include <fstream>
#include <haywire.h>
#include "lmdb.h"
#include "common.h"
#include "storage/store.h"
#include "storage/lmdb_store.h"
#include "haywire.h"
#include "hellcat.h"
#include <mutex>
#include <thread>
#include <chrono>
#include <unordered_map>
#include <uuid/uuid.h>
#include <boost/thread/shared_mutex.hpp>

using namespace std;
using namespace hellcat::storage;

void create_http_endpoint(int http_listen_port, int thread_count, int response_batch_size);
void get_root(http_request* request, hw_http_response* response, void* user_data);
void response_complete(void* user_data);

static unique_ptr<Store> store;

static boost::shared_mutex transaction_mutex;

struct txn
{
    hcat_transaction* tx;
    int transactions_count = 0;
};
static std::unordered_map<uuid_t*, txn*> transactions;

static hcat_transaction* tx;

void commit_thread_loop()
{
    while(true)
    {
        std::this_thread::sleep_for(std::chrono::seconds(2));
        transaction_mutex.lock();
        tx->commit();
        store->begin_transaction(&tx, 1);
        transaction_mutex.unlock();
        store->sync();
    }
}

int main(int args, char* argsv[]) {
    store = unique_ptr<Store>(new LMDBStore());
    int rc = store->open("/ramdisk", true);
    
    int http_listen_port = 8000;
    int thread_count = 0;
    int response_batch_size = 1;
    
    if (args > 1)
    {
        http_listen_port = atoi(argsv[1]);
    }
    
    if (args > 2)
    {
        thread_count = atoi(argsv[2]);
    }
    if (args > 3)
    {
        response_batch_size = atoi(argsv[3]);
    }
    
    std::thread commit_thread(commit_thread_loop);
    
    create_http_endpoint(http_listen_port, thread_count, response_batch_size);
    
    return 0;
}

void create_http_endpoint(int http_listen_port, int thread_count, int response_batch_size)
{
    store->begin_transaction(&tx, 1);
    
    char route[] = "/";
    configuration config;
    config.http_listen_address = "0.0.0.0";
    config.http_listen_port = 8000;
    config.thread_count = thread_count;
    config.response_batch_size = response_batch_size;
    
    hw_init_with_config(&config);
    hw_http_add_route(route, get_root, NULL);
    hw_http_open();
}

#define CRLF "\r\n"

void response_complete(void* user_data)
{
}

void get_root(http_request* request, hw_http_response* response, void* user_data)
{
    int rc = 0;
    hw_string status_code;
    hw_string body;
    body.length = 0;
    
    if (request->method == HW_HTTP_GET)
    {
        // Process GET request.
        hcat_keypair pair;
        pair.keyspace = string_ref(hw_get_header(request, "keyspace"));
        pair.key = string_ref(hw_get_header(request, "key"));
        
        if (pair.keyspace.data() != NULL)
        {
            hcat_transaction* tx;
            store->begin_transaction(&tx, 1);
            rc = tx->get(&pair);
            tx->commit();
            
            if (rc == 0)
            {
                SETSTRING(status_code, HTTP_STATUS_200);
                body.value = (char*)pair.value;
                body.length = pair.value_length;
            }
            else if (rc != 0)
            {
                SETSTRING(status_code, HTTP_STATUS_404);
                SETSTRING(body, "hello world");
            }
            delete tx;
        }
        else
        {
            SETSTRING(status_code, HTTP_STATUS_404);
            SETSTRING(body, "FAIL");
        }
    }
    else if (request->method == HW_HTTP_PUT || request->method == HW_HTTP_POST)
    {
        // Process PUT or POST request.
        hcat_keypair pair;
        pair.keyspace = string_ref(hw_get_header(request, "keyspace"));
        pair.key = string_ref(hw_get_header(request, "key"));
        string_ref val = string_ref(hw_get_header(request, "value"));
        pair.value = (void*)val.data();
        pair.value_length = val.length();
        
        if (pair.keyspace.length() != 0 && pair.key.length() != 0 && pair.value_length != 0)
        {
            transaction_mutex.lock_shared();
            rc = tx->set(&pair);
            transaction_mutex.unlock_shared();
            
            SETSTRING(status_code, HTTP_STATUS_200);
            SETSTRING(body, "OK");
        }
        else
        {
            //tx->abort();
            char* key = hw_get_header(request, "key");
            pair.key = string_ref(hw_get_header(request, "key"));
            SETSTRING(status_code, HTTP_STATUS_404);
            SETSTRING(body, "FAIL");
        }
        //delete tx;
    }
    
    hw_string content_type_name;
    hw_string content_type_value;
    hw_string keep_alive_name;
    hw_string keep_alive_value;
    
    SETSTRING(content_type_name, "Content-Type");
    
    SETSTRING(content_type_value, "text/html");
    hw_set_response_header(response, &content_type_name, &content_type_value);
    
    hw_set_response_status_code(response, &status_code);
    hw_set_body(response, &body);
    
    if (request->keep_alive)
    {
        SETSTRING(keep_alive_name, "Connection");
        
        SETSTRING(keep_alive_value, "Keep-Alive");
        hw_set_response_header(response, &keep_alive_name, &keep_alive_value);
    }
    else
    {
        hw_set_http_version(response, 1, 0);
    }
    
    hw_http_response_send(response, NULL, response_complete);
}