/*
 * Author: Catalin(ux) M BOIE <catab at embedromix.ro>
 * Date: 2004-2008
 * Description:	Some functions to help writing network servers and clients,
 *		both ipv4 and ipv6.
 * Licence: LGPL
 */

#include "Conn_engine_core.h"


/* Visible variables */
void		(*Conn_accept_cb)(struct Conn *C) = NULL;
void		(*Conn_recv_cb)(struct Conn *C) = NULL;
void		(*Conn_send_cb)(struct Conn *C) = NULL;
void		(*Conn_data_cb)(struct Conn *C) = NULL;
void		(*Conn_close_cb)(struct Conn *C) = NULL;
void		(*Conn_trigger_cb)(struct Conn *C) = NULL;
void		(*Conn_error_cb)(struct Conn *C) = NULL;
void		(*Conn_connected_cb)(struct Conn *C) = NULL;
void		(*Conn_accept_error_cb)(struct Conn *C) = NULL;

char		*(*Conn_status_slot_html_cb)(const struct Conn *C);
char		*(*Conn_status_cb)(void);


unsigned int		Conn_max_reached = 0;
unsigned int		Conn_default_ibuf = 128;
unsigned int		Conn_default_obuf = 128;
unsigned int		Conn_max_ibuf = 4096000;
unsigned int		Conn_max_obuf = 4096000;

/* Max bytes enqueued on one send/recv call */
unsigned int		Conn_max_send = 32 * 1024;
unsigned int		Conn_max_recv = 32 * 1024;

unsigned int		Conn_no = 0;
unsigned int		Conn_work_to_do = 0;
unsigned int		Conn_max = 0;
unsigned long		Conn_total = 0;
unsigned int		Conn_start = 0;
unsigned int		Conn_pending = 0;
struct timeval		Conn_now;
unsigned short		Conn_level = 0;			/* debug level */
unsigned int		Conn_accept_is_allowed;
unsigned int		Conn_accept_is_allowed_last;
/* memory stuff */
unsigned long long	Conn_mem_buffers_in = 0;
unsigned long long	Conn_mem_buffers_out = 0;
unsigned long long	Conn_mem_structs = 0;

struct Conn		*Conns = NULL;
unsigned int		Conn_inited = 0;
unsigned int		Conn_allocated = 0;
unsigned long long	Conn_id = 1;
unsigned int		Conn_must_stop = 0;

char			Conn_error[512];

FILE			*Conn_Log = NULL;
int			debug_band = 11;

/* Flags */
unsigned int		CONN_POLLIN;
unsigned int		CONN_POLLOUT;
unsigned int		CONN_POLLPRI;
unsigned int		CONN_POLLERR;
unsigned int		CONN_POLLHUP;
unsigned int		CONN_POLLNVAL;
unsigned int		CONN_POLLRDNORM;
unsigned int		CONN_POLLRDBAND;

/* queues */
struct Conn_queue	Conn_queue_free;


/* Functions */

char *Conn_strerror(void)
{
	return Conn_error;
}

/*
 * Difference between two timeval strutures, in milliseconds
 */
long long Conn_time_diff(const struct timeval *t1, const struct timeval *t2)
{
	return (t1->tv_sec - t2->tv_sec) * 1000
		+ (t1->tv_usec - t2->tv_usec) / 1000;
}

/*
 * Returns string representation of errno code
 */
char *Conn_errno(const struct Conn *C)
{
	static char buf[256];
	char *is;
	unsigned int slot;

	slot = C->slot;

	switch (Conns[slot].error_state) {
		case CONN_ERROR_USERREQ:	is = "user"; break;
		case CONN_ERROR_POLL:		is = "poll"; break;
		case CONN_ERROR_RECV:		is = "recv"; break;
		case CONN_ERROR_SEND:		is = "send"; break;
		case CONN_ERROR_SOCKET:		is = "socket"; break;
		case CONN_ERROR_HANGUP:		is = "hangup"; break;
		case CONN_ERROR_GETADDRINFO:	is = "lookup error"; break;
		case CONN_ERROR_EXPIRED:	is = "expired"; break;
		case CONN_ERROR_ACCEPT:		is = "accept"; break;
		case CONN_ERROR_MEM:		is = "allocation failed"; break;
		case CONN_ERROR_CONNECT:	is = "connect"; break;
		case CONN_ERROR_READ_TIMEOUT:	is = "read timeout"; break;
		case CONN_ERROR_CONN_TIMEOUT:	is = "conn timeout"; break;
		case CONN_ERROR_INTERNAL:	is = "internal error"; break;
		default:			is = "?"; break;
	}

	snprintf(buf, sizeof(buf), "%s (%s)",
		is, (Conns[slot].xerrno > 0) ? strerror(Conns[slot].xerrno) : "-");

	return buf;
}

/*
 * Raise an error. It is just a little helper.
 */
void Conn_error_raise(const unsigned int slot, const int err)
{
	if (err != 0)
		Conns[slot].xerrno = err;

	if (Conns[slot].cb_error)
		Conns[slot].cb_error(&Conns[slot]);
	else if (Conn_error_cb)
		Conn_error_cb(&Conns[slot]);
}

/* set noblocking */
int Conn_setnonblock(const int fd)
{
	return fcntl(fd, F_SETFL, O_NONBLOCK | O_RDWR);
}

