ELAPI Adding file provider and CSV format

This patch creates the infrastructure for
logging of the event from the top of the interface
to the bottom. It is a start. A lot of functionality
is left aside.
The attempt of this patch is pass event from caller
of the ELAPI interface via targets to sinks
then to providers and do serialization creating
entity that is ready to be written to a file.
It also implements more specific provider related
configuration parameters.
Also it addresses couple suggestions that were
brought up against previous patch.

ELAPI Correcting issues

This patch addresses the issues found during the
review of the previous patches and addresses
ticket #166.

6 files changed, 1768 insertions, 45 deletions

diff --git a/common/elapi/providers/file/file_fmt_csv.c b/common/elapi/providers/file/file_fmt_csv.c
new file mode 100644
index 000000000..a81111330
--- /dev/null
+++ b/common/elapi/providers/file/file_fmt_csv.c
@@ -0,0 +1,536 @@
+/*
+    ELAPI
+
+    Module contains functions related to outputting events in CSV format.
+
+    Copyright (C) Dmitri Pal <dpal@redhat.com> 2009
+
+    This program is free software; you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation; either version 3 of the License, or
+    (at your option) any later version.
+    This program is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+    You should have received a copy of the GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#define _GNU_SOURCE
+#include <errno.h>      /* for errors */
+#include <stdlib.h>     /* for free() */
+#include <string.h>     /* for strcmp() */
+
+#include "collection.h"
+#include "file_fmt_csv.h"
+#include "collection_tools.h"
+#include "ini_config.h"
+#include "trace.h"
+#include "config.h"
+
+/* Reasonable size for one event */
+/* FIXME: may be it would make sense to make it configurable ? */
+#define FILE_CSV_BLOCK      256
+
+/* Calculate the potential size of the item */
+static unsigned file_csv_data_len(struct file_csv_cfg *cfg,
+                                  int type,
+                                  int raw_len)
+{
+    int serialized_len = 0;
+
+    TRACE_FLOW_STRING("col_get_data_len", "Entry point");
+
+    switch (type) {
+    case COL_TYPE_INTEGER:
+    case COL_TYPE_UNSIGNED:
+    case COL_TYPE_LONG:
+    case COL_TYPE_ULONG:
+        serialized_len = MAX_LONG_STRING_LEN;
+        break;
+
+    case COL_TYPE_STRING:
+        if ((cfg->csvqualifier) &&
+            (cfg->csvescchar)) serialized_len = raw_len * 2;
+        else serialized_len = raw_len;
+        break;
+
+    case COL_TYPE_BINARY:
+        serialized_len = raw_len * 2;
+        break;
+
+    case COL_TYPE_DOUBLE:
+        serialized_len = MAX_DOUBLE_STRING_LEN;
+        break;
+
+    case COL_TYPE_BOOL:
+        serialized_len = MAX_BOOL_STRING_LEN;
+        break;
+
+    default:
+        serialized_len = 0;
+        break;
+    }
+
+    if (cfg->csvqualifier) serialized_len += 2;
+
+    TRACE_FLOW_STRING("col_get_data_len","Exit point");
+    return (uint32_t)serialized_len;
+}
+
+/* Copy data escaping characters */
+int file_copy_esc(char *dest,
+                  const char *source,
+                  unsigned char what_to_esc,
+                  unsigned char what_to_use)
+{
+    int i = 0;
+    int j = 0;
+
+    while (source[i]) {
+        if ((source[i] == what_to_use) ||
+            (source[i] == what_to_esc)) {
+
+            dest[j] = what_to_use;
+            j++;
+
+        }
+        dest[j] = source[i];
+        i++;
+        j++;
+    }
+
+    return j;
+}
+
+/* Serialize item into the csv format */
+int file_serialize_csv(struct elapi_data_out *out_data,
+                       int type,
+                       int length,
+                       void *data,
+                       void *mode_cfg)
+{
+    int error = EOK;
+    struct file_csv_cfg *cfg;
+    uint32_t projected_len;
+    uint32_t used_len;
+    int first = 1;
+    int i;
+
+    TRACE_FLOW_STRING("file_serialize_csv", "Entry");
+
+    cfg = (struct file_csv_cfg *)mode_cfg;
+
+    /* Get projected length of the item */
+    projected_len = file_csv_data_len(cfg, type, length);
+
+    TRACE_INFO_NUMBER("Expected data length: ", projected_len);
+
+    /* Make sure we have enough space */
+    if (out_data->buffer != NULL) {
+        TRACE_INFO_STRING("Not a first time use.", "Adding length overhead");
+        if (cfg->csvseparator) projected_len++;
+        projected_len += cfg->csvnumsp;
+        first = 0;
+    }
+    else {
+        /* Add null terminating zero */
+        projected_len++;
+    }
+
+    /* Grow buffer if needed */
+    error = elapi_grow_data(out_data,
+                            projected_len,
+                            FILE_CSV_BLOCK);
+    if (error) {
+        TRACE_ERROR_NUMBER("Error. Failed to allocate memory.", error);
+        return error;
+    }
+
+    /* Now everything should fit */
+    if (!first) {
+        /* Add separator if any */
+        if (cfg->csvseparator) {
+            out_data->buffer[out_data->length] = cfg->csvseparator;
+            out_data->length++;
+        }
+
+        /* Add spaces if any */
+        memset(&out_data->buffer[out_data->length],
+               cfg->csvspace,
+               cfg->csvnumsp);
+    }
+
+    /* Add qualifier */
+    if (cfg->csvqualifier) {
+        out_data->buffer[out_data->length] = cfg->csvqualifier;
+        out_data->length++;
+    }
+
+    /* Add the value */
+    switch (type) {
+    case COL_TYPE_STRING:
+
+        if ((cfg->csvqualifier) && (cfg->csvescchar)) {
+            /* Qualify and escape */
+            used_len = file_copy_esc((char *)&out_data->buffer[out_data->length],
+                                     (const char *)(data),
+                                     cfg->csvqualifier,
+                                     cfg->csvescchar);
+        }
+        else {
+            /* No escaping so just copy without trailing 0 */
+            /* Item's length includes trailing 0 for data items */
+            used_len = length - 1;
+            memcpy(&out_data->buffer[out_data->length],
+                   (const char *)(data),
+                   used_len);
+        }
+        break;
+
+    case COL_TYPE_BINARY:
+
+        for (i = 0; i < length; i++)
+            sprintf((char *)&out_data->buffer[out_data->length + i * 2],
+                    "%02X", (unsigned int)(((const unsigned char *)(data))[i]));
+        used_len = length * 2;
+        break;
+
+    case COL_TYPE_INTEGER:
+        used_len = sprintf((char *)&out_data->buffer[out_data->length],
+                           "%d", *((const int *)(data)));
+        break;
+
+    case COL_TYPE_UNSIGNED:
+        used_len = sprintf((char *)&out_data->buffer[out_data->length],
+                           "%u", *((const unsigned int *)(data)));
+        break;
+
+    case COL_TYPE_LONG:
+        used_len = sprintf((char *)&out_data->buffer[out_data->length],
+                           "%ld", *((const long *)(data)));
+        break;
+
+    case COL_TYPE_ULONG:
+        used_len = sprintf((char *)&out_data->buffer[out_data->length],
+                           "%lu", *((const unsigned long *)(data)));
+        break;
+
+    case COL_TYPE_DOUBLE:
+        used_len = sprintf((char *)&out_data->buffer[out_data->length],
+                           "%.4f", *((const double *)(data)));
+        break;
+
+    case COL_TYPE_BOOL:
+        used_len = sprintf((char *)&out_data->buffer[out_data->length],
+                           "%s",
+                           (*((const unsigned char *)(data))) ? "true" : "false");
+        break;
+
+    default:
+        out_data->buffer[out_data->length] = '\0';
+        used_len = 0;
+        break;
+    }
+
+    /* Adjust length */
+    out_data->length += used_len;
+
+    /* Add qualifier */
+    if (cfg->csvqualifier) {
+        out_data->buffer[out_data->length] = cfg->csvqualifier;
+        out_data->length++;
+    }
+
+    /* The "length" member of the structure does not account
+     * for the 0 symbol but we made sure that it fits
+     * when we asked for the memory at the top.
+     */
+    out_data->buffer[out_data->length] = '\0';
+
+    TRACE_INFO_STRING("Data: ", out_data->buffer);
+
+    TRACE_FLOW_STRING("file_serialize_csv.", "Exit");
+    return error;
+
+}
+
+/* Function that reads the specific configuration
+ * information about the format of the output
+ */
+int file_get_csv_cfg(void **storage,
+                     const char *name,
+                     struct collection_item *ini_config,
+                     const char *appname)
+{
+    int error = EOK;
+    struct collection_item *cfg_item = NULL;
+    struct file_csv_cfg *cfg= NULL;
+    const char *qual;
+    const char *sep;
+    const char *esc;
+    const char *space;
+
+    TRACE_FLOW_STRING("file_get_csv_cfg", "Entry");
+
+    /* Allocate memory for configuration */
+    cfg = (struct file_csv_cfg *)malloc(sizeof(struct file_csv_cfg));
+    if (cfg == NULL) {
+        TRACE_ERROR_NUMBER("Failed to allocate storage for CSV configuration", ENOMEM);
+        return ENOMEM;
+    }
+
+    /*********** Qualifier *************/
+
+    /* Get qualifier */
+    error = get_config_item(name,
+                            FILE_CSV_QUAL,
+                            ini_config,
+                            &cfg_item);
+    if (error) {
+        TRACE_ERROR_NUMBER("Attempt to read qualifier attribute returned error", error);
+        free(cfg);
+        return error;
+    }
+
+    /* Do we have qualifier? */
+    if (cfg_item == NULL) {
+        /* There is no qualifier - use default */
+        cfg->csvqualifier = FILE_CSV_DEF_QUAL;
+    }
+    else {
+        /* Get qualifier from configuration */
+        error = EOK;
+        qual = get_const_string_config_value(cfg_item, &error);
+        if (error) {
+            TRACE_ERROR_STRING("Failed to get value from configuration.", "Fatal Error!");
+            free(cfg);
+            return error;
+        }
+
+        if (qual[0] == '\0') cfg->csvqualifier = '\0';
+        else if(qual[1] != '\0') {
+            TRACE_ERROR_STRING("Qualifier has more than one symbol.", "Fatal Error!");
+            free(cfg);
+            return EINVAL;
+        }
+        else cfg->csvqualifier = qual[0];
+    }
+
+    /*********** Separator *************/
+
+    /* Get separator */
+    error = get_config_item(name,
+                            FILE_CSV_SEP,
+                            ini_config,
+                            &cfg_item);
+    if (error) {
+        TRACE_ERROR_NUMBER("Attempt to read separator attribute returned error", error);
+        free(cfg);
+        return error;
+    }
+
+    /* Do we have separator? */
+    if (cfg_item == NULL) {
+        /* There is no separator - use default */
+        cfg->csvseparator = FILE_CSV_DEF_SEP;
+    }
+    else {
+        /* Get separator from configuration */
+        error = EOK;
+        sep = get_const_string_config_value(cfg_item, &error);
+        if (error) {
+            TRACE_ERROR_STRING("Failed to get value from configuration.", "Fatal Error!");
+            free(cfg);
+            return error;
+        }
+
+        if (sep[0] == '\0') cfg->csvseparator = '\0';
+        else if(sep[1] != '\0') {
+            TRACE_ERROR_STRING("Separator has more than one symbol.", "Fatal Error!");
+            free(cfg);
+            return EINVAL;
+        }
+        else cfg->csvseparator = sep[0];
+    }
+
+    /*********** Escape symbol *************/
+
+    /* Get escape symbol */
+    error = get_config_item(name,
+                            FILE_CSV_ESCSYM,
+                            ini_config,
+                            &cfg_item);
+    if (error) {
+        TRACE_ERROR_NUMBER("Attempt to read esc symbol attribute returned error", error);
+        free(cfg);
+        return error;
+    }
+
+    /* Do we have esc symbol? */
+    if (cfg_item == NULL) {
+        /* There is no esc symbol - use default */
+        cfg->csvescchar = FILE_CSV_DEF_ESC;
+    }
+    else {
+        /* Get esc symbol from configuration */
+        error = EOK;
+        esc = get_const_string_config_value(cfg_item, &error);
+        if (error) {
+            TRACE_ERROR_STRING("Failed to get value from configuration.", "Fatal Error!");
+            free(cfg);
+            return error;
+        }
+
+        if (esc[0] == '\0') cfg->csvescchar = '\0';
+        else if(esc[1] != '\0') {
+            TRACE_ERROR_STRING("Esc symbol has more than one symbol.", "Fatal Error!");
+            free(cfg);
+            return EINVAL;
+        }
+        else cfg->csvescchar = esc[0];
+    }
+
+    /*********** Space *************/
+
+    /* Get space */
+    error = get_config_item(name,
+                            FILE_CSV_SPACE,
+                            ini_config,
+                            &cfg_item);
+    if (error) {
+        TRACE_ERROR_NUMBER("Attempt to read space attribute returned error", error);
+        free(cfg);
+        return error;
+    }
+
+    /* Do we have space? */
+    if (cfg_item == NULL) {
+        /* There is no esc symbol - use default */
+        cfg->csvspace = FILE_CSV_DEF_SPC;
+    }
+    else {
+        /* Get file name from configuration */
+        error = EOK;
+        space = get_const_string_config_value(cfg_item, &error);
+        if (error) {
+            TRACE_ERROR_STRING("Failed to get value from configuration.", "Fatal Error!");
+            free(cfg);
+            return error;
+        }
+
+        /* Determine what to use as a space symbol */
+        if (space[0] == '\0') cfg->csvspace = ' ';
+        else if(strcmp(space, FILE_CSV_SP) == 0) cfg->csvspace = ' ';
+        else if(strcmp(space, FILE_CSV_TAB) == 0) cfg->csvspace = '\t';
+        else if(strcmp(space, FILE_CSV_CR) == 0) cfg->csvspace = '\n';
+        else {
+            TRACE_ERROR_STRING("Esc symbol has more than one symbol.", "Fatal Error!");
+            free(cfg);
+            return EINVAL;
+        }
+    }
+
+    /*********** Number of spaces *************/
+    /* Get number of spaces */
+
+    cfg_item = NULL;
+    error = get_config_item(name,
+                            FILE_CSV_NUMSP,
+                            ini_config,
+                            &cfg_item);
+    if (error) {
+        TRACE_ERROR_NUMBER("Attempt to read number of spaces attribute returned error", error);
+        free(cfg);
+        return error;
+    }
+
+    /* Do we have number of spaces? */
+    if (cfg_item == NULL) {
+        /* There is no attribute - assume default */
+        TRACE_INFO_STRING("No attribute.", "Assume no spaces");
+        cfg->csvnumsp = 0;
+    }
+    else {
+        cfg->csvnumsp = (uint32_t) get_unsigned_config_value(cfg_item, 1, 0, &error);
+        if (error) {
+            TRACE_ERROR_STRING("Invalid number of spaces value", "Fatal Error!");
+            free(cfg);
+            return EINVAL;
+        }
+        /* Check for right range */
+        if (cfg->csvnumsp > FILE_MAXSPACE) {
+            TRACE_ERROR_STRING("Too many spaces - not allowed", "Fatal Error!");
+            free(cfg);
+            return ERANGE;
+        }
+    }
+
+    /*********** Header *************/
+    /* Next is header field */
+
+    cfg_item = NULL;
+    error = get_config_item(name,
+                            FILE_CSV_HEADER,
+                            ini_config,
+                            &cfg_item);
+    if (error) {
+        TRACE_ERROR_NUMBER("Attempt to read header attribute returned error", error);
+        free(cfg);
+        return error;
+    }
+
+    /* Do we have header? */
+    if (cfg_item == NULL) {
+        /* There is no attribute - assume default */
+        TRACE_INFO_STRING("No attribute.", "Assume no header");
+        cfg->csvheader = 0;
+    }
+    else {
+        cfg->csvheader = (uint32_t) get_bool_config_value(cfg_item, '\0', &error);
+        if (error) {
+            TRACE_ERROR_STRING("Invalid csv header value", "Fatal Error!");
+            free(cfg);
+            return EINVAL;
+        }
+    }
+
+    *((struct file_csv_cfg **)storage) = cfg;
+
+    TRACE_FLOW_STRING("file_get_csv_cfg", "Entry");
+    return error;
+}
+
+#ifdef ELAPI_VERBOSE
+
+void file_print_fmt_csv(void *data)
+{
+    struct file_csv_cfg *cfg;
+
+    cfg = (struct file_csv_cfg *)(data);
+    if (cfg == NULL) {
+        printf("CSV Configuration is undefined!\n");
+        return;
+    }
+
+    printf("CSV Configuration:\n");
+    printf("  Qualifier: ");
+    if (cfg->csvqualifier != '\0') printf("[%c]\n", cfg->csvqualifier);
+    else printf("[undefined]\n");
+
+    printf("  Separator: ");
+    if (cfg->csvseparator != '\0') printf("[%c]\n", cfg->csvseparator);
+    else printf("[undefined]\n");
+
+    printf("  Escape: ");
+    if (cfg->csvescchar != '\0') printf("[%c]\n", cfg->csvescchar);
+    else printf("[undefined]\n");
+
+    printf("  Space: [%c] [ASCII: %d]\n", cfg->csvspace, (int)(cfg->csvspace));
+    printf("  Number of spaces: [%d]\n", cfg->csvnumsp);
+    printf("  Header: [%s]\n", ((cfg->csvheader > 0) ? "yes" : "no"));
+    printf("CSV Configuration END\n");
+
+}
+#endif
diff --git a/common/elapi/providers/file/file_fmt_csv.h b/common/elapi/providers/file/file_fmt_csv.h
new file mode 100644
index 000000000..6cc52745d
--- /dev/null
+++ b/common/elapi/providers/file/file_fmt_csv.h
@@ -0,0 +1,82 @@
+/*
+    ELAPI
+
+    Module contains functions related to outputting events in CSV format.
+
+    Copyright (C) Dmitri Pal <dpal@redhat.com> 2009
+
+    This program is free software; you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation; either version 3 of the License, or
+    (at your option) any later version.
+    This program is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+    You should have received a copy of the GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef ELAPI_FILE_FMT_CSV_H
+#define ELAPI_FILE_FMT_CSV_H
+
+#include <stdint.h>
+#include "collection.h"
+#include "elapi_basic.h"
+
+/* Format specific configuration parameters */
+/* CSV:                                     */
+#define FILE_CSV_QUAL       "csvqual"
+#define FILE_CSV_SEP        "csvsep"
+#define FILE_CSV_ESCSYM     "csvescsym"
+#define FILE_CSV_SPACE      "csvspace"
+#define FILE_CSV_NUMSP      "csvnumsp"
+#define FILE_CSV_HEADER     "csvheader"
+
+/* Strings from config that will be recognized */
+#define FILE_CSV_SP         "space"
+#define FILE_CSV_TAB        "tab"
+#define FILE_CSV_CR         "cr"
+
+
+/* Default values for configuration parameters */
+#define FILE_CSV_DEF_QUAL   '"'
+#define FILE_CSV_DEF_SEP    ','
+#define FILE_CSV_DEF_ESC    '\\'
+#define FILE_CSV_DEF_SPC    ' '
+
+/* Try catch corrupted configuration 80 is more than enough */
+#define FILE_MAXSPACE   80
+
+/* Configuration for the CSV output */
+struct file_csv_cfg {
+    uint32_t csvheader;             /* Include csv header or not? */
+    uint32_t csvnumsp;              /* How many spaces ? */
+    unsigned char csvqualifier;     /* What is the qualifier? */
+    unsigned char csvseparator;     /* What is the separator? */
+    unsigned char csvescchar;       /* What is the escape character? */
+    unsigned char csvspace;         /* What is the space character? */
+};
+
+/* Function that reads the specific configuration
+ * information about the CSV format of the output
+ */
+int file_get_csv_cfg(void **storage,
+                     const char *name,
+                     struct collection_item *ini_config,
+                     const char *appname);
+
+/* Serialize an item into the csv format */
+int file_serialize_csv(struct elapi_data_out *out_data,
+                       int type,
+                       int length,
+                       void *data,
+                       void *mode_cfg);
+
+
+#ifdef ELAPI_VERBOSE
+/* Function for debugging */
+void file_print_fmt_csv(void *data);
+
+#endif
+#endif
diff --git a/common/elapi/providers/file/file_provider.c b/common/elapi/providers/file/file_provider.c
index 589ed5eb7..09d6261eb 100644
--- a/common/elapi/providers/file/file_provider.c
+++ b/common/elapi/providers/file/file_provider.c
@@ -20,34 +20,530 @@
 #define _GNU_SOURCE
 #include <errno.h>      /* for errors */
 #include <stdlib.h>     /* for free() */
