* New rspamd protocol (changed replies)

* Add ability to attach string lists to symbols
* Check destructors and do not add identical destructors (argument and function) to pool's destructors
* Remove 2 warnings when building with gmime22
* Attach url names to surbl symbols
* Fix bug with blocking on read (I think it is linux specific thought)

10 files changed, 173 insertions, 93 deletions

diff --git a/perl/Rspamd/Message.xs b/perl/Rspamd/Message.xs
index e8f0571bc..c9a21886e 100644
--- a/perl/Rspamd/Message.xs
+++ b/perl/Rspamd/Message.xs
@@ -65,7 +65,7 @@ rspamd_message_get_recipients(message, type)
 	CODE:
 		retav = newAV();
 #ifndef GMIME24
-		rcpt = g_mime_message_get_recipients (message, type);
+		rcpt = (InternetAddressList *)g_mime_message_get_recipients (message, type);
 		while (rcpt) {
 		  SV * address = newSViv(0);
 		  sv_setref_pv(address, "Mail::Rspamd::InternetAddress", (Mail__Rspamd__InternetAddress)(rcpt->address));
diff --git a/perl/Rspamd/Object.xs b/perl/Rspamd/Object.xs
index cf5fffb11..2f56fb995 100644
--- a/perl/Rspamd/Object.xs
+++ b/perl/Rspamd/Object.xs
@@ -15,7 +15,7 @@ rspamd_object_get_content_type (mime_object)
 		char *			textdata;
 		GMimeContentType	*ct;
 	CODE:
-		ct = g_mime_object_get_content_type (mime_object);
+		ct = (GMimeContentType *)g_mime_object_get_content_type (mime_object);
 		textdata = g_mime_content_type_to_string (ct);
 		RETVAL = g_mime_content_type_new_from_string (textdata);
 		plist = g_list_prepend (plist, RETVAL);
diff --git a/perl/Rspamd/Task.xs b/perl/Rspamd/Task.xs
index dec521660..16719ef7b 100644
--- a/perl/Rspamd/Task.xs
+++ b/perl/Rspamd/Task.xs
@@ -55,7 +55,7 @@ rspamd_task_insert_result (task, metric, symbol, flag)
 		const char *symbol
 		double flag
 	CODE:
-		insert_result (task, metric, symbol, flag);
+		insert_result (task, metric, symbol, flag, NULL);
 
 Mail::Rspamd::Conf
 rspamd_task_get_conf (task)
diff --git a/src/buffer.c b/src/buffer.c
index 8883e314d..5864da382 100644
--- a/src/buffer.c
+++ b/src/buffer.c
@@ -122,7 +122,7 @@ write_buffers (int fd, rspamd_io_dispatcher_t *d)
 }
 
 static void
-read_buffers (int fd, rspamd_io_dispatcher_t *d)
+read_buffers (int fd, rspamd_io_dispatcher_t *d, gboolean skip_read)
 {
 	ssize_t r;
 	GError *err;
@@ -150,7 +150,7 @@ read_buffers (int fd, rspamd_io_dispatcher_t *d)
 			return;
 		}
 	}
-	else {
+	else if (!skip_read) {
 		/* Try to read the whole buffer */
 		r = read (fd, d->in_buf->pos, BUFREMAIN (d->in_buf));
 		if (r == -1 && errno != EAGAIN) {
@@ -213,7 +213,7 @@ read_buffers (int fd, rspamd_io_dispatcher_t *d)
 						len = d->in_buf->data->len;
 						if (d->policy != saved_policy) {
 							msg_debug ("read_buffers: policy changed during callback, restart buffer's processing");
-							read_buffers (fd, d);
+							read_buffers (fd, d, TRUE);
 							return;
 						}
 						continue;
@@ -239,7 +239,7 @@ read_buffers (int fd, rspamd_io_dispatcher_t *d)
 						len = d->in_buf->data->len;
 						if (d->policy != saved_policy) {
 							msg_debug ("read_buffers: policy changed during callback, restart buffer's processing");
-							read_buffers (fd, d);
+							read_buffers (fd, d, TRUE);
 							return;
 						}
 						continue;
@@ -282,7 +282,7 @@ dispatcher_cb (int fd, short what, void *arg)
 			}
 			break;
 		case EV_READ:
-			read_buffers (fd, d);
+			read_buffers (fd, d, FALSE);
 			break;
 	}
 }
