Merge 1.0

bin/arguments.cc

@@ -39,6 +39,7 @@
 #include "gear_config.h"
 
 #include <cstdlib>
+#include <cstdio>
 #include <cstring>
 #include <string>
 #include <vector>
@@ -64,11 +65,13 @@ Args::Args(int p_argc, char *p_argv[]) :
   _strip_newline(false),
   _worker(false),
   _suppress_input(false),
+  _verbose(false),
   _prefix(false),
   _background(false),
   _daemon(false),
   _usage(false),
   _is_error(false),
+  _arg_error(NULL),
   _priority(GEARMAN_JOB_PRIORITY_NORMAL),
   _timeout(-1),
   argv(p_argv),
@@ -79,6 +82,10 @@ Args::Args(int p_argc, char *p_argv[]) :
 
 Args::~Args()
 {
+  if (_arg_error)
+  {
+    free(_arg_error);
+  }
 }
 
 void Args::init(int argc)
@@ -87,7 +94,7 @@ void Args::init(int argc)
 
   opterr= 0;
 
-  while ((c = getopt(argc, argv, "bc:f:h:HILnNp:Pst:u:wi:d")) != -1)
+  while ((c = getopt(argc, argv, "bc:f:h:HILnNp:Pst:u:vwi:d")) != -1)
   {
     switch(c)
     {
@@ -158,11 +165,36 @@ void Args::init(int argc)
 
     case 'H':
       _usage= true;
-      return;
+      break;
+
+    case 'v':
+      _verbose= true;
+      break;
 
     default:
       _is_error= true;
-      return;
+
+      if (optarg)
+      {
+        size_t length= snprintf(NULL, 0, "-%c %s", char(c), optarg);
+        ++length; // Add in space for null
+        _arg_error= (char*)malloc(length);
+        if (_arg_error)
+        {
+          snprintf(_arg_error, length, "-%c %s", char(c), optarg);
+        }
+      }
+      else
+      {
+        size_t length= snprintf(NULL, 0, "-%c", char(c));
+        length++;
+        _arg_error= (char*)malloc(length);
+        if (_arg_error)
+        {
+          snprintf(_arg_error, length, "-%c", char(c));
+        }
+      }
+      break;
     }
   }
 

bin/arguments.h

@@ -102,6 +102,7 @@ public:
   {
     return _functions.end();
   }
+
   bool strip_newline() const
   {
     return _strip_newline;
@@ -150,6 +151,11 @@ public:
     return false;
   }
 
+  bool verbose() const
+  {
+    return _verbose;
+  }
+
   bool suppress_input() const
   {
     return _suppress_input;
@@ -165,16 +171,22 @@ public:
     return argv;
   }
 
-  bool is_error() const
+  bool is_error()
   {
     if (_functions.empty())
     {
+      _arg_error= strdup("No Functions were provided");
       return true;
     }
 
     return _is_error;
   }
 