void Log(const unsigned short level, char *format, ...)
{
	va_list	ap;
	FILE *out;

	if (level > Conn_level)
		return;

	if (Conn_Log == NULL)
		out = stderr;
	else
		out = Conn_Log;
	fprintf(out, "%ld.%06ld ",
		Conn_now.tv_sec, Conn_now.tv_usec);

	va_start(ap, format);
	vfprintf(out, format, ap);
	va_end(ap);
}

char *Conn_dump(const char *buf_src, const int len_src)
{
	int i, j;
	char tmp[3];
	char *buf_dst;
	unsigned char c;

	if (len_src < 0)
		return strdup("[Error: len < 0]");

	Log(30, "\tConn_dump(%p, len=%d)\n",
		buf_src, len_src);

	buf_dst = malloc(len_src * 4 + 1);
	if (buf_dst == NULL)
		return strdup("Memory allocation error1!");

	j = 0;
	for (i = 0; i < len_src; i++) {
		c = buf_src[i];
		if ((c < 32) || (c > 127)) {
			buf_dst[j++] = '[';
			snprintf(tmp, sizeof(tmp), "%02x", c);
			buf_dst[j++] = tmp[0];
			buf_dst[j++] = tmp[1];
			buf_dst[j++] = ']';
		} else {
			buf_dst[j++] = c;
		}
	}

	buf_dst[j] = '\0';

	/*
	Log(0, "%s ([%s], %d, [%s], %d\n",
		__FUNCTION__, buf_src, len_src, buf_dst, len_dst);
	*/

	return buf_dst;
}

char *Conn_dumphex(const char *buf_src, const int len_src)
{
	int i, j;
	char tmp[3];
	char *buf_dst;
	unsigned char c;

	if (len_src < 0)
		return strdup("[Error: len < 0]");

	Log(30, "\tConn_dumphex(%p, len=%d)\n",
		buf_src, len_src);

	buf_dst = malloc(len_src * 2 + 1);
	if (buf_dst == NULL)
		return strdup("Memory allocation error1!");

	j = 0;
	for (i = 0; i < len_src; i++) {
		c = buf_src[i];
		snprintf(tmp, sizeof(tmp), "%02x", c);
		buf_dst[j++] = tmp[0];
		buf_dst[j++] = tmp[1];
	}

	buf_dst[j] = '\0';

	return buf_dst;
}

void Conn_debug(FILE *f, const unsigned short debug)
{
	Conn_Log = f;
	Conn_level = debug;
}

char *Conn_state(const struct Conn *C)
{
	unsigned int slot;

	slot = C->slot;

	switch (Conns[slot].state) {
		case CONN_STATE_FREE:		return "FREE";
		case CONN_STATE_EMPTY:		return "EMPTY";
		case CONN_STATE_OPEN:		return "OPEN";
		case CONN_STATE_LISTEN:		return "LISTEN";
		case CONN_STATE_CONNECT_0:	return "CONN0";
		case CONN_STATE_CONNECT_a:	return "CONNa";
		case CONN_STATE_CONNECT_b:	return "CONNb";
		case CONN_STATE_ERROR:		return "ERROR";
		default:			return "BUG?";
	}
}

/*
 * Expand the requested buffer
 * what = 0 for out buffer, what = 1 for input buffer
 * returns 0 if OK, -1 on error
 */
int Conn_try_expand_buf(const unsigned int slot, const int what,
	const unsigned int needed)
{
	char *p;
	unsigned int hm;
	unsigned int old_size, amount, head, tail;
	unsigned int default_buf, buf_size, max_buf;
	char *pbuf;

	if (what == 0) {
		head = Conns[slot].obuf_head;
		tail = Conns[slot].obuf_tail;
		default_buf = Conn_default_obuf;
		old_size = Conns[slot].obuf_size;
		buf_size = Conns[slot].obuf_size;
		max_buf = Conn_max_obuf;
		pbuf = Conns[slot].obuf;
	} else {
		head = Conns[slot].ibuf_head;
		tail = Conns[slot].ibuf_tail;
		default_buf = Conn_default_ibuf;
		old_size = Conns[slot].ibuf_size;
		buf_size = Conns[slot].ibuf_size;
		max_buf = Conn_max_ibuf;
		pbuf = Conns[slot].ibuf;
	}

	/* Do we have enough room? */
	if (buf_size - tail >= needed)
		return 0;

	Log(10, "\tTry to expand buffer on slot=%u for [%s] needed=%d head=%u tail=%u.\n",
		slot, what == 0 ? "o" : "i", needed,
		head, tail);

	amount = needed - (buf_size - tail);

	/* Do not alloc less than 128 bytes */
	if (amount < 128)
		amount = 128;

	hm = buf_size + amount;
	if (hm > max_buf)
		hm = max_buf;

	/* Seems we are not allowed to grow larger */
	if (hm <= buf_size)
		return -1;

	p = realloc(pbuf, hm);
	if (p == NULL) {
		Log(3, "Cannot realloc obuf!\n");
		return -1;
	}

	if (what == 0) {
		Conns[slot].obuf = p;
		Conns[slot].obuf_size = hm;
		Conn_mem_buffers_out += hm - old_size;
	} else {
		Conns[slot].ibuf = p;
		Conns[slot].ibuf_size = hm;
		Conn_mem_buffers_in += hm - old_size;
	}


	Log(10, "\tSucces. Old/new size = %u/%u.\n",
		old_size, hm);

	return 0;
}

