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

diff --git a/cfg_file.h b/cfg_file.h
index 6e93a5f44bd8bcda1233a63d0d89d24e5df455a2..6eb441c710dc608c7e248184e2760d321d78354c 100644 (file)
--- 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 9130a9e9c7f26875ba2bd4b826bb053d5eea422f..a69a9efa907ef459dba3b513ea0f516db73d14a7 100644 (file)
--- 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 9f0a96da97d3f9728fc32874a2d34bf674ae9ddf..d7f494dc64dd881b63fbf98e71638a7889c9d098 100644 (file)
--- 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 94eceebcfb7e5a7cbf3a80ca5a9cc0b3d5100574..5877d12d0fc94ee05e9434293ef2b707b1a25421 100644 (file)
--- 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;
 };