* Write test case for async memcached library

* Fix memcached async library to pass test :)

7 files changed, 130 insertions, 42 deletions

diff --git a/memcached.c b/memcached.c
index 3228d791f..05ae16617 100644
--- a/memcached.c
+++ b/memcached.c
@@ -22,6 +22,7 @@
 #include <fcntl.h>
 #include <sys/uio.h>
 #include <event.h>
+#include <glib.h>
 
 #include "memcached.h"
 
@@ -59,8 +60,8 @@ memc_log (const memcached_ctx_t *ctx, int line, const char *fmt, ...)
 	va_list args;
 	if (ctx->options & MEMC_OPT_DEBUG) {
 		va_start (args, fmt);
-		syslog (LOG_DEBUG, "memc_debug(%d): host: %s, port: %d", line, inet_ntoa (ctx->addr), ntohs (ctx->port));
-		vsyslog (LOG_DEBUG, fmt, args);
+		g_log (G_LOG_DOMAIN, G_LOG_LEVEL_DEBUG, "memc_debug(%d): host: %s, port: %d", line, inet_ntoa (ctx->addr), ntohs (ctx->port));
+		g_logv (G_LOG_DOMAIN, G_LOG_LEVEL_DEBUG, fmt, args);
 		va_end (args);
 	}
 }
@@ -104,7 +105,7 @@ write_handler (int fd, short what, memcached_ctx_t *ctx)
 			iov[2].iov_len = ctx->param->bufsize - ctx->param->bufpos;
 			iov[3].iov_base = CRLF;
 			iov[3].iov_len = sizeof (CRLF) - 1;
-			ctx->param->bufpos = writev (ctx->sock, iov, 4);
+			writev (ctx->sock, iov, 4);
 		}
 		else {
 			iov[0].iov_base = read_buf;
@@ -113,18 +114,11 @@ write_handler (int fd, short what, memcached_ctx_t *ctx)
 			iov[1].iov_len = ctx->param->bufsize - ctx->param->bufpos;
 			iov[2].iov_base = CRLF;
 			iov[2].iov_len = sizeof (CRLF) - 1;
-			ctx->param->bufpos = writev (ctx->sock, iov, 3);	
-		}
-		if (ctx->param->bufpos == ctx->param->bufsize) {
-			event_del (&ctx->mem_ev);
-			event_set (&ctx->mem_ev, ctx->sock, EV_READ | EV_PERSIST | EV_TIMEOUT, socket_callback, (void *)ctx);
-			event_add (&ctx->mem_ev, &ctx->timeout);
-		}
-		else {
-			event_del (&ctx->mem_ev);
-			event_set (&ctx->mem_ev, ctx->sock, EV_WRITE | EV_TIMEOUT, socket_callback, (void *)ctx);
-			event_add (&ctx->mem_ev, &ctx->timeout);
+			writev (ctx->sock, iov, 3);
 		}
+		event_del (&ctx->mem_ev);
+		event_set (&ctx->mem_ev, ctx->sock, EV_READ | EV_PERSIST | EV_TIMEOUT, socket_callback, (void *)ctx);
+		event_add (&ctx->mem_ev, &ctx->timeout);
 	}
 	else if (what == EV_READ) {
 		/* Read header */
@@ -135,8 +129,8 @@ write_handler (int fd, short what, memcached_ctx_t *ctx)
 			iov[1].iov_base = read_buf;
 			iov[1].iov_len = READ_BUFSIZ;
 			if ((r = readv (ctx->sock, iov, 2)) == -1) {
-				ctx->callback (ctx, SERVER_ERROR, ctx->callback_data);
 				event_del (&ctx->mem_ev);
+				ctx->callback (ctx, SERVER_ERROR, ctx->callback_data);
 			}
 			if (header.req_id != ctx->count && retries < MAX_RETRIES) {
 				retries ++;
@@ -148,8 +142,10 @@ write_handler (int fd, short what, memcached_ctx_t *ctx)
 		if (ctx->protocol != UDP_TEXT) {
 			r = read (ctx->sock, read_buf, READ_BUFSIZ - 1);
 		}
+		memc_log (ctx, __LINE__, "memc_write: read reply from memcached: %s", read_buf);
 		/* Increment count */
 		ctx->count++;
+		event_del (&ctx->mem_ev);
 		if (strncmp (read_buf, STORED_TRAILER, sizeof (STORED_TRAILER) - 1) == 0) {
 			ctx->callback (ctx, OK, ctx->callback_data);
 		}
@@ -162,11 +158,10 @@ write_handler (int fd, short what, memcached_ctx_t *ctx)
 		else {
 			ctx->callback (ctx, SERVER_ERROR, ctx->callback_data);
 		}
-		event_del (&ctx->mem_ev);
 	}
 	else if (what == EV_TIMEOUT) {
-		ctx->callback (ctx, SERVER_TIMEOUT, ctx->callback_data);
 		event_del (&ctx->mem_ev);
+		ctx->callback (ctx, SERVER_TIMEOUT, ctx->callback_data);
 	}
 }
 