diff --git a/src/filter.c b/src/filter.c
index 56d7d9945..8e0569e6f 100644
--- a/src/filter.c
+++ b/src/filter.c
@@ -38,12 +38,11 @@
 #include "tokenizers/tokenizers.h"
 
 void
-insert_result (struct worker_task *task, const char *metric_name, const char *symbol, double flag)
+insert_result (struct worker_task *task, const char *metric_name, const char *symbol, double flag, GList *opts)
 {
 	struct metric *metric;
 	struct metric_result *metric_res;
-	double *fl = memory_pool_alloc (task->task_pool, sizeof (double));
-	*fl = flag;
+	struct symbol *s;
 
 	metric = g_hash_table_lookup (task->worker->srv->cfg->metrics, metric_name);
 	if (metric == NULL) {
@@ -61,7 +60,33 @@ insert_result (struct worker_task *task, const char *metric_name, const char *sy
 		g_hash_table_insert (task->results, (gpointer)metric_name, metric_res);
 	}
 	
-	g_hash_table_insert (metric_res->symbols, (gpointer)symbol, fl);
+	if ((s = g_hash_table_lookup (metric_res->symbols, symbol)) != NULL) {
+		if (s->options && opts) {
+			/* Append new options */
+			s->options = g_list_concat (s->options, opts);
+			/* 
+			 * Note that there is no need to add new destructor of GList as elements of appended
+			 * GList are used directly, so just free initial GList
+			 */
+		}
+		else if (opts) {
+			s->options = opts;
+			memory_pool_add_destructor (task->task_pool, (pool_destruct_func)g_list_free, s->options);
+		}
+
+		s->score = flag;
+	}
+	else {
+		s = memory_pool_alloc (task->task_pool, sizeof (struct symbol));
+		s->score = flag;
+		s->options = opts;
+
+		if (opts) {
+			memory_pool_add_destructor (task->task_pool, (pool_destruct_func)g_list_free, s->options);
+		}
+
+		g_hash_table_insert (metric_res->symbols, (gpointer)symbol, s);
+	}
 }
 
 /*
@@ -76,17 +101,17 @@ static void
 consolidation_callback (gpointer key, gpointer value, gpointer arg)
 {
 	double *factor;
-	double val = *(double *)value;
+	struct symbol *s = (struct symbol *)value;
 	struct consolidation_callback_data *data = (struct consolidation_callback_data *)arg;
 	
 	factor = g_hash_table_lookup (data->task->worker->srv->cfg->factors, key);
 	if (factor == NULL) {
-		msg_debug ("consolidation_callback: got %.2f score for metric %s, factor: 1", val, (char *)key);
-		data->score += val;
+		msg_debug ("consolidation_callback: got %.2f score for metric %s, factor: 1", s->score, (char *)key);
+		data->score += s->score;
 	}
 	else {
-		data->score += *factor * val;
-		msg_debug ("consolidation_callback: got %.2f score for metric %s, factor: %.2f", val, (char *)key, *factor);
+		data->score += *factor * s->score;
+		msg_debug ("consolidation_callback: got %.2f score for metric %s, factor: %.2f", s->score, (char *)key, *factor);
 	}
 }
 
@@ -232,8 +257,7 @@ continue_process_filters (struct worker_task *task)
 				}
 				cur = LIST_NEXT (cur, next);
 			}
-			/* Process all metrics */
-			g_hash_table_foreach (task->results, metric_process_callback, task);
+			/* Process all statfiles */
 			process_statfiles (task);
 			/* XXX: ugly direct call */
 			task->dispatcher->write_callback (task);
@@ -306,7 +330,7 @@ composites_foreach_callback (gpointer key, gpointer value, void *data)
 	GQueue *stack;
 	GList *symbols = NULL, *s;
 	gsize cur, op1, op2;
-	double *res;
+	struct symbol *res;
 	
 	stack = g_queue_new ();
 
@@ -360,8 +384,9 @@ composites_foreach_callback (gpointer key, gpointer value, void *data)
 				s = g_list_next (s);
 			}
 			/* Add new symbol */