/*
 * Do not use it yet, it sucks (the paras)
 */
void Conn_poll_status(const short ev, char *ret)
{
	int i = 0;

	strcpy(ret, "");

	if (ev & CONN_POLLIN)		ret[i++] = 'I';
	if (ev & CONN_POLLPRI)		ret[i++] = 'P';
	if (ev & CONN_POLLOUT)		ret[i++] = 'O';
	if (ev & CONN_POLLERR)		ret[i++] = 'E';
	if (ev & CONN_POLLHUP)		ret[i++] = 'H';
	if (ev & CONN_POLLNVAL)		ret[i++] = 'V';
	if (ev & CONN_POLLRDNORM)	ret[i++] = 'r';
	if (ev & CONN_POLLRDBAND)	ret[i++] = 'R';
	ret[i++] = '\0';
}

char *Conn_domain(const struct Conn *C)
{
	unsigned int slot;

	slot = C->slot;

	switch (Conns[slot].sock_domain) {
		case PF_INET:		return "IPv4";
		case PF_INET6:		return "IPv6";
		case PF_PACKET:		return "PACKET";

		default:		return "?";
	}
}

char *Conn_type(const struct Conn *C)
{
	unsigned int slot;

	slot = C->slot;

	switch (Conns[slot].sock_type) {
		case SOCK_STREAM:	return "stream";
		case SOCK_DGRAM:	return "dgram";
		case SOCK_RAW:		return "raw";

		default:		return "?";
	}
}

char *Conn_get_socket_protocol(const struct Conn *C)
{
	unsigned int slot;

	slot = C->slot;

	switch (Conns[slot].sock_protocol) {
		case IPPROTO_IP:	return "IP";

		default:		return "?";
	}
}

static char *Conn_socktype(const struct Conn *C)
{
	unsigned int slot;

	slot = C->slot;

	switch (Conns[slot].type) {
		case CONN_TYPE_UNK:		return "unk";
		case CONN_TYPE_MASTER:		return "master";
		case CONN_TYPE_P2P:		return "p2p";
		default:			return "?";
	}
}

/*
 * Returns a nice speed
 */
void Conn_speed(char *dst, const unsigned int dst_len, const unsigned int speed)
{
	float sp;

	sp = speed;

	if (speed < 1000)
		snprintf(dst, dst_len, "%.2fBps", sp);
	else if (speed < 1000 * 1000)
		snprintf(dst, dst_len, "%.2fKBps", sp / 1000);
	else
		snprintf(dst, dst_len, "%.2fMBps", sp / 1000 / 1000);
}

char *Conn_status_slot(const unsigned int slot)
{
	static char tmp[1024];
	char polle[16], pollr[16];
	char speedi[32], speedo[32];
	unsigned int dT, si, so;
	char flags[128], flags_prefix[3], flags_postfix[2];
	char *local_addr, *remote_addr;
	int local_port, remote_port;
	char flags_tmp[64];

	/* flags */
	strcpy(flags, "");
	strcpy(flags_prefix, " [");
	strcpy(flags_postfix, "");

	if (Conns[slot].flags & CONN_FLAGS_AUTO_RECONNECT) {
		strcat(flags, flags_prefix);
		snprintf(flags_tmp, sizeof(flags_tmp), "autoreconnect_in_%ld/%u",
			(Conns[slot].tryat == 0) ? 0 : Conns[slot].tryat - Conn_now.tv_sec,
			Conns[slot].delay);
		strcat(flags, flags_tmp);
		strcpy(flags_prefix, " ");
		strcpy(flags_postfix, "]");
	}
	if (Conns[slot].flags & CONN_FLAGS_CLOSE_AFTER_SEND) {
		strcat(flags, flags_prefix);
		strcat(flags, "close_after_send");
		strcpy(flags_prefix, " ");
		strcpy(flags_postfix, "]");
	}

	strcat(flags, flags_postfix);

	Conn_poll_status(Conns[slot].events, polle);
	Conn_poll_status(Conns[slot].revents, pollr);

	dT = Conn_now.tv_sec - Conns[slot].start;
	if (dT == 0)
		dT = 1;
	si = Conns[slot].bi / dT;
	so = Conns[slot].bo / dT;

	Conn_speed(speedi, sizeof(speedi), si);
	Conn_speed(speedo, sizeof(speedo), so);

	Conn_set_address(&Conns[slot], 0);
	Conn_set_address(&Conns[slot], 1);

	local_addr = "-";
	local_port = 0;
	remote_addr = "-";
	remote_port = 0;
	if (Conns[slot].type == CONN_TYPE_MASTER) {
		local_addr = Conns[slot].bind_addr;
		local_port = Conns[slot].bind_port;
	} else if (Conns[slot].type == CONN_TYPE_P2P) {
		if (strlen(Conns[slot].bind_addr) > 0) {
			local_addr = Conns[slot].bind_addr;
			local_port = Conns[slot].bind_port;
		}
		remote_addr = Conns[slot].addr;
		remote_port = Conns[slot].port;
	}

	snprintf(tmp, sizeof(tmp), "id=%llu slot=%d fd=%d"
		" %s/%s/%s"
		" %s %s"
		" %s/%d <-> %s/%d"
		" via=%llu [%s][%s] IO=%llu/%llu"
		" BS=%u/%u S=%s/%s"
		" T=%ld bw=%u f=%u tk=%u"
		"%s\n",
		Conns[slot].id, slot, Conns[slot].fd,
		Conn_domain(&Conns[slot]), Conn_type(&Conns[slot]),
		Conn_get_socket_protocol(&Conns[slot]),
		Conn_socktype(&Conns[slot]), Conn_state(&Conns[slot]),
		local_addr, local_port, remote_addr, remote_port,
		Conns[slot].via, polle, pollr, Conns[slot].bi, Conns[slot].bo,
		Conns[slot].ibuf_size, Conns[slot].obuf_size, speedi, speedo,
		Conn_now.tv_sec - Conns[slot].start,
		Conns[slot].band_width, Conns[slot].band_factor, Conns[slot].band_tokens,
		flags);

	return tmp;
}

