diff --git a/lib/Makefile.am b/lib/Makefile.am
index 83434df..8e4a625 100644
--- a/lib/Makefile.am
+++ b/lib/Makefile.am
@@ -17,7 +17,7 @@ pkginclude_HEADERS = afinter.h center.h compat.h filter-expr-parser.h logmatcher
   logreader.h logwriter.h msg-format.h pragma-parser.h str-format.h timeutils.h \
   block-ref-parser.h children.h driver.h gsocket.h logpipe.h \
   logrewrite.h mainloop.h nvtable.h rewrite-expr-parser.h \
-  syslog-names.h utils.h cfg-grammar.h cfg-lex.h
+  syslog-names.h utils.h cfg-grammar.h cfg-lex.h logqueue_base.h
 
 libsyslog_ng_la_SOURCES = \
 	misc.c misc.h utils.c utils.h messages.c messages.h syslog-names.c syslog-names.h \
diff --git a/lib/cfg-grammar.y b/lib/cfg-grammar.y
index 1516b2b..5075b50 100644
--- a/lib/cfg-grammar.y
+++ b/lib/cfg-grammar.y
@@ -26,6 +26,7 @@
 
 /* YYSTYPE and YYLTYPE is defined by the lexer */
 #include "cfg-lexer.h"
+#include "driver.h"
 
 }
 
@@ -96,6 +97,7 @@
 %token LL_CONTEXT_PRAGMA              11
 %token LL_CONTEXT_FORMAT              12
 %token LL_CONTEXT_TEMPLATE_FUNC       13
+%token LL_CONTEXT_INNER_DEST          14
 
 /* statements */
 %token KW_SOURCE                      10000
@@ -299,6 +301,7 @@
 LogDriver *last_driver;
 LogReaderOptions *last_reader_options;
 LogWriterOptions *last_writer_options;
+LogDestDriverOptions *last_dest_driver_options;
 LogTemplate *last_template;
 GList *last_rewrite_expr;
 GList *last_parser_expr;
@@ -757,7 +760,7 @@ dest_writer_options
 
 dest_writer_option
 	: KW_FLAGS '(' dest_writer_options_flags ')' { last_writer_options->options = $3; }
-	| KW_LOG_FIFO_SIZE '(' LL_NUMBER ')'	{ last_writer_options->mem_fifo_size = $3; }
+	| KW_LOG_FIFO_SIZE '(' LL_NUMBER ')'	{ last_writer_options->mem_fifo_size = last_dest_driver_options->qoverflow_size = $3; }
 	| KW_FLUSH_LINES '(' LL_NUMBER ')'		{ last_writer_options->flush_lines = $3; }
 	| KW_FLUSH_TIMEOUT '(' LL_NUMBER ')'	{ last_writer_options->flush_timeout = $3; }
         | KW_SUPPRESS '(' LL_NUMBER ')'            { last_writer_options->suppress = $3; }
diff --git a/lib/cfg-lexer.c b/lib/cfg-lexer.c
index 60e2d5c..bf9ca11 100644
--- a/lib/cfg-lexer.c
+++ b/lib/cfg-lexer.c
@@ -858,6 +858,7 @@ static const gchar *lexer_contexts[] =
   [LL_CONTEXT_PRAGMA] = "pragma",
   [LL_CONTEXT_FORMAT] = "format",
   [LL_CONTEXT_TEMPLATE_FUNC] = "template-func",
+  [LL_CONTEXT_INNER_DEST] = "inner-dest",
 };
 
 gint
diff --git a/lib/driver.c b/lib/driver.c
index 2dc866c..017526d 100644
--- a/lib/driver.c
+++ b/lib/driver.c
@@ -39,6 +39,16 @@ log_drv_init_instance(LogDriver *self)
   self->super.free_fn = log_drv_free;
 }
 
+extern LogDestDriverOptions *last_dest_driver_options;
+
+void
+log_dest_drv_init_instance(LogDestDriver *self)
+{
+  self->options = last_dest_driver_options = g_new0(LogDestDriverOptions,1);
+  self->options->qoverflow_size = configuration->log_fifo_size;
+  log_drv_init_instance(&self->super);
+}
+
 void
 log_drv_free(LogPipe *s)
 {
diff --git a/lib/driver.h b/lib/driver.h
index 1360c65..d1741a5 100644
--- a/lib/driver.h
+++ b/lib/driver.h
@@ -27,6 +27,7 @@
 
 #include "syslog-ng.h"
 #include "logpipe.h"
+#include "logqueue.h"
 #include "cfg.h"
 
 typedef struct _LogDriver
@@ -38,14 +39,39 @@ typedef struct _LogDriver
   struct _LogDriver *drv_next;
 } LogDriver;
 
+typedef struct _LogDestDriverOptions
+{
+  gint qoverflow_size;
+} LogDestDriverOptions;
+
+typedef struct _LogDestDriver
+{
+  LogDriver super;
+  LogDestDriverOptions *options;
+  LogQueue* (*create_queue)(struct _LogDestDriver *dd);
+  gchar * (*format_persist_name)(struct _LogDestDriver *dd, gboolean file);
+  guint32 *stored_messages;
+} LogDestDriver;
+
 
 void log_drv_append(LogDriver *self, LogDriver *next);
-void log_drv_set_next_hop(LogDriver *self, LogPipe *next_hop);
-void log_drv_set_fifo_size(LogDriver *self, gint fifo_size);
 
 void log_drv_init_instance(LogDriver *self);
+void log_dest_drv_init_instance(LogDestDriver *self);
 void log_drv_free(LogPipe *self);
 
