Merge lp:~tgates/gearmand/nuodb-queue-adapter into lp:gearmand

[Tom Gates]

Add support for using NuoDB as a persistent queue for Gearman.

Unmerged revisions

895. By Tom Gates on 2014-01-10

Add support for NuoDB persistent queues

=== modified file 'configure.ac'
--- configure.ac	2014-02-16 02:15:49 +0000
+++ configure.ac	2014-05-09 18:35:29 +0000
@@ -104,6 +104,11 @@
       [AC_DEFINE([HAVE_LIBSQLITE3],[0],[Have the SQLITE3 library])])
 AM_CONDITIONAL([HAVE_LIBSQLITE3],[test "x${WANT_SQLITE3}" = "xyes"])
 
+AX_LIB_NUODB
+AS_IF([test "x${WANT_NUODB}" = "xyes"],
+      [AC_DEFINE([HAVE_NUODB],[1],[Have the NuoDB library])],
+      [AC_DEFINE([HAVE_NUODB],[0],[Have the NuoDB library])])
+AM_CONDITIONAL([HAVE_NUODB],[test "x${WANT_NUODB}" = "xyes"])
 
 PANDORA_ENABLE_DTRACE
 AX_HAVE_LIBPQ
@@ -345,6 +350,7 @@
 echo "   * Debug enabled:             $ax_enable_debug"
 echo "   * Warnings as failure:       $ac_cv_warnings_as_errors"
 echo "   * Building with libsqlite3   $WANT_SQLITE3"
+echo "   * Building with nuodb        $WANT_NUODB"
 echo "   * Building with libdrizzle   $ac_enable_libdrizzle"
 echo "   * Building with libmemcached $ax_enable_libmemcached"
 echo "   * Building with libpq        $ac_cv_libpq"
 
=== modified file 'libgearman-server/plugins.cc'
--- libgearman-server/plugins.cc	2013-10-20 11:00:25 +0000
+++ libgearman-server/plugins.cc	2014-05-09 18:35:29 +0000
@@ -71,6 +71,13 @@
   }
 #endif
 