char *Conn_status_slot_html(const unsigned int slot)
{
	static char tmp[1024];
	char polle[16], pollr[16], *ext = "";
	char speedi[32], speedo[32];
	unsigned int dT, si, so;

	Conn_poll_status(Conns[slot].events, polle);
	Conn_poll_status(Conns[slot].revents, pollr);

	dT = Conn_now.tv_sec - Conns[slot].start;
	if (dT == 0)
		dT = 1;
	si = Conns[slot].bi / dT;
	so = Conns[slot].bo / dT;

	Conn_speed(speedi, sizeof(speedi), si);
	Conn_speed(speedo, sizeof(speedo), so);

	if (Conn_status_slot_html_cb)
		ext = Conn_status_slot_html_cb(&Conns[slot]);

	snprintf(tmp, sizeof(tmp), "<td>%llu</td><td>%d</td><td>%d</td>"
		"<td>%s</td><td>%s</td><td>%s</td>"
		"<td>%s</td><td>%s</td>"
		"<td>%s/%d</td>"
		"<td>%llu</td><td>%s</td><td>%s</td><td>%llu / %llu</td>"
		"<td>%u / %u</td><td>%s / %s</td><td>%ld</td>"
		"<td>%u</td><td>%u</td><td>%u</td>"
		"%s\n",
		Conns[slot].id, slot, Conns[slot].fd,
		Conn_domain(&Conns[slot]), Conn_type(&Conns[slot]),
		Conn_get_socket_protocol(&Conns[slot]),
		Conn_socktype(&Conns[slot]), Conn_state(&Conns[slot]),
		Conns[slot].addr, Conns[slot].port, Conns[slot].via, polle, pollr, Conns[slot].bi, Conns[slot].bo,
		Conns[slot].ibuf_size, Conns[slot].obuf_size,
		speedi, speedo, Conn_now.tv_sec - Conns[slot].start,
		Conns[slot].band_width, Conns[slot].band_factor, Conns[slot].band_tokens,
		ext);

	return tmp;
}

/* flags: bit 1 = 1 - html */
char *Conn_status(const unsigned int flags)
{
	unsigned int len = 0, slot, max, tmp_len;
	char tmp[512];
	char polle[16], pollr[16];
	char *buf, *per_slot, *ext = "";
	char speedi[32], speedo[32];
	unsigned long long bi, bo, dT;

	max = (Conn_no + 1) * 512 - 1;
	buf = malloc(max + 1);
	if (!buf)
		return strdup("No enough memory!");

	strcpy(buf, "");

	gettimeofday(&Conn_now, NULL);
	/* TODO: "len += " is incorrect */
	tmp_len = snprintf(tmp, sizeof(tmp),
		"Conn_pending=%d  Conn_no/Conn_max=%d/%d  Conn_total=%lu"
		"  Conn_uptime=%lus  Conn_allocated=%d  Conn_work_to_do=%u"
		"  Conn_mem_structs=%llu  Conn_mem_buffers_in/out=%llu/%llu\n",
		Conn_pending, Conn_no, Conn_max, Conn_total,
		Conn_now.tv_sec - Conn_start, Conn_allocated, Conn_work_to_do,
		Conn_mem_structs, Conn_mem_buffers_in, Conn_mem_buffers_out);
	if ((tmp_len > 0) && (len + tmp_len < max)) {
		strcat(buf, tmp);
		len += tmp_len;
	}

	if (flags & 1)
		if (Conn_status_cb)
			ext = Conn_status_cb();

	if (flags & 1) {
		strcat(buf, "<table border=\"0\" cellspacing=\"1\" cellpadding=\"3\" bgcolor=\"#aaaaaa\">\n");
		strcat(buf, "<tr bgcolor=\"ffffff\">\n");
		strcat(buf, "<td>ID</td>");
		strcat(buf, "<td>Slot</td>");
		strcat(buf, "<td>FD</td>");
		strcat(buf, "<td>Dom</td>");
		strcat(buf, "<td>Type</td>");
		strcat(buf, "<td>Protocol</td>");
		strcat(buf, "<td>SType</td>");
		strcat(buf, "<td>State</td>");
		strcat(buf, "<td>Addr/port</td>");
		strcat(buf, "<td>Via</td>");
		strcat(buf, "<td>Polle</td>");
		strcat(buf, "<td>Pollr</td>");
		strcat(buf, "<td>BI/BO</td>");
		strcat(buf, "<td>BUF I/O</td>");
		strcat(buf, "<td>Speed I/O</td>");
		strcat(buf, "<td>Elap (s)</td>");
		strcat(buf, "<td>Band</td>");
		strcat(buf, "<td>F</td>");
		strcat(buf, "<td>Tks</td>");
		strcat(buf, ext);
		strcat(buf, "</tr>\n");
	} else {
		strcat(buf, ext);
	}

	bi = 0; bo = 0; dT = 0;
	for (slot = 0; slot < Conn_no; slot++) {
		if (Conns[slot].state == CONN_STATE_FREE)
			continue;

		if (Conns[slot].type == CONN_TYPE_P2P) {
			bi += Conns[slot].bi;
			bo += Conns[slot].bo;
			dT += Conn_now.tv_sec - Conns[slot].start;
		}

		if (flags & 1)
			strcat(buf, "<tr bgcolor=\"ffffff\">\n");

		Conn_poll_status(Conns[slot].events, polle);
		Conn_poll_status(Conns[slot].revents, pollr);

		if ((flags & 1) == 0)
			per_slot = Conn_status_slot(slot);
		else
			per_slot = Conn_status_slot_html(slot);
		len += snprintf(tmp, sizeof(tmp), "%s", per_slot);
		if (len < max)
			strcat(buf, tmp);

		if (flags & 1)
			strcat(buf, "</tr>\n");
	}

	if (flags & 1)
		strcat(buf, "</table>\n");

	if (dT == 0)
		dT = 1;

	Conn_speed(speedi, sizeof(speedi), bi / dT);
	Conn_speed(speedo, sizeof(speedo), bo / dT);

	tmp_len = snprintf(tmp, sizeof(tmp), "Total speed I/O: %s/%s."
		" Total bytes I/O: %llu/%llu\n",
		speedi, speedo, bi, bo);
	if (len + tmp_len < max) {
		strcat(buf, tmp);
		len += tmp_len;
	}

	return buf;
}