@@ -182,7 +177,7 @@ read_handler (int fd, short what, memcached_ctx_t *ctx)
 	size_t datalen;
 	struct memc_udp_header header;
 	struct iovec iov[2];
-	int retries = 0;
+	int retries = 0, t;
 	
 	if (what == EV_WRITE) {
 		/* Send command to memcached */
@@ -216,7 +211,9 @@ read_handler (int fd, short what, memcached_ctx_t *ctx)
 			iov[1].iov_base = read_buf;
 			iov[1].iov_len = READ_BUFSIZ;
 			if ((r = readv (ctx->sock, iov, 2)) == -1) {
+				event_del (&ctx->mem_ev);
 				ctx->callback (ctx, SERVER_ERROR, ctx->callback_data);
+				return;
 			}
 			memc_log (ctx, __LINE__, "memc_read: got read_buf: %s", read_buf);
 			if (header.req_id != ctx->count && retries < MAX_RETRIES) {
@@ -234,22 +231,25 @@ read_handler (int fd, short what, memcached_ctx_t *ctx)
 		if (r > 0) {
 			read_buf[r] = 0;
 			if (ctx->param->bufpos == 0) {
-				r = memc_parse_header (read_buf, &datalen, &p);
-				if (r < 0) {
+				t = memc_parse_header (read_buf, &datalen, &p);
+				if (t < 0) {
+					event_del (&ctx->mem_ev);
 					memc_log (ctx, __LINE__, "memc_read: cannot parse memcached reply");
 					ctx->callback (ctx, SERVER_ERROR, ctx->callback_data);
-					goto cleanup;
+					return;
 				}
-				else if (r == 0) {
+				else if (t == 0) {
 					memc_log (ctx, __LINE__, "memc_read: record does not exists");
+					event_del (&ctx->mem_ev);
 					ctx->callback (ctx, NOT_EXISTS, ctx->callback_data);
-					goto cleanup;
+					return;
 				}
 
 				if (datalen > ctx->param->bufsize) {
 					memc_log (ctx, __LINE__, "memc_read: user's buffer is too small: %zd, %zd required", ctx->param->bufsize, datalen);
+					event_del (&ctx->mem_ev);
 					ctx->callback (ctx, WRONG_LENGTH, ctx->callback_data);
-					goto cleanup;
+					return;
 				}
 				/* Check if we already have all data in buffer */
 				if (r >= datalen + sizeof (END_TRAILER) + sizeof (CRLF) - 2) {
@@ -257,8 +257,9 @@ read_handler (int fd, short what, memcached_ctx_t *ctx)
 					memcpy (ctx->param->buf + ctx->param->bufpos, p, datalen);
 					/* Increment count */
 					ctx->count++;
+					event_del (&ctx->mem_ev);
 					ctx->callback (ctx, OK, ctx->callback_data);
-					goto cleanup;
+					return;
 				}
 				/* Subtract from sum parsed header's length */
 				r -= p - read_buf;
@@ -271,8 +272,9 @@ read_handler (int fd, short what, memcached_ctx_t *ctx)
 						END_TRAILER, sizeof (END_TRAILER) - 1) == 0) {
 				r -= sizeof (END_TRAILER) - sizeof (CRLF) - 2;
 				memcpy (ctx->param->buf + ctx->param->bufpos, p, r);
+				event_del (&ctx->mem_ev);
 				ctx->callback (ctx, OK, ctx->callback_data);
-				goto cleanup;
+				return;
 			}
 			/* Store this part of data in param's buffer */
 			memcpy (ctx->param->buf + ctx->param->bufpos, p, r);
@@ -280,21 +282,18 @@ read_handler (int fd, short what, memcached_ctx_t *ctx)
 		}
 		else {
 			memc_log (ctx, __LINE__, "memc_read: read(v) failed: %d, %m", r);
+			event_del (&ctx->mem_ev);
 			ctx->callback (ctx, SERVER_ERROR, ctx->callback_data);
-			goto cleanup;
+			return;
 		}
 
 		ctx->count++;
 	}
 	else if (what == EV_TIMEOUT) {
+		event_del (&ctx->mem_ev);
 		ctx->callback (ctx, SERVER_TIMEOUT, ctx->callback_data);
-		goto cleanup;
 	}
 	
-	return;
-
-cleanup:
-	event_del (&ctx->mem_ev);
 }
 
 /*
@@ -343,8 +342,9 @@ delete_handler (int fd, short what, memcached_ctx_t *ctx)
 			iov[1].iov_base = read_buf;
 			iov[1].iov_len = READ_BUFSIZ;
 			if ((r = readv (ctx->sock, iov, 2)) == -1) {
-				ctx->callback (ctx, SERVER_ERROR, ctx->callback_data);
 				event_del (&ctx->mem_ev);
+				ctx->callback (ctx, SERVER_ERROR, ctx->callback_data);
+				return;
 			}
 			if (header.req_id != ctx->count && retries < MAX_RETRIES) {
 				retries ++;
@@ -358,6 +358,7 @@ delete_handler (int fd, short what, memcached_ctx_t *ctx)
 		}
 		/* Increment count */
 		ctx->count++;
+		event_del (&ctx->mem_ev);
 		if (strncmp (read_buf, DELETED_TRAILER, sizeof (STORED_TRAILER) - 1) == 0) {
 			ctx->callback (ctx, OK, ctx->callback_data);
 		}
@@ -367,11 +368,10 @@ delete_handler (int fd, short what, memcached_ctx_t *ctx)
 		else {
 			ctx->callback (ctx, SERVER_ERROR, ctx->callback_data);
 		}
-		event_del (&ctx->mem_ev);
 	}
 	else if (what == EV_TIMEOUT) {
-		ctx->callback (ctx, SERVER_TIMEOUT, ctx->callback_data);
 		event_del (&ctx->mem_ev);
+		ctx->callback (ctx, SERVER_TIMEOUT, ctx->callback_data);
 	}
 }
 
