#define _FILE_OFFSET_BITS 64

#include <assert.h>

#include <string.h>

#include <stdio.h>

#include <stdlib.h>

#include <unistd.h>

#include <fcntl.h>

#include <sys/types.h>

#include <sys/mman.h>

#include <signal.h>

#include <netdb.h>

#include "crc32.h"

#include "kdb-data-common.h"

#include "net-events.h"

#include "net-buffers.h"

#include "server-functions.h"

#include "net-connections.h"

#include "net-memcache-server.h"

#include "net-memcache-client.h"

#include "net-crypto-aes.h"

#include "net-http-server.h"

#include "word-split.h"

#include "stemmer-new.h"

#include "resolver.h"

#include "net-rpc-server.h"

#include "net-rpc-client.h"

#include "net-rpc-common.h"

#include "md5.h"

#include "net-http-server.h"

#include "net-crypto-aes.h"

#include "queue-data.h"

#define MAX_VALUE_LEN (1 << 20)

#define VERSION "0.5"

#define VERSION_STR "queue "VERSION

#define TCP_PORT 11211

#define UDP_PORT 11211

int port = TCP_PORT, udp_port = UDP_PORT;

int watchcat_port = 6670, engine_id = 0, engine_n = 1000,

news_port = 6671;

long max_memory = MAX_MEMORY;

struct in_addr settings_addr;

int interactive = 0;

volatile int sigpoll_cnt;

char *hostname = "localhost";

long long cmd_get, cmd_set, cmd_stats, cmd_version;

long long rpc_failed, rpc_sent, rpc_received, rpc_received_news_subscr, rpc_received_news_redirect;

double cmd_get_time = 0.0, cmd_set_time = 0.0, load_time;

double max_cmd_get_time = 0.0, max_cmd_set_time = 0.0;

char max_cmd_get_query[3000], max_cmd_set_query[3000];

int header_size;

long long rpc_received_size, rpc_sent_size, http_sent_size, http_sent;

long long rpc_received_news_redirect_size, rpc_received_news_subscr_size;

double cmd_http_time = 0.0;

double max_cmd_http_time = 0.0;

char max_cmd_http_query[3000];

#define STATS_BUFF_SIZE (1 << 14)

int stats_buff_len;

char stats_buff[STATS_BUFF_SIZE];

int queue_prepare_stats (void);

long long http_queries_ok, http_queries_delayed, http_failed[4];

long long buff_overflow_cnt;

int pending_http_queries;

long long sent_queries;

conn_type_t ct_queue_engine_server = {

.magic = CONN_FUNC_MAGIC,

.title = "queue_engine_server",

.accept = accept_new_connections,

.init_accepted = mcs_init_accepted,

.run = server_read_write,

.reader = server_reader,

.writer = server_writer,

.parse_execute = mcs_parse_execute,

.close = server_close_connection,

.free_buffers = free_connection_buffers,

.init_outbound = server_failed,

.connected = server_failed,

.wakeup = server_failed,

.alarm = server_failed,

#ifndef NOAES

.crypto_init = aes_crypto_init,

.crypto_free = aes_crypto_free,

.crypto_encrypt_output = aes_crypto_encrypt_output,

.crypto_decrypt_input = aes_crypto_decrypt_input,

.crypto_needed_output_bytes = aes_crypto_needed_output_bytes,

};

int memcache_store (struct connection *c, int op, const char *key, int key_len, int flags, int delay, int size);

int memcache_get (struct connection *c, const char *key, int key_len);

int memcache_delete (struct connection *c, const char *key, int key_len);

int memcache_incr (struct connection *c, int op, const char *key, int key_len, long long arg);

int memcache_stats (struct connection *c);

int memcache_version (struct connection *c);