+#include <string.h>     /* for strlen() */
+#include <unistd.h>     /* for close() */
 
 #include "file_provider.h"
+#include "file_util.h"
+#include "file_fmt_csv.h"
 #include "ini_config.h"
 #include "trace.h"
 #include "config.h"
-/* FIXME: temporary for debugging */
+
+/* NOTE: Each format module has its own header */
+#include "file_fmt_csv.h"
+/* Add headers for new formats here... */
+
+/*******************************************************************/
+/* SECTION FOR INTERNAL CONDITIONALLY COMPILED DEBUGGING FUNCTIONS */
+/*******************************************************************/
+#ifdef ELAPI_VERBOSE
 #include "collection_tools.h"
 
+/* Function to debug format configurations */
+void file_print_fmt_cfg(uint32_t mode, void *fmt_cfg)
+{
+    switch(mode) {
+    case FILE_MODE_CSV:
+        file_print_fmt_csv(fmt_cfg);
+        break;
+    /* FIXME : add other formats later */
+/*
+    case FILE_MODE_FORMAT:
+        error = file_print_fmt_format(fmt_cfg);
+        break;
+    case FILE_MODE_HTML:
+        error = file_print_fmt_html(fmt_cfg);
+        break;
+    case FILE_MODE_XML:
+        error = file_print_fmt_xml(fmt_cfg);
+        break;
+    case FILE_MODE_JSON:
+        error = file_print_fmt_json(fmt_cfg);
+        break;
+    case FILE_MODE_KVP:
+        error = file_print_fmt_kvp(fmt_cfg);
+        break;
+*/
+    default:
+        printf("Unsupported mode!\n");
+    }
+}
+
+
+/* Function for debugging configuration */
+void file_print_cfg(struct file_prvdr_cfg *cfg)
+{
+    printf("File provider configuration\n");
+
+    printf("  File name: [%s]\n", ((cfg->filename != NULL) ? cfg->filename : "NULL"));
+    printf("  Own file : [%s]\n", ((cfg->ownfile > 0) ? "yes" : "no"));
+    printf("  Keep open: [%s]\n", ((cfg->keepopen > 0) ? "yes" : "no"));
+
+    if (cfg->fsyncmode == 0) {
+        printf("  Sync mode: [no flush]\n");
+    }
+    else if (cfg->fsyncmode > 0) {
+        printf("  Sync mode: every [%d] event\n", cfg->fsyncmode);
+    }
+    else {
+        printf("  Sync mode: every [%d] second\n", 0 - cfg->fsyncmode);
+    }
+
+    if (cfg->set) {
+        printf("  There is a set of predefined fields\n");
+        col_print_collection(cfg->set);
+        printf("  Use leftovers: [%s]\n", ((cfg->use_leftovers > 0) ? "yes" : "no"));
+        printf("  Jam leftovers: [%s]\n", ((cfg->jam_leftovers > 0) ? "yes" : "no"));
+        if (cfg->use_leftovers > 0) {
+            printf("Leftovers configuration:\n");
+            file_print_fmt_cfg(cfg->mode_leftovers, cfg->lo_fmt_cfg);
+            printf("Leftovers configuration END\n");
+        }
+    }
+    else printf("All fields go into the output.\n");
+
+
+    printf("Main configuration:\n");
+    file_print_fmt_cfg(cfg->outmode, cfg->main_fmt_cfg);
+    printf("Main configuration END:\n");
+
+    printf("File provider configuration END\n");
+
+}
+
+/* Function to debug context */
+void file_print_ctx(struct file_prvdr_ctx *ctx)
+{
+    if (ctx == NULL) {
+        printf("No file provider context!\n");
+        return;
+    }
+
+    printf("File Provider Context\n");
+
+    /* Print configuration */
+    file_print_cfg(&(ctx->config));
+
+    /* Print other parts of the context */
+    printf("File is currently: [%s]\n", ((ctx->outfile >= 0) ? "open" : "closed"));
+    printf("File Provider Context END\n\n");
+
+}
+#endif
+
+/*******************************************************************/
+/*                     MAIN MODULE FUNCTIONS                       */
+/*******************************************************************/
+
+/* Function that reads the specific configuration
+ * information about the format of the output
+ */
+static int file_read_fmt_cfg(void **storage,
+                             uint32_t mode,
+                             const char *name,
+                             struct collection_item *ini_config,
+                             const char *appname)
+{
+    int error = EOK;
+
+    TRACE_FLOW_STRING("file_read_fmt_cfg", "Entry");
+
+    switch(mode) {
+    case FILE_MODE_CSV:
+        error = file_get_csv_cfg(storage, name, ini_config, appname);
+        break;
+    /* FIXME : add other formats later */
+/*
+    case FILE_MODE_FORMAT:
+        error = file_get_format_cfg(storage, name, ini_config, appname);
+        break;
+    case FILE_MODE_HTML:
+        error = file_get_html_cfg(storage, name, ini_config, appname);
+        break;
+    case FILE_MODE_XML:
+        error = file_get_xml_cfg(storage, name, ini_config, appname);
+        break;
+    case FILE_MODE_JSON:
+        error = file_get_json_cfg(storage, name, ini_config, appname);
+        break;
+    case FILE_MODE_KVP:
+        error = file_get_kvp_cfg(storage, name, ini_config, appname);
+        break;
+*/
+    default:
+        TRACE_ERROR_STRING("Unsupported mode", "Fatal error!");
+        error = EINVAL;
+
+    }
+    TRACE_FLOW_NUMBER("file_read_fmt_cfg. Exit. Returning:", error);
+    return error;
+}
+
+/* Function to build the set object from the configuration data */
+static int file_build_set(struct file_prvdr_cfg *file_cfg,
+                          struct collection_item *cfg_item)
+{
+    int error = EOK;
+    char **fields;
+    char *field;
+    int size;
+    int count;
+    struct collection_item *dummy = NULL;
+    struct collection_item *set = NULL;
+
+    TRACE_FLOW_STRING("file_build_set", "Entry");
+
+    /* Get fields array from config field */
+    fields = get_string_config_array(cfg_item, NULL, &size, &error);
+    if (error) {
+        TRACE_ERROR_NUMBER("Attempt to get set items returned error", error);
+        return error;
+    }
+
+    if (size > 0) {
+
+        TRACE_INFO_STRING("We have the set of required fields", "");
+
+        /* Create collection */
+        error = col_create_collection(&set, FILE_FIELDSET_COL, FILE_FIELDSET_CLASS);
+        if (error) {
+            TRACE_ERROR_NUMBER("Attempt to create collection failed", error);
+            return error;
+        }
+
+        for (count = 0; count < size; count++) {
+            field = fields[count];
+            TRACE_INFO_STRING("FIELD:", field);
+
+            if (field[0] == FILE_SET_END) {
+                TRACE_INFO_STRING("Leftovers field found.", "");
+                if (count != (size - 1)) {
+                    /* We found an end list field in the middle - error */
+                    TRACE_ERROR_NUMBER("More fields after end list field.", EINVAL);
+                    col_destroy_collection(set);
+                    free_string_config_array(fields);
+                    return EINVAL;
+                }
+
+                file_cfg->use_leftovers = 1;
+
+                /* What format to use leftovers ? */
+                /* NOTE: Is we ever support more than 10 formats
+                 * this logic needs to change
+                 */
+                if ((field[1] >= '0') &&
+                    (field[1] <= ('0' + FILE_MAXMODE)) &&
+                    (field[2] == '\0')) {
+                    /* We have a format specifier */
+                    file_cfg->mode_leftovers = (uint32_t)(field[1] - '0');
+                    file_cfg->jam_leftovers = 1;
+                    TRACE_INFO_NUMBER("Use mode for leftovers:", file_cfg->mode_leftovers);
+                }
+                else {
+                    /* Wrong format */
+                    TRACE_ERROR_NUMBER("Leftover field has invalid format.", EINVAL);
+                    col_destroy_collection(set);
+                    free_string_config_array(fields);
+                    return EINVAL;
+                }
+
+            }
+            else {
+                error = col_add_binary_property(set,
+                                                NULL,
+                                                field,
+                                                &dummy,
+                                                sizeof(struct collection_item *));
+                if (error) {
+                    TRACE_ERROR_NUMBER("Error adding item to the set.", error);
+                    col_destroy_collection(set);
+                    free_string_config_array(fields);
+                    return error;
+                }
+            }
+        }
+
+        file_cfg->set = set;
+    }
+
+    /* Free the list */
+    free_string_config_array(fields);
+
+    TRACE_FLOW_STRING("file_build_set", "Exit");
+    return error;
+}
+
 
 /* Function to read configuration */
