* Add filter processing function which implements new filter's logic

4 files changed, 182 insertions, 13 deletions

diff --git a/cfg_file.h b/cfg_file.h
index 6e93a5f44..6eb441c71 100644
--- a/cfg_file.h
+++ b/cfg_file.h
@@ -50,13 +50,6 @@
 
 enum { VAL_UNDEF=0, VAL_TRUE, VAL_FALSE };
 
-enum script_type {
-	SCRIPT_HEADER,
-	SCRIPT_MIME,
-	SCRIPT_URL,
-	SCRIPT_MESSAGE,
-};
-
 struct memcached_server {
 	struct upstream up;
 	struct in_addr addr;
diff --git a/filter.c b/filter.c
index 9130a9e9c..a69a9efa9 100644
--- a/filter.c
+++ b/filter.c
@@ -7,6 +7,7 @@
 #include "filter.h"
 #include "main.h"
 #include "cfg_file.h"
+#include "perl.h"
 
 void
 insert_result (struct worker_task *task, const char *metric_name, const char *symbol, u_char flag)
@@ -39,13 +40,13 @@ insert_result (struct worker_task *task, const char *metric_name, const char *sy
 /*
  * Default consolidation function based on factors in config file
  */
-int
+double
 factor_consolidation_func (struct worker_task *task, const char *metric_name)
 {
 	struct metric_result *metric_res;
 	struct filter_result *result;
 	double *factor;
-	int res = 0;
+	double res = 0.;
 
 	metric_res = g_hash_table_lookup (task->results, metric_name);
 	if (metric_res == NULL) {
@@ -68,8 +69,174 @@ factor_consolidation_func (struct worker_task *task, const char *metric_name)
 	return res;
 }
 
+/* 
+ * Call perl or C module function for specified part of message 
+ */
+static void
+call_filter_by_name (struct worker_task *task, const char *name, enum script_type sc_type, enum filter_type filt_type)
+{
+	struct module_ctx *c_module;
+
+	switch (filt_type) {
+		case C_FILTER:
+			c_module = g_hash_table_lookup (task->worker->srv->cfg->c_modules, name);
+			if (c_module) {
+				switch (filt_type) {
+					case SCRIPT_HEADER:
+						c_module->header_filter (task);
+						break;
+					case SCRIPT_MIME:
+						c_module->mime_filter (task);
+						break;
+					case SCRIPT_URL:
+						c_module->url_filter (task);
+						break;
+					case SCRIPT_MESSAGE:
+						c_module->message_filter (task);
+						break;
+				}
+			}
+			break;
+		case PERL_FILTER:
+			switch (filt_type) {
+				case SCRIPT_HEADER:
+					perl_call_header_filter (name, task);
+					break;
+				case SCRIPT_MIME:
+					perl_call_mime_filter (name, task);
+					break;
+				case SCRIPT_URL:
+					perl_call_url_filter (name, task);
+					break;
+				case SCRIPT_MESSAGE:
+					perl_call_message_filter (name, task);
+					break;
+			}
+			break;
+	}
+}
+
+static void
+metric_process_callback (gpointer key, gpointer value, void *data)
+{
+	struct worker_task *task = (struct worker_task *)data;
+	struct metric_result *metric_res = (struct metric_result *)value;
+	
+	if (metric_res->metric->func != NULL) {
+		metric_res->score = metric_res->metric->func (task, metric_res->metric->name);
+	}
+	else {
+		metric_res->score = factor_consolidation_func (task, metric_res->metric->name);
+	}
+}
+
+static int
+continue_process_filters (struct worker_task *task)
+{
+	struct filter *cur = task->save.entry;
+	
+	cur = LIST_NEXT (cur, next);
+	/* Note: no breaks in this case! */
+	switch (task->save.type) {
+		case SCRIPT_HEADER:
+			while (cur) {
+				call_filter_by_name (task, cur->func_name, cur->type, SCRIPT_HEADER);
+				if (task->save.saved) {
+					task->save.entry = cur;
+					task->save.type = SCRIPT_HEADER;
+					return 0;
+				}
+				cur = LIST_NEXT (cur, next);
+			}
+			/* Process mime filters */
+			cur = LIST_FIRST (&task->worker->srv->cfg->mime_filters);
+		case SCRIPT_MIME:
+			while (cur) {
+				call_filter_by_name (task, cur->func_name, cur->type, SCRIPT_MIME);
+				if (task->save.saved) {
+					task->save.entry = cur;
+					task->save.type = SCRIPT_MIME;
+					return 0;
+				}
+				cur = LIST_NEXT (cur, next);
+			}
+			/* Process url filters */
+			cur = LIST_FIRST (&task->worker->srv->cfg->url_filters);
+		case SCRIPT_URL:
+			while (cur) {
+				call_filter_by_name (task, cur->func_name, cur->type, SCRIPT_URL);
+				if (task->save.saved) {
+					task->save.entry = cur;
+					task->save.type = SCRIPT_URL;
+					return 0;
+				}
+				cur = LIST_NEXT (cur, next);
+			}
+			/* Process message filters */
+			cur = LIST_FIRST (&task->worker->srv->cfg->message_filters);
+		case SCRIPT_MESSAGE:
+			while (cur) {
+				call_filter_by_name (task, cur->func_name, cur->type, SCRIPT_MESSAGE);
+				if (task->save.saved) {
+					task->save.entry = cur;
+					task->save.type = SCRIPT_MESSAGE;
+					return 0;
+				}
+				cur = LIST_NEXT (cur, next);
+			}
+			/* All done */
+			return 1;
+	}
+}
+
 int 
 process_filters (struct worker_task *task)
 {
-	/* TODO: Implement new logic of filters chains */
+	struct filter *cur;
+
+	if (task->save.saved) {
+		task->save.saved = 0;
+		return continue_process_filters (task);
+	}
+
+	/* Process filters in order that they are listed in config file */
+	LIST_FOREACH (cur, &task->worker->srv->cfg->header_filters, next) {
+		call_filter_by_name (task, cur->func_name, cur->type, SCRIPT_HEADER);
+		if (task->save.saved) {
+			task->save.entry = cur;
+			task->save.type = SCRIPT_HEADER;
+			return 0;
+		}
+	}
+
+	LIST_FOREACH (cur, &task->worker->srv->cfg->mime_filters, next) {
+		call_filter_by_name (task, cur->func_name, cur->type, SCRIPT_MIME);
+		if (task->save.saved) {
+			task->save.entry = cur;
+			task->save.type = SCRIPT_MIME;
+			return 0;
+		}
+	}
+
+	LIST_FOREACH (cur, &task->worker->srv->cfg->url_filters, next) {
+		call_filter_by_name (task, cur->func_name, cur->type, SCRIPT_URL);
+		if (task->save.saved) {
+			task->save.entry = cur;
+			task->save.type = SCRIPT_URL;
+			return 0;
+		}
+	}
+
+	LIST_FOREACH (cur, &task->worker->srv->cfg->message_filters, next) {
+		call_filter_by_name (task, cur->func_name, cur->type, SCRIPT_MESSAGE);
+		if (task->save.saved) {
+			task->save.entry = cur;
+			task->save.type = SCRIPT_MESSAGE;
+			return 0;
+		}
+	}
+
+	/* Process all metrics */
+	g_hash_table_foreach (task->results, metric_process_callback, task);
+	return 1;
 }
diff --git a/filter.h b/filter.h
index 9f0a96da9..d7f494dc6 100644
--- a/filter.h
+++ b/filter.h
@@ -11,7 +11,7 @@
 
 struct worker_task;
 
-typedef int (*metric_cons_func)(struct worker_task *task, const char *metric_name);
+typedef double (*metric_cons_func)(struct worker_task *task, const char *metric_name);
 typedef void (*filter_func)(struct worker_task *task);
 
 enum filter_type { C_FILTER, PERL_FILTER };
@@ -37,11 +37,12 @@ struct filter_result {
 
 struct metric_result {
 	struct metric *metric;
+	double score;
 	LIST_HEAD (resultq, filter_result) results;
 };
 
 int process_filters (struct worker_task *task);
 void insert_result (struct worker_task *task, const char *metric_name, const char *symbol, u_char flag);
-int factor_consolidation_func (struct worker_task *task, const char *metric_name);
+double factor_consolidation_func (struct worker_task *task, const char *metric_name);
 
 #endif
diff --git a/main.h b/main.h
index 94eceebcf..5877d12d0 100644
--- a/main.h
+++ b/main.h
@@ -48,6 +48,14 @@ enum process_type {
 	TYPE_WORKER,
 };
 
+/* Filter type */
+enum script_type {
+	SCRIPT_HEADER,
+	SCRIPT_MIME,
+	SCRIPT_URL,
+	SCRIPT_MESSAGE,
+};
+
 /* Worker process structure */
 struct rspamd_worker {
 	pid_t pid;
@@ -83,7 +91,7 @@ struct mime_part {
 
 struct save_point {
 	void *entry;
-	void *chain;
+	enum script_type type;
 	unsigned int saved;
 };