-			res = memory_pool_alloc (cd->task->task_pool, sizeof (double));
-			*res = 1;
+			res = memory_pool_alloc (cd->task->task_pool, sizeof (struct symbol));
+			res->score = 1.;
+			res->options = NULL;
 			g_hash_table_insert (cd->metric_res->symbols, key, res);
 		}
 	}
@@ -473,7 +498,7 @@ statfiles_results_callback (gpointer key, gpointer value, void *arg)
 
 	w = memory_pool_alloc (task->task_pool, sizeof (double));
 	filename = classifier->result_file_func (res->ctx, w);
-	insert_result (task, res->metric->name, classifier->name, *w);
+	insert_result (task, res->metric->name, classifier->name, *w, NULL);
 	msg_debug ("statfiles_results_callback: got total weight %.2f for metric %s", *w, res->metric->name);
 
 }
diff --git a/src/filter.h b/src/filter.h
index c0b113dbd..02fc60c63 100644
--- a/src/filter.h
+++ b/src/filter.h
@@ -25,6 +25,14 @@ struct filter {
 };
 
 /**
+ * Rspamd symbol
+ */
+struct symbol {
+	double score;									/**< symbol's score					*/
+	GList *options;									/**< list of symbol's options		*/
+};
+
+/**
  * Common definition of metric
  */
 struct metric {
@@ -63,8 +71,9 @@ void process_statfiles (struct worker_task *task);
  * @param metric_name metric's name to which we need to insert result
  * @param symbol symbol to insert
  * @param flag numeric weight for symbol
+ * @param opts list of symbol's options
  */
-void insert_result (struct worker_task *task, const char *metric_name, const char *symbol, double flag);
+void insert_result (struct worker_task *task, const char *metric_name, const char *symbol, double flag, GList *opts);
 
 /**
  * Process all results and form composite metrics from existent metrics as it is defined in config
diff --git a/src/mem_pool.c b/src/mem_pool.c
index 23f25b5f3..db8e3f351 100644
--- a/src/mem_pool.c
+++ b/src/mem_pool.c
@@ -331,10 +331,20 @@ void memory_pool_unlock_shared (memory_pool_t *pool, void *pointer)
 void
 memory_pool_add_destructor (memory_pool_t *pool, pool_destruct_func func, void *data)
 {
-	struct _pool_destructors *cur;
+	struct _pool_destructors *cur, *tmp;
 
 	cur = memory_pool_alloc (pool, sizeof (struct _pool_destructors));
 	if (cur) {
+		/* Check whether we have identical destructor in pool */
+		tmp = pool->destructors;
+		while (tmp) {
+			if (tmp->func == func && tmp->data == data) {
+				/* Do not add identical destructors, they must be unique */
+				return;
+			}
+			tmp = tmp->prev;
+		}
+
 		cur->func = func;
 		cur->data = data;
 		cur->prev = pool->destructors;
diff --git a/src/plugins/regexp.c b/src/plugins/regexp.c
index 8606be77c..a2079bffa 100644
--- a/src/plugins/regexp.c
+++ b/src/plugins/regexp.c
@@ -249,7 +249,7 @@ process_regexp_item (struct regexp_module_item *item, struct worker_task *task)
 		op1 = GPOINTER_TO_SIZE (g_queue_pop_head (stack));
 		if (op1) {
 			/* Add symbol to results */
-			insert_result (task, regexp_module_ctx->metric, item->symbol, op1);
+			insert_result (task, regexp_module_ctx->metric, item->symbol, op1, NULL);
 		}
 	}
 
diff --git a/src/plugins/surbl.c b/src/plugins/surbl.c
index 44b5d3fb8..1e466719c 100644
--- a/src/plugins/surbl.c
+++ b/src/plugins/surbl.c
@@ -325,7 +325,7 @@ make_surbl_requests (struct uri* url, struct worker_task *task)
 }
 
 static void