+  const char* arg_error() const
+  {
+    return _arg_error;
+  }
+
   bool is_valid() const
   {
     return _functions.size();
@@ -190,12 +202,14 @@ private:
   bool _strip_newline;
   bool _worker;
   bool _suppress_input;
+  bool _verbose;
 
   bool _prefix;
   bool _background;
   bool _daemon;
   bool _usage;
   bool _is_error;
+  char *_arg_error;
   gearman_job_priority_t _priority;
   int _timeout;
   char **argv;

bin/client.h

@@ -1,69 +0,0 @@
-/*  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 bin {
-
-class Client
-{
-public:
-  Client() :
-    _client()
-  {
-    if (gearman_client_create(&_client) == NULL)
-    {
-      std::cerr << "Failed memory allocation while initializing client." << std::endl;
-      abort();
-    }
-  }
-
-  ~Client()
-  {
-    gearman_client_free(&_client);
-  }
-
-  gearman_client_st &client()
-  {
-    return _client;
-  }
-
-private:
-  gearman_client_st _client;
-};
-
-} // namespace bin

bin/error.cc

@@ -83,14 +83,14 @@ void message(const std::string &arg, gearman_return_t rc)
   std::cerr << "gearman: " << arg << " : " << gearman_strerror(rc) << std::endl;
 }
 
-void message(const char *arg, const gearman_client_st &client)
+void message(const char *arg, const gearman_client_st* client)
 {
-  std::cerr << "gearman: " << arg << " : " << gearman_client_error(&client) << std::endl;
+  std::cerr << "gearman: " << arg << " : " << gearman_client_error(client) << std::endl;
 }
 
-void message(const char *arg, const gearman_worker_st &worker)
+void message(const char *arg, const gearman_worker_st* worker)
 {
-  std::cerr << "gearman: " << arg << " : " << gearman_worker_error(&worker) << std::endl;
+  std::cerr << "gearman: " << arg << " : " << gearman_worker_error(worker) << std::endl;
 }
 
 } // namespace error

bin/error.h

@@ -50,8 +50,8 @@ void perror(const char *);
 void message(const char *);
 void message(const std::string &arg, gearman_return_t rc);
 void message(const char *arg, const char *arg2);
-void message(const char *arg, const gearman_client_st &);
-void message(const char *arg, const gearman_worker_st &);
+void message(const char *arg, const gearman_client_st*);
+void message(const char *arg, const gearman_worker_st*);
 
 } // namespace error
 

bin/gearman.cc

@@ -54,11 +54,14 @@
 #include <vector>
 
 
-#include <libgearman/gearman.h>
+#include <libgearman-1.0/gearman.h>
 
 #include "bin/arguments.h"
-#include "bin/client.h"
-#include "bin/worker.h"
+
+#include "libgearman/client.hpp"
+#include "libgearman/worker.hpp"
+using namespace org::gearmand;
+
 #include "util/pidfile.hpp"
 #include "bin/error.h"
 
@@ -87,9 +90,11 @@ static void _client(Args &args);
 /**
  * Run client jobs.
  */
-static void _client_run(gearman_client_st& client, Args &args,
+static void _client_run(libgearman::Client& client, Args &args,
                         const void *workload, size_t workload_size);
 
+static gearman_return_t _client_created(gearman_task_st *task);
+
 /**
  * Client data/complete callback function.
  */
@@ -128,7 +133,7 @@ static void _read_workload(int fd, Bytes& workload);
 /**
  * Print usage information.
  */
-static void usage(char *name);
+static void usage(Args&, char *name);
 
 extern "C"
 {
@@ -150,13 +155,13 @@ int main(int argc, char *argv[])
 
   if (args.usage())
   {
-    usage(argv[0]);
+    usage(args, argv[0]);
     return EXIT_SUCCESS;
   }
 
   if (args.is_error())
   {
-    usage(argv[0]);
+    usage(args, argv[0]);
     return EXIT_FAILURE;
   }
 
@@ -237,8 +242,7 @@ int main(int argc, char *argv[])
 
 void _client(Args &args)
 {
-  bin::Client local_client;
-  gearman_client_st &client= local_client.client();
+  libgearman::Client client;
   Bytes workload;
   if (args.timeout() >= 0)
   {
@@ -249,16 +253,17 @@ void _client(Args &args)
   {
     if (gearman_failed(gearman_client_add_servers(&client, getenv("GEARMAN_SERVER"))))
     {
-      error::message("Error occurred while parsing GEARMAN_SERVER", client);
+      error::message("Error occurred while parsing GEARMAN_SERVER", &client);
       return;
     }
   }
   else if (gearman_failed(gearman_client_add_server(&client, args.host(), args.port())))
   {
-    error::message("gearman_client_add_server", client);
+    error::message("gearman_client_add_server", &client);
     return;
   }
 
+  gearman_client_set_created_fn(&client, _client_created);
   gearman_client_set_data_fn(&client, _client_data);
   gearman_client_set_warning_fn(&client, _client_warning);
   gearman_client_set_status_fn(&client, _client_status);
@@ -308,7 +313,7 @@ void _client(Args &args)
   }
 }
 
-void _client_run(gearman_client_st& client, Args &args,
+void _client_run(libgearman::Client& client, Args &args,
                  const void *workload, size_t workload_size)
 {
   gearman_return_t ret;
@@ -404,16 +409,28 @@ void _client_run(gearman_client_st& client, Args &args,
 
     if (gearman_failed(ret))
     {
-      error::message("gearman_client_add_task", client);
+      error::message("gearman_client_add_task", &client);
     }
   }
 
   if (gearman_failed(gearman_client_run_tasks(&client)))
   {
-    error::message("gearman_client_run_tasks", client);
+    error::message("gearman_client_run_tasks", &client);
   }
 }
 
+static gearman_return_t _client_created(gearman_task_st *task)
+{
+  const Args *args= static_cast<const Args*>(gearman_task_context(task));
+
+  if (args->verbose())
+  {
+    fprintf(stdout, "Task created: %s\n", gearman_task_job_handle(task));
+  }
+
+  return GEARMAN_SUCCESS;
+}
+
 static gearman_return_t _client_data(gearman_task_st *task)
 {
   const Args *args= static_cast<const Args*>(gearman_task_context(task));
@@ -482,8 +499,7 @@ static void _worker_free(void *, void *)
 
 void _worker(Args &args)
 {
-  bin::Worker local_worker;
-  gearman_worker_st &worker= local_worker.worker();
+  libgearman::Worker worker;
 
   if (args.timeout() >= 0)
   {
@@ -494,13 +510,13 @@ void _worker(Args &args)
   {
     if (gearman_failed(gearman_worker_add_servers(&worker, getenv("GEARMAN_SERVER"))))
     {
-      error::message("Error occurred while parsing GEARMAN_SERVER", worker);
+      error::message("Error occurred while parsing GEARMAN_SERVER", &worker);
       return;
     }
   }
   else if (gearman_failed(gearman_worker_add_server(&worker, args.host(), args.port())))
   {
-    error::message("gearman_worker_add_server", worker);
+    error::message("gearman_worker_add_server", &worker);
     _exit(EXIT_FAILURE);
   }
 
@@ -514,7 +530,7 @@ void _worker(Args &args)
     worker_argument_t pass(args, *iter);
     if (gearman_failed(gearman_worker_add_function(&worker, function.name(), 0, _worker_cb, &pass)))
     {
-      error::message("gearman_worker_add_function", worker);
+      error::message("gearman_worker_add_function", &worker);
       _exit(EXIT_FAILURE);
     }
   }
@@ -523,7 +539,7 @@ void _worker(Args &args)
   {
     if (gearman_failed(gearman_worker_work(&worker)))
     {
-      error::message("gearman_worker_work", worker);
+      error::message("gearman_worker_work", &worker);
     }
 
     if (args.count() > 0)
@@ -743,32 +759,45 @@ void _read_workload(int fd, Bytes& workload)
   }
 }
 
-static void usage(char *name)
+static void usage(Args& args, char *name)
 {
-  printf("Client mode: %s [options] [<data>]\n", name);
-  printf("Worker mode: %s -w [options] [<command> [<args> ...]]\n", name);
-
-  printf("\nCommon options to both client and worker modes.\n");
-  printf("\t-f <function> - Function name to use for jobs (can give many)\n");
-  printf("\t-h <host>     - Job server host\n");
-  printf("\t-H            - Print this help menu\n");
-  printf("\t-p <port>     - Job server port\n");
-  printf("\t-t <timeout>  - Timeout in milliseconds\n");
-  printf("\t-i <pidfile>  - Create a pidfile for the process\n");
-
-  printf("\nClient options:\n");
-  printf("\t-b            - Run jobs in the background\n");
-  printf("\t-I            - Run jobs as high priority\n");
-  printf("\t-L            - Run jobs as low priority\n");
-  printf("\t-n            - Run one job per line\n");
-  printf("\t-N            - Same as -n, but strip off the newline\n");
-  printf("\t-P            - Prefix all output lines with functions names\n");
-  printf("\t-s            - Send job without reading from standard input\n");
-  printf("\t-u <unique>   - Unique key to use for job\n");
-
-  printf("\nWorker options:\n");
-  printf("\t-c <count>    - Number of jobs for worker to run before exiting\n");
-  printf("\t-n            - Send data packet for each line\n");
-  printf("\t-N            - Same as -n, but strip off the newline\n");
-  printf("\t-w            - Run in worker mode\n");
+  FILE *stream;
+  if (args.is_error())
+  {
+    stream= stderr;
+    fprintf(stream, "\n%s\tError in usage(%s).\n\n", name, args.arg_error());
+  }
+  else
+  {
+    stream= stdout;
+    fprintf(stream, "\n%s\tError in usage(%s).\n\n", name, args.arg_error());
+  }
+
+  fprintf(stream, "Client mode: %s [options] [<data>]\n", name);
+  fprintf(stream, "Worker mode: %s -w [options] [<command> [<args> ...]]\n", name);
+
+  fprintf(stream, "\nCommon options to both client and worker modes.\n");
+  fprintf(stream, "\t-f <function> - Function name to use for jobs (can give many)\n");
+  fprintf(stream, "\t-h <host>     - Job server host\n");
+  fprintf(stream, "\t-H            - Print this help menu\n");
+  fprintf(stream, "\t-v            - Print diagnostic information to stdout(%s)\n", args.verbose() ? "true" : "false");
+  fprintf(stream, "\t-p <port>     - Job server port\n");
+  fprintf(stream, "\t-t <timeout>  - Timeout in milliseconds\n");
+  fprintf(stream, "\t-i <pidfile>  - Create a pidfile for the process\n");
+
+  fprintf(stream, "\nClient options:\n");
+  fprintf(stream, "\t-b            - Run jobs in the background(%s)\n", args.background() ? "true" : "false");
+  fprintf(stream, "\t-I            - Run jobs as high priority\n");
+  fprintf(stream, "\t-L            - Run jobs as low priority\n");
+  fprintf(stream, "\t-n            - Run one job per line(%s)\n", args.job_per_newline() ? "true" : "false");
+  fprintf(stream, "\t-N            - Same as -n, but strip off the newline(%s)\n", args.strip_newline() ? "true" : "false");
+  fprintf(stream, "\t-P            - Prefix all output lines with functions names\n");
+  fprintf(stream, "\t-s            - Send job without reading from standard input\n");
+  fprintf(stream, "\t-u <unique>   - Unique key to use for job\n");
+
+  fprintf(stream, "\nWorker options:\n");
+  fprintf(stream, "\t-c <count>    - Number of jobs for worker to run before exiting\n");
+  fprintf(stream, "\t-n            - Send data packet for each line(%s)\n", args.job_per_newline() ? "true" : "false");
+  fprintf(stream, "\t-N            - Same as -n, but strip off the newline(%s)\n", args.strip_newline() ? "true" : "false");
+  fprintf(stream, "\t-w            - Run in worker mode(%s)\n", args.worker() ? "true" : "false");
 }

bin/include.am

@@ -12,12 +12,9 @@
 # Included from Top Level Makefile.am
 # All paths should be given relative to the root
 
-noinst_HEADERS+= \
-		 bin/arguments.h \
-		 bin/error.h \
-		 bin/function.h \
-		 bin/client.h \
-		 bin/worker.h
+noinst_HEADERS+= bin/arguments.h
+noinst_HEADERS+= bin/error.h
+noinst_HEADERS+= bin/function.h
 
 bin_PROGRAMS+= bin/gearman
 bin_PROGRAMS+= bin/gearadmin

bin/worker.h

@@ -1,68 +0,0 @@
-/*  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 bin {
-
-class Worker
-{
-public:
-  Worker() :
-    _worker()
-  {
-    if (gearman_worker_create(&_worker) == NULL)
-    {
-      std::cerr << "Failed memory allocation while initializing memory." << std::endl;
-      abort();
-    }
-  }
-
-  ~Worker()
-  {
-    gearman_worker_free(&_worker);
-  }
-
-  gearman_worker_st &worker()
-  {
-    return _worker;
-  }
-
-private:
-  gearman_worker_st _worker;
-};
-} //namespace bin

configure.ac

@@ -67,7 +67,7 @@ AC_DEFINE([HAVE_GEARMAND_BINARY],[1],[gearmand binary was found])
 AC_DEFUN([CHECK_CURL],
          [AX_LIB_CURL([7.21.7],
                       [AX_CHECK_LIBRARY([LIBCURL],[curl/curl.h],[curl])],
-                      [AC_DEFINE([HAVE_LIBCURL],[1],[Have the LIBCURL library])
+                      [AC_DEFINE([HAVE_LIBCURL],[0],[Have the LIBCURL library])
                       ])
          ])
 CHECK_CURL

libgearman-server/plugins/protocol/http/protocol.cc

@@ -214,13 +214,13 @@ public:
                                     "HTTP/1.0 200 OK\r\n"
                                     "X-Gearman-Job-Handle: %.*s\r\n"
                                     "X-Gearman-Command: %s\r\n"
-                                    "Content-Length: %" PRIu64 "\r\n"
+                                    "Content-Length: %d\r\n"
                                     "Server: Gearman/" PACKAGE_VERSION "\r\n"
                                     "\r\n%.*s",
                                     packet->command == GEARMAN_COMMAND_JOB_CREATED ?  int(packet->arg_size[0]) : int(packet->arg_size[0] - 1),
-                                    (const char *)packet->arg[0],
+                                    (const char *)packet->arg[0], // Job handle
                                     gearman_strcommand(packet->command),
-                                    uint64_t(content.size()),
+                                    int(content.size()), // Content-length
                                     int(content.size()), &content[0]);
       }
       else