@@ -560,6 +560,7 @@ memc_write (memcached_ctx_t *ctx, const char *cmd, memcached_param_t *param, int
 	ctx->cmd = cmd;
 	ctx->op = CMD_WRITE;
 	ctx->param = param;
+	param->expire = expire;
 	event_set (&ctx->mem_ev, ctx->sock, EV_WRITE | EV_TIMEOUT, socket_callback, (void *)ctx);
 	event_add (&ctx->mem_ev, &ctx->timeout);
 
diff --git a/memcached.h b/memcached.h
index c815884a3..46bcae465 100644
--- a/memcached.h
+++ b/memcached.h
@@ -63,8 +63,6 @@ typedef struct memcached_ctx_s {
 	short options;
 	/* Current operation */
 	memc_opt_t op;
-	/* Event structure */
-	struct event mem_ev;
 	/* Current command */
 	const char *cmd;
 	/* Current param */
@@ -73,6 +71,8 @@ typedef struct memcached_ctx_s {
 	void (*callback) (struct memcached_ctx_s *ctx, memc_error_t error, void *data);
 	/* Data for callback function */
 	void *callback_data;
+	/* Event structure */
+	struct event mem_ev;
 } memcached_ctx_t;
 
 typedef void (*memcached_callback_t) (memcached_ctx_t *ctx, memc_error_t error, void *data);
diff --git a/test/Makefile.in b/test/Makefile.in
index 1d46e2e2f..ddd298c7c 100644
--- a/test/Makefile.in
+++ b/test/Makefile.in
@@ -2,8 +2,8 @@
 
 all: rspamd_test_suite
 
-rspamd_test_suite: $(OBJECTS) ../url.o ../util.o
-	$(CC) $(PTHREAD_LDFLAGS) $(LDFLAGS) $(OBJECTS) ../url.o ../util.o $(LIBS) -o rspamd_test_suite
+rspamd_test_suite: $(OBJECTS) ../url.o ../util.o ../memcached.o
+	$(CC) $(PTHREAD_LDFLAGS) $(LDFLAGS) $(OBJECTS) ../url.o ../util.o ../memcached.o $(LIBS) -o rspamd_test_suite
 
 run_test: rspamd_test_suite
 	gtester --verbose -k -o=rspamd_test.xml ./rspamd_test_suite
diff --git a/test/rspamd_memcached_test.c b/test/rspamd_memcached_test.c
new file mode 100644
index 000000000..1993c756e
--- /dev/null
+++ b/test/rspamd_memcached_test.c
@@ -0,0 +1,85 @@
+#include <sys/types.h>
+#include <sys/time.h>
+#include <sys/wait.h>
+#include <sys/param.h>
+
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <netdb.h>
+#include <syslog.h>
+#include <fcntl.h>
+#include <stdlib.h>
+#include <string.h>
+#include <event.h>
+
+#include "../config.h"
+#include "../main.h"
+#include "../cfg_file.h"
+#include "../memcached.h"
+#include "tests.h"
+
+u_char *buf = "test";
+
+static void 
+memcached_callback (memcached_ctx_t *ctx, memc_error_t error, void *data)
+{
+	struct timeval tv;
+
+	switch (ctx->op) {
+		case CMD_CONNECT:
+			g_assert (error == OK);
+			msg_debug ("Connect ok");
+			memc_set (ctx, ctx->param, 60);
+			break;
+		case CMD_READ:
+			g_assert (error == OK);
+			g_assert (!strcmp(ctx->param->buf, buf));
+			msg_debug ("Read ok");
+			memc_close_ctx (ctx);
+			tv.tv_sec = 0;
+			tv.tv_usec = 0;
+			event_loopexit (&tv);
+			break;
+		case CMD_WRITE:
+			g_assert (error == OK);
+			msg_debug ("Write ok");
+			ctx->param->buf = g_malloc (sizeof (buf));
+			bzero (ctx->param->buf, sizeof (buf));
+			memc_get (ctx, ctx->param);
+			break;
+	}
+}
+			
+void
+rspamd_memcached_test_func ()
+{
+	memcached_ctx_t *ctx;
+	memcached_param_t *param;
+	struct in_addr addr;
+
+	ctx = g_malloc (sizeof (memcached_ctx_t));
+	param = g_malloc (sizeof (memcached_param_t));
+	bzero (ctx, sizeof (memcached_ctx_t));
+	bzero (param, sizeof (memcached_param_t));
+
+	event_init ();
+
+	ctx->callback = memcached_callback;
+	ctx->callback_data = (void *)param;
+	ctx->protocol = TCP_TEXT;
+	inet_aton ("127.0.0.1", &addr);
+	memcpy (&ctx->addr, &addr, sizeof (struct in_addr));
+	ctx->port = htons (11211);
+	ctx->timeout.tv_sec = 1;
+	ctx->timeout.tv_usec = 0;
+	ctx->sock = -1;
+	ctx->options = MEMC_OPT_DEBUG;
+	strlcpy (param->key, buf, sizeof (param->key));
+	param->buf = buf;
+	param->bufsize = strlen (buf);
+	ctx->param = param;
+	g_assert (memc_init_ctx (ctx) != -1);
+
+	event_loop (0);
+}
+
diff --git a/test/rspamd_test_suite.c b/test/rspamd_test_suite.c
index 7b699f40b..09db78496 100644
--- a/test/rspamd_test_suite.c
+++ b/test/rspamd_test_suite.c
@@ -21,8 +21,7 @@ main (int argc, char **argv)
 	g_test_init (&argc, &argv, NULL);
 
 	g_test_add_func ("/rspamd/url", rspamd_url_test_func);
+	g_test_add_func ("/rspamd/memcached", rspamd_memcached_test_func);
 
 	g_test_run ();
-	
-	g_mem_profile ();
 }
diff --git a/test/tests.h b/test/tests.h
index c0e1858a2..85e111b2b 100644
--- a/test/tests.h
+++ b/test/tests.h
@@ -8,4 +8,7 @@
 /* URL parser test */
 void rspamd_url_test_func ();
 
+/* Memceched library test */
+void rspamd_memcached_test_func ();
+
 #endif
diff --git a/utils/url_extracter.c b/utils/url_extracter.c
index 8cd8498df..eadf922c2 100644
--- a/utils/url_extracter.c
+++ b/utils/url_extracter.c
@@ -27,6 +27,7 @@ mime_foreach_callback (GMimeObject *part, gpointer user_data)
 	GMimeDataWrapper *wrapper;
 	GMimeStream *part_stream;
 	GByteArray *part_content;
+	GMimeMessage *message;
 	
 	/* 'part' points to the current part node that g_mime_message_foreach_part() is iterating over */
 	
@@ -34,7 +35,6 @@ mime_foreach_callback (GMimeObject *part, gpointer user_data)
 	if (GMIME_IS_MESSAGE_PART (part)) {
 		/* message/rfc822 or message/news */
 		printf ("Message part found\n");
-		GMimeMessage *message;
 		
 		/* g_mime_message_foreach_part() won't descend into
                    child message parts, so if we want to count any