-process_dns_results (struct worker_task *task, struct suffix_item *suffix, uint32_t addr)
+process_dns_results (struct worker_task *task, struct suffix_item *suffix, char *url, uint32_t addr)
 {
 	char *c, *symbol;
 	GList *cur;
@@ -345,18 +345,21 @@ process_dns_results (struct worker_task *task, struct suffix_item *suffix, uint3
 				symbol = memory_pool_alloc (task->task_pool, len);
 				snprintf (symbol, len, "%s%s%s", suffix->symbol, bit->symbol, c + 2);
 				*c = '%';
-				insert_result (task, surbl_module_ctx->metric, symbol, 1);
+				insert_result (task, surbl_module_ctx->metric, symbol, 1, 
+							g_list_prepend (NULL, memory_pool_strdup (task->task_pool, url)));
 				found = 1;
 			}
 			cur = g_list_next (cur);
 		}
 
 		if (!found) {
-			insert_result (task, surbl_module_ctx->metric, suffix->symbol, 1);
+			insert_result (task, surbl_module_ctx->metric, symbol, 1, 
+							g_list_prepend (NULL, memory_pool_strdup (task->task_pool, url)));
 		}
 	}
 	else {
-		insert_result (task, surbl_module_ctx->metric, suffix->symbol, 1);
+		insert_result (task, surbl_module_ctx->metric, symbol, 1, 
+							g_list_prepend (NULL, memory_pool_strdup (task->task_pool, url)));
 	}
 }
 
