/*
* server.cpp
*
* Created on: 2010-2-4
* Author: xiang lei
* web: http://www.jeehe.com
*/
#include "BdbQueue.h"
#include "QueueStats.h"
#include <sys/types.h>
#include <sys/time.h>
#include <sys/queue.h>
#include <sys/types.h>
#include <ctype.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <getopt.h>
#include <unistd.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/mman.h>
#include <netinet/in.h>
#include <net/if.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <time.h>
#include <sys/ioctl.h>
#include <errno.h>
#include <assert.h>
#include <signal.h>
#include <stdbool.h>
#include <err.h>
#include <event.h>
#include <evhttp.h>
#include <iostream>
#include <sstream>
#define VERSION "1.0"
using std::cout;
using std::endl;
using std::string;
//
char *urldecode(char *input_str);
void http_bdbqueue_handler(struct evhttp_request *req, void *arg);
static void show_help();
static void kill_signal(const int sig);
BdbQueue *bdb_queue;
char* http_keep_alive = "120";//second
/**
*
*/
int main(int argc, char **argv) {
int http_port = 1985;
char* http_ip = "127.0.0.1";
char* queue_path = "/home/bdb";
char* queue_name = "bdb.db";
int cache_size = 64;// m
int page_size = 4;//k
bool http_daemon = false;
int http_time_out = 1;// second
int flag;
while ((flag = getopt(argc, argv, "p:i:f:n:c:a:t:k:dh")) != -1) {
switch (flag) {
case 'p':
http_port = atoi(optarg);
break;
case 'i':
http_ip = strdup(optarg);
break;
case 'f':
queue_path = strdup(optarg);
break;
case 'n':
queue_name = strdup(optarg);
break;
case 'c':
cache_size = atoi(optarg);
break;
case 'a':
page_size = atoi(optarg);
break;
case 't':
http_time_out = atoi(optarg);
break;
case 'k':
http_keep_alive = strdup(optarg);
break;
case 'd':
http_daemon = true;
break;
default:
show_help();
break;
}
}
if (http_daemon) {//
pid_t pid;
pid = fork();
if (pid < 0) {
exit(EXIT_FAILURE);
}
if (pid > 0) {
exit(EXIT_SUCCESS);
}
}
signal(SIGPIPE, SIG_IGN);
signal(SIGINT, kill_signal);
signal(SIGKILL, kill_signal);
signal(SIGQUIT, kill_signal);
signal(SIGTERM, kill_signal);
signal(SIGHUP, kill_signal);
bdb_queue = new BdbQueue(queue_path, queue_name, cache_size, page_size);
struct evhttp *httpd;
event_init();
httpd = evhttp_start(http_ip, http_port);
evhttp_set_timeout(httpd, http_time_out);
evhttp_set_gencb(httpd, http_bdbqueue_handler, NULL);
cout<< "bdb queue server start success;\nAuthor: xiang lei\nWeb: http://www.jeehe.com"<< endl;
event_dispatch();
evhttp_free(httpd);
return 0;
}
/**
*
*/
static void show_help() {
string msg ="./bdbqueue -d -n test.db -p 1500 -i 10.0.0.76 -f /home -c 64 -a 4 -t 1 -k 120 ";
string help="基于bdb存贮的简单fifo队列，支持多个队列，采用bdb的btree方式组织数据。\n"
"启动参数说明：\n"
"-n <db文件名> 数据库的文件名，默认为：xianglei.db\n"
"-p <端口号> http访问的端口号，默认：1985\n"
"-i <ip地址> http访问的ip地址，默认：127.0.0.1\n"
"-f <db文件目录> 数据文件存放的目录，目录权限为读写。默认：/home\n"
"-c <cache size> bdb的cache size 单位：m 默认64m\n"
"-a <page size> bdb的page zise 单位：k 默认4k\n"
"-t <http time out> http请求超时时间，单位s 默认1s\n"
"-k <http keep alive> http keep alive时间，单位s 默认120s\n"
"-h 查看帮助\n"
"-d 守护进程运行\n"
"注意：在终止程序的时候慎用pkill -9或者kill -9 PID\n"
"这样做可能导致bdb数据丢失或者其他异常。请用killall或者pkill或者kill PID\n"
"执行以上命令可能要过段时间才能终止程序，因为bdb需要把数据从缓存持久化到硬盘\n";
cout<<help<<endl;
cout << msg << endl;
exit(1);
}
/**
*
*/
char *urldecode(char *input_str) {
int len = strlen(input_str);
char *str = strdup(input_str);
char *dest = str;
char *data = str;
int value;
int c;
while (len--) {
if (*data == '+') {
*dest = ' ';
} else if (*data == '%' && len >= 2 && isxdigit((int) *(data + 1))
&& isxdigit((int) *(data + 2))) {
c = ((unsigned char *) (data + 1))[0];
if (isupper(c))
c = tolower(c);
value = (c >= '0' && c <= '9' ? c - '0' : c - 'a' + 10) * 16;
c = ((unsigned char *) (data + 1))[1];
if (isupper(c))
c = tolower(c);
value += c >= '0' && c <= '9' ? c - '0' : c - 'a' + 10;
*dest = (char) value;
data += 2;
len -= 2;
} else {
*dest = *data;
}
data++;
dest++;
}
*dest = '\0';
return str;
}
/**
*
*/
static void kill_signal(const int sig) {
bdb_queue->sync();
delete bdb_queue;
bdb_queue = NULL;
exit(0);
}
/**
*
*/
void http_bdbqueue_handler(struct evhttp_request *req, void *arg) {
struct evbuffer *buf;
buf = evbuffer_new();
char *decode_uri = strdup((char*) evhttp_request_uri(req));
struct evkeyvalq httpsqs_http_query;
evhttp_parse_query(decode_uri, &httpsqs_http_query);
free(decode_uri);
decode_uri = NULL;
const char *queue_name = evhttp_find_header(&httpsqs_http_query, "name"); /* 队列名称 */
const char *queue_opt = evhttp_find_header(&httpsqs_http_query, "opt"); /* 操作类别 */
const char *queue_data = evhttp_find_header(&httpsqs_http_query, "data"); /* 队列数据*/
evhttp_add_header(req->output_headers, "Content-Type",
"text/plain;charset=UTF-8");
evhttp_add_header(req->output_headers, "Keep-Alive", http_keep_alive);
if (queue_name != NULL && queue_opt != NULL && strlen(queue_name) <= 256) {
/* 放入队列 */
if ((strcmp(queue_opt, "put") == 0) && queue_data != NULL && strlen(
queue_data) > 0) {
int buffer_data_len = strlen(queue_data);
char *input_postbuffer;
char *buffer_data = (char *) malloc(buffer_data_len + 1);
memset(buffer_data, '\0', buffer_data_len + 1);
memcpy(buffer_data, queue_data, buffer_data_len);
input_postbuffer = urldecode(buffer_data);
string queue_name_str = queue_name;
string queue_data_str = buffer_data;
if (bdb_queue->put(queue_name_str, queue_data_str)) {
free(input_postbuffer);
free(buffer_data);
input_postbuffer = NULL;
buffer_data = NULL;
evbuffer_add_printf(buf, "%s", "PUT_OK");
} else {
free(input_postbuffer);
free(buffer_data);
input_postbuffer = NULL;
buffer_data = NULL;
evbuffer_add_printf(buf, "%s", "PUT_ERROR");
}
} else if (strcmp(queue_opt, "get") == 0) {
string queue_name_str = queue_name;
char* data = bdb_queue->get(queue_name_str);
if (data == NULL) {
evbuffer_add_printf(buf, "%s", "GET_END");
} else {
evbuffer_add_printf(buf, "%s", data);
free(data);
data = NULL;
}
} else if (strcmp(queue_opt, "delete") == 0) {
string queue_name_str = queue_name;
if (bdb_queue->deleteKey(queue_name_str)) {
evbuffer_add_printf(buf, "%s", "DELETE_OK");
} else {
evbuffer_add_printf(buf, "%s", "DELETE_ERROR");
}
} else if (strcmp(queue_opt, "stats") == 0) {
QueueStats *queueStats = new QueueStats();
bdb_queue->queueStats(queueStats);
std::stringstream outstr;
outstr << (queueStats->getCacheSize());
string cacheSizeStr = outstr.str();
std::stringstream outstrtwo;
outstrtwo << (queueStats->getPageSize());
string pageSizeStr = outstrtwo.str();
string stats = "cacheSize=" + cacheSizeStr + "M\r\n" + "pageSize="
+ pageSizeStr + "K\r\n" + "dbName="
+ (queueStats->getQueueDbName()) + "\r\n" + "queuePath="
+ (queueStats->getQueuePath());
;
delete queueStats;
queueStats = NULL;
evbuffer_add_printf(buf, "%s", stats.c_str());
} else {
/* 命令错误 */
evbuffer_add_printf(buf, "%s", "COMMAND_ERROR");
}
} else {
/* 客户端错误 */
evbuffer_add_printf(buf, "%s", "CLIENT_ERROR");
}
evhttp_send_reply(req, HTTP_OK, "OK", buf);
evhttp_clear_headers(&httpsqs_http_query);
evbuffer_free(buf);
}