+#if defined(HAVE_NUODB) && HAVE_NUODB
+  if (HAVE_NUODB)
+  {
+    queue::initialize_nuodb();
+  }
+#endif
+
 #if defined(HAVE_LIBPQ) && HAVE_LIBPQ
   if (HAVE_LIBPQ)
   {
 
=== modified file 'libgearman-server/plugins/queue.h'
--- libgearman-server/plugins/queue.h	2012-11-23 09:17:06 +0000
+++ libgearman-server/plugins/queue.h	2014-05-09 18:35:29 +0000
@@ -52,3 +52,5 @@
 #include <libgearman-server/plugins/queue/redis/queue.h>
 
 #include <libgearman-server/plugins/queue/mysql/queue.h>
+
+#include <libgearman-server/plugins/queue/nuodb/queue.h>
 
=== modified file 'libgearman-server/plugins/queue/include.am'
--- libgearman-server/plugins/queue/include.am	2012-11-23 09:17:06 +0000
+++ libgearman-server/plugins/queue/include.am	2014-05-09 18:35:29 +0000
@@ -21,3 +21,4 @@
 include libgearman-server/plugins/queue/sqlite/include.am
 include libgearman-server/plugins/queue/tokyocabinet/include.am
 include libgearman-server/plugins/queue/mysql/include.am
+include libgearman-server/plugins/queue/nuodb/include.am
 
=== added directory 'libgearman-server/plugins/queue/nuodb'
=== added file 'libgearman-server/plugins/queue/nuodb/include.am'
--- libgearman-server/plugins/queue/nuodb/include.am	1970-01-01 00:00:00 +0000
+++ libgearman-server/plugins/queue/nuodb/include.am	2014-05-09 18:35:29 +0000
@@ -0,0 +1,23 @@
+# vim:ft=automake
+# Gearman
+# Copyright (C) 2011 Data Differential, http://datadifferential.com/
+# All rights reserved.
+#
+# Use and distribution licensed under the BSD license.  See
+# the COPYING file in the parent directory for full text.
+#
+# All paths should be given relative to the root
+#
+
+noinst_HEADERS+= libgearman-server/plugins/queue/nuodb/queue.h
+
+if HAVE_NUODB
+
+libgearman_server_libgearman_server_la_SOURCES+= libgearman-server/plugins/queue/nuodb/queue.cc
+libgearman_server_libgearman_server_la_LIBADD+= $(NUODB_LDFLAGS)
+libgearman_server_libgearman_server_la_CXXFLAGS+= $(NUODB_CFLAGS)
+
+gearmand_gearmand_LDADD+= $(NUODB_LDFLAGS)
+
+endif
+
 
=== added file 'libgearman-server/plugins/queue/nuodb/queue.cc'
--- libgearman-server/plugins/queue/nuodb/queue.cc	1970-01-01 00:00:00 +0000
+++ libgearman-server/plugins/queue/nuodb/queue.cc	2014-05-09 18:35:29 +0000
@@ -0,0 +1,550 @@
+/*  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+ *
+ *  Gearmand client and server library.
+ *
+ *  Copyright (C) 2011 Data Differential, http://datadifferential.com/
+ *  Copyright (C) 2008 Brian Aker, Eric Day
+ *  All rights reserved.
+ *
+ *  Redistribution and use in source and binary forms, with or without
+ *  modification, are permitted provided that the following conditions are
+ *  met:
+ *
+ *      * Redistributions of source code must retain the above copyright
+ *  notice, this list of conditions and the following disclaimer.
+ *
+ *      * Redistributions in binary form must reproduce the above
+ *  copyright notice, this list of conditions and the following disclaimer
+ *  in the documentation and/or other materials provided with the
+ *  distribution.
+ *
+ *      * The names of its contributors may not be used to endorse or
+ *  promote products derived from this software without specific prior
+ *  written permission.
+ *
+ *  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ *  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ *  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ *  A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ *  OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ *  SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ *  LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ *  DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ *  THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ *  (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ *  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+/**
+ * @file
+ * @brief NuoDB Queue Storage Definitions
+ */
+
+#include <gear_config.h>
+#include <libgearman-server/common.h>
+#include <libgearman-server/byte.h>
+#include <libgearman-server/plugins/queue/nuodb/queue.h>
+#include <libgearman-server/plugins/queue/base.h>
+
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wundef"
+#pragma GCC diagnostic ignored "-Wnon-virtual-dtor"
+#pragma GCC diagnostic ignored "-Woverloaded-virtual"
+#pragma GCC diagnostic ignored "-Wreorder"
+#pragma GCC diagnostic ignored "-Wunused-parameter"
+#pragma GCC diagnostic ignored "-Wold-style-cast"
+
+#include <NuoDB.h>
+#include <cerrno>
+
+/**
+ * @addtogroup plugins::queue::NuoDB Static NuoDB Queue Storage Definitions
+ * @ingroup gearman_queue_nuodb
+ * @{
+ */
+
+/**
+ * Default values.
+ */
+#define GEARMAND_QUEUE_NUODB_DEFAULT_SCHEMA "gearman"
+#define GEARMAND_QUEUE_NUODB_DEFAULT_TABLE "queue"
+
+namespace gearmand { namespace plugins { namespace  queue { class Nuodb; }}}
+
+static gearmand_error_t _initialize(gearman_server_st& server, gearmand::plugins::queue::Nuodb *queue);
+
+namespace gearmand {
+namespace plugins {
+namespace queue {
+
+class Nuodb : public gearmand::plugins::Queue {
+public:
+  Nuodb();
+  ~Nuodb();
+
+  gearmand_error_t initialize();
+
+  const std::string &insert() {
+    return _insert_query;
+  }
+
+  const std::string &select() {
+    return _select_query;
+  }
+
+  const std::string &create() {
+    return _create_query;
+  }
+
+  NuoDB::Connection *con;
+  std::string nuodb_connect_string;
+  std::string nuodb_user;
+  std::string nuodb_pw;
+  std::string schema;
+  std::string table;
+  NuoDB::PreparedStatement *_insert_stmt;
+  NuoDB::PreparedStatement *_delete_stmt;
+  NuoDB::PreparedStatement *_select_stmt;
+
+public:
+  std::string _insert_query;
+  std::string _select_query;
+  std::string _create_query;
+
+  NuoDB::PreparedStatement *insert_stmt() {
+    return _insert_stmt;
+  }
+
+  NuoDB::PreparedStatement *delete_stmt() {
+    return _delete_stmt;
+  }
+
+  NuoDB::PreparedStatement *select_stmt() {
+    return _select_stmt;
+  }
+
+};
+
+Nuodb::Nuodb() :
+  Queue("NuoDB"),
+  con(NULL),
+  nuodb_connect_string(""),
+  nuodb_user(""),
+  nuodb_pw(""),
+  table(""),
+  schema("")
+{
+  command_line_options().add_options()
+    ("nuodb-conninfo", boost::program_options::value(&nuodb_connect_string)->default_value(""), "NuoDB connection information string.")
+    ("nuodb-user", boost::program_options::value(&nuodb_user)->default_value(""), "NuoDB connection username.")
+    ("nuodb-password", boost::program_options::value(&nuodb_pw)->default_value(""), "NuoDB connection password.")
+    ("nuodb-table", boost::program_options::value(&table)->default_value(GEARMAND_QUEUE_NUODB_DEFAULT_TABLE), "Table to use.")
+    ("nuodb-schema", boost::program_options::value(&schema)->default_value(GEARMAND_QUEUE_NUODB_DEFAULT_SCHEMA), "Schema to use.");
+}
+
+Nuodb::~Nuodb ()
+{
+  if (con) {
+    con->close();
+  }
+  con = NULL;
+}
+
+gearmand_error_t Nuodb::initialize()
+{
+  _create_query+= "CREATE TABLE " +table +" (unique_key VARCHAR" +"(" + TOSTRING(GEARMAN_UNIQUE_SIZE) +"), ";
+  _create_query+= "function_name VARCHAR(255), priority INTEGER, data BLOB, when_to_run INTEGER, UNIQUE (unique_key, function_name))";
+  _insert_query+= "INSERT INTO " +table +" (priority, unique_key, function_name, data, when_to_run) VALUES(?,?,?,?,?)";
+  _select_query+= "SELECT unique_key,function_name,priority,data,when_to_run FROM " +table;
+  gearmand_error_t ret= _initialize(Gearmand()->server, this);
+  return ret;
+}
+
+void initialize_nuodb()
+{
+  static Nuodb local_instance;
+}
+
+} // namespace queue
+} // namespace plugins
+} // namespace gearmand
+
+
+/* Queue callback functions. */
+static gearmand_error_t _nuodb_add(gearman_server_st *server, void *context,
+                                   const char *unique, size_t unique_size,
+                                   const char *function_name,
+                                   size_t function_name_size,
+                                   const void *data, size_t data_size,
+                                   gearman_job_priority_t priority,
+                                   int64_t when);
+
+static gearmand_error_t _nuodb_flush(gearman_server_st *server, void *context);
+
+static gearmand_error_t _nuodb_done(gearman_server_st *server, void *context,
+                                    const char *unique,
+                                    size_t unique_size,
+                                    const char *function_name,
+                                    size_t function_name_size);
+
+static gearmand_error_t _nuodb_replay(gearman_server_st *server, void *context,
+                                      gearman_queue_add_fn *add_fn,
+                                      void *add_context);
+
+/** @} */
+
+/*
+ * Public definitions
+ */
+
+gearmand_error_t _initialize(gearman_server_st& server,
+                             gearmand::plugins::queue::Nuodb *queue)
+{
+  // TODO: use nuodb_connnect_string
+  const char *database = queue->nuodb_connect_string.c_str();
+  const char *user = queue->nuodb_user.c_str();
+  const char *pw = queue->nuodb_pw.c_str();
+  const char *schema = queue->schema.c_str();
+
+  gearmand_info("Initializing nuodb module");
+
+  gearman_server_set_queue(server, queue, _nuodb_add, _nuodb_flush, _nuodb_done, _nuodb_replay);
+
+  try {
+    queue->con = NuoDB::Connection::createUtf8();
+    NuoDB::Properties* properties = queue->con->allocProperties();
+    properties->putValue("user", user);
+    properties->putValue("password", pw);
+    properties->putValue("schema", schema);
+    queue->con->openDatabase(database, properties);
+  } catch (NuoDB::SQLException &e) {
+    gearman_server_set_queue(server, NULL, NULL, NULL, NULL, NULL);
+    return gearmand_log_gerror(GEARMAN_DEFAULT_LOG_PARAM, GEARMAND_QUEUE_ERROR, "NuoDB Connection: [%s] %s", e.getSqlcode(), e.getText());
+  } catch (...) {
+    // Unknown error
+    queue->con = NULL;
+  }
+  if (queue->con == NULL)
+  {
+    gearman_server_set_queue(server, NULL, NULL, NULL, NULL, NULL);
+    return gearmand_log_gerror(GEARMAN_DEFAULT_LOG_PARAM, GEARMAND_QUEUE_ERROR, "NuoDB Connection: %s", "Unknown error");
+  }
+
+
+  std::string query("SELECT TABLENAME FROM SYSTEM.TABLES WHERE TYPE = 'TABLE' AND SCHEMA = '" +queue->schema +"' AND TABLENAME = '" +queue->table + "'");
+
+  NuoDB::PreparedStatement *stmt = NULL;
+  NuoDB::ResultSet *rs = NULL;
+
+  try {
+    stmt = queue->con->prepareStatement(query.c_str());
+    stmt->execute();
+    rs = stmt->getResultSet();
+  } catch (NuoDB::SQLException &e) {
+    std::string error_string= "NuoDB:";
+    error_string += "Queue error: [";
+    error_string += e.getSqlcode();
+    error_string += "] ";
+    error_string += e.getText();
+    gearmand_gerror(error_string.c_str(), GEARMAND_QUEUE_ERROR);
+    return GEARMAND_QUEUE_ERROR;
+  } catch (...) {
+    // Unknown error
+    rs = NULL;
+  }
+
+  if (rs == NULL) {
+            std::string error_string= "NuoDB:";
+            error_string+= "No results for query: ";
+            error_string+= query.c_str();
+            gearmand_gerror(error_string.c_str(), GEARMAND_QUEUE_ERROR);
+            return GEARMAND_QUEUE_ERROR;
+  }
+
+  if (!rs->next())
+  {
+    gearmand_log_info(GEARMAN_DEFAULT_LOG_PARAM, "nuodb module creating table '%s'", queue->table.c_str());
+
+    bool result = 1;
+    const char *sql = queue->create().c_str();
+    try {
+      stmt = queue->con->prepareStatement(sql);
+      result = stmt->execute();
+    } catch (NuoDB::SQLException &e) {
+      std::string error_string= "NuoDB:";
+      error_string += "Table create error: [";
+      error_string += e.getSqlcode();
+      error_string += "] ";
+      error_string += e.getText();
+      gearmand_gerror(error_string.c_str(), GEARMAND_QUEUE_ERROR);
+      return GEARMAND_QUEUE_ERROR;
+    } catch (...) {
+      // Unknown error
+      std::string error_string= "NuoDB:";
+      error_string+= "Table create error for statement: ";
+      error_string+= query.c_str();
+      gearmand_gerror(error_string.c_str(), GEARMAND_QUEUE_ERROR);
+      return GEARMAND_QUEUE_ERROR;
+    }
+
+    if (result == 1)
+    {
+      gearmand_log_error(GEARMAN_DEFAULT_LOG_PARAM,
+                         "NuoDB:Problem creating table: %s", sql);
+      gearman_server_set_queue(server, NULL, NULL, NULL, NULL, NULL);
+      return GEARMAND_QUEUE_ERROR;
+    }
+  }
+
+  const char *insert_sql = queue->insert().c_str();
+  try {
+    queue->_insert_stmt = queue->con->prepareStatement(insert_sql);
+  } catch (NuoDB::SQLException &e) {
+    std::string error_string= "NuoDB:";
+    error_string += "Error preparing insert statement: [";
+    error_string += e.getSqlcode();
+    error_string += "] ";
+    error_string += e.getText();
+    gearmand_gerror(error_string.c_str(), GEARMAND_QUEUE_ERROR);
+    return GEARMAND_QUEUE_ERROR;
+  } catch (...) {
+    // Unknown error
+    std::string error_string= "NuoDB:";
+    error_string+= "Error preparing insert statement: ";
+    error_string+= insert_sql;
+    gearmand_gerror(error_string.c_str(), GEARMAND_QUEUE_ERROR);
+    return GEARMAND_QUEUE_ERROR;
+  }
+
+  std::string delete_query;
+  delete_query+= "DELETE FROM ";
+  delete_query+= queue->table;
+  delete_query+= " WHERE unique_key = ?";
+  delete_query+= " AND function_name = ?";
+  const char *delete_sql = delete_query.c_str();
+  try {
+    queue->_delete_stmt = queue->con->prepareStatement(delete_sql);
+  } catch (NuoDB::SQLException &e) {
+    std::string error_string= "NuoDB:";
+    error_string += "Error preparing delete statement: [";
+    error_string += e.getSqlcode();
+    error_string += "] ";
+    error_string += e.getText();
+    gearmand_gerror(error_string.c_str(), GEARMAND_QUEUE_ERROR);
+    return GEARMAND_QUEUE_ERROR;
+  } catch (...) {
+    // Unknown error
+    std::string error_string= "NuoDB:";
+    error_string+= "Error preparing delete statement: ";
+    error_string+= insert_sql;
+    gearmand_gerror(error_string.c_str(), GEARMAND_QUEUE_ERROR);
+    return GEARMAND_QUEUE_ERROR;
+  }
+
+  std::string select_query("SELECT unique_key,function_name,priority,data,when_to_run FROM " + queue->table);
+  const char *select_sql = select_query.c_str();
+  try {
+    queue->_select_stmt = queue->con->prepareStatement(select_sql);
+  } catch (NuoDB::SQLException &e) {
+    std::string error_string= "NuoDB:";
+    error_string += "Error preparing select statement: [";
+    error_string += e.getSqlcode();
+    error_string += "] ";
+    error_string += e.getText();
+    gearmand_gerror(error_string.c_str(), GEARMAND_QUEUE_ERROR);
+    return GEARMAND_QUEUE_ERROR;
+  } catch (...) {
+    // Unknown error
+    std::string error_string= "NuoDB:";
+    error_string+= "Error preparing select statement: ";
+    error_string+= insert_sql;
+    gearmand_gerror(error_string.c_str(), GEARMAND_QUEUE_ERROR);
+    return GEARMAND_QUEUE_ERROR;
+  }
+
+  rs->close();
+  rs = NULL;
+  stmt->close();
+  stmt = NULL;
+  return GEARMAND_SUCCESS;
+}
+
+/*
+ * Static definitions
+ */
+
+static gearmand_error_t _nuodb_add(gearman_server_st*, void *context,
+                                   const char *unique, size_t unique_size,
+                                   const char *function_name,
+                                   size_t function_name_size,
+                                   const void *data, size_t data_size,
+                                   gearman_job_priority_t priority,
+                                   int64_t when)
+{
+  gearmand::plugins::queue::Nuodb *queue= (gearmand::plugins::queue::Nuodb *)context;
+
+  gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM, "nuodb add: %.*s", (uint32_t)unique_size, (char *)unique);
+
+
+  NuoDB::PreparedStatement *stmt = queue->insert_stmt();
+
+  try {
+    stmt->setInt(1, static_cast<uint32_t>(priority));
+    stmt->setString(2, unique);
+    stmt->setString(3, function_name);
+    stmt->setBytes(4, data_size, (const void *) data);
+    stmt->setLong(5, when);
+    stmt->execute();
+  } catch (NuoDB::SQLException &e) {
+    std::string error_string= "NuoDB:";
+    error_string += "Add queue error: [";
+    error_string += e.getSqlcode() + "] ";
+    error_string += e.getText();
+    gearmand_gerror(error_string.c_str(), GEARMAND_QUEUE_ERROR);
+    return GEARMAND_QUEUE_ERROR;
+  } catch (...) {
+    // Unknown error
+    std::string error_string= "NuoDB:";
+    error_string+= "Unable to add to queue ";
+    gearmand_gerror(error_string.c_str(), GEARMAND_QUEUE_ERROR);
+    return GEARMAND_QUEUE_ERROR;
+  }
+
+  return GEARMAND_SUCCESS;
+}
+
+static gearmand_error_t _nuodb_flush(gearman_server_st *, void *)
+{
+  gearmand_debug("nuodb flush");
+
+  return GEARMAND_SUCCESS;
+}
+
+static gearmand_error_t _nuodb_done(gearman_server_st*, void *context,
+                                    const char *unique,
+                                    size_t unique_size,
+                                    const char *function_name,
+                                    size_t function_name_size)
+{
+  (void)function_name_size;
+  gearmand::plugins::queue::Nuodb *queue= (gearmand::plugins::queue::Nuodb *)context;
+  gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM, "nuodb done: %.*s", (uint32_t)unique_size, (char *)unique);
+
+  NuoDB::PreparedStatement *stmt = queue->delete_stmt();
+
+  try {
+    stmt->setString(1, unique);
+    stmt->setString(2, function_name);
+    stmt->execute();
+  } catch (NuoDB::SQLException &e) {
+    std::string error_string= "NuoDB:";
+    error_string += "Delete queue error: [";
+    error_string += e.getSqlcode();
+    error_string += "] ";
+    error_string += e.getText();
+    gearmand_gerror(error_string.c_str(), GEARMAND_QUEUE_ERROR);
+    return GEARMAND_QUEUE_ERROR;
+  } catch (...) {
+    // Unknown error
+    std::string error_string= "NuoDB:";
+    error_string+= "Unable to delete from queue ";
+    gearmand_gerror(error_string.c_str(), GEARMAND_QUEUE_ERROR);
+    return GEARMAND_QUEUE_ERROR;
+  }
+
+  return GEARMAND_SUCCESS;
+}
+
+static gearmand_error_t _nuodb_replay(gearman_server_st *server, void *context,
+                                      gearman_queue_add_fn *add_fn,
+                                      void *add_context)
+{
+  gearmand::plugins::queue::Nuodb *queue= (gearmand::plugins::queue::Nuodb *)context;
+
+  gearmand_info("nuodb replay start");
+
+
+  NuoDB::PreparedStatement *stmt = queue->select_stmt();
+  NuoDB::ResultSet *rs = NULL;
+
+
+  try {
+    stmt->execute();
+    rs = stmt->getResultSet();
+  } catch (NuoDB::SQLException &e) {
+    std::string error_string= "NuoDB:";
+    error_string += "Select error: [";
+    error_string += e.getSqlcode();
+    error_string += "] ";
+    error_string += e.getText();
+    gearmand_gerror(error_string.c_str(), GEARMAND_QUEUE_ERROR);
+    return GEARMAND_QUEUE_ERROR;
+  } catch (...) {
+    // Unknown error
+    rs = NULL;
+  }
+
+  if (rs == NULL)
+  {
+    std::string error_string= "NuoDB:";
+    error_string+= "No results for query ";
+    gearmand_gerror(error_string.c_str(), GEARMAND_QUEUE_ERROR);
+    return GEARMAND_QUEUE_ERROR;
+  }
+
+
+  while(rs->next())
+  {
+    const char *unique_key;
+    const char *function_name;
+    int priority;
+    size_t data_length;
+    char *data;
+    int when_to_run;
+
+    unique_key = rs->getString(1);
+    gearmand_log_debug(GEARMAN_DEFAULT_LOG_PARAM,
+                       "nuodb replay: %s",
+                       unique_key);
+
+    NuoDB::Blob *blob = rs->getBlob(4);
+    if (blob == NULL)
+    {
+      data= NULL;
+      data_length= 0;
+    }
+    else
+    {
+      data_length = size_t(blob->length());
+      data= (char *)malloc(data_length);
+      if (data == NULL)
+      {
+        return gearmand_perror(errno, "malloc");
+      }
+      blob->getBytes(0, data_length, (unsigned char *)data);
+    }
+
+    gearmand_error_t ret;
+    function_name = rs->getString(2);
+    priority = rs->getInt(3);
+    when_to_run = rs->getInt(5);
+    ret= (*add_fn)(server, add_context, unique_key,
+                   (size_t) strlen(unique_key),
+                   function_name,
+                   (size_t) strlen(function_name),
+                   data, data_length,
+                   (gearman_job_priority_t) priority,
+                   when_to_run);
+    if (gearmand_failed(ret))
+    {
+      rs->close();
+      return ret;
+    }
+  }
+
+  rs->close();
+  return GEARMAND_SUCCESS;
+}
+
+#pragma GCC diagnostic pop
 