struct memcache_server_functions memcache_methods = {

.execute = mcs_execute,

.mc_store = memcache_store,

.mc_get_start = mcs_get_start,

.mc_get = memcache_get,

.mc_get_end = mcs_get_end,

.mc_incr = mcs_incr,

.mc_delete = mcs_delete,

.mc_version = memcache_version,

.mc_stats = memcache_stats,

.mc_check_perm = mcs_default_check_perm,

.mc_init_crypto = mcs_init_crypto

};

int rpcs_execute (struct connection *c, int op, int len);

struct rpc_server_functions queue_rpc_server = {

.execute = rpcs_execute,

.check_ready = server_check_ready,

.flush_packet = rpcs_flush_packet,

.rpc_check_perm = rpcs_default_check_perm,

.rpc_init_crypto = rpcs_init_crypto,

.memcache_fallback_type = &ct_queue_engine_server,

.memcache_fallback_extra = &memcache_methods

};

char buf[MAX_VALUE_LEN + 50];

#define MAX_WID_LEN 10000

long long wids[MAX_WID_LEN];

int wn;

#define MAX_SUBS_LEN 20000

int subs[MAX_SUBS_LEN];

int sn;

static inline void eat_at (const char *key, int key_len, char **new_key, int *new_len) {

if (*key == '^') {

key++;

key_len--;

}

*new_key = (char *)key;

*new_len = key_len;

if ((*key >= '0' && *key <= '9') || (*key == '-' && key[1] >= '0' && key[1] <= '9')) {

key++;

while (*key >= '0' && *key <= '9') {

key++;

}

if (*key++ == '@') {

if (*key == '^') {

key++;

}

*new_len -= (key - *new_key);

*new_key = (char *)key;

}

}

}

static inline void safe_read_in (netbuffer_t *H, char *data, int len) {

assert (read_in (H, data, len) == len);

int i;

for (i = 0; i < len; i++) {

if (data[i] == 0) {

data[i] = ' ';

}

}

}

char query[3000];

#define QRETURN(x, y) \

cmd_time += mytime() - 1e-6; \

cmd_ ## x ## _time += cmd_time; \