-int file_read_cfg(struct file_prvdr_cfg *file_cfg,
-                  char *name,
-                  struct collection_item *ini_config)
+static int file_read_cfg(struct file_prvdr_cfg *file_cfg,
+                         const char *name,
+                         struct collection_item *ini_config,
+                         const char *appname)
 {
     int error = EOK;
+    struct collection_item *cfg_item = NULL;
+    const char *filename;
+    int use_default_name = 0;
+
     TRACE_FLOW_STRING("file_read_cfg", "Entry point");
 
-    /* FIXME: read configuration items */
+    /*********** Filename *************/
+
+    /* Get file name */
+    error = get_config_item(name,
+                            FILE_OUTNAME,
+                            ini_config,
+                            &cfg_item);
+    if (error) {
+        TRACE_ERROR_NUMBER("Attempt to read \"filename\" attribute returned error", error);
+        return error;
+    }
 
+    /* Do we have file name? */
+    if (cfg_item == NULL) use_default_name = 1;
+    else {
+        /* Get file name from configuration */
+        error = EOK;
+        filename = get_const_string_config_value(cfg_item, &error);
+        if (error) {
+            TRACE_ERROR_STRING("Failed to get value from configuration.", "Fatal Error!");
+            return error;
+        }
+        /* Check if file name is empty */
+        if (filename[0] == '\0') use_default_name = 1;
+        else {
+            /* Now get a copy */
+            file_cfg->filename = get_string_config_value(cfg_item, &error);
+            if (error) {
+                TRACE_ERROR_STRING("Failed to copy value from configuration.", "Fatal Error!");
+                return error;
+            }
+        }
+    }
+
+    if (use_default_name) {
+        /* There is no file name - use default */
+        file_cfg->filename = malloc(strlen(appname) + sizeof(FILE_SUFFIX));
+        if (file_cfg->filename == NULL) {
+            TRACE_ERROR_STRING("Failed to allocate memory for file name.", "Fatal Error!");
+            return ENOMEM;
+        }
+        /* Appname is validated in the elapi_log.c */
+        /* This should be safe to do */
+        strcpy(file_cfg->filename, appname);
+        strcat(file_cfg->filename, FILE_SUFFIX);
+
+        file_cfg->ownfile = 1;
+    }
+    else if (strcmp(filename, FILE_STDERR) != 0) file_cfg->ownfile = 1;
+    else file_cfg->ownfile = 0;
+
+    /*********** Keep open *************/
+    /* Next is "keepopen" field */
+
+    cfg_item = NULL;
+    error = get_config_item(name,
+                            FILE_KEEPOPEN,
+                            ini_config,
+                            &cfg_item);
+    if (error) {
+        TRACE_ERROR_NUMBER("Attempt to read \"keepopen\" attribute returned error", error);
+        return error;
+    }
+
+    /* Do we have "keepopen"? */
+    if (cfg_item == NULL) {
+        /* There is no attribute - assume default */
+        TRACE_INFO_STRING("No \"keepopen\" attribute.", "Assume open on each entry");
+        file_cfg->keepopen = 0;
+    }
+    else {
+        file_cfg->keepopen = (uint32_t) get_bool_config_value(cfg_item, '\0', &error);
+        if (error) {
+            TRACE_ERROR_STRING("Invalid \"keepopen\" value", "Fatal Error!");
+            return EINVAL;
+        }
+    }
+
+    /*********** Outmode *************/
+    /* Next is "outmode" field */
+
+    cfg_item = NULL;
+    error = get_config_item(name,
+                            FILE_OUTMODE,
+                            ini_config,
+                            &cfg_item);
+    if (error) {
+        TRACE_ERROR_NUMBER("Attempt to read \"outmode\" attribute returned error", error);
+        return error;
+    }
+
+    /* Do we have "outmode"? */
+    if (cfg_item == NULL) {
+        /* There is no attribute - assume default */
+        TRACE_INFO_STRING("No \"outmode\" attribute.", "Assume CSV kind");
+        file_cfg->outmode = 0;
+    }
+    else {
+        file_cfg->outmode = (uint32_t) get_unsigned_config_value(cfg_item, 1, 0, &error);
+        if (error) {
+            TRACE_ERROR_STRING("Invalid \"outmode\" value", "Fatal Error!");
+            return EINVAL;
+        }
+        /* Check for right range */
+        if (file_cfg->outmode > FILE_MAXMODE) {
+            TRACE_ERROR_STRING("Invalid \"outmode\" value - out of range", "Fatal Error!");
+            return ERANGE;
+        }
+    }
+
+    /*********** Sync mode *************/
+    /* Next is sync mode field */
+
+    cfg_item = NULL;
+    error = get_config_item(name,
+                            FILE_FLUSH,
+                            ini_config,
+                            &cfg_item);
+    if (error) {
+        TRACE_ERROR_NUMBER("Attempt to read \"fsyncmode\" attribute returned error", error);
+        return error;
+    }
+
+    /* Do we have "fsyncmode"? */
+    if (cfg_item == NULL) {
+        /* There is no attribute - assume default */
+        TRACE_INFO_STRING("No \"fsyncmode\" attribute.", "Assume CSV kind");
+        file_cfg->fsyncmode = 0;
+    }
+    else {
+        file_cfg->fsyncmode = (int32_t) get_int_config_value(cfg_item, 1, 0, &error);
+        if (error) {
+            TRACE_ERROR_STRING("Invalid \"fsyncmode\" value", "Fatal Error!");
+            return EINVAL;
+        }
+    }
+
+    /*********** Set *************/
+    /* Next is the "set" field */
+    cfg_item = NULL;
+    error = get_config_item(name,
+                            FILE_FIELDSET,
+                            ini_config,
+                            &cfg_item);
+    if (error) {
+        TRACE_ERROR_NUMBER("Attempt to read \"set\" attribute returned error", error);
+        return error;
+    }
+
+    file_cfg->use_leftovers = 0;
+    file_cfg->jam_leftovers = 0;
+    file_cfg->mode_leftovers = file_cfg->outmode;
+
+    /* Do we have "required"? */
+    if (cfg_item == NULL) {
+        /* There is no attribute - assume default */
+        TRACE_INFO_STRING("No \"set\" attribute.", "Assume all fields as specified");
+        file_cfg->set = NULL;
+    }
+    else {
+        error = file_build_set(file_cfg, cfg_item);
+        if (error) {
+            TRACE_ERROR_STRING("Invalid \"set\" value", "Fatal Error!");
+            return EINVAL;
+        }
+    }
+
+    /*********** Format specific configurations *************/
+    /* Read the main format configuration details */
+    error = file_read_fmt_cfg((void **)(&(file_cfg->main_fmt_cfg)),
+                              file_cfg->outmode,
+                              name,
+                              ini_config,
+                              appname);
+    if (error) {
+        TRACE_ERROR_NUMBER("Failed to read main format configuration", error);
+        return error;
+    }
+
+    if (file_cfg->use_leftovers) {
+        /* If we use same mode for leftovers and main do not read things again */
+        if (file_cfg->mode_leftovers == file_cfg->outmode) {
+            TRACE_INFO_STRING("Output modes are the same", "");
+            file_cfg->lo_fmt_cfg = file_cfg->main_fmt_cfg;
+        }
+        else {
+            TRACE_INFO_STRING("Output modes are the different", "");
+            TRACE_INFO_NUMBER("Main mode", file_cfg->outmode);
+            TRACE_INFO_NUMBER("Left over's mode", file_cfg->mode_leftovers);
+
+            /* Read the leftover's format configuration details */
+            error = file_read_fmt_cfg((void **)(&(file_cfg->lo_fmt_cfg)),
+                                      file_cfg->mode_leftovers,
+                                      name,
+                                      ini_config,
+                                      appname);
+            if (error) {
+                TRACE_ERROR_NUMBER("Failed to read main format configuration", error);
+                return error;
+            }
+        }
+    }
     TRACE_FLOW_STRING("file_read_cfg", "Exit");
     return error;
 }
 