=== added file 'libgearman-server/plugins/queue/nuodb/queue.h'
--- libgearman-server/plugins/queue/nuodb/queue.h	1970-01-01 00:00:00 +0000
+++ libgearman-server/plugins/queue/nuodb/queue.h	2014-05-09 18:35:29 +0000
@@ -0,0 +1,49 @@
+/*  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+ * 
+ *  Gearmand client and server library.
+ *
+ *  Copyright (C) 2011 Data Differential, http://datadifferential.com/
+ *  All rights reserved.
+ *
+ *  Redistribution and use in source and binary forms, with or without
+ *  modification, are permitted provided that the following conditions are
+ *  met:
+ *
+ *      * Redistributions of source code must retain the above copyright
+ *  notice, this list of conditions and the following disclaimer.
+ *
+ *      * Redistributions in binary form must reproduce the above
+ *  copyright notice, this list of conditions and the following disclaimer
+ *  in the documentation and/or other materials provided with the
+ *  distribution.
+ *
+ *      * The names of its contributors may not be used to endorse or
+ *  promote products derived from this software without specific prior
+ *  written permission.
+ *
+ *  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ *  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ *  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ *  A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ *  OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ *  SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ *  LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ *  DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ *  THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ *  (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ *  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#pragma once
+
+
+namespace gearmand {
+namespace plugins {
+namespace queue {
+
+void initialize_nuodb();
+
+} // namespace queue
+} // namespace plugin
+} // namespace gearmand
 
=== added file 'm4/ax_lib_nuodb.m4'
--- m4/ax_lib_nuodb.m4	1970-01-01 00:00:00 +0000
+++ m4/ax_lib_nuodb.m4	2014-05-09 18:35:29 +0000
@@ -0,0 +1,130 @@
+#
+# SYNOPSIS
+#
+#   AX_LIB_NUODB([MINIMUM-VERSION])
+#
+# DESCRIPTION
+#
+#   This macro provides tests of availability of NuoDB 'libRemote' library
+#   of particular version or newer.
+#
+#   AX_LIB_NUODB macro takes only one argument which is optional. If
+#   there is no required version passed, then macro does not run version
+#   test.
+#
+#   The --with-nuodb option takes one of three possible values:
+#
+#   no - do not check for NuoDB client library
+#
+#   yes - do check for NuoDB client library in standard locations
+#
+#   path - complete path to NuoDB home directory
+#
+#   This macro calls:
+#
+#     AC_SUBST(NUODB_CFLAGS)
+#     AC_SUBST(NUODB_LDFLAGS)
+#     AC_SUBST(NUODB_VERSION)
+#
+#   And sets:
+#
+#     HAVE_NUODB
+#
+
+AC_DEFUN([AX_LIB_NUODB],
+[
+    AC_ARG_WITH([nuodb],
+        AS_HELP_STRING([--with-nuodb=@<:@ARG@:>@],
+            [use NuoDB libNuoRemote library @<:@default=yes@:>@, optionally specify path to NuoDB Home Directory]
+        ),
+        [
+        if test "$withval" = "no"; then
+            WANT_NUODB="no"
+        elif test "$withval" = "yes"; then
+            WANT_NUODB="yes"
+	    NUOHOME="/opt/nuodb"
+        else
+            WANT_NUODB="yes"
+            NUOHOME="$withval"
+        fi
+        ],
+        [WANT_NUODB="yes"]
+    )
+
+    NUODB_CFLAGS=""
+    NUODB_LDFLAGS=""
+    NUODB_VERSION=""
+
+    dnl
+    dnl Check NuoDB libraries (libNuoRemote)
+    dnl
+
+    if test "$WANT_NUODB" = "yes"; then
+
+        if test -z "$NUOHOME" -o test; then
+
+        AC_MSG_CHECKING([for NuoDB libraries])
+
+        NUODB_CFLAGS="-I$NUOHOME/include"
+        NUODB_LDFLAGS="-L$NUOHOME/lib64 -lNuoRemote"
+        NUODB_VERSION=`$NUOHOME/bin/nuochk --version | sed -e 's#NuoDB Server build ##'`
+        AC_DEFINE([HAVE_NUODB], [1],
+                [Define to 1 if NuoDB libraries are available])
+
+        found_nuodb="yes"
+            AC_MSG_RESULT([yes])
+        else
+            found_nuodb="no"
+            AC_MSG_RESULT([no])
+        fi
+    fi
+
+    dnl
+    dnl Check if required version of NuoDB is available
+    dnl
+
+
+    nuodb_version_req=ifelse([$1], [], [], [$1])
+
+    if test "$found_nuodb" = "yes" -a -n "$nuodb_version_req"; then
+
+        AC_MSG_CHECKING([if NuoDB version is >= $nuodb_version_req])
+
+        dnl Decompose required version string of NuoDB
+        dnl and calculate its number representation
+        nuodb_version_req_major=`expr $nuodb_version_req : '\([[0-9]]*\)'`
+        nuodb_version_req_minor=`expr $nuodb_version_req : '[[0-9]]*\.\([[0-9]]*\)'`
+        nuodb_version_req_micro=`expr $nuodb_version_req : '[[0-9]]*\.[[0-9]]*\.\([[0-9]]*\)'`
+        if test "x$nuodb_version_req_micro" = "x"; then
+            nuodb_version_req_micro="0"
+        fi
+
+        nuodb_version_req_number=`expr $nuodb_version_req_major \* 1000000 \
+                                   \+ $nuodb_version_req_minor \* 1000 \
+                                   \+ $nuodb_version_req_micro`
+
+        dnl Decompose version string of installed NuoDB
+        dnl and calculate its number representation
+        nuodb_version_major=`expr $NUODB_VERSION : '\([[0-9]]*\)'`
+        nuodb_version_minor=`expr $NUODB_VERSION : '[[0-9]]*\.\([[0-9]]*\)'`
+        nuodb_version_micro=`expr $NUODB_VERSION : '[[0-9]]*\.[[0-9]]*\.\([[0-9]]*\)'`
+        if test "x$nuodb_version_micro" = "x"; then
+            nuodb_version_micro="0"
+        fi
+
+        nuodb_version_number=`expr $nuodb_version_major \* 1000000 \
+                                   \+ $nuodb_version_minor \* 1000 \
+                                   \+ $nuodb_version_micro`
+
+        nuodb_version_check=`expr $nuodb_version_number \>\= $nuodb_version_req_number`
+        if test "$nuodb_version_check" = "1"; then
+            AC_MSG_RESULT([yes])
+        else
+            AC_MSG_RESULT([no])
+        fi
+    fi
+
+    AC_SUBST([NUODB_VERSION])
+    AC_SUBST([NUODB_CFLAGS])
+    AC_SUBST([NUODB_LDFLAGS])
+])