/*
 * Returns the number of bytes in 'in' buffer
 */
unsigned int Conn_iqlen(const struct Conn *C)
{
	return C->ibuf_tail - C->ibuf_head;
}

/*
 * Returns the number of bytes in 'out' buffer
 */
unsigned int Conn_oqlen(const struct Conn *C)
{
	return C->obuf_tail - C->obuf_head;
}

/*
 * Returns the number of bytes in 'in' buffer (obsolete)
 */
unsigned int Conn_qlen(const struct Conn *C)
{
	return Conn_iqlen(C);
}

/*
 * Returns 1 if we can ignore this connection
 */
int Conn_ignore(const unsigned int slot)
{
	if (Conns[slot].error_state > 0)
		return 1;

	return 0;
}

/*
 * Close a connection if it exceeded maximum idle time or got a timeout
 */
void Conn_expire(const unsigned int slot)
{
	long long diff_ms;

	if (Conns[slot].trigger > 0) {
		/* We do not trigger first time */
		if (Conns[slot].last_trigger == 0)
			Conns[slot].last_trigger = Conn_now.tv_sec;

		if ((Conns[slot].last_trigger > 0)
			&& (Conns[slot].last_trigger + Conns[slot].trigger < Conn_now.tv_sec)) {
			if (Conns[slot].cb_trigger)
				Conns[slot].cb_trigger(&Conns[slot]);
			else if (Conn_trigger_cb)
				Conn_trigger_cb(&Conns[slot]);
			Conns[slot].last_trigger = Conns[slot].last_trigger + Conns[slot].trigger;
		}
	}

	if ((Conns[slot].idle > 0) && (Conns[slot].trecv.tv_sec + Conns[slot].idle < Conn_now.tv_sec)) {
		Conns[slot].error_state = CONN_ERROR_EXPIRED;
	} else if ((Conns[slot].read_timeout > 0) && (Conns[slot].tsend.tv_sec > 0)
		&& (Conns[slot].tsend.tv_sec > Conns[slot].trecv.tv_sec)) {
		diff_ms = Conn_time_diff(&Conn_now, &Conns[slot].tsend);
		if (diff_ms > Conns[slot].read_timeout) {
			Conns[slot].error_state = CONN_ERROR_READ_TIMEOUT;
		}
	} else if ((Conns[slot].conn_timeout > 0) && (Conns[slot].state == CONN_STATE_CONNECT_b)) {
		diff_ms = Conn_time_diff(&Conn_now, &Conns[slot].conn_syn);
		if (diff_ms > Conns[slot].conn_timeout) {
			/* connection attempt expired */
			Conns[slot].error_state = CONN_ERROR_CONN_TIMEOUT;
		}
	}
}

/*
 * Set NODELAY on socket
 */
int Conn_nodelay(const struct Conn *C)
{
	int i = 1;
	unsigned int slot;

	slot = C->slot;

	return setsockopt(Conns[slot].fd, SOL_TCP, TCP_NODELAY, &i, sizeof(i));
}

void Conn_rollback(struct Conn *C, const unsigned int bytes)
{
	unsigned int slot;

	slot = C->slot;

	if (Conns[slot].obuf_tail - Conns[slot].obuf_head <= bytes)
		Conns[slot].obuf_tail -= bytes;
}

/*
 * Returns a pointer to current in buffer
 */
char *Conn_ibuf(const struct Conn *C)
{
	unsigned int slot;

	slot = C->slot;

	return Conns[slot].ibuf + Conns[slot].ibuf_head;
}

/*
 * Returns a pointer to current out buffer
 */
char *Conn_obuf(const struct Conn *C)
{
	unsigned int slot;

	slot = C->slot;

	return Conns[slot].obuf + Conns[slot].obuf_head;
}