+/* Function to destroy the context */
+static void file_destroy_ctx(struct file_prvdr_ctx **file_ctx)
+{
+    TRACE_FLOW_STRING("file_destroy_ctx", "Entry");
+
+    if ((file_ctx) && (*file_ctx)) {
+        /* Close file if it is open */
+        if (((*file_ctx)->outfile >= 0) && ((*file_ctx)->config.ownfile)) {
+            TRACE_INFO_STRING("File was open", "");
+            close((*file_ctx)->outfile);
+        }
+
+        /* Free file name if it is not NULL */
+        if ((*file_ctx)->config.filename) {
+            TRACE_INFO_STRING("Freeing file name", (*file_ctx)->config.filename);
+            free((*file_ctx)->config.filename);
+        }
+
+        /* Free set if any */
+        if ((*file_ctx)->config.set) {
+            TRACE_INFO_NUMBER("Freeing set", (*file_ctx)->config.set);
+            col_destroy_collection((*file_ctx)->config.set);
+        }
+
+        /* Free main format configuration if it is not NULL */
+        if (((*file_ctx)->config.main_fmt_cfg) &&
+            ((*file_ctx)->config.main_fmt_cfg != (*file_ctx)->config.lo_fmt_cfg)) {
+            TRACE_INFO_NUMBER("Freeing main format config.", (*file_ctx)->config.main_fmt_cfg);
+            free((*file_ctx)->config.main_fmt_cfg);
+        }
+
+        /* Free left over format configuration if it is not NULL */
+        if ((*file_ctx)->config.lo_fmt_cfg) {
+            TRACE_INFO_NUMBER("Freeing leftover format config.", (*file_ctx)->config.lo_fmt_cfg);
+            free((*file_ctx)->config.lo_fmt_cfg);
+        }
+
+        TRACE_FLOW_STRING("Freeing file context", "Entry");
+        free(*file_ctx);
+        *file_ctx = NULL;
+    }
+
+    TRACE_FLOW_STRING("file_destroy_ctx", "Exit");
+}
 
 /* Function to create context */
