HttpStorage.Tc

#include <math.h>
#include "logging.h"
#include "arpc.h"
#include "tame_aio.h"
#include "libconfig.h++"
#include "HttpStorage.h"
#include "tame.h"
#include "tame_io.h"
#include "async.h"


using namespace std;
using namespace tame;

const unsigned int MAX_BUF = 4096;

HttpStorage::HttpStorage(log4cpp::Appender *app, int lighttpd_port)
{
	LOG.setAdditivity(false);
	LOG.setAppender(app);
	port = lighttpd_port;
}

HttpStorage::~HttpStorage()
{
}

tamed void HttpStorage::getFD(cbi ret_int) {	

	tvars {
		int fd;
	}
	
	if (conn_pool.empty()) {
		twait { tcpconnect ("127.0.0.1", port, mkevent(fd)); }
		LOG_DEBUG << "fd: " << fd << "\n";
		TRIGGER(ret_int, fd);
	} else {
		fd = conn_pool.front();
		conn_pool.pop();
		LOG_DEBUG << "fd: " << fd << "\n";
		TRIGGER(ret_int, fd);
	}
	
}

static blob make_blob(const char * str) {
	blob to_ret;
        bzero(&to_ret, sizeof(blob));
	to_ret.setsize(strlen(str));
	for(int i=0; i<strlen(str); i++) {
		to_ret[i] = str[i];
	}
	return to_ret;
}

static int xtoi(char *str) {
	
	LOG_INFO << "in xtoi";
	int return_val = 0;
	int length = strlen(str);
	for (int i = 0; i < length; i++) {
		int letter = str[length - i - 1];
		if (letter <= '9') {
			letter -= '0';
		} else {
			letter -='a';
			letter += 10;
		}
		return_val += letter * (int) pow(16, i);
	}
	return return_val;
	
}

tamed void HttpStorage::get(ID_Value key, cb_blob ret_blob) {
	tvars {
		int fd;
		strbuf data;
		int rc;
		strbuf resp;
		int line_len;
		char buff[MAX_BUF];
		strbuf output;
		char *s_out;
		str str_out;
		char *crlf;
		char *crlf2;
		char *content;
		char *httpHeader;
		blob value;
		int prev_size;
		int count;
		int value_size;
	}
	
	LOG_INFO << "in http get";
	
	prev_size = 0;
	count = 0;

	//attempt to connect to the server over and over again until we succeed
	//the only reason we might not is that the file descriptor is expired
	while (true) {
		//get a new file descriptor
		twait { getFD (mkevent(fd)); }
		if (fd >= 0) {
			LOG_DEBUG << "Connection succeeded!\n";
		} else {
			LOG_DEBUG << "Connection failed!\n";
			continue;
		}
		
		//construct the GET request
		data << "GET /";
		data << key.toString().c_str();
		data << " HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n";
		LOG_DEBUG << "size of write: " << data.tosuio()->resid() << "\n";
		
		//write/flush the GET request
		while(data.tosuio()->resid()) {
			twait { tame::waitwrite(fd, mkevent()); }
			if (data.tosuio()->output(fd) < 0) {
				LOG_DEBUG << "error writing to socket";
				twait { getFD (mkevent(fd)); }
			}
		}
		
		//read the response until we have all of it
		while(true) {
			twait { tame::waitread (fd, mkevent()); }
			if( (rc = resp.tosuio()->input(fd)) < 0 && errno != EAGAIN) {
				LOG_DEBUG << "error reading from socket"; //this happens if file descriptor has expired
				break;
			}
			line_len = resp.tosuio()->linelen();
			while(line_len > 0) {
				if(line_len < MAX_BUF) {
					resp.tosuio()->copyout(buff, line_len);
					buff[line_len] = '\0';
					output << buff;
				} else {
					LOG_FATAL << "exceeded buffer\n";
				}
				resp.tosuio()->rembytes(line_len);
				line_len = resp.tosuio()->linelen();
			}
			str_out = str(output);
			s_out = (char *) str_out.cstr();
			if ((prev_size == strlen(s_out)) || (strlen(s_out) <= 0)) {
				count++;
			} else {
				count = 0;
				prev_size = strlen(s_out);
			}
			if (count > 10) {
				LOG_DEBUG << "preemptive return\n";
				break;
			}
			httpHeader = strstr(s_out, "HTTP/1.1");
			if (httpHeader == NULL) {
				continue;
			}
			crlf = strstr(httpHeader, "\r\n\r\n");
			if (crlf == NULL) {
 				continue;
			}
			crlf2 = strstr(crlf + 4, "\r\n");
			if (crlf2 == NULL) {
                                continue;
			}
			content = crlf2 + 2;
			crlf += 4;
			*crlf2 = '\0';
			value_size = xtoi(crlf);
			if (strlen(content) >= value_size) {
				content[value_size] = '\0';
				value = make_blob(str(content));
				conn_pool.push(fd);
				TRIGGER(ret_blob, New refcounted<blob>(value));
				return;
			}
		}
	}

}