/*
 * Returns the id of a connection
 */
unsigned long long Conn_getid(const struct Conn *C)
{
	unsigned int slot;

	slot = C->slot;

	return Conns[slot].id;
}

/*
 * Returns a Conn* searching by id
 */
struct Conn *Conn_get(const unsigned long long id)
{
	struct Conn *R = NULL;
	int i;

	for (i = Conn_no - 1; i >= 0; i--) {
		if (Conns[i].id == id) {
			R = &Conns[i];
			break;
		}
	}

	return R;
}

/*
 * Returns the fd associated with C
 */
int Conn_get_fd(const struct Conn *C)
{
	unsigned int slot;

	slot = C->slot;

	return Conns[slot].fd;
}

/*
 * Returns the timeval of the last packet
 */
void Conn_last_time(const struct Conn *C, struct timeval *tv)
{
	unsigned int slot;

	slot = C->slot;

	*tv = Conns[slot].trecv;
}

/*
 * Set a callback
 */
int Conn_set_cb(struct Conn *C, const unsigned int cb_type, void (*f)(struct Conn *))
{
	unsigned int slot;

	slot = C->slot;

	switch (cb_type) {
		case CONN_CB_ACCEPT:	Conns[slot].cb_accept = f; break;
		case CONN_CB_RECV:	Conns[slot].cb_recv = f; break;
		case CONN_CB_SEND:	Conns[slot].cb_send = f; break;
		case CONN_CB_DATA:	Conns[slot].cb_data = f; break;
		case CONN_CB_CLOSE:	Conns[slot].cb_close = f; break;
		case CONN_CB_TRIGGER:	Conns[slot].cb_trigger = f; break;
		case CONN_CB_ERROR:	Conns[slot].cb_error = f; break;
		case CONN_CB_CONNECTED:	Conns[slot].cb_connected = f; break;
		case CONN_CB_ACCEPT_ERROR:	Conns[slot].cb_accept_error = f; break;

		default:
			return -1;
	}

	return 0;
}

/*
 * Search for str in active buffer from a given offset
 * Returns pointer to string if match or NUll if doesn't.
 * @flags: bit 0 == 1 => case insensitive
 */
char *Conn_ostrstr(struct Conn *C, const unsigned int off, const char *str,
	const unsigned int flags)
{
	unsigned int len, str_len, i;
	char *buf, *ret = NULL;
	int err;
	unsigned int slot;

	slot = C->slot;

	len = Conns[slot].ibuf_tail - Conns[slot].ibuf_head - off;
	buf = Conns[slot].ibuf + Conns[slot].ibuf_head + off;
	str_len = strlen(str);

	if (len < str_len)
		return NULL;

	i = 0;
	while (i <= len - str_len) {
		if (flags & 1)
			err = strncasecmp(buf + i, str, str_len);
		else
			err = strncmp(buf + i, str, str_len);
		if (err == 0) {
			ret = buf + i;
			break;
		}

		i++;
	}

	return ret;
}

/*
 * Search for str in active buffer
 * Returns pointer to string if match or NUll if doesn't.
 */
char *Conn_strstr(struct Conn *C, const char *str)
{
	return Conn_ostrstr(C, 0, str, 0);
}

/*
 * Search for str in active buffer (case insensitive)
 * Returns pointer to string if match or NUll if doesn't.
 */
char *Conn_strcasestr(struct Conn *C, const char *str)
{
	return Conn_ostrstr(C, 0, str, 1);
}

/*
 * Returns a '\0' terminated line, modifying received buffer
 */
char *Conn_get_line(struct Conn *C)
{
	char *cr;

	cr = Conn_ostrstr(C, 0, "\n", 1);
	if (!cr)
		return NULL;

	*cr = '\0';

	return Conn_ibuf(C);
}

/*
 * Helper help building text line daemons
 */
void Conn_for_every_line(struct Conn *C, int (*cb)(struct Conn *C, char *line))
{
	int ret = 0;
	char *line;
	unsigned int line_size;
	unsigned int slot;

	slot = C->slot;

	if (cb == NULL)
		return;

	while (1) {
		line = Conn_get_line(&Conns[slot]);
		if (line == NULL)
			break;

		line_size = strlen(line) + 1;

		Conn_rtrim(line, "\r");

		ret = cb(&Conns[slot], line);
		if (ret != 0)
			break;

		Conn_eat(&Conns[slot], line_size);
	}
}

/*
 * Eat @bytes from head of input buffer
 */
void Conn_eat(struct Conn *C, const unsigned int bytes)
{
	unsigned int slot;

	slot = C->slot;

	/* advance head */
	Conns[slot].ibuf_head += bytes;
	if (Conns[slot].ibuf_head >= Conns[slot].ibuf_tail) {
		Conns[slot].ibuf_head = 0;
		Conns[slot].ibuf_tail = 0;
	}

	Log(10, "Conn_eat(slot=%u, %u) head=%u tail=%u qlen=%u\n",
		slot, bytes, Conns[slot].ibuf_head, Conns[slot].ibuf_tail,
		Conn_qlen(&Conns[slot]));
}

/*
 * Eat all input buffer
 */
void Conn_eatall(struct Conn *C)
{
	Conn_eat(C, Conn_qlen(C));
}

/*
 * If put buffer is empty, just mark for closing.
 * If we have data, set the flag to do the closing after send.
 */