+static inline LogQueue* log_dest_drv_create_queue(LogDestDriver *dd)
+{
+  if(dd->create_queue)
+    {
+      return dd->create_queue(dd);
+    }
+  else
+    {
+      return log_queue_new(dd->options->qoverflow_size);
+    }
+}
+
 static inline LogDriver *
 log_drv_ref(LogDriver *s)
 {
diff --git a/lib/logqueue.c b/lib/logqueue.c
index 8732cd6..54defce 100644
--- a/lib/logqueue.c
+++ b/lib/logqueue.c
@@ -45,6 +45,7 @@
 
 struct _LogQueue
 {
+  LogQueueBase super;
   GMutex *lock;
   GQueue *qoverflow;   /* entries that did not fit to the disk based queue */
   GQueue *qbacklog;    /* entries that were sent but not acked yet */
@@ -60,9 +61,13 @@ struct _LogQueue
 };
 
 /* NOTE: this is inherently racy */
-gint64 
+gint64
 log_queue_get_length(LogQueue *self)
 {
+  if (self->super.get_length)
+    /* call the 'inherited' method */
+    return self->super.get_length(&self->super);
+
   return (self->qoverflow->length / 2);
 }
 
@@ -103,8 +108,12 @@ log_queue_push_notify(LogQueue *self)
 gboolean
 log_queue_push_tail(LogQueue *self, LogMessage *msg, const LogPathOptions *path_options)
 {
+  if (self->super.push_tail)
+    /* call the 'inherited' method */
+    return self->super.push_tail(&self->super, msg, path_options);
+
   LogPathOptions local_options = *path_options;
-  
+
   g_mutex_lock(self->lock);
   if ((self->qoverflow->length / 2) < self->qoverflow_size)
     {
@@ -130,6 +139,10 @@ log_queue_push_tail(LogQueue *self, LogMessage *msg, const LogPathOptions *path_
 gboolean
 log_queue_push_head(LogQueue *self, LogMessage *msg, const LogPathOptions *path_options)
 {
+  if (self->super.push_head)
+    /* call the 'inherited' method */
+    return self->super.push_head(&self->super, msg, path_options);
+
   g_mutex_lock(self->lock);
   g_queue_push_head(self->qoverflow, LOG_PATH_OPTIONS_TO_POINTER(path_options));
   g_queue_push_head(self->qoverflow, msg);
@@ -143,6 +156,10 @@ log_queue_push_head(LogQueue *self, LogMessage *msg, const LogPathOptions *path_
 void
 log_queue_reset_parallel_push(LogQueue *self)
 {
+  if (self->super.reset_parallel_push)
+    /* call the 'inherited' method */
+    return self->super.reset_parallel_push(&self->super);
+
   g_mutex_lock(self->lock);
   self->parallel_push_notify = NULL;
   self->parallel_push_data = NULL;
@@ -161,6 +178,10 @@ log_queue_check_items(LogQueue *self, gint batch_items, gboolean *partial_batch,
 {
   gint64 num_elements;
 
+  if (self->super.check_items)
+    /* call the 'inherited' method */
+    return self->super.check_items(&self->super, batch_items, partial_batch, timeout, parallel_push_notify, user_data, user_data_destroy);
+
   g_mutex_lock(self->lock);
 
   /* drop reference to the previous callback/userdata */
@@ -258,6 +279,10 @@ log_queue_check_items(LogQueue *self, gint batch_items, gboolean *partial_batch,
 gboolean
 log_queue_pop_head(LogQueue *self, LogMessage **msg, LogPathOptions *path_options, gboolean push_to_backlog)
 {
+  if (self->super.pop_head)
+    /* call the 'inherited' method */
+    return self->super.pop_head(&self->super, msg, path_options, push_to_backlog);
+
   g_mutex_lock(self->lock);
   if (self->throttle && self->throttle_buckets == 0)
     {
@@ -290,10 +315,14 @@ log_queue_pop_head(LogQueue *self, LogMessage **msg, LogPathOptions *path_option
 void
 log_queue_ack_backlog(LogQueue *self, gint n)
 {
+  if (self->super.ack_backlog)
+    /* call the 'inherited' method */
+    return self->super.ack_backlog(&self->super, n);
+
   LogMessage *msg;
   LogPathOptions path_options = LOG_PATH_OPTIONS_INIT;
   gint i;
-  
+
   g_mutex_lock(self->lock);
   for (i = 0; i < n; i++)
     {
@@ -315,6 +344,10 @@ log_queue_ack_backlog(LogQueue *self, gint n)
 void
 log_queue_rewind_backlog(LogQueue *self)
 {
+  if (self->super.rewind_backlog)
+    /* call the 'inherited' method */
+    return self->super.rewind_backlog(&self->super);
+
   LogPathOptions path_options = LOG_PATH_OPTIONS_INIT;
   LogMessage *msg;
   gint i, n;
@@ -325,7 +358,7 @@ log_queue_rewind_backlog(LogQueue *self)
     {
       msg = g_queue_pop_head(self->qbacklog);
       POINTER_TO_LOG_PATH_OPTIONS(g_queue_pop_head(self->qbacklog), &path_options);
-      
+
       /* NOTE: reverse order as we are pushing to the head */
       g_queue_push_head(self->qoverflow, LOG_PATH_OPTIONS_TO_POINTER(&path_options));
       g_queue_push_head(self->qoverflow, msg);
@@ -337,6 +370,10 @@ log_queue_rewind_backlog(LogQueue *self)
 void
 log_queue_set_throttle(LogQueue *self, gint throttle)
 {
+  if (self->super.set_throttle)
+    /* call the 'inherited' method */
+    return self->super.set_throttle(&self->super, throttle);
+
   self->throttle = throttle;
   self->throttle_buckets = throttle;
 }
@@ -345,7 +382,7 @@ LogQueue *
 log_queue_new(gint qoverflow_size)
 {
   LogQueue *self = g_new0(LogQueue, 1);
-  
+
   self->qoverflow = g_queue_new();
   self->qbacklog = g_queue_new();
   self->qoverflow_size = qoverflow_size;
@@ -360,7 +397,7 @@ log_queue_free_queue(GQueue *q)
     {
       LogMessage *lm;
       LogPathOptions path_options = LOG_PATH_OPTIONS_INIT;
-      
+
       lm = g_queue_pop_head(q);
       POINTER_TO_LOG_PATH_OPTIONS(g_queue_pop_head(q), &path_options);
       log_msg_ack(lm, &path_options);
@@ -372,10 +409,42 @@ log_queue_free_queue(GQueue *q)
 void
 log_queue_free(LogQueue *self)
 {
+  if (self->super.free)
+    /* call the 'inherited' method */
+    return self->super.free(&self->super);
+
   log_queue_free_queue(self->qoverflow);
   log_queue_free_queue(self->qbacklog);
   g_mutex_free(self->lock);
   g_free(self);
 }
 
+void
+log_queue_drop_message_warning(LogQueue *self, gchar *property)
+{
+  if (self->super.drop_message_warning)
+    /* call the 'inherited' method */
+    self->super.drop_message_warning(&self->super, property);
+  else {
+    char msg[128];
+    if (!property)
+      property = "";
+    snprintf(msg, 128, "Destination queue full, dropping %smessage", property);
+    msg_debug(msg,
+           evt_tag_int("queue_len", log_queue_get_length(self)),
+           evt_tag_int("log_fifo_size", self->qoverflow_size),
+           NULL);
+    }
+}
+
+gboolean
+log_queue_save_queue(LogQueue *self, gboolean *persistent)
+{
+  if (self->super.save_queue)
+    /* call the 'inherited' method */
+    return self->super.save_queue(&self->super, persistent);
 
+  /* default behaviour: save the world  - do nothing */
+  *persistent = TRUE;
+  return TRUE;
+}
diff --git a/lib/logqueue.h b/lib/logqueue.h
index 3fbbcc6..4b9c7d6 100644
--- a/lib/logqueue.h
+++ b/lib/logqueue.h
@@ -26,11 +26,10 @@
 #define LOGQUEUE_H_INCLUDED
 
 #include "logmsg.h"
+#include "logqueue_base.h"
 
 typedef struct _LogQueue LogQueue;
 
-typedef void (*LogQueuePushNotifyFunc)(gpointer user_data);
-
 void log_queue_reset_parallel_push(LogQueue *self);
 gboolean log_queue_check_items(LogQueue *self, gint batch_items, gboolean *partial_batch, gint *timeout, LogQueuePushNotifyFunc parallel_push_notify, gpointer user_data, GDestroyNotify user_data_destroy);
 gint64 log_queue_get_length(LogQueue *self);
@@ -44,4 +43,7 @@ LogQueue *log_queue_new(gint qoverflow_size);
 void log_queue_set_throttle(LogQueue *self, gint throttle);
 void log_queue_free(LogQueue *self);
 
+gboolean log_queue_save_queue(LogQueue *self, gboolean *persistent);
+
+void log_queue_drop_message_warning(LogQueue *self, gchar *property);
 #endif
diff --git a/lib/logqueue_base.h b/lib/logqueue_base.h
new file mode 100644
index 0000000..ce6404b
--- /dev/null
+++ b/lib/logqueue_base.h
@@ -0,0 +1,27 @@
+/*
+ * COPYRIGHTHERE
+ */
+#ifndef LOGQUEUE_BASE_H_INCLUDED
+#define LOGQUEUE_BASE_H_INCLUDED
+
+typedef void (*LogQueuePushNotifyFunc)(gpointer user_data);
+
+typedef struct _LogQueueBase LogQueueBase;
+
+struct _LogQueueBase
+{
+  gint64 (*get_length)(LogQueueBase *self);
+  gboolean (*push_tail)(LogQueueBase *self, LogMessage *msg, const LogPathOptions *path_options);
+  gboolean (*push_head)(LogQueueBase *self, LogMessage *msg, const LogPathOptions *path_options);
+  void (*reset_parallel_push)(LogQueueBase *self);
+  gboolean (*check_items)(LogQueueBase *self, gint batch_items, gboolean *partial_batch, gint *timeout, LogQueuePushNotifyFunc parallel_push_notify, gpointer user_data, GDestroyNotify user_data_destroy);
+  gboolean (*pop_head)(LogQueueBase *self, LogMessage **msg, LogPathOptions *path_options, gboolean push_to_backlog);
+  void (*ack_backlog)(LogQueueBase *self, gint n);
+  void (*rewind_backlog)(LogQueueBase *self);
+  void (*set_throttle)(LogQueueBase *self, gint throttle);
+  void (*free)(LogQueueBase *self);
+  void (*drop_message_warning)(LogQueueBase *self, gchar *msg_property);
+  gboolean (*save_queue)(LogQueueBase *self, gboolean *persistent);
+};
+
+#endif
diff --git a/lib/logwriter.c b/lib/logwriter.c
index 3bf5a05..4ee75a4 100644
--- a/lib/logwriter.c
+++ b/lib/logwriter.c
@@ -70,6 +70,7 @@ struct _LogWriter
   LogProto *proto;
   gboolean watches_running:1, suspended:1, working:1, flush_waiting_for_timeout:1;
   time_t last_msg_time;
+  LogDestDriver *dest_driver;
 };
 
 /**
@@ -360,7 +361,8 @@ log_writer_stop_watches(LogWriter *self)
       if (iv_task_registered(&self->immed_io_task))
         iv_task_unregister(&self->immed_io_task);
 
-      log_queue_reset_parallel_push(self->queue);
+      if (self->queue)
+        log_queue_reset_parallel_push(self->queue);
 
       self->watches_running = FALSE;
     }
@@ -483,9 +485,7 @@ log_writer_last_msg_flush(LogWriter *self)
   if (!log_queue_push_tail(self->queue, m, &path_options))
     {
       stats_counter_inc(self->dropped_messages);
-      msg_debug("Destination queue full, dropping suppressed message",
-                evt_tag_int("log_fifo_size", self->options->mem_fifo_size),
-                NULL);
+      log_queue_drop_message_warning(self->queue, "suppressed ");
       log_msg_drop(m, &path_options);
     }
 
@@ -609,9 +609,7 @@ log_writer_queue(LogPipe *s, LogMessage *lm, const LogPathOptions *path_options)
       /* we don't send a message here since the system is draining anyway */
       
       stats_counter_inc(self->dropped_messages);
-      msg_debug("Destination queue full, dropping message",
-                evt_tag_int("log_fifo_size", self->options->mem_fifo_size),
-                NULL);
+      log_queue_drop_message_warning(self->queue, "");
       log_msg_drop(lm, path_options);
       return;
     }
@@ -873,7 +871,7 @@ log_writer_flush(LogWriter *self, gboolean flush_all)
       LogPathOptions path_options = LOG_PATH_OPTIONS_INIT;
       gboolean consumed = FALSE;
       
-      if (!log_queue_pop_head(self->queue, &lm, &path_options, FALSE))
+      if (!self->queue || !log_queue_pop_head(self->queue, &lm, &path_options, FALSE))
         {
           /* no more items are available */
           break;
@@ -964,7 +962,7 @@ log_writer_init(LogPipe *s)
   iv_event_register(&self->queue_filled);
 
   if (!self->queue)
-    self->queue = log_queue_new(self->options->mem_fifo_size);
+    self->queue = log_dest_drv_create_queue(self->dest_driver);
   log_queue_set_throttle(self->queue, self->options->throttle);
   
   if ((self->options->options & LWO_NO_STATS) == 0 && !self->dropped_messages)
@@ -1013,12 +1011,17 @@ log_writer_free(LogPipe *s)
   
   if (self->line_buffer)
     g_string_free(self->line_buffer, TRUE);
+  self->line_buffer = NULL;
   if (self->queue)
     log_queue_free(self->queue);
+  self->queue = NULL;
   if (self->last_msg)
     log_msg_unref(self->last_msg);
+  self->last_msg = NULL;
   g_free(self->stats_id);
+  self->stats_id = NULL;
   g_free(self->stats_instance);
+  self->stats_instance  = NULL;
   log_pipe_free_method(s);
 }
 
@@ -1066,6 +1069,23 @@ log_writer_reopen(LogPipe *s, LogProto *proto)
   main_loop_call((MainLoopTaskFunc) log_writer_reopen_deferred, args, TRUE);
 }
 
+
+gboolean
+log_writer_save_queue(LogPipe *s)
+{
+  LogWriter *self = (LogWriter *) s;
+  gboolean success;
+  gboolean persistent;
+
+  success = log_queue_save_queue(self->queue, &persistent);
+  if (success && persistent)
+    {
+      log_queue_free(self->queue);
+      self->queue = NULL;
+    }
+  return success;
+}
+
 void
 log_writer_set_options(LogWriter *self, LogPipe *control, LogWriterOptions *options, gint stats_level, gint stats_source, const gchar *stats_id, const gchar *stats_instance)
 {
@@ -1076,11 +1096,11 @@ log_writer_set_options(LogWriter *self, LogPipe *control, LogWriterOptions *opti
   self->stats_source = stats_source;
   self->stats_id = stats_id ? g_strdup(stats_id) : NULL;
   self->stats_instance = stats_instance ? g_strdup(stats_instance) : NULL;
-
+  init_sequence_number(&self->seq_num);
 }
 
 LogPipe *
-log_writer_new(guint32 flags)
+log_writer_new(guint32 flags, LogDestDriver *dest_driver)
 {
   LogWriter *self = g_new0(LogWriter, 1);
   
@@ -1090,6 +1110,7 @@ log_writer_new(guint32 flags)
   self->super.queue = log_writer_queue;
   self->super.free_fn = log_writer_free;
   self->flags = flags;
+  self->dest_driver = dest_driver;
   self->line_buffer = g_string_sized_new(128);
 
   log_writer_init_watches(self);
@@ -1097,6 +1118,11 @@ log_writer_new(guint32 flags)
   return &self->super;
 }
 
+void log_writer_set_dest_driver(LogWriter *self, LogDestDriver *dest_driver)
+{
+  self->dest_driver = dest_driver;
+}
+
 void 
 log_writer_options_defaults(LogWriterOptions *options)
 {
diff --git a/lib/logwriter.h b/lib/logwriter.h
index 18a60a7..97826ae 100644
--- a/lib/logwriter.h
+++ b/lib/logwriter.h
@@ -28,6 +28,7 @@
 #include "logpipe.h"
 #include "templates.h"
 #include "logqueue.h"
+#include "driver.h"
 #include "logproto.h"
 #include "timeutils.h"
 
@@ -81,7 +82,9 @@ void log_writer_format_log(LogWriter *self, LogMessage *lm, GString *result);
 gboolean log_writer_has_pending_writes(LogWriter *self);
 gboolean log_writer_opened(LogWriter *self);
 void log_writer_reopen(LogPipe *s, LogProto *proto);
-LogPipe *log_writer_new(guint32 flags);
+gboolean log_writer_save_queue(LogPipe *);
+LogPipe *log_writer_new(guint32 flags, LogDestDriver *dest_driver);
+void log_writer_set_dest_driver(LogWriter *self, LogDestDriver *dest_driver);
 
 void log_writer_options_set_template_escape(LogWriterOptions *options, gboolean enable);
 void log_writer_options_defaults(LogWriterOptions *options);
diff --git a/lib/plugin.c b/lib/plugin.c
index 9cdd7ce..8b3ad93 100644
--- a/lib/plugin.c
+++ b/lib/plugin.c
@@ -130,6 +130,46 @@ plugin_parse_config(Plugin *self, GlobalConfig *cfg, YYLTYPE *yylloc)
   return instance;
 }
 
+gpointer
+plugin_parse_config_user_data(Plugin *self, GlobalConfig *cfg, YYLTYPE *yylloc, gpointer user_data)
+{
+  gpointer instance = user_data;
+
+  g_assert(self->construct == NULL);
+
+  /* make sure '_' and '-' are handled equally in plugin name */
+  if (!self->setup_context)
+    {
+      CfgTokenBlock *block;
+      YYSTYPE token;
+
+      block = cfg_token_block_new();
+
+      memset(&token, 0, sizeof(token));
+      token.type = LL_TOKEN;
+      token.token = self->type;
+      cfg_token_block_add_token(block, &token);
+      cfg_lexer_push_context(cfg->lexer, self->parser->context, self->parser->keywords, self->parser->name);
+      cfg_lexer_lookup_keyword(cfg->lexer, &token, yylloc, self->name);
+      cfg_lexer_pop_context(cfg->lexer);
+      cfg_token_block_add_token(block, &token);
+
+      cfg_lexer_inject_token_block(cfg->lexer, block);
+    }
+  else
+    {
+      (self->setup_context)(self, cfg, self->type, self->name);
+    }
+
+  if (!cfg_parser_parse(self->parser, cfg->lexer, &instance))
+    {
+      cfg_parser_cleanup(self->parser, instance);
+      instance = NULL;
+    }
+
+  return instance;
+}
+
 gboolean
 plugin_load_module(const gchar *module_name, GlobalConfig *cfg, CfgArgs *args)
 {
diff --git a/lib/plugin.h b/lib/plugin.h
index f3bcfbd..79b23df 100644
--- a/lib/plugin.h
+++ b/lib/plugin.h
@@ -50,6 +50,7 @@ struct _Plugin
 Plugin *plugin_find(GlobalConfig *cfg, gint plugin_type, const gchar *plugin_name);
 gpointer plugin_construct(Plugin *self, GlobalConfig *cfg, gint plugin_type, const gchar *plugin_name);
 gpointer plugin_parse_config(Plugin *plugin, GlobalConfig *cfg, YYLTYPE *yylloc);
+gpointer plugin_parse_config_user_data(Plugin *self, GlobalConfig *cfg, YYLTYPE *yylloc, gpointer user_data);
 
 
 /* plugin side API */
diff --git a/modules/affile/affile-grammar.ym b/modules/affile/affile-grammar.ym
index 34aeef9..98a2612 100644
--- a/modules/affile/affile-grammar.ym
+++ b/modules/affile/affile-grammar.ym
@@ -39,6 +39,7 @@
 
 extern LogReaderOptions *last_reader_options;
 extern LogWriterOptions *last_writer_options;
+extern LogDestDriverOptions *last_dest_driver_options;
 
 }
 
diff --git a/modules/affile/affile.c b/modules/affile/affile.c
index 39f6b77..eec4cd5 100644
--- a/modules/affile/affile.c
+++ b/modules/affile/affile.c
@@ -557,7 +557,7 @@ affile_dw_reopen(AFFileDestWriter *self)
                 evt_tag_str("filename", self->filename->str),
                 evt_tag_errno(EVT_TAG_OSERROR, errno),
                 NULL);
-      return self->owner->super.optional;
+      return self->owner->super.super.optional;
     }
   return TRUE;
 }
@@ -591,8 +591,8 @@ affile_dw_init(LogPipe *s)
 
   if (!self->writer)
     {
-      self->writer = log_writer_new(LW_FORMAT_FILE | ((self->owner->flags & AFFILE_PIPE || !is_file_regular(self->filename->str)) ? 0 : LW_FILE_FD));
-      log_writer_set_options((LogWriter *) self->writer, s, &self->owner->writer_options, 1, self->owner->flags & AFFILE_PIPE ? SCS_PIPE : SCS_FILE, self->owner->super.id, self->filename->str);
+      self->writer = log_writer_new(LW_FORMAT_FILE | ((self->owner->flags & AFFILE_PIPE || !is_file_regular(self->filename->str)) ? 0 : LW_FILE_FD), &self->owner->super);
+      log_writer_set_options((LogWriter *) self->writer, s, &self->owner->writer_options, 1, self->owner->flags & AFFILE_PIPE ? SCS_PIPE : SCS_FILE, self->owner->super.super.id, self->filename->str);
 
       if (!log_pipe_init(self->writer, NULL))
         {
@@ -657,11 +657,11 @@ static void
 affile_dw_set_owner(AFFileDestWriter *self, AFFileDestDriver *owner)
 {
   if (self->owner)
-    log_pipe_unref(&self->owner->super.super);
-  log_pipe_ref(&owner->super.super);
+    log_pipe_unref(&self->owner->super.super.super);
+  log_pipe_ref(&owner->super.super.super);
   self->owner = owner;
   if (self->writer)
-    log_writer_set_options((LogWriter *) self->writer, &self->super, &owner->writer_options, 1, SCS_FILE, self->owner->super.id, self->filename->str);
+    log_writer_set_options((LogWriter *) self->writer, &self->super, &owner->writer_options, 1, SCS_FILE, self->owner->super.super.id, self->filename->str);
   
 }
 
@@ -673,7 +673,7 @@ affile_dw_free(LogPipe *s)
   log_pipe_unref(self->writer);
   self->writer = NULL;
   g_string_free(self->filename, TRUE);
-  log_pipe_unref(&self->owner->super.super);
+  log_pipe_unref(&self->owner->super.super.super);
   log_pipe_free_method(s);
 }
 
@@ -688,7 +688,7 @@ affile_dw_new(AFFileDestDriver *owner, GString *filename)
   self->super.deinit = affile_dw_deinit;
   self->super.free_fn = affile_dw_free;  
   self->super.queue = affile_dw_queue;
-  log_pipe_ref(&owner->super.super);
+  log_pipe_ref(&owner->super.super.super);
   self->owner = owner;
   self->time_reopen = 60;
 
@@ -994,7 +994,7 @@ static LogPipe *
 affile_dd_open_writer(gpointer args[])
 {
   AFFileDestDriver *self = args[0];
-  GlobalConfig *cfg = log_pipe_get_config(&self->super.super);
+  GlobalConfig *cfg = log_pipe_get_config(&self->super.super.super);
   AFFileDestWriter *next;
 
   main_loop_assert_main_thread();
@@ -1153,13 +1153,13 @@ affile_dd_new(gchar *filename, guint32 flags)
 {
   AFFileDestDriver *self = g_new0(AFFileDestDriver, 1);
 
-  log_drv_init_instance(&self->super);
+  log_dest_drv_init_instance(&self->super);
   if ((flags & AFFILE_PIPE) == 0)
-    self->super.super.flags |= PIF_SOFT_FLOW_CONTROL;
-  self->super.super.init = affile_dd_init;
-  self->super.super.deinit = affile_dd_deinit;
-  self->super.super.queue = affile_dd_queue;
-  self->super.super.free_fn = affile_dd_free;
+    self->super.super.super.flags |= PIF_SOFT_FLOW_CONTROL;
+  self->super.super.super.init = affile_dd_init;
+  self->super.super.super.deinit = affile_dd_deinit;
+  self->super.super.super.queue = affile_dd_queue;
+  self->super.super.super.free_fn = affile_dd_free;
   self->filename_template = log_template_new(configuration, NULL, filename);
   self->flags = flags;
   self->file_uid = self->file_gid = -1;
@@ -1174,5 +1174,5 @@ affile_dd_new(gchar *filename, guint32 flags)
   self->time_reap = -1;
   log_template_options_defaults(&self->template_fname_options);
   g_static_mutex_init(&self->lock);
-  return &self->super;
+  return &self->super.super;
 }
diff --git a/modules/affile/affile.h b/modules/affile/affile.h
index 9754847..3c5412c 100644
--- a/modules/affile/affile.h
+++ b/modules/affile/affile.h
@@ -55,7 +55,7 @@ typedef struct _AFFileDestWriter AFFileDestWriter;
 
 typedef struct _AFFileDestDriver
 {
-  LogDriver super;
+  LogDestDriver super;
   GStaticMutex lock;
   LogTemplateOptions template_fname_options;
   LogTemplate *filename_template;
diff --git a/modules/afprog/afprog-grammar.ym b/modules/afprog/afprog-grammar.ym
index 1e9f05b..398b204 100644
--- a/modules/afprog/afprog-grammar.ym
+++ b/modules/afprog/afprog-grammar.ym
@@ -38,6 +38,7 @@
 
 extern LogReaderOptions *last_reader_options;
 extern LogWriterOptions *last_writer_options;
+extern LogDestDriverOptions *last_dest_driver_options;
 
 }
 
diff --git a/modules/afprog/afprog.c b/modules/afprog/afprog.c
index d3e0d6d..45ba0ee 100644
--- a/modules/afprog/afprog.c
+++ b/modules/afprog/afprog.c
@@ -247,8 +247,8 @@ afprogram_dd_exit(pid_t pid, int status, gpointer s)
                   evt_tag_int("status", status),
                   NULL);
       self->pid = -1;
-      log_pipe_deinit(&self->super.super);
-      log_pipe_init(&self->super.super, NULL);
+      log_pipe_deinit(&self->super.super.super);
+      log_pipe_init(&self->super.super.super, NULL);
     }
 }
 
@@ -271,17 +271,17 @@ afprogram_dd_init(LogPipe *s)
 
   /* parent */
   
-  child_manager_register(self->pid, afprogram_dd_exit, log_pipe_ref(&self->super.super), (GDestroyNotify) log_pipe_unref);
+  child_manager_register(self->pid, afprogram_dd_exit, log_pipe_ref(&self->super.super.super), (GDestroyNotify) log_pipe_unref);
   
   g_fd_set_nonblock(fd, TRUE);
   if (!self->writer)
     {
-      self->writer = log_writer_new(LW_FORMAT_FILE);
-      log_writer_set_options((LogWriter *) self->writer, s, &self->writer_options, 0, SCS_PROGRAM, self->super.id, self->cmdline->str);
+      self->writer = log_writer_new(LW_FORMAT_FILE, &self->super);
+      log_writer_set_options((LogWriter *) self->writer, s, &self->writer_options, 0, SCS_PROGRAM, self->super.super.id, self->cmdline->str);
     }
   log_writer_reopen(self->writer, log_proto_text_client_new(log_transport_plain_new(fd, 0)));
   log_pipe_init(self->writer, NULL);
-  log_pipe_append(&self->super.super, self->writer);
+  log_pipe_append(&self->super.super.super, self->writer);
   return TRUE;
 }
 
@@ -332,13 +332,13 @@ LogDriver *
 afprogram_dd_new(gchar *cmdline)
 {
   AFProgramDestDriver *self = g_new0(AFProgramDestDriver, 1);
-  log_drv_init_instance(&self->super);
+  log_dest_drv_init_instance(&self->super);
   
-  self->super.super.init = afprogram_dd_init;
-  self->super.super.deinit = afprogram_dd_deinit;
-  self->super.super.free_fn = afprogram_dd_free;
-  self->super.super.notify = afprogram_dd_notify;
+  self->super.super.super.init = afprogram_dd_init;
+  self->super.super.super.deinit = afprogram_dd_deinit;
+  self->super.super.super.free_fn = afprogram_dd_free;
+  self->super.super.super.notify = afprogram_dd_notify;
   self->cmdline = g_string_new(cmdline);
   log_writer_options_defaults(&self->writer_options);
-  return &self->super;
+  return &self->super.super;
 }
diff --git a/modules/afprog/afprog.h b/modules/afprog/afprog.h
index 8112174..502de75 100644
--- a/modules/afprog/afprog.h
+++ b/modules/afprog/afprog.h
@@ -39,7 +39,7 @@ typedef struct _AFProgramSourceDriver
 
 typedef struct _AFProgramDestDriver
 {
-  LogDriver super;
+  LogDestDriver super;
   GString *cmdline;
   LogPipe *writer;
   pid_t pid;
diff --git a/modules/afsocket/afinet.c b/modules/afsocket/afinet.c
index e6fb8ce..a3f06b9 100644
--- a/modules/afsocket/afinet.c
+++ b/modules/afsocket/afinet.c
@@ -546,9 +546,9 @@ afinet_dd_new(gint af, gchar *host, gint port, guint flags)
     self->super.transport = g_strdup("udp");
   else if (self->super.flags & AFSOCKET_STREAM)
     self->super.transport = g_strdup("tcp");
-  self->super.super.super.init = afinet_dd_init;
-  self->super.super.super.queue = afinet_dd_queue;
-  self->super.super.super.free_fn = afinet_dd_free;
+  self->super.super.super.super.init = afinet_dd_init;
+  self->super.super.super.super.queue = afinet_dd_queue;
+  self->super.super.super.super.free_fn = afinet_dd_free;
   if (af == AF_INET)
     {
       self->super.bind_addr = g_sockaddr_inet_new("0.0.0.0", 0);
@@ -564,5 +564,5 @@ afinet_dd_new(gint af, gchar *host, gint port, guint flags)
 #endif
     }
   self->super.setup_socket = afinet_dd_setup_socket;
-  return &self->super.super;
+  return &self->super.super.super;
 }
diff --git a/modules/afsocket/afsocket-grammar.ym b/modules/afsocket/afsocket-grammar.ym
index 8081b03..5adc672 100644
--- a/modules/afsocket/afsocket-grammar.ym
+++ b/modules/afsocket/afsocket-grammar.ym
@@ -36,6 +36,7 @@
 #include "afinet.h"
 #include "messages.h"
 #include "syslog-names.h"
+#include "plugin.h"
 
 #if ENABLE_SSL /* BEGIN MARK: tls */
 #include "tlscontext.h"
@@ -48,6 +49,7 @@ static SocketOptions *last_sock_options;
 static gint last_addr_family = AF_INET;
 extern LogReaderOptions *last_reader_options;
 extern LogWriterOptions *last_writer_options;
+extern LogDestDriverOptions *last_dest_driver_options;
 
 #if ENABLE_SSL
 TLSContext *last_tls_context;
@@ -375,6 +377,19 @@ dest_afinet_option
 	| inet_socket_option
 	| dest_writer_option
 	| dest_afsocket_option
+  | LL_IDENTIFIER  {
+            Plugin *p;
+            gint context = LL_CONTEXT_INNER_DEST;
+
+            p = plugin_find(configuration, context, $1);
+            CHECK_ERROR(p, @1, "%s plugin %s not found", cfg_lexer_lookup_context_name_by_type(context), $1);
+            gpointer val = plugin_parse_config_user_data(p, configuration, &@1, *last_driver);
+            free($1);
+            if (!val)
+              {
+                YYERROR;
+              }
+        }
 	;
 
 dest_afinet_udp_option
diff --git a/modules/afsocket/afsocket.c b/modules/afsocket/afsocket.c
index 5620141..5207a2b 100644
--- a/modules/afsocket/afsocket.c
+++ b/modules/afsocket/afsocket.c
@@ -821,15 +821,17 @@ afsocket_dd_set_keep_alive(LogDriver *s, gint enable)
 }
 
 
-static inline gchar *
-afsocket_dd_format_persist_name(AFSocketDestDriver *self, const gchar *dest_name, gboolean qfile)
+static gchar *
+afsocket_dd_format_persist_name(LogDestDriver *base, gboolean qfile)
 {
+  AFSocketDestDriver *self = (AFSocketDestDriver *)base;
+
   static gchar persist_name[128];
 
   g_snprintf(persist_name, sizeof(persist_name),
              qfile ? "afsocket_dd_qfile(%s,%s)" : "afsocket_dd_connection(%s,%s)",
              !!(self->flags & AFSOCKET_STREAM) ? "stream" : "dgram",
-             dest_name);
+             self->dest_name);
   return persist_name;
 }
 
@@ -1099,6 +1101,10 @@ gboolean
 afsocket_dd_init(LogPipe *s)
 {
   AFSocketDestDriver *self = (AFSocketDestDriver *) s;
+
+  self->super.format_persist_name = afsocket_dd_format_persist_name;
+  self->super.stored_messages = &self->stat_counter;
+
   GlobalConfig *cfg = log_pipe_get_config(s);
 
 #if ENABLE_SSL
@@ -1121,7 +1127,7 @@ afsocket_dd_init(LogPipe *s)
        * even while the connection is not established */
 
       if ((self->flags & AFSOCKET_KEEP_ALIVE))
-        self->writer = cfg_persist_config_fetch(cfg, afsocket_dd_format_persist_name(self, self->dest_name, FALSE));
+        self->writer = cfg_persist_config_fetch(cfg, afsocket_dd_format_persist_name((LogDestDriver *)self, FALSE));
 
       if (!self->writer)
         self->writer = log_writer_new(LW_FORMAT_PROTO |
@@ -1130,10 +1136,12 @@ afsocket_dd_init(LogPipe *s)
 #else
                                       ((self->flags & AFSOCKET_STREAM) ? LW_DETECT_EOF : 0) |
 #endif
-                                      (self->flags & AFSOCKET_SYSLOG_PROTOCOL ? LW_SYSLOG_PROTOCOL : 0));
-      log_writer_set_options((LogWriter *) self->writer, &self->super.super, &self->writer_options, 0, afsocket_dd_stats_source(self), self->super.id, afsocket_dd_stats_instance(self));
+                                      (self->flags & AFSOCKET_SYSLOG_PROTOCOL ? LW_SYSLOG_PROTOCOL : 0), &self->super);
+      else
+        log_writer_set_dest_driver((LogWriter *)self->writer, &self->super);
+      log_writer_set_options((LogWriter *) self->writer, &self->super.super.super, &self->writer_options, 0, afsocket_dd_stats_source(self), self->super.super.id, afsocket_dd_stats_instance(self));
       log_pipe_init(self->writer, NULL);
-      log_pipe_append(&self->super.super, self->writer);
+      log_pipe_append(&self->super.super.super, self->writer);
     }
 
   afsocket_dd_reconnect(self);
@@ -1149,12 +1157,13 @@ afsocket_dd_deinit(LogPipe *s)
   afsocket_dd_stop_watches(self);
   if (self->writer)
     {
+      log_writer_save_queue(self->writer);
       log_pipe_deinit(self->writer);
     }
 
   if (self->flags & AFSOCKET_KEEP_ALIVE)
     {
-      cfg_persist_config_add(cfg, afsocket_dd_format_persist_name(self, self->dest_name, FALSE), self->writer, (GDestroyNotify) log_pipe_unref, FALSE);
+      cfg_persist_config_add(cfg, afsocket_dd_format_persist_name((LogDestDriver *)self, FALSE), self->writer, (GDestroyNotify) log_pipe_unref, FALSE);
       self->writer = NULL;
     }
 
@@ -1207,14 +1216,14 @@ afsocket_dd_free(LogPipe *s)
 void
 afsocket_dd_init_instance(AFSocketDestDriver *self, SocketOptions *sock_options, guint32 flags, gchar *hostname, gchar *dest_name)
 {
-  log_drv_init_instance(&self->super);
+  log_dest_drv_init_instance(&self->super);
 
   log_writer_options_defaults(&self->writer_options);
-  self->super.super.init = afsocket_dd_init;
-  self->super.super.deinit = afsocket_dd_deinit;
-  self->super.super.queue = log_pipe_forward_msg;
-  self->super.super.free_fn = afsocket_dd_free;
-  self->super.super.notify = afsocket_dd_notify;
+  self->super.super.super.init = afsocket_dd_init;
+  self->super.super.super.deinit = afsocket_dd_deinit;
+  self->super.super.super.queue = log_pipe_forward_msg;
+  self->super.super.super.free_fn = afsocket_dd_free;
+  self->super.super.super.notify = afsocket_dd_notify;
   self->setup_socket = afsocket_dd_setup_socket;
   self->sock_options_ptr = sock_options;
   self->flags = flags  | AFSOCKET_KEEP_ALIVE;
diff --git a/modules/afsocket/afsocket.h b/modules/afsocket/afsocket.h
index 9578e8f..06c7a70 100644
--- a/modules/afsocket/afsocket.h
+++ b/modules/afsocket/afsocket.h
@@ -99,7 +99,7 @@ void afsocket_sd_free(LogPipe *self);
 
 struct _AFSocketDestDriver
 {
-  LogDriver super;
+  LogDestDriver super;
   guint32 flags;
   gint fd;
   LogPipe *writer;
@@ -118,6 +118,7 @@ struct _AFSocketDestDriver
   struct iv_timer reconnect_timer;
   SocketOptions *sock_options_ptr;
   gboolean (*setup_socket)(AFSocketDestDriver *s, gint fd);
+  guint32 stat_counter;
 };
 
 
diff --git a/modules/afsocket/afunix.c b/modules/afsocket/afunix.c
index e784f30..c001db3 100644
--- a/modules/afsocket/afunix.c
+++ b/modules/afsocket/afunix.c
@@ -135,5 +135,5 @@ afunix_dd_new(gchar *filename, guint flags)
     self->super.transport = g_strdup("unix-stream");
   self->super.bind_addr = g_sockaddr_unix_new(NULL);
   self->super.dest_addr = g_sockaddr_unix_new(filename);
-  return &self->super.super;
+  return &self->super.super.super;
 }
diff --git a/modules/afsql/afsql-grammar.ym b/modules/afsql/afsql-grammar.ym
index 699ae74..e5f46e5 100644
--- a/modules/afsql/afsql-grammar.ym
+++ b/modules/afsql/afsql-grammar.ym
@@ -33,7 +33,11 @@
 #include "cfg-parser.h"
 #include "afsql-grammar.h"
 #include "messages.h"
+#include "plugin.h"
+#include "logwriter.h"
 
+extern LogWriterOptions *last_writer_options;
+extern LogDestDriverOptions *last_dest_driver_options;
 }
 
 %name-prefix "afsql_"
@@ -94,7 +98,6 @@ dest_afsql_option
         | KW_COLUMNS '(' string_list ')'	{ afsql_dd_set_columns((*last_driver), $3); }
         | KW_INDEXES '(' string_list ')'        { afsql_dd_set_indexes((*last_driver), $3); }
         | KW_VALUES '(' dest_afsql_values ')'		{ afsql_dd_set_values((*last_driver), $3); }
-	| KW_LOG_FIFO_SIZE '(' LL_NUMBER ')'	{ afsql_dd_set_mem_fifo_size((*last_driver), $3); }
         | KW_FRAC_DIGITS '(' LL_NUMBER ')'         { afsql_dd_set_frac_digits((*last_driver), $3); }
 	| KW_TIME_ZONE '(' string ')'           { afsql_dd_set_send_time_zone((*last_driver),$3); free($3); }
 	| KW_LOCAL_TIME_ZONE '(' string ')'     { afsql_dd_set_local_time_zone((*last_driver),$3); free($3); }
@@ -104,6 +107,20 @@ dest_afsql_option
         | KW_FLUSH_TIMEOUT '(' LL_NUMBER ')'    { afsql_dd_set_flush_timeout((*last_driver), $3); }
         | KW_SESSION_STATEMENTS '(' string_list ')' { afsql_dd_set_session_statements((*last_driver), $3); }
         | KW_FLAGS '(' dest_afsql_flags ')'     { afsql_dd_set_flags((*last_driver), $3); }
+        | LL_IDENTIFIER {
+            Plugin *p;
+            gint context = LL_CONTEXT_INNER_DEST;
+
+            p = plugin_find(configuration, context, $1);
+            CHECK_ERROR(p, @1, "%s plugin %s not found", cfg_lexer_lookup_context_name_by_type(context), $1);
+            gpointer val = plugin_parse_config_user_data(p, configuration, &@1, *last_driver);
+            free($1);
+            if (!val)
+              {
+                YYERROR;
+              }
+        }
+        | KW_LOG_FIFO_SIZE '(' LL_NUMBER ')'	{ last_writer_options->mem_fifo_size = last_dest_driver_options->qoverflow_size = $3; }
         | KW_ENDIF {
 #endif /* ENABLE_SQL */
 }
diff --git a/modules/afsql/afsql.c b/modules/afsql/afsql.c
index e4eca4d..da90320 100644
--- a/modules/afsql/afsql.c
+++ b/modules/afsql/afsql.c
@@ -66,7 +66,7 @@ typedef struct _AFSqlField
  **/
 typedef struct _AFSqlDestDriver
 {
-  LogDriver super;
+  LogDestDriver super;
   /* read by the db thread */
   gchar *type;
   gchar *host;
@@ -80,7 +80,6 @@ typedef struct _AFSqlDestDriver
   GList *values;
   GList *indexes;
   LogTemplate *table;
-  gint mem_fifo_size;
   gint64 disk_fifo_size;
   gboolean use_time_recvd;
   gint fields_len;
@@ -229,14 +228,6 @@ afsql_dd_set_null_value(LogDriver *s, const gchar *null)
 }
 
 void
-afsql_dd_set_mem_fifo_size(LogDriver *s, gint mem_fifo_size)
-{
-  AFSqlDestDriver *self = (AFSqlDestDriver *) s;
-
-  self->mem_fifo_size = mem_fifo_size;
-}
-
-void
 afsql_dd_set_retries(LogDriver *s, gint num_retries)
 {
   AFSqlDestDriver *self = (AFSqlDestDriver *) s;
@@ -840,7 +831,7 @@ afsql_dd_database_thread(gpointer arg)
   gboolean sigexit = FALSE;
 
   msg_verbose("Database thread started",
-              evt_tag_str("driver", self->super.id),
+              evt_tag_str("driver", self->super.super.id),
               NULL);
   while (!self->db_thread_terminate)
     {
@@ -919,7 +910,7 @@ afsql_dd_database_thread(gpointer arg)
       afsql_dd_disconnect(self);
 
       msg_verbose("Database thread finished",
-                  evt_tag_str("driver", self->super.id),
+                  evt_tag_str("driver", self->super.super.id),
                   NULL);
     }
 
@@ -946,8 +937,10 @@ afsql_dd_stop_thread(AFSqlDestDriver *self)
 
 /* file param is true then its a file else its a queue*/
 static gchar *
-afsql_dd_format_persist_name(AFSqlDestDriver *self, gboolean file)
+afsql_dd_format_persist_name(LogDestDriver *dd, gboolean file)
 {
+  AFSqlDestDriver *self = (AFSqlDestDriver *)dd;
+
   static gchar persist_name[256];
 
   g_snprintf(persist_name, sizeof(persist_name),
@@ -989,14 +982,19 @@ afsql_dd_init(LogPipe *s)
       return FALSE;
     }
 
-  stats_register_counter(0, SCS_SQL | SCS_DESTINATION, self->super.id, afsql_dd_format_stats_instance(self), SC_TYPE_STORED, &self->stored_messages);
-  stats_register_counter(0, SCS_SQL | SCS_DESTINATION, self->super.id, afsql_dd_format_stats_instance(self), SC_TYPE_DROPPED, &self->dropped_messages);
+  stats_register_counter(0, SCS_SQL | SCS_DESTINATION, self->super.super.id, afsql_dd_format_stats_instance(self), SC_TYPE_STORED, &self->stored_messages);
+  stats_register_counter(0, SCS_SQL | SCS_DESTINATION, self->super.super.id, afsql_dd_format_stats_instance(self), SC_TYPE_DROPPED, &self->dropped_messages);
 
   if (!self->queue)
     {
-      self->queue = cfg_persist_config_fetch(cfg, afsql_dd_format_persist_name(self, FALSE));
-      if (!self->queue)
-        self->queue = log_queue_new(self->mem_fifo_size);
+      if (!self->super.create_queue)
+        {
+          self->queue = cfg_persist_config_fetch(cfg, afsql_dd_format_persist_name((LogDestDriver *)self, FALSE));
+          if (!self->queue)
+            self->queue = log_queue_new(cfg->log_fifo_size);
+        }
+      else
+        self->queue = self->super.create_queue(&self->super);
     }
 
   if (!self->fields)
@@ -1093,8 +1091,8 @@ afsql_dd_init(LogPipe *s)
 
  error:
 
-  stats_unregister_counter(SCS_SQL | SCS_DESTINATION, self->super.id, afsql_dd_format_stats_instance(self), SC_TYPE_STORED, &self->stored_messages);
-  stats_unregister_counter(SCS_SQL | SCS_DESTINATION, self->super.id, afsql_dd_format_stats_instance(self), SC_TYPE_DROPPED, &self->dropped_messages);
+  stats_unregister_counter(SCS_SQL | SCS_DESTINATION, self->super.super.id, afsql_dd_format_stats_instance(self), SC_TYPE_STORED, &self->stored_messages);
+  stats_unregister_counter(SCS_SQL | SCS_DESTINATION, self->super.super.id, afsql_dd_format_stats_instance(self), SC_TYPE_DROPPED, &self->dropped_messages);
 
   return FALSE;
 }
@@ -1106,11 +1104,12 @@ afsql_dd_deinit(LogPipe *s)
   GlobalConfig *cfg = log_pipe_get_config(s);
 
   afsql_dd_stop_thread(self);
-  cfg_persist_config_add(cfg, afsql_dd_format_persist_name(self, FALSE), self->queue, (GDestroyNotify) log_queue_free, FALSE);
+  if (!self->super.create_queue)
+    cfg_persist_config_add(cfg, afsql_dd_format_persist_name((LogDestDriver *)self, FALSE), self->queue, (GDestroyNotify) log_queue_free, FALSE);
   self->queue = NULL;
 
-  stats_unregister_counter(SCS_SQL | SCS_DESTINATION, self->super.id, afsql_dd_format_stats_instance(self), SC_TYPE_STORED, &self->stored_messages);
-  stats_unregister_counter(SCS_SQL | SCS_DESTINATION, self->super.id, afsql_dd_format_stats_instance(self), SC_TYPE_DROPPED, &self->dropped_messages);
+  stats_unregister_counter(SCS_SQL | SCS_DESTINATION, self->super.super.id, afsql_dd_format_stats_instance(self), SC_TYPE_STORED, &self->stored_messages);
+  stats_unregister_counter(SCS_SQL | SCS_DESTINATION, self->super.super.id, afsql_dd_format_stats_instance(self), SC_TYPE_DROPPED, &self->dropped_messages);
 
   return TRUE;
 }
@@ -1140,11 +1139,7 @@ afsql_dd_queue(LogPipe *s, LogMessage *msg, const LogPathOptions *path_options)
       /* we don't send a message here since the system is draining anyway */
 
       stats_counter_inc(self->dropped_messages);
-      msg_debug("Destination queue full, dropping message",
-                evt_tag_int("queue_len", log_queue_get_length(self->queue)),
-                evt_tag_int("mem_fifo_size", self->mem_fifo_size),
-                evt_tag_int("disk_fifo_size", self->disk_fifo_size),
-                NULL);
+      log_queue_drop_message_warning(self->queue, "");
       log_msg_drop(msg, path_options);
     }
 }
@@ -1224,11 +1219,13 @@ afsql_dd_new(void)
 {
   AFSqlDestDriver *self = g_new0(AFSqlDestDriver, 1);
 
-  log_drv_init_instance(&self->super);
-  self->super.super.init = afsql_dd_init;
-  self->super.super.deinit = afsql_dd_deinit;
-  self->super.super.queue = afsql_dd_queue;
-  self->super.super.free_fn = afsql_dd_free;
+  log_dest_drv_init_instance(&self->super);
+  self->super.super.super.init = afsql_dd_init;
+  self->super.super.super.deinit = afsql_dd_deinit;
+  self->super.super.super.queue = afsql_dd_queue;
+  self->super.super.super.free_fn = afsql_dd_free;
+  self->super.format_persist_name = afsql_dd_format_persist_name;
+  self->super.stored_messages = self->stored_messages;
 
   self->type = g_strdup("mysql");
   self->host = g_strdup("");
@@ -1245,7 +1242,6 @@ afsql_dd_new(void)
   self->uses_default_columns = TRUE;
   self->uses_default_values = TRUE;
   self->uses_default_indexes = TRUE;
-  self->mem_fifo_size = 1000;
   self->disk_fifo_size = 0;
   self->failed_message_counter = 0;
 
@@ -1259,7 +1255,7 @@ afsql_dd_new(void)
 
   log_template_options_defaults(&self->template_options);
   init_sequence_number(&self->seq_num);
-  return &self->super;
+  return &self->super.super;
 }
 
 gint
diff --git a/modules/afsql/afsql.h b/modules/afsql/afsql.h
index abef461..8fe68af 100644
--- a/modules/afsql/afsql.h
+++ b/modules/afsql/afsql.h
@@ -43,7 +43,6 @@ void afsql_dd_set_columns(LogDriver *s, GList *columns);
 void afsql_dd_set_values(LogDriver *s, GList *values);
 void afsql_dd_set_null_value(LogDriver *s, const gchar *null);
 void afsql_dd_set_indexes(LogDriver *s, GList *indexes);
-void afsql_dd_set_mem_fifo_size(LogDriver *s, gint mem_fifo_size);
 void afsql_dd_set_disk_fifo_size(LogDriver *s, gint64 disk_fifo_size);
 void afsql_dd_set_frac_digits(LogDriver *s, gint frac_digits);
 void afsql_dd_set_local_time_zone(LogDriver *s, const gchar *local_time_zone);
@@ -68,7 +67,6 @@ void afsql_dd_set_retries(LogDriver *s, gint num_retries);
 #define afsql_dd_set_columns(s, c)
 #define afsql_dd_set_values(s, v)
 #define afsql_dd_set_null_value(s, v)
-#define afsql_dd_set_mem_fifo_size(s, f)
 #define afsql_dd_set_disk_fifo_size(s, f)
 
 #define afsql_dd_new() 0
diff --git a/modules/afuser/afuser.c b/modules/afuser/afuser.c
index 922e664..97e4bc4 100644
--- a/modules/afuser/afuser.c
+++ b/modules/afuser/afuser.c
@@ -36,7 +36,7 @@
 
 typedef struct _AFUserDestDriver
 {
-  LogDriver super;
+  LogDestDriver super;
   GString *username;
   time_t disable_until;
 } AFUserDestDriver;
@@ -129,10 +129,10 @@ afuser_dd_new(gchar *user)
 {
   AFUserDestDriver *self = g_new0(AFUserDestDriver, 1);
   
-  log_drv_init_instance(&self->super);
-  self->super.super.init = afuser_dd_init;
-  self->super.super.deinit = afuser_dd_deinit;
-  self->super.super.queue = afuser_dd_queue;
+  log_dest_drv_init_instance(&self->super);
+  self->super.super.super.init = afuser_dd_init;
+  self->super.super.super.deinit = afuser_dd_deinit;
+  self->super.super.super.queue = afuser_dd_queue;
   self->username = g_string_new(user);
-  return &self->super;
+  return &self->super.super;
 }
diff --git a/tests/unit/test_logwriter.c b/tests/unit/test_logwriter.c
index c3c3da4..1fa1656 100644
--- a/tests/unit/test_logwriter.c
+++ b/tests/unit/test_logwriter.c
@@ -47,6 +47,7 @@ testcase(gchar *msg_string, gchar *template, gboolean use_syslog_protocol,gchar
   GString *res = g_string_sized_new(128);
   GError *error = NULL;
   LogMessage *msg = NULL;
+  LogDestDriver *dd = NULL;
   LogWriterOptions opt = {0};
   guint writer_flags = 0;
 
@@ -69,7 +70,9 @@ testcase(gchar *msg_string, gchar *template, gboolean use_syslog_protocol,gchar
     }
   opt.template = templ;
   msg = init_msg(msg_string, use_syslog_protocol);
-  writer = (LogWriter*)log_writer_new(writer_flags);
+  dd = g_new0(LogDestDriver, 1);
+  log_dest_drv_init_instance(dd);
+  writer = (LogWriter*)log_writer_new(writer_flags, dd);
   if (writer)
     {
       log_writer_set_options(writer, NULL, &opt, 0, 0, NULL, NULL);