@@ -372,7 +375,7 @@ dns_callback (int result, char type, int count, int ttl, void *addresses, void *
 	/* If we have result from DNS server, this url exists in SURBL, so increase score */
 	if (result == DNS_ERR_NONE && type == DNS_IPv4_A) {
 		msg_info ("surbl_check: url %s is in surbl %s", param->url->host, param->suffix->suffix);
-		process_dns_results (param->task, param->suffix, (uint32_t)(((in_addr_t *)addresses)[0]));
+		process_dns_results (param->task, param->suffix, param->url->host, (uint32_t)(((in_addr_t *)addresses)[0]));
 	}
 	else {
 		msg_debug ("surbl_check: url %s is not in surbl %s", param->url->host, param->suffix->suffix);
diff --git a/src/protocol.c b/src/protocol.c
index 9999dc279..598c851c6 100644
--- a/src/protocol.c
+++ b/src/protocol.c
@@ -22,7 +22,9 @@
  * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  */
 
+#include "config.h"
 #include "main.h"
+#include "cfg_file.h"
 
 /* Max line size as it is defined in rfc2822 */
 #define OUTBUFSIZ 1000
@@ -348,112 +350,145 @@ show_url_header (struct worker_task *task)
 }
 
 static void
-show_metric_result (gpointer metric_name, gpointer metric_value, void *user_data)
+metric_symbols_callback (gpointer key, gpointer value, void *user_data)
 {
 	struct worker_task *task = (struct worker_task *)user_data;
-	int r;
+	int r = 0;
 	char outbuf[OUTBUFSIZ];
-	struct metric_result *metric_res = (struct metric_result *)metric_value;
-	int is_spam = 0;
+	struct symbol *s = (struct symbol *)value;	
+	GList *cur;
 
-	if (metric_res->score >= metric_res->metric->required_score) {
-		is_spam = 1;
-	}
-	if (task->proto == SPAMC_PROTO) {
-		r = snprintf (outbuf, sizeof (outbuf), "Spam: %s ; %.2f / %.2f" CRLF,
-					(is_spam) ? "True" : "False", metric_res->score, metric_res->metric->required_score);
+	if (s->options) {
+		r = snprintf (outbuf, OUTBUFSIZ, "Symbol: %s; ", (char *)key);
+		cur = s->options;
+		while (cur) {
+			if (g_list_next (cur)) {
+				r += snprintf (outbuf + r, OUTBUFSIZ - r, "%s,", (char *)cur->data);
+			}
+			else {
+				r += snprintf (outbuf + r, OUTBUFSIZ - r, "%s" CRLF, (char *)cur->data);
+			}
+			cur = g_list_next (cur);
+		}
+		/* End line with CRLF strictly */
+		if (r >= OUTBUFSIZ - 1) {
+			outbuf[OUTBUFSIZ - 2] = '\r';
+			outbuf[OUTBUFSIZ - 1] = '\n';
+		}
 	}
 	else {
-		r = snprintf (outbuf, sizeof (outbuf), "%s: %s ; %.2f / %.2f" CRLF, (char *)metric_name,
-					(is_spam) ? "True" : "False", metric_res->score, metric_res->metric->required_score);
+		r = snprintf (outbuf, OUTBUFSIZ, "Symbol: %s" CRLF, (char *)key);
 	}
+
 	rspamd_dispatcher_write (task->dispatcher, outbuf, r, FALSE);
 }
 
-static int
-write_check_reply (struct worker_task *task)
+static void
+show_metric_symbols (struct metric_result *metric_res, struct worker_task *task)
 {
-	int r;
+	int r = 0;
+	GList *symbols, *cur;
 	char outbuf[OUTBUFSIZ];
-	struct metric_result *metric_res;
 
-	r = snprintf (outbuf, sizeof (outbuf), "%s 0 %s" CRLF, (task->proto == SPAMC_PROTO) ? SPAMD_REPLY_BANNER : RSPAMD_REPLY_BANNER, "OK");
-	rspamd_dispatcher_write (task->dispatcher, outbuf, r, TRUE);
 	if (task->proto == SPAMC_PROTO) {
-		/* Ignore metrics, just write report for 'default' metric */
-		metric_res = g_hash_table_lookup (task->results, "default");
-		if (metric_res == NULL) {
-			return -1;
-		}
-		else {
-			show_metric_result ((gpointer)"default", (gpointer)metric_res, (void *)task);
+		symbols = g_hash_table_get_keys (metric_res->symbols);
+		cur = symbols;
+		while (cur) {
+			if (g_list_next (cur) != NULL) {
+				r += snprintf (outbuf + r, sizeof (outbuf) - r, "%s,", (char *)cur->data);
+			}
+			else {
+				r += snprintf (outbuf + r, sizeof (outbuf) - r, "%s" CRLF, (char *)cur->data);
+			}
+			cur = g_list_next (cur);
 		}
+		g_list_free (symbols);
+		rspamd_dispatcher_write (task->dispatcher, outbuf, r, FALSE);
 	}
 	else {
-		/* Write result for each metric separately */
-		g_hash_table_foreach (task->results, show_metric_result, task);
-		/* URL stat */
-		show_url_header (task);
+		g_hash_table_foreach (metric_res->symbols, metric_symbols_callback, task);
 	}
-	rspamd_dispatcher_write (task->dispatcher, CRLF, sizeof (CRLF) - 1, FALSE);
-
-	return 0;
 }
 
 static void
-show_metric_symbols (gpointer metric_name, gpointer metric_value, void *user_data)
+show_metric_result (gpointer metric_name, gpointer metric_value, void *user_data)
 {
 	struct worker_task *task = (struct worker_task *)user_data;
-	int r = 0;
+	int r;
 	char outbuf[OUTBUFSIZ];
-	GList *symbols = NULL, *cur;
 	struct metric_result *metric_res = (struct metric_result *)metric_value;
-
-	if (task->proto == RSPAMC_PROTO) {
-		r = snprintf (outbuf, sizeof (outbuf), "%s: ", (char *)metric_name);
+	struct metric *m;
+	int is_spam = 0;
+	
+	if (metric_name == NULL || metric_value == NULL) {
+		m = g_hash_table_lookup (task->cfg->metrics, "default");
+		if (task->proto == SPAMC_PROTO) {
+			r = snprintf (outbuf, sizeof (outbuf), "Spam: False ; 0 / %.2f" CRLF,
+						m != NULL ? m->required_score : 0);
+		}
+		else {
+			r = snprintf (outbuf, sizeof (outbuf), "Metric: default; False; 0 / %.2f" CRLF,
+						m != NULL ? m->required_score : 0);
+		}
 	}
-
-	symbols = g_hash_table_get_keys (metric_res->symbols);
-	cur = symbols;
-	while (cur) {
-		if (g_list_next (cur) != NULL) {
-			r += snprintf (outbuf + r, sizeof (outbuf) - r, "%s,", (char *)cur->data);
+	else {
+		if (metric_res->score >= metric_res->metric->required_score) {
+			is_spam = 1;
+		}
+		if (task->proto == SPAMC_PROTO) {
+			r = snprintf (outbuf, sizeof (outbuf), "Spam: %s ; %.2f / %.2f" CRLF,
+						(is_spam) ? "True" : "False", metric_res->score, metric_res->metric->required_score);
 		}
 		else {
-			r += snprintf (outbuf + r, sizeof (outbuf) - r, "%s", (char *)cur->data);
+			r = snprintf (outbuf, sizeof (outbuf), "Metric: %s; %s; %.2f / %.2f" CRLF, (char *)metric_name,
+						(is_spam) ? "True" : "False", metric_res->score, metric_res->metric->required_score);
 		}
-		cur = g_list_next (cur);
 	}
-	g_list_free (symbols);
-	msg_debug ("show_metric_symbols: write symbols line: %s", outbuf);
-	outbuf[r++] = '\r'; outbuf[r++] = '\n';
 	rspamd_dispatcher_write (task->dispatcher, outbuf, r, FALSE);
+
+	if (task->cmd == CMD_SYMBOLS && metric_value != NULL) {
+		show_metric_symbols (metric_res, task);
+	}
 }
 
 static int
-write_symbols_reply (struct worker_task *task)
+write_check_reply (struct worker_task *task)
 {
+	int r;
+	char outbuf[OUTBUFSIZ];
 	struct metric_result *metric_res;
-	
-	/* First of all write normal results by calling write_check_reply */
-	if (write_check_reply (task) == -1) {
-		return -1;
-	}
-	/* Now write symbols */
+
+	r = snprintf (outbuf, sizeof (outbuf), "%s 0 %s" CRLF, (task->proto == SPAMC_PROTO) ? SPAMD_REPLY_BANNER : RSPAMD_REPLY_BANNER, "OK");
+	rspamd_dispatcher_write (task->dispatcher, outbuf, r, TRUE);
 	if (task->proto == SPAMC_PROTO) {
 		/* Ignore metrics, just write report for 'default' metric */
 		metric_res = g_hash_table_lookup (task->results, "default");
 		if (metric_res == NULL) {
-			return -1;
+			/* Implicit metric result */
+			show_metric_result (NULL, NULL, (void *)task);
 		}
 		else {
-			show_metric_symbols ((gpointer)"default", (gpointer)metric_res, (void *)task);
+			show_metric_result ((gpointer)"default", (gpointer)metric_res, (void *)task);
 		}
 	}
 	else {
+		/* Show default metric first */
+		metric_res = g_hash_table_lookup (task->results, "default");
+		if (metric_res == NULL) {
+			/* Implicit metric result */
+			show_metric_result (NULL, NULL, (void *)task);
+		}
+		else {
+			show_metric_result ((gpointer)"default", (gpointer)metric_res, (void *)task);
+		}
+		g_hash_table_remove (task->results, "default");
+
 		/* Write result for each metric separately */
-		g_hash_table_foreach (task->results, show_metric_symbols, task);
+		g_hash_table_foreach (task->results, show_metric_result, task);
+		/* URL stat */
+		show_url_header (task);
 	}
+	rspamd_dispatcher_write (task->dispatcher, CRLF, sizeof (CRLF) - 1, FALSE);
 
 	return 0;
 }
@@ -498,10 +533,8 @@ write_reply (struct worker_task *task)
 			case CMD_REPORT_IFSPAM:
 			case CMD_REPORT:
 			case CMD_CHECK:
-				return write_check_reply (task);
-				break;
 			case CMD_SYMBOLS:
-				return write_symbols_reply (task);
+				return write_check_reply (task);
 				break;
 			case CMD_PROCESS:
 				return write_process_reply (task);