-int file_create_ctx(struct file_prvdr_ctx **file_ctx,
-                    char *name,
-                    struct collection_item *ini_config)
+static int file_create_ctx(struct file_prvdr_ctx **file_ctx,
+                           const char *name,
+                           struct collection_item *ini_config,
+                           const char *appname)
 {
     int error = EOK;
     struct file_prvdr_ctx *ctx = NULL;
@@ -62,12 +558,15 @@ int file_create_ctx(struct file_prvdr_ctx **file_ctx,
 
     /* Init allocatable items */
     ctx->config.filename = NULL;
+    ctx->config.main_fmt_cfg = NULL;
+    ctx->config.lo_fmt_cfg = NULL;
+    ctx->outfile = -1;
 
     /* Read configuration data */
-    error = file_read_cfg(&(ctx->config), name, ini_config);
+    error = file_read_cfg(&(ctx->config), name, ini_config, appname);
     if (error) {
         TRACE_ERROR_NUMBER("Error reading sink configuration", error);
-        free(ctx);
+        file_destroy_ctx(&ctx);
         return error;
     }
 
@@ -80,8 +579,9 @@ int file_create_ctx(struct file_prvdr_ctx **file_ctx,
 
 /* File init function */
 int file_init(void **priv_ctx,
-              char *name,
-              struct collection_item *ini_config)
+              const char *name,
+              struct collection_item *ini_config,
+              const char *appname)
 {
     int error = EOK;
     TRACE_FLOW_STRING("file_init", "Entry point");
@@ -89,15 +589,21 @@ int file_init(void **priv_ctx,
     /* Start with creating context */
     error = file_create_ctx((struct file_prvdr_ctx **)priv_ctx,
                             name,
-                            ini_config);
+                            ini_config,
+                            appname);
     if (error) {
         TRACE_ERROR_NUMBER("Failed to create context", error);
         return error;
     }
 
-	/* Open file */
+    /* Open file */
     /* FIXME: ... */
 
+#ifdef ELAPI_VERBOSE
+    printf("Initializaing file provider for sink: [%s]\n", name);
+    file_print_ctx(*((struct file_prvdr_ctx **)priv_ctx));
+#endif
+
     TRACE_FLOW_STRING("file_init", "Exit");
     return error;
 }
@@ -111,18 +617,11 @@ void file_close(void **priv_ctx)
 
     ctx = (struct file_prvdr_ctx **)priv_ctx;
 
-    /* Close file */
-    /* FIXME: ... */
-
-    /* If we allocated file name free it */
-    if ((*ctx)->config.filename != NULL) {
-        TRACE_INFO_STRING("Freeing file name", (*ctx)->config.filename);
-        free((*ctx)->config.filename);
-    }
+#ifdef ELAPI_VERBOSE
+    file_print_ctx(*ctx);
+#endif
 
-    /* Free and indicate that the context is freed */
-    free(*ctx);
-    *ctx = NULL;
+    file_destroy_ctx(ctx);
 
     TRACE_FLOW_STRING("file_close", "Exit");
 }
@@ -131,11 +630,38 @@ void file_close(void **priv_ctx)
 int file_submit(void *priv_ctx, struct collection_item *event)
 {
     int error = EOK;
+    struct file_prvdr_ctx *ctx = (struct file_prvdr_ctx *)priv_ctx;
+    struct elapi_data_out *out_data;
+
     TRACE_FLOW_STRING("file_submit", "Entry point");
 
+#ifdef ELAPI_VERBOSE
+    file_print_ctx(ctx);
 
     /* FIXME: Placeholder for now */
     col_print_collection(event);
+#endif
+
+    /* FIXME: Open file here if it is closed */
+
+    error = file_prep_data(&out_data, ctx, event);
+    if (error) {
+        TRACE_ERROR_NUMBER("Failed to prepare data", error);
+        return error;
+    }
+
+    /* FIXME: just print it for now!!! */
+
+    printf("EVENT: [%*s]\n", out_data->length, out_data->buffer);
+
+
+    /* FIXME: write data base on the synch or not synch mode of the sink */
+    /* For now we will just assume synch */
+    /* This function will probably be a part of the common callbacks */
+    /* elapi_write_to_fd(out_data, ctx_>outfile); */
+
+    /* This one is temporary here too */
+    elapi_free_serialized_data(out_data);
 
     TRACE_FLOW_STRING("file_sumbit", "Exit");
     return error;
diff --git a/common/elapi/providers/file/file_provider.h b/common/elapi/providers/file/file_provider.h
index 218f69f58..f5e6753df 100644
--- a/common/elapi/providers/file/file_provider.h
+++ b/common/elapi/providers/file/file_provider.h
@@ -24,48 +24,75 @@
 
 #include "elapi_sink.h"
 
+/* Common configuration parameters */
+#define FILE_OUTNAME        "filename"
+#define FILE_KEEPOPEN       "keepopen"
+#define FILE_OUTMODE        "outmode"
+#define FILE_FIELDSET       "set"
+#define FILE_FORMAT         "format"
+#define FILE_FLUSH          "fsyncmode"
+
+
+/* Max supported mode */
+/* NOTE: Increase this value when you add a new mode.
+ * If it ever gets to 10 the logic in the
+ * function that builds the set needs to change.
+ */
+#define FILE_MAXMODE        5
+/* Modes: */
+#define FILE_MODE_CSV       0
+#define FILE_MODE_FORMAT    1
+#define FILE_MODE_HTML      2
+#define FILE_MODE_XML       3
+#define FILE_MODE_JSON      4
+#define FILE_MODE_KVP       5
+
+
+/* FIXME: Should it be a compile time switch? */
+#define FILE_SUFFIX ".log"
+#define FILE_SET_END '@'
+
+/* Field set collection */
+#define FILE_FIELDSET_COL   "set"
+#define FILE_FIELDSET_CLASS  21000
+
+/* Special file name - stderr is handled differently */
+#define FILE_STDERR "stderr"
+
 /* Structure that holds internal configuration of the file
  * provider.
  */
 struct file_prvdr_cfg {
     char *filename;                 /* File name */
+    uint32_t ownfile;               /* Do I own the file handle? */
     uint32_t keepopen;              /* Do we need to keep file open */
-    uint32_t fsyncmode;             /* How frequently data is fsynced */
+    int32_t fsyncmode;              /* How frequently data is fsynced */
     uint32_t outmode;               /* Output mode */
     struct collection_item *set;    /* Field set without leftovers symbol */
     uint32_t use_leftovers;         /* Was there a leftover symbol */
     uint32_t jam_leftovers;         /* leftovers should be serialized into one field */
     uint32_t mode_leftovers;        /* Format for the leftover fields */
-    uint32_t csvheader;             /* Include csv header or not? */
-    char csvqualifier;              /* What is the qualifier? */
-    char csvseparator;              /* What is the separator? */
-    uint32_t csvescape;             /* Do we need to escape strings ? */
-    char csvescchar;                /* What is the escape character? */
+    void *main_fmt_cfg;             /* Configuration data for the main format */
+    void *lo_fmt_cfg;               /* Configuration data for leftovers format */
+    /* FIXME add other config data strutures here */
+
+    /* FIXME: Rotation rules ? */
 };
 
+
 /* File context */
 struct file_prvdr_ctx {
     struct file_prvdr_cfg config; /* Configuration */
     int outfile;                  /* File handle */
+    uint32_t smode;               /* Sink's synch mode */
     /* FIXME - other things go here */
 };
 
-
-
-/* Function to read configuration */
-int file_read_cfg(struct file_prvdr_cfg *file_cfg,
-                  char *name,
-                  struct collection_item *ini_config);
-
-/* Function to create context */
-int file_create_ctx(struct file_prvdr_ctx **file_ctx,
-                    char *name,
-                    struct collection_item *ini_config);
-
 /* File init function */
 int file_init(void **priv_ctx,
-              char *name,
-              struct collection_item *ini_config);
+              const char *name,
+              struct collection_item *ini_config,
+              const char *appname);
 
 /* File close function */
 void file_close(void **priv_ctx);
diff --git a/common/elapi/providers/file/file_util.c b/common/elapi/providers/file/file_util.c
new file mode 100644
index 000000000..4508e35d1
--- /dev/null
+++ b/common/elapi/providers/file/file_util.c
@@ -0,0 +1,504 @@
+/*
+    ELAPI
+
+    Module contains internal utility functions for the file provider.
+
+    Copyright (C) Dmitri Pal <dpal@redhat.com> 2009
+
+    This program is free software; you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation; either version 3 of the License, or
+    (at your option) any later version.
+    This program is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+    You should have received a copy of the GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#define _GNU_SOURCE
+#include <errno.h>      /* for errors */
+#include <stdlib.h>     /* for free() */
+#include <string.h>     /* for strlen() */
+
+/* To be able to serialize on needs to know the guts
+ * of the collection structure so have to include
+ * private header here.
+ */
+#include "collection_priv.h"
+#include "file_provider.h"
+#include "file_util.h"
+#include "ini_config.h"
+#include "trace.h"
+#include "config.h"
+
+#ifdef ELAPI_VERBOSE
+/* FIXME: remove when api is stable */
+#include "collection_tools.h"
+#endif
+
+char empty[] = "";
+
+/* Callback to prepare set for splitting */
+static int file_set_clean_cb(const char *property,
+                             int property_len,
+                             int type,
+                             void *data,
+                             int length,
+                             void *custom_data,
+                             int *stop)
+{
+    int error = EOK;
+    TRACE_FLOW_STRING("file_set_clean_cb", "Entry");
+
+    /* Skip header */
+    if (type == COL_TYPE_COLLECTION) return EOK;
+
+    /* Clean data */
+    *((struct collection_item **)(data)) = NULL;
+
+    TRACE_FLOW_STRING("file_set_clean_cb", "Exit");
+    return error;
+}
+
+/* Function to split event into two parts by given set */
+static int file_split_by_set(struct collection_item **leftovers,
+                             struct file_prvdr_cfg *cfg,
+                             struct collection_item *event)
+{
+    int error = EOK;
+    struct collection_item *item_event;
+    struct collection_item *item_set;
+    struct collection_iterator *it_event;
+    struct collection_iterator *it_set;
+    struct collection_item *lo = NULL;
+    int found = 0;
+    TRACE_FLOW_STRING("file_split_by_set", "Entry");
+
+    /* First prepare set for use */
+    error = col_traverse_collection(cfg->set,
+                                    COL_TRAVERSE_ONELEVEL,
+                                    file_set_clean_cb,
+                                    NULL);
+    if (error) {
+        TRACE_ERROR_NUMBER("Traverse set failed.", error);
+        return error;
+    }
+
+    /* If we are going to use leftovers create a collection */
+    if (cfg->use_leftovers) {
+        error = col_create_collection(&lo,
+                                      FILE_LO_NAME,
+                                      FILE_LO_CLASS);
+        if (error) {
+            TRACE_ERROR_NUMBER("Faild to create collection.", error);
+            return error;
+        }
+    }
+
+    /* Now all items from the set are NULLs */
+    /* Split the event in two parts */
+    /* We need to iterate through the event rather than use a callback. */
+    /* Bind iterator */
+    error =  col_bind_iterator(&it_event, event, COL_TRAVERSE_FLAT);
+    if (error) {
+        TRACE_ERROR_NUMBER("Error bind iterator for event failed:", error);
+        /* Here and below it is safe to destroy it event if is NULL. */
+        col_destroy_collection(lo);
+        return error;
+    }
+
+    while(1) {
+        /* Loop through the event */
+        error = col_iterate_collection(it_event, &item_event);
+        if (error) {
+            TRACE_ERROR_NUMBER("Error iterating event:", error);
+            col_unbind_iterator(it_event);
+            col_destroy_collection(lo);
+            return error;
+        }
+
+        /* Are we done ? */
+        if (item_event == NULL) break;
+
+        /* Skip headers */
+        if (item_event->type == COL_TYPE_COLLECTION) continue;
+
+        /* For each item in the event find an item in the set */
+        error =  col_bind_iterator(&it_set, cfg->set, COL_TRAVERSE_ONELEVEL);
+        if (error) {
+            TRACE_ERROR_NUMBER("Error bind iterator for set failed:", error);
+            col_unbind_iterator(it_event);
+            col_destroy_collection(lo);
+            return error;
+        }
+
+        found = 0;
+        while(1) {
+            /* Loop through the event */
+            error = col_iterate_collection(it_set, &item_set);
+            if (error) {
+                TRACE_ERROR_NUMBER("Error iterating set:", error);
+                col_unbind_iterator(it_event);
+                col_unbind_iterator(it_set);
+                col_destroy_collection(lo);
+                return error;
+            }
+
+            /* Are we done ? */
+            if (item_set == NULL) break;
+
+            /* Skip headers */
+            if (item_set->type == COL_TYPE_COLLECTION) continue;
+
+            /* Hashes should match and the data in the set should be NULL,
+             * and legths should be same.
+             */
+            if ((item_event->phash == item_set->phash) &&
+                (*((struct collection_item **)(item_set->data)) == NULL) &&
+                (item_event->property_len == item_set->property_len)) {
+                /* This is a candidate for match - compare strings */
+                TRACE_INFO_STRING("Found a good candidate for match.","");
+                TRACE_INFO_STRING("Set item:", item_set->property);
+                TRACE_INFO_STRING("Event:", item_event->property);
+
+                if (strncasecmp(item_set->property,
+                                item_event->property,
+                                item_event->property_len) == 0) {
+                    TRACE_INFO_STRING("Match found!","");
+                    TRACE_INFO_STRING("Set item:", item_set->property);
+                    TRACE_INFO_STRING("Event:", item_event->property);
+
+                    *((struct collection_item **)(item_set->data)) = item_event;
+                    found = 1;
+                    break;
+                }
+            }
+        }
+        /* Done with the set */
+        col_unbind_iterator(it_set);
+
+        /* Is it a leftover ? */
+        if ((!found) && (cfg->use_leftovers)) {
+            /* We need to put it in the leftovers pile */
+            /* To save time and space we do not care about property name.
+             * The property name is going to be in the referenced item.
+             */
+            error = col_add_binary_property(lo,
+                                            NULL,
+                                            "",
+                                            (void *)(&item_event),
+                                            sizeof(struct collection_item *));
+            if (error) {
+                TRACE_ERROR_NUMBER("Error addding item to leftovers:", error);
+                col_unbind_iterator(it_event);
+                col_destroy_collection(lo);
+                return error;
+            }
+        }
+    }
+
+    /* Done with the event */
+    col_unbind_iterator(it_event);
+
+    /* Save leftovers if any */
+    *leftovers = lo;
+
+    TRACE_FLOW_STRING("file_spserialized_lo->bufferlit_by_set", "Exit");
+    return error;
+}
+
+/* Function to serialize one item */
+static int file_serialize_item(struct elapi_data_out *out_data,
+                               int type,
+                               int length,
+                               void *data,
+                               uint32_t mode,
+                               void *mode_cfg)
+{
+    int error = EOK;
+    TRACE_FLOW_STRING("file_serialize_item", "Entry");
+
+    switch(mode) {
+    case FILE_MODE_CSV:
+        error = file_serialize_csv(out_data,
+                                   type,
+                                   length,
+                                   data,
+                                   mode_cfg);
+        break;
+/* FIXME : add other iterative formats later */
+/*
+    case FILE_MODE_HTML:
+        error = file_serialize_html(out_data,
+                                    type,
+                                    length,
+                                    data,
+                                    mode_cfg);
+        break;
+    case FILE_MODE_XML:
+        error = file_serialize_xml(out_data,
+                                   type,
+                                   length,
+                                   data,
+                                   mode_cfg);
+        break;
+    case FILE_MODE_JSON:
+        error = file_serialize_json(out_data,
+                                    type,
+                                    length,
+                                    data,
+                                    mode_cfg);
+        break;
+    case FILE_MODE_KVP:
+        error = file_serialize_kvp(out_data,
+                                   type,
+                                   length,
+                                   data,
+                                   mode_cfg);
+        break;
+*/
+    default:
+        TRACE_ERROR_STRING("Unsupported mode", "Fatal error!");
+        error = EINVAL;
+
+    }
+
+    TRACE_FLOW_STRING("file_serialize_item", "Exit");
+    return error;
+
+}
+
+
+
+/* Function to serialize the list */
+static int file_serialize_list(struct elapi_data_out **out_data,
+                               int append,
+                               int reference,
+                               struct collection_item *input,
+                               uint32_t mode,
+                               void *mode_cfg)
+{
+    int error = EOK;
+    struct elapi_data_out *allocated = NULL;
+    struct elapi_data_out *to_use = NULL;
+    struct collection_iterator *iterator;
+    struct collection_item *item;
+
+    TRACE_FLOW_STRING("file_serialize_list", "Entry");
+
+    /* Allocate storage if we are not appending */
+    if (!append) {
+        error = elapi_alloc_serialized_data(&allocated);
+        if (error) {
+            TRACE_ERROR_NUMBER("Failed to allocated serialized data", error);
+            return error;
+        }
+        TRACE_INFO_STRING("Allocated new out data", "");
+        to_use = allocated;
+    }
+    else {
+        TRACE_INFO_STRING("Appening, use passed in output data", "");
+        to_use = *out_data;
+    }
+
+    /* FIXME: This logic works for iterative formats only. */
+    /* When we implement the free form format this
+     * logic should be augmented. */
+
+#ifdef ELAPI_VERBOSE
+    /* FIXME: remove when stable */
+    col_debug_collection(input, COL_TRAVERSE_FLAT);
+#endif
+
+
+    /* Start iterating */
+    error =  col_bind_iterator(&iterator, input, COL_TRAVERSE_FLAT);
+    if (error) {
+        TRACE_ERROR_NUMBER("Error bind iterator failed:", error);
+        return error;
+    }
+
+    while(1) {
+        /* Loop through the collection */
+        error = col_iterate_collection(iterator, &item);
+        if (error) {
+            TRACE_ERROR_NUMBER("Error iterating event:", error);
+            col_unbind_iterator(iterator);
+            /* Free allocated data if we allocated it */
+            elapi_free_serialized_data(allocated);
+            return error;
+        }
+
+        /* Are we done ? */
+        if (item == NULL) break;
+
+        /* Skip headers */
+        if (item->type == COL_TYPE_COLLECTION) continue;
+
+        /* Got item */
+        if (reference) {
+            /* Derefernce the item before using */
+            item = *((struct collection_item **)(item->data));
+        }
+
+        if (item) {
+            TRACE_ERROR_NUMBER("Item property", item->property);
+
+            /* Serialize this item */
+            error = file_serialize_item(to_use,
+                                        item->type,
+                                        item->length,
+                                        item->data,
+                                        mode,
+                                        mode_cfg);
+        }
+        else {
+            /* Serialize this item */
+            error = file_serialize_item(to_use,
+                                        COL_TYPE_BINARY,
+                                        0,
+                                        NULL,
+                                        mode,
+                                        mode_cfg);
+        }
+
+        if (error) {
+            TRACE_ERROR_NUMBER("Failed to serialize item", error);
+            col_unbind_iterator(iterator);
+            /* Free allocated data if we allocated it */
+            elapi_free_serialized_data(allocated);
+            return error;
+        }
+    }
+    col_unbind_iterator(iterator);
+
+    *out_data = to_use;
+
+    TRACE_FLOW_STRING("file_serialize_list", "Exit");
+    return error;
+}
+
+/* Function to log event into sink */
+int file_prep_data(struct elapi_data_out **out_data,
+                   struct file_prvdr_ctx *ctx,
+                   struct collection_item *event)
+{
+    int error = EOK;
+    struct elapi_data_out *serialized = NULL;
+    struct elapi_data_out *serialized_lo = NULL;
+    struct collection_item *leftovers = NULL;
+
+    TRACE_FLOW_STRING("file_prep_data", "Entry");
+
+    /* Do we need to split the data into two parts by set ? */
+    if (ctx->config.set) {
+        /* Split collection based on the configured set of fields */
+        error = file_split_by_set(&leftovers,
+                                  &(ctx->config),
+                                  event);
+        if (error) {
+            TRACE_ERROR_NUMBER("Split collection returned error", error);
+            return error;
+        }
+
+        /* Serialize main items */
+        error = file_serialize_list(&serialized,
+                                    FILE_SER_NEW,
+                                    FILE_ITEM_REF,
+                                    ctx->config.set,
+                                    ctx->config.outmode,
+                                    ctx->config.main_fmt_cfg);
+        if (error) {
+            TRACE_ERROR_NUMBER("Failed to serialize main set", error);
+            col_destroy_collection(leftovers);
+            return error;
+        }
+
+        if (ctx->config.use_leftovers) {
+            /* Do we have to jam leftovers? */
+            if (ctx->config.jam_leftovers) {
+                /* Serialise leftovers into one field */
+                error = file_serialize_list(&serialized_lo,
+                                            FILE_SER_NEW,
+                                            FILE_ITEM_REF,
+                                            leftovers,
+                                            ctx->config.mode_leftovers,
+                                            ctx->config.lo_fmt_cfg);
+                if (error) {
+                    TRACE_ERROR_NUMBER("Failed to serialize main set", error);
+                    col_destroy_collection(leftovers);
+                    elapi_free_serialized_data(serialized);
+                    return error;
+                }
+
+                /* Check if we go anything */
+                if (serialized_lo->length) {
+                    /* Append leftovers item */
+                    error = file_serialize_item(serialized,
+                                                COL_TYPE_STRING,
+                                                serialized_lo->length + 1,
+                                                serialized_lo->buffer,
+                                                ctx->config.outmode,
+                                                ctx->config.main_fmt_cfg);
+                }
+                else {
+                    /* Put empty item */
+                    error = file_serialize_item(serialized,
+                                                COL_TYPE_BINARY,
+                                                0,
+                                                NULL,
+                                                ctx->config.outmode,
+                                                ctx->config.main_fmt_cfg);
+                }
+                if (error) {
+                    TRACE_ERROR_NUMBER("Failed to serialize main set", error);
+                    col_destroy_collection(leftovers);
+                    elapi_free_serialized_data(serialized);
+                    elapi_free_serialized_data(serialized_lo);
+                    return error;
+                }
+
+                /* Done with the jammed leftovers */
+                elapi_free_serialized_data(serialized_lo);
+            }
+            else {
+                /* Leftovers are added as normal fields */
+                error = file_serialize_list(&serialized,
+                                            FILE_SER_APPEND,
+                                            FILE_ITEM_REF,
+                                            leftovers,
+                                            ctx->config.outmode,
+                                            ctx->config.main_fmt_cfg);
+                if (error) {
+                    TRACE_ERROR_NUMBER("Failed to serialize main set", error);
+                    col_destroy_collection(leftovers);
+                    elapi_free_serialized_data(serialized);
+                    return error;
+                }
+            }
+            /* Do not need leftovers */
+            col_destroy_collection(leftovers);
+        }
+    }
+    else {
+        /* No set is defined - the whole event is processed */
+        error = file_serialize_list(&serialized,
+                                    FILE_SER_NEW,
+                                    FILE_ITEM_DIRECT,
+                                    event,
+                                    ctx->config.outmode,
+                                    ctx->config.main_fmt_cfg);
+        if (error) {
+            TRACE_ERROR_NUMBER("Failed to serialize event", error);
+            return error;
+        }
+    }
+
+    *out_data = serialized;
+
+    TRACE_FLOW_STRING("file_prep_data", "Exit");
+    return error;
+
+}
diff --git a/common/elapi/providers/file/file_util.h b/common/elapi/providers/file/file_util.h
new file mode 100644
index 000000000..f4c0e4bf5
--- /dev/null
+++ b/common/elapi/providers/file/file_util.h
@@ -0,0 +1,48 @@
+/*
+    ELAPI
+
+    Header for file provider utility functions.
+
+    Copyright (C) Dmitri Pal <dpal@redhat.com> 2009
+
+    This program is free software; you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation; either version 3 of the License, or
+    (at your option) any later version.
+    This program is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+    You should have received a copy of the GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef FILE_UTIL_H
+#define FILE_UTIL_H
+
+#include "file_provider.h"
+#include "elapi_basic.h"
+#include "collection.h"
+
+/* Sepcific format related includes */
+#include "file_fmt_csv.h"
+
+/* Leftovers' class and name */
+#define FILE_LO_NAME    "lo"
+#define FILE_LO_CLASS   20300
+
+/* Allocate a new one or add to existing */
+#define FILE_SER_NEW      0
+#define FILE_SER_APPEND   1
+
+/* Denotes how data is referenced */
+#define FILE_ITEM_DIRECT 0  /* Data is in the collection */
+#define FILE_ITEM_REF    1  /* Collection contains references */
+
+
+/* Function to prepare data for logging */
+int file_prep_data(struct elapi_data_out **out_data,
+                   struct file_prvdr_ctx *ctx,
+                   struct collection_item *event);
+
+#endif