if (cmd_time > max_cmd_ ## x ## _time && \

now > start_time + 600) { \

strcpy (max_cmd_ ## x ## _query, query); \

max_cmd_ ## x ## _time = cmd_time; \

} \

return y;

int memcache_store (struct connection *c, int op, const char *old_key, int old_key_len, int flags, int delay, int size) {

cmd_set++;

double cmd_time = -mytime();

if (verbosity > 1) {

fprintf (stderr, "memcache_store: key='%s', key_len=%d, value_len=%d\n", old_key, old_key_len, size);

}

dl_log_add (LOG_HISTORY, 1, "memcache_store: key='%s', key_len=%d, value_len=%d\n", old_key, old_key_len, size);

sprintf (query, "key='%s', value_len=%d", old_key, size);

if (size + 1 < MAX_VALUE_LEN) {

char *key;

int key_len;

eat_at (old_key, old_key_len, &key, &key_len);

if (key_len >= 5 && !strncmp (key, "queue", 5)) {

int add = 0, x, y;

if (sscanf(key + 5, "%d,%d%n", &x, &y, &add) != 2) {

x = y = add = 0;

}

if (key[key_len - 1] != ')' || key[5 + add] != '(' || (key_len - 6 >= STATS_BUFF_SIZE)) {

QRETURN(set, -2);

}

safe_read_in (&c->In, buf, size);

buf[size] = 0;

int len = key_len - 7 - add;

memcpy (stats_buff, key + 6 + add, len);

stats_buff[len] = 0;

if (!do_add_event (stats_buff, len, buf, size, x, y, TTL_EVENT)) {

QRETURN(set, 0)

}

QRETURN(set, 1);

}

if (!strcmp (key, "entry")) {

safe_read_in (&c->In, buf, size);

buf[size] = 0;

char *s = buf;

wn = 0;

while (*s) {

long long x = 0;

int f = 0;

if (*s == '-') {

f = 1;

s++;

}

while (*s >= '0' && *s <= '9') {

x = x * 10 + *s++ - '0';

}

if (f) {

x = -x;

}

if (wn < MAX_WID_LEN) {

wids[wn++] = x;

}

if (*s != ',') {

break;

}

s++;

}

if (*s != ';') {

QRETURN(set, 0);

}

s++;

add_event_to_watchcats (wids, wn, s);

QRETURN(set, 1);

}

if (key_len >= 10 && !strncmp (key, "news_local", 10)) {

ll id;

int n, x, y, ttl;

if (sscanf (key + 10, "%lld,%d,%d,%d,%d", &id, &x, &y, &ttl, &n) == 5) {

read_in (&c->In, buf, size); // no safe_read_in

buf[size] = 0;

if (size >= (int)sizeof (pli) * n) {

add_event_to_news (id, x, y, ttl, (pli *)buf, n, buf + sizeof (pli) * n, 0);

QRETURN(set, 1);

}

QRETURN(set, 0);

}

QRETURN(set, -2);

}

if (key_len >= 4 && !strncmp (key, "news", 4)) {

ll id;

int x, y;

if (sscanf (key + 4, "%lld,%d,%d", &id, &x, &y) == 3) {

safe_read_in (&c->In, buf, size);

buf[size] = 0;

dbg ("NEWS %lld %d %d: [%s]\n", id, x, y, buf);

redirect_news (id, x, y, TTL_NEWS_EVENT, buf, size);

QRETURN(set, 1);

}

QRETURN(set, -2);

}

int tp = -1;

if (key_len >= 15 && !strncmp (key, "add_news_subscr", 15)) {

tp = 0;

} else if (key_len >= 15 && !strncmp (key, "del_news_subscr", 15)) {

tp = 1;

} else if (key_len >= 10 && !strncmp (key, "subscr_add", 10)) {

tp = 3;

} else if (key_len >= 10 && !strncmp (key, "subscr_del", 10)) {

tp = 4;

} else if (key_len >= 6 && !strncmp (key, "subscr", 6)) {

tp = 2;

}

static int _len[] = {15, 15, 6, 10, 10};

if (tp != -1) {

read_in (&c->In, buf, size); // no safe_read_in

buf[size] = 0;

char *s = key + _len[tp];

ll id;

int is_rev = 0;

if (*s == '!') {

is_rev = 1;

s++;

}

if (sscanf (s, "%lld", &id) == 1) {

int res = 0;

if (tp >= 2) {

dbg ("%lld[%lld] %% %d == %d?\n", id, dl_abs(id), engine_n, engine_id);

if (dl_abs (id) % engine_n == engine_id) {

dbg ("OK [%s]\n", buf);

sn = 0;

int i, mul = 1;

ll x = 0;

for (i = 0; i <= size - !size && sn + 2 < MAX_SUBS_LEN; i++) {

if (buf[i] == ',' || buf[i] == 0) {

if (sn % 3 == 0 || tp == 4) {

*(ll *)&subs[sn] = x * mul;

sn += 2;

} else {

subs[sn++] = x * mul;

}

x = 0;

mul = 1;

} else if (buf[i] == '-') {

mul = -1;

} else {

x = x * 10 + buf[i] - '0';

if (buf[i] < '0' || buf[i] > '9') {

QRETURN(set, 0);

}

}

}

if (tp != 4) {

if (sn % 3 != 0) {

dbg ("FAIL sn = %d\n", sn);

QRETURN(set, 0);

}

sn /= 3;

} else {

if (sn % 2 != 0) {

dbg ("FAIL sn = %d\n", sn);

QRETURN(set, 0);

}

sn /= 2;

}

if (tp == 2) {

res = set_news_subscr (id, (pli *)subs, sn);

} else if (tp == 3) {

res = set_news_subscr_add (id, (pli *)subs, sn);

} else if (tp == 4) {

res = set_news_subscr_del (id, (ll *)subs, sn);

} else {

assert (0);

}

QRETURN(set, res);

}

QRETURN(set, 0);

}

if (size % sizeof (pli) != 0) {

QRETURN(set, 0);

}

int wn = size / sizeof (pli);

if (wn >= 1) {

if (tp == 0) {

if (is_rev) {

res = subscribers_add_new_rev (id, (pli *)buf, wn);

} else {

res = subscribers_add_new (id, (pli *)buf, wn);

}

} else if (tp == 1) {

if (is_rev) {

res = subscribers_del_rev (id, (pli *)buf, wn);

} else {

res = subscribers_del_old (id, (pli *)buf, wn);

}

}

QRETURN(set, res);

}

}

QRETURN(set, 0);

}

}

QRETURN(set, -2);

}

int memcache_get (struct connection *c, const char *old_key, int old_key_len) {

if (verbosity > 1) {

fprintf (stderr, "memcache_get: key='%s', key_len=%d\n", old_key, old_key_len);

}

dl_log_add (LOG_HISTORY, 1, "memcache_get: key='%s', key_len=%d\n", old_key, old_key_len);

sprintf (query, "key='%s', key_len=%d", old_key, old_key_len);

char *key;

int key_len;

eat_at (old_key, old_key_len, &key, &key_len);

if (key_len >= 5 && !strncmp (key, "stats", 5)) {

int len = queue_prepare_stats();

int len2 = prepare_stats (c, stats_buff + len, STATS_BUFF_SIZE - len);

return_one_key (c, old_key, stats_buff, len + len2 - 1);

return 0;

}

GET_LOG;

SET_LOG_VERBOSITY;

double cmd_time = -mytime();

cmd_get++;

if (key_len >= 13 && !strncmp (key, "timestamp_key", 13)) {

int id, timeout, st;

long long ip;

if (sscanf (key, "timestamp_key%d,%lld,%d%n", &id, &ip, &timeout, &st) == 3 && key[st] == '(' && key[key_len - 1] == ')' && key_len - st - 2 < STATS_BUFF_SIZE) {

char *s = stats_buff;

int len = key_len - st - 2;

memcpy (s, key + st + 1, len);

s[len] = 0;

dl_log_add (LOG_HISTORY, 1, "GET KEY (%s) id = %d, ip = %lld\n", s, id, ip);

s = get_timestamp_key (s, id, ip, timeout, NULL, Q_DEF);

return_one_key (c, old_key, s, strlen (s));

}

QRETURN(get, 0);

}

if (key_len >= 5 && !strncmp (key, "alias", 5)) {

int st = 5, is_news = 0;

if (key_len >= 10 && !strncmp (key + 5, "_news", 5)) {

st += 5;

is_news = 1;

}

if (key[st] == '(' && key[key_len - 1] == ')' && key_len - st - 2 < STATS_BUFF_SIZE) {

char *s = stats_buff;

int len = key_len - st - 2;

memcpy (s, key + st + 1, len);

s[len] = 0;

dl_log_add (LOG_HISTORY, 1, "GET ALIAS (%s)\n", s);

ll res;

if (is_news) {

ll id;

if (sscanf (s, "%lld", &id) != 1 || !get_queue_news_alias (id, &res)) {//TODO get alias_news(1qwerty)

QRETURN(get, 0);

}

} else {

if (!is_news && !get_queue_alias (s, &res)) {

QRETURN(get, 0);

}

}

sprintf (s, "%lld", res);

return_one_key (c, old_key, s, strlen (s));

}

QRETURN(get, 0);

}

if (key_len >= 14 && !strncmp (key, "qname_by_alias", 14)) {

ll id;

int st;

if (sscanf (key, "qname_by_alias%lld%n", &id, &st) == 1 && key[st] == 0) {

queue *q = get_queue_by_alias (id);

if (q != NULL) {

sprintf (buf, "%s : ref_cnt = %d, ev_first = %p, keys_cnt = %d, subscr_cnt = %d", q->name, q->ref_cnt, q->ev_first, q->keys_cnt, q->subscr_cnt);

return_one_key (c, old_key, buf, strlen (buf));

}

}

QRETURN(get, 0);

}

if (key_len >= 10 && !strncmp (key, "queue_info", 10)) {

if (key[10] == '(' && key[key_len - 1] == ')') {

key[key_len - 1] = 0;

char *qname = key + 11;

queue *q = get_queue (qname, 0);

if (q != NULL) {

sprintf (buf,

"name\t%s\n"

"ref_cnt\t%d\n"

"key_cnt\t%d\n"

"subscr_cnt\t%d\n"

"ev_frist\t%p\n"

, q->name, q->ref_cnt, q->keys_cnt, q->subscr_cnt, q->ev_first);

return_one_key (c, old_key, buf, strlen (buf));

}

}

QRETURN (get, 0);

}

if (key_len >= 12 && !strncmp (key, "watchcat_key", 12)) {

int id, timeout, st;

long long ip;

if (sscanf (key, "watchcat_key%d,%lld,%d%n", &id, &ip, &timeout, &st) == 3 && key[st] == '(' && key[key_len - 1] == ')' && key_len - st - 2 < STATS_BUFF_SIZE) {

char *s = stats_buff;

int len = key_len - st - 2;

memcpy (s, key + st + 1, len);

s[len] = 0;

char *tmp = s;

while (*tmp) {

if (*tmp == '?') {

*tmp = 0x1f;

}

tmp++;

}

s = get_watchcat_key (s, id, ip, timeout);

if (s != NULL) {

return_one_key (c, old_key, s, strlen (s));

}

}

QRETURN(get, 0);

}

if (key_len >= 8 && !strncmp (key, "news_key", 8)) {

int id, timeout;

long long ip;

ll uid;

if (sscanf (key, "news_key%d,%lld,%d(%lld)", &id, &ip, &timeout, &uid) == 4) {

char *s;

dl_log_add (LOG_HISTORY, 1, "GET NEWS KEY (%lld) id = %d, ip = %lld\n", uid, id, ip);

s = get_news_key (id, ip, timeout, uid);

if (s != NULL) {

return_one_key (c, old_key, s, strlen (s));

}

}

QRETURN(get, 0);

}

if (key_len >= 5 && !strncmp (key, "queue", 5)) {

int id, ts, st;

long long ip;

if (sscanf (key, "queue%d,%lld,%d%n", &id, &ip, &ts, &st) == 3 && key[st] == '(' && key[key_len - 1] == ')' && key_len - st - 2 < STATS_BUFF_SIZE) {

char *s = stats_buff, *ans = "{\"failed\":2}";

int len = key_len - st - 2;

if (len == KEY_LEN) {

memcpy (s, key + st + 1, len);

s[len] = 0;

qkey *k;

//fprintf (stderr, "mc validate : %s %d %lld %d\n", s, id, ip, ts);

char err;

if ((k = validate_key (s, id, ip, ts, 0, &err)) != NULL) {

ans = get_events_http (k);

}

}

return_one_key (c, old_key, ans, strlen (ans));

}

QRETURN(get, 0);

}

if (key_len >= 10 && !strncmp (key, "upd_secret", 10)) {

int id;

if (sscanf (key, "upd_secret%d", &id) == 1) {

upd_secret (id);

}

return_one_key (c, old_key, "OK", 2);

QRETURN(get, 0);

}

QRETURN(get, 0);

}

int memcache_version (struct connection *c) {

cmd_version++;

write_out (&c->Out, "VERSION "VERSION"\r\n", 9 + sizeof (VERSION));

return 0;

}

int queue_prepare_stats (void) {

cmd_stats++;

char *s = stats_buff;

#define W(args...) WRITE_ALL (s, ## args)

W ("heap_used\t%ld\n", (long) (dyn_cur - dyn_first));

W ("heap_max\t%ld\n", (long) (dyn_last - dyn_first));

W ("load_time\t%lf\n", load_time);

W ("pid\t%d\n", getpid());

W ("version\t%s\n", VERSION);

W ("pointer_size\t%d\n", (int)(sizeof (void *) * 8));

W ("current_memory_used\t%lld\n", dl_get_memory_used());

W ("cmd_get\t%lld\n", cmd_get);

W ("cmd_get_time\t%.7lf\n", cmd_get_time);

W ("max_cmd_get_time\t%.7lf\n", max_cmd_get_time);

W ("max_cmd_get_query\t%s\n", max_cmd_get_query);

W ("cmd_set\t%lld\n", cmd_set);

W ("cmd_set_time\t%.7lf\n", cmd_set_time);

W ("max_cmd_set_time\t%.7lf\n", max_cmd_set_time);

W ("max_cmd_set_query\t%s\n", max_cmd_set_query);

W ("cmd_http_time\t%.7lf\n", cmd_http_time);

W ("max_cmd_http_time\t%.7lf\n", max_cmd_http_time);

W ("max_cmd_http_query\t%s\n", max_cmd_http_query);

W ("cmd_stats\t%lld\n", cmd_stats);

W ("cmd_version\t%lld\n", cmd_version);

W ("limit_max_dynamic_memory\t%ld\n", max_memory);

W ("events_memory\t%ld\n", events_memory);

W ("events_total\t%d\n", events_cnt);

W ("events_created\t%lld\n", events_created);

W ("events_sent\t%lld\n", events_sent);

W ("keys_memory\t%ld\n", keys_memory);

W ("keys_count\t%d\n", keys_cnt);

W ("queues_memory\t%ld\n", queues_memory);

W ("queues_count\t%d\n", queues_cnt);

W ("subscribers_memory\t%ld\n", subscribers_memory);

W ("subscribers_count\t%ld\n", subscribers_cnt);

W ("treap_memory\t%ld\n", (long)treap_get_memory());

W ("treap_cnt\t%d\n", treap_cnt);

W ("treap_cnt_rev\t%d\n", treap_cnt_rev);

W ("str_memory\t%ld\n", str_memory);

W ("htbl_memory\t%ld\n", get_htbls_memory());

W ("extra_memory\t%lld\n", dl_get_memory_used() - events_memory - keys_memory - subscribers_memory

- treap_get_memory() - str_memory - get_htbls_memory()

- crypto_memory - queues_memory - time_keys_memory);

W ("sent_queries\t%lld\n", sent_queries);

STAT_OUT (send_create_watchcat);

STAT_OUT (send_changes_cnt);

STAT_OUT (process_changes_cnt);

STAT_OUT (changes_len_max);

STAT_OUT (process_changes_total_len);

STAT_OUT (changes_add_rev_cnt);

STAT_OUT (changes_add_rev_len);

STAT_OUT (changes_add_cnt);

STAT_OUT (changes_add_len);

STAT_OUT (changes_del_rev_cnt);

STAT_OUT (changes_del_rev_len);

STAT_OUT (changes_del_cnt);

STAT_OUT (changes_del_len);

STAT_OUT (send_news_cnt);

STAT_OUT (redirect_news_twice_cnt);

STAT_OUT (redirect_news_cnt);

STAT_OUT (redirect_news_len);

STAT_OUT (to_add_overflow);

STAT_OUT (to_del_overflow);

STAT_OUT (buff_overflow_cnt);

W ("engine_id\t%d\n", engine_id);

W ("engine_n\t%d\n", engine_n);

STAT_OUT (news_subscr_cnt);

STAT_OUT (news_subscr_missed);

W ("news_subscr_cache_missed\t%.6lf\n", (double)news_subscr_missed / news_subscr_cnt);

STAT_OUT (news_subscr_len);

STAT_OUT (news_subscr_actual_len);

W ("news_subscr_cache_missed\t%.6lf\n", (double)news_subscr_actual_len / news_subscr_len);

STAT_OUT (rpc_received_news_subscr);

STAT_OUT (rpc_received_news_subscr_size);

W ("rpc_received_news_subcr_avg\t%.2lf\n", (double)rpc_received_news_subscr_size / rpc_received_news_subscr);

STAT_OUT (rpc_received_news_redirect);

STAT_OUT (rpc_received_news_redirect_size);

W ("rpc_received_news_redirect_avg\t%.2lf\n", (double)rpc_received_news_redirect_size / rpc_received_news_redirect);

STAT_OUT (rpc_failed);

STAT_OUT (rpc_sent);

STAT_OUT (rpc_sent_size);

W ("rpc_sent_avg\t%.2lf\n", (double)rpc_sent_size / rpc_sent);

STAT_OUT (rpc_received);

STAT_OUT (rpc_received_size);

W ("rpc_received_avg\t%.2lf\n", (double)rpc_received_size / rpc_received);

STAT_OUT (http_queries);

STAT_OUT (http_queries_size);

W ("http_queries_avg\t%lf\n", (double)http_queries_size / http_queries);

STAT_OUT (http_sent);

STAT_OUT (http_sent_size);

W ("http_sent_avg\t%lf\n", (double)http_sent_size / http_sent);

W ("http_connections\t%d\n", http_connections);

W ("pending_http_queries\t%d\n", pending_http_queries);

W ("http_queries_ok\t%lld\n", http_queries_ok);

W ("http_queries_delayed\t%lld\n", http_queries_delayed);

W ("http_bad_headers\t%lld\n", http_bad_headers);

W ("http_success\t%lld\n", http_failed[0]);

W ("http_failed1\t%lld\n", http_failed[1]);

W ("http_failed2\t%lld\n", http_failed[2]);

W ("http_failed3\t%lld\n", http_failed[3]);

W ("http_qps\t%.6f\n", now > start_time ? http_queries * 1.0 / (now - start_time) : 0);

W ("max_delay\t%.6f\n", max_delay);

W ("mean_delay\t%.6f\n", cnt_delay ? sum_delay / cnt_delay : 0);

W ("version\t" VERSION_STR " compiled at " __DATE__ " " __TIME__ " by gcc " __VERSION__ "\n");

#undef W

int len = s - stats_buff;

assert (len < STATS_BUFF_SIZE);

return len;

}

int memcache_stats (struct connection *c) {

int len = queue_prepare_stats();

int len2 = prepare_stats (c, stats_buff + len, STATS_BUFF_SIZE - len);

write_out (&c->Out, stats_buff, len + len2);

write_out (&c->Out, "END\r\n", 5);

return 0;

}

int hts_wakeup (struct connection *c);

int hts_close (struct connection *c, int who);

int delete_history_query (struct conn_query *q);

struct conn_query_functions history_cq_func = {

.magic = CQUERY_FUNC_MAGIC,

.title = "queue-events-query",

.wakeup = delete_history_query, // history_query_timeout

.close = delete_history_query,

.complete = delete_history_query

};

int create_history_query (queue *U, struct connection *c, double timeout,

struct conn_query **rq, char *kname) {

struct conn_query *q = zmalloc (sizeof (struct conn_query));

if (verbosity > 1) {

fprintf (stderr, "OPPA\n");

fprintf (stderr, "create_history_query (%p[%s], key = %p[%s], %p[%d]): q=%p\n", U, U->name,

((qkey *)HTS_DATA (c)->extra)->name, ((qkey *)HTS_DATA (c)->extra)->name,

c, c->fd, q);

}

q->custom_type = 0;

q->outbound = (struct connection *) U;

q->requester = c;

q->start_time = mytime();

str_memory += strlen (kname) + 1;

q->extra = dl_strdup (kname);

q->cq_type = &history_cq_func;

q->timer.wakeup_time = (timeout > 0 ? q->start_time + timeout : 0);

pending_http_queries++;

insert_conn_query (q);

if (verbosity > 1) {

fprintf (stderr, "after insert_conn_query()\n");

}

*rq = q;

return 1;

}

int create_history_query_group (qkey_group *g, struct connection *c, double timeout) {

int i;

if (verbosity > 2) {

fprintf (stderr, "create_history_query_group (size = %d) (timeout = %lf)\n", g->n, timeout);

}

for (i = 0; i < g->n; i++) {

queue *q = g->k[i]->q;

assert (q != NULL);

HTS_DATA (c)->extra = g->k[i];

create_history_query (q, c, timeout, &g->k[i]->conn, g->k[i]->name);

g->k[i]->lock++;

g->k[i]->subscribed = get_utime (CLOCK_MONOTONIC);

}

c->pending_queries = 1;

return 1;

}

int delete_history_query (struct conn_query *q) {

if (verbosity > 1) {

fprintf (stderr, "delete_history_query (%p,%p)\n", q, q->requester);

}

char *kname = (char *)q->extra;

qkey_clear_conn (kname);

str_memory -= strlen (kname) + 1;

dl_strfree (kname);

pending_http_queries--;

struct connection *c = q->requester;

int req_generation = q->req_generation;

delete_conn_query (q);

if (c->generation == req_generation) {

c->generation = ++conn_generation;

c->pending_queries = 0;

}

zfree (q, sizeof (*q));

return 0;

}

struct http_server_functions http_methods = {

.execute = hts_execute,

.ht_wakeup = hts_wakeup,

.ht_alarm = hts_wakeup,

.ht_close = hts_close

};

static char *qPost, *qGet, *qUri, *qHeaders;

static int qPostLen, qGetLen, qUriLen, qHeadersLen;

int getArgFrom (char *buffer, int b_len, const char *arg_name, char *where, int where_len) {

char *where_end = where + where_len;

int arg_len = strlen (arg_name);

while (where < where_end) {

char *start = where;

while (where < where_end && (*where != '=' && *where != '&')) {

++where;

}

if (where == where_end) {

buffer[0] = 0;

return -1;

}

if (*where == '=') {

if (arg_len == where - start && !memcmp (arg_name, start, arg_len)) {

start = ++where;

while (where < where_end && *where != '&') {

++where;

}

b_len--;

if (where - start < b_len) {

b_len = where - start;

}

memcpy (buffer, start, b_len);

buffer[b_len] = 0;

return b_len;

}

++where;

}

while (where < where_end && *where != '&') {

++where;

}

if (where < where_end) {

++where;

}

}

buffer[0] = 0;

return -1;

}

int getArg (char *buffer, int b_len, const char *arg_name) {

int res = getArgFrom (buffer, b_len, arg_name, qGet, qGetLen);

if (res < 0) {

res = getArgFrom (buffer, b_len, arg_name, qPost, qPostLen);

}

return res;

}

#define KEY_MAX_CNT 100

static char key_buff[KEY_LEN * KEY_MAX_CNT + 1], ip_buff[64 + 1];

static char no_cache_headers[] =

"Pragma: no-cache\r\n"

"Cache-Control: no-store\r\n";

void http_return (struct connection *c, const char *str) {

assert (str != NULL);

int len = strlen (str);

http_sent++;

http_sent_size += len;

write_basic_http_header (c, 200, 0, len, no_cache_headers, "text/javascript; charset=UTF-8");

write_out (&c->Out, str, len);

}

int http_return_history (struct connection *c, const char *ans) {

if (ans == NULL) {

fprintf (stderr, "buffer overflow\n");

return -500;

}

http_return (c, ans);

return 0;

}

#define MAX_POST_SIZE 4096

inline int conv_hex_digit (int c) {

if ('0' <= c && c <= '9') {

return c - '0';

}

return (c | 0x20) - 'a' + 10;

}