void Conn_close(struct Conn *C)
{
	unsigned int slot;

	slot = C->slot;

	Log(10, "%s: Mark slot=%u, id=%llu for closing...\n",
		__FUNCTION__, slot, Conns[slot].id);

	if (Conns[slot].obuf_head == Conns[slot].obuf_tail)
		Conns[slot].error_state = CONN_ERROR_USERREQ;
	else
		Conns[slot].flags |= CONN_FLAGS_CLOSE_AFTER_SEND;
}

/*
 * Instructs Conn to stop
 */
void Conn_stop(void)
{
	Conn_must_stop = 1;
}

/*
 * Set some internal parameters
 */
void Conn_set(struct Conn *C, const unsigned int var, const int val)
{
	int fd;
	unsigned int slot;

	slot = C->slot;

	fd = Conn_get_fd(&Conns[slot]);

	switch (var) {
	case CONN_PARA_AUTO_RECONNECT:
		Conns[slot].flags |= (val == 0) ? 0 : CONN_FLAGS_AUTO_RECONNECT;
		break;
	case CONN_PARA_RECONNECT_DELAY:
		Conns[slot].delay = val;
		break;
	case CONN_PARA_IDLE_TIME:
		Conns[slot].idle = val;
		break;
	case CONN_PARA_READ_TIMEOUT:
		Conns[slot].read_timeout = val;
		break;
	case CONN_PARA_CONN_TIMEOUT:
		Conns[slot].conn_timeout = val;
		break;
	case CONN_PARA_TRIGGER:
		Conns[slot].trigger = val;
		Conns[slot].last_trigger = 0;
		break;
	case CONN_PARA_IBUF:
		setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &val, sizeof(val));
		break;
	case CONN_PARA_OBUF:
		setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &val, sizeof(val));
		break;
	}
}

/*
 * Init a queue
 */
void Conn_queue_init(struct Conn_queue *q)
{
	q->head = q->tail = NULL;
}

/*
 * Add a slot in a queue
 */
int Conn_queue_add(struct Conn_queue *q, const unsigned int slot)
{
	struct Conn_queue_entry *p;

	p = (struct Conn_queue_entry *) calloc(1, sizeof(struct Conn_queue_entry));
	if (!p)
		return -1;

	p->slot = slot;
	p->next = NULL;

	if (q->head == NULL) {
		q->head = p;
	} else {
		q->tail->next = p;
	}

	q->tail = p;

	return 0;
}

/*
 * Destroys a queue
 */
void Conn_queue_destroy(struct Conn_queue *q)
{
	struct Conn_queue_entry *p, *next;

	p = q->head;
	while (p) {
		next = p->next;
		free(p);
		p = next;
	}

	q->head = q->tail = NULL;
}

/* Misc */
/*
 * Returns the string representation of an socket address
 * @flags: bit0==0 => local address, bit0==1 => peer address
 */
int Conn_set_address(struct Conn *C, const int flags)
{
	int err;
	struct sockaddr *psa;
	struct sockaddr_in sa4;
	struct sockaddr_in6 sa6;
	socklen_t sa_len;
	char *paddr;
	size_t addr_size;
	int *pport;
	unsigned int slot;

	/* Test if we need to regenerate. */
	if (flags & 1) {
		if (!(C->flags & CONN_ADDR_LOCAL_DIRTY))
			return 0;
        } else {
		if (!(C->flags & CONN_ADDR_REMOTE_DIRTY))
			return 0;
	}

	slot = C->slot;

	switch (Conns[slot].sock_domain) {
		case PF_INET:
			psa = (struct sockaddr *) &sa4;
			sa_len = sizeof(struct sockaddr_in);
			break;
		case PF_INET6:
			psa = (struct sockaddr *) &sa6;
			sa_len = sizeof(struct sockaddr_in6);
			break;
		default:
			return -1;
	}

	if (flags & 1) {
		/* peer */
		paddr = Conns[slot].addr;
		addr_size = sizeof(Conns[slot].addr);
		pport = &Conns[slot].port;
		err = getpeername(Conns[slot].fd, psa, &sa_len);
	} else {
		/* local */
		paddr = Conns[slot].bind_addr;
		addr_size = sizeof(Conns[slot].bind_addr);
		pport = &Conns[slot].bind_port;
		err = getsockname(Conns[slot].fd, psa, &sa_len);
	}

	if (err != 0)
		return -1;

	switch (Conns[slot].sock_domain) {
		case PF_INET:
			inet_ntop(Conns[slot].sock_domain, &sa4.sin_addr,
				paddr, addr_size);
			*pport = ntohs(sa4.sin_port);
			break;
		case PF_INET6:
			inet_ntop(Conns[slot].sock_domain, &sa6.sin6_addr,
				paddr, addr_size);
			*pport = ntohs(sa6.sin6_port);
			break;
		default:
			return -1;
	}

	return 0;
}

/*
 * Returns the address family for address stored in @addr.
 */
int Conn_addr_family(const char *addr)
{
	struct addrinfo hints, *results = NULL;
	int ret;

	memset(&hints, 0, sizeof(struct addrinfo));
	hints.ai_family = AF_UNSPEC;
	hints.ai_socktype = 0;
	hints.ai_flags = AI_NUMERICHOST;
	hints.ai_protocol = 0;
	hints.ai_canonname = NULL;
	hints.ai_addr = NULL;
	hints.ai_next = NULL;

	ret = getaddrinfo(addr, NULL, &hints, &results);
	if (ret != 0) {
		snprintf(Conn_error, sizeof(Conn_error),
			"getaddrinfo error on %s (%s)",
			addr, gai_strerror(ret));
		if (results)
			freeaddrinfo(results);
		return -1;
	}

	ret = results->ai_family;

	freeaddrinfo(results);

	return ret;
}