tamed void HttpStorage::set(ID_Value key, const blob* data, cbb ret_blob) {
	tvars {
		int fd;
		strbuf request;
		int rc;
		strbuf resp;
		int line_len;
		char buff[MAX_BUF];
		strbuf output;
		const char *s_out;
		str str_out;
		char *crlf;
		const char *start;
		char *end;
	}
	
	LOG_INFO << "in http set";
	
	while (true) {
		twait { getFD (mkevent(fd)); }
		if (fd >= 0) {
			LOG_WARN << "Connection succeeded!\n";
		} else {
			LOG_FATAL << "Connection failed!\n";
		}
		
		start = data->base();
		end = (char *) (data->base() + data->size());
		*end = '\0';
	
		request << "PUT /";
		request << key.toString().c_str();
		request << " HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-length: ";
		request << data->size();
		request << "\r\n\r\n";
		request << start;
		while(request.tosuio()->resid()) {
			twait { tame::waitwrite(fd, mkevent()); }
			if(request.tosuio()->output(fd) < 0) {
				LOG_DEBUG << "error writing to socket\n";
				twait { getFD (mkevent(fd)); }
			}
		}
		
		while(true) {
			twait { tame::waitread (fd, mkevent()); }
			if( (rc = resp.tosuio()->input(fd)) < 0 && errno != EAGAIN) {
				LOG_DEBUG << "error reading from socket\n";
				break;
			}
			line_len = resp.tosuio()->linelen();
			while(line_len > 0) {
				if(line_len < MAX_BUF) {
					resp.tosuio()->copyout(buff, line_len);
					buff[line_len] = '\0';
					output << buff;
				} else {
					LOG_FATAL << "exceeded buffer\n";
				}
				resp.tosuio()->rembytes(line_len);
				line_len = resp.tosuio()->linelen();
			}
			str_out = str(output);
			s_out = (char *) str_out.cstr();
			crlf = (char *) strstr(s_out, "\r\n\r\n");
			if (crlf == NULL && strlen(s_out) > 0) {
				cout << "crlf == NULL\n";
				cout << s_out << "\n";
				continue;
			}
			conn_pool.push(fd);
			TRIGGER(ret_blob, strstr(s_out, "HTTP/1.1 200 OK") != NULL);
			return;
			
		}
	}
}

tamed void HttpStorage::add(ID_Value key, const blob* data, cbb ret_blob) {
	tvars {
		
		ptr<blob> get_result;
		bool set_result;
		
	}
	
	twait { get(key, mkevent(get_result)); }
	
	if (get_result == NULL) {
		twait { set(key, data, mkevent(set_result)); }
		TRIGGER(ret_blob, set_result);
	} else {
		TRIGGER(ret_blob, false);
	}
}

tamed void HttpStorage::replace(ID_Value key, const blob* data, cbb ret_blob) {
	tvars {
			
		ptr<blob> get_result;
		bool set_result;
		
	}
	
	twait { get(key, mkevent(get_result)); }
	
	if (get_result == NULL) {
		TRIGGER(ret_blob, false);
	} else {
		twait { set(key, data, mkevent(set_result)); }
		TRIGGER(ret_blob, set_result);
	}
}


tamed void HttpStorage::del(ID_Value key, cbb ret_bool) {
	tvars {
		int fd;
		strbuf request;
		int rc;
		strbuf resp;
		int line_len;
		char buff[MAX_BUF];
		strbuf output;
		const char *s_out;
		char *crlf;
	}
	
	while (true) {
		twait { getFD (mkevent(fd)); }
		if (fd >= 0) {
			LOG_WARN << "Connection succeeded!\n";
		} else {
			LOG_FATAL << "Connection failed!\n";
		}
	
		request << "DELETE /";
		request << key.toString().c_str();
		request << " HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n";
		while(request.tosuio()->resid()) {
			twait { tame::waitwrite(fd, mkevent()); }
			if(request.tosuio()->output(fd) < 0) {
				LOG_DEBUG << "error writing to socket\n";
				twait { getFD (mkevent(fd)); }
			}
		}
		
		while(true) {
			twait { tame::waitread (fd, mkevent()); }
			if( (rc = resp.tosuio()->input(fd)) < 0 && errno != EAGAIN) {
				LOG_DEBUG << "error reading from socket\n";
				break;
			}
			line_len = resp.tosuio()->linelen();
			while(line_len > 0) {
				if(line_len < MAX_BUF) {
					resp.tosuio()->copyout(buff, line_len);
					buff[line_len] = '\0';
					output << buff;
				} else {
					LOG_FATAL << "exceeded buffer\n";
				}
				resp.tosuio()->rembytes(line_len);
				line_len = resp.tosuio()->linelen();
			}
			s_out = str(output).cstr();
			crlf = (char *) strstr(s_out, "\r\n\r\n");
			if (crlf == NULL && strlen(s_out) > 0) {
				continue;
			}
			conn_pool.push(fd);
			TRIGGER(ret_bool, strstr(s_out, "HTTP/1.1 200 OK") != NULL);
			return;
		}
	}
}