/* Splitting stuff */

/*
 * Cut from right, in place, the chars specified in @chars.
 */
void Conn_rtrim(char *s, const char *chars)
{
	char *e;

	if (!s || (*s == '\0'))
		return;

	e = s + strlen(s) - 1;
	while ((e >= s) && (strchr(chars, *e))) {
		*e = '\0';
		e--;
	}
}

/*
 * Free a prealocated structure
 */
void Conn_split_free(struct Conn_split **s)
{
	struct Conn_split *p;
	struct Conn_split_cell *q, *next;

	if (!s)
		return;

	p = *s;
	if (!p)
		return;

	q = p->head;
	while (q) {
		next = q->next;
		free(q);
		q = next;
	}

	if (p->line)
		free(p->line);

	free(p);
}

/*
 * Split a buffer pointed by C to var,value pairs.
 */
struct Conn_split *Conn_split(const char *line0)
{
	char *p;
	struct Conn_split *ret = NULL;
	struct Conn_split_cell *q;
	unsigned int i;
	char search_for;
	char *left, *right;
	unsigned int right_len;

	ret = (struct Conn_split *) calloc(1, sizeof(struct Conn_split));
	if (!ret) {
		snprintf(Conn_error, sizeof(Conn_error),
			"cannot alloc memory for Conn_split!\n");
		return NULL;
	}

	ret->line = strdup(line0);
	if (!ret->line) {
		snprintf(Conn_error, sizeof(Conn_error),
			"cannot alloc memory for line duplication!\n");
		goto free;
	}

	Conn_rtrim(ret->line, "\r\n \t");

	/* do the spliting */
	p = ret->line;
	while (*p != '\0') {
		/* skip empty space */
		while ((*p == ' ') || (*p == '\t'))
			p++;

		if (*p == '\0')
			break;

		/* Init */
		right = "";
		right_len = 0;

		/* Building left */
		left = p;
		i = 0;
		while ((*p != '\0') && (*p != '='))
			p++;
		if (*p != '\0') {
			*p = '\0';

			/* skip '=' */
			p++;

			/* Building right */
			right = p;

			search_for = ' ';
			if (*p == '"') {
				search_for = '"';
				p++;
			}

			i = 0;
			while ((*p != '\0') && (*p != search_for)) {
				right_len++;
				p++;
			}

			if (*p != '\0') {
				*p = '\0';
				p++;
			}
		}

		/* alloc data and fill it */
		q = (struct Conn_split_cell *) calloc(1, sizeof(struct Conn_split_cell));
		if (!q) {
			snprintf(Conn_error, sizeof(Conn_error),
				"cannot alloc memory!\n");
			goto free;
		}
		q->left = left;
		q->right = right;
		q->right_len = right_len;

		if (ret->head == NULL)
			ret->head = q;
		else
			ret->tail->next = q;

		ret->tail = q;
	}

	return ret;

	free:
	Conn_split_free(&ret);

	return NULL;
}

/*
 * Search for a string and return the value
 */
char *Conn_split_get_size(const struct Conn_split *s, const char *left,
	unsigned int *size)
{
	struct Conn_split_cell *p;

	p = s->head;
	while (p) {
		if (strcmp(left, p->left) == 0) {
			if (size != NULL)
				*size = p->right_len;
			return p->right;
		}
		p = p->next;
	}

	return NULL;
}

/*
 * Search for a string and return the value
 */
char *Conn_split_get_e(const struct Conn_split *s, const char *l)
{
	return Conn_split_get_size(s, l, NULL);
}

/*
 * Search for a string and return the value or "" if not found
 */
char *Conn_split_get(const struct Conn_split *s, const char *l)
{
	char *r;

	r = Conn_split_get_size(s, l, NULL);
	if (!r)
		r = "";

	return r;
}

/*
 * Return a value as unsigned long
 */
unsigned long Conn_split_get_ul(const struct Conn_split *s, const char *l,
	unsigned int base)
{
	char *r;
	unsigned long ret = 0;

	r = Conn_split_get_e(s, l);
	if (r)
		ret = strtoul(r, NULL, base);

	return ret;
}

/*
 * Return a value as unsigned long long
 */
unsigned long long Conn_split_get_ull(const struct Conn_split *s, const char *l,
	unsigned int base)
{
	char *r;
	unsigned long long ret = 0;

	r = Conn_split_get_e(s, l);
	if (r)
		ret = strtoull(r, NULL, base);

	return ret;
}

/*
 * Return a value as double
 */
double Conn_split_get_d(const struct Conn_split *s, const char *l)
{
	char *r;
	double ret = 0;

	r = Conn_split_get_e(s, l);
	if (r)
		ret = strtod(r, NULL);

	return ret;
}

/*
 * Returns 1 if the string contains only 0-9a-zA-Z. Else 0
 */
int Conn_alphanum(const char *s)
{
	size_t i, len;

	len = strlen(s);
	for (i = 0; i < len; i++)
		if (isalnum(s[i]) == 0)
			return 0;

	return 1;
}