Merge lp:~brianaker/gearmand/finish into lp:gearmand

[Brian Aker]

=== modified file 'libgearman/error.hpp'
--- libgearman/error.hpp	2013-02-04 06:00:39 +0000
+++ libgearman/error.hpp	2013-07-03 01:51:27 +0000
@@ -63,3 +63,4 @@
                                               const char *position);
 
 void gearman_worker_reset_error(gearman_worker_st *worker);
+void gearman_worker_reset_error(Worker&);
 
=== modified file 'libgearman/function/function_v1.hpp'
--- libgearman/function/function_v1.hpp	2012-08-20 17:07:47 +0000
+++ libgearman/function/function_v1.hpp	2013-07-03 01:51:27 +0000
@@ -63,7 +63,7 @@
     }
 
     job->error_code= GEARMAN_SUCCESS;
-    job->worker->impl()->work_result= _worker_fn(job, context_arg, &(job->worker->impl()->work_result_size), &job->error_code);
+    job->_worker.work_result= _worker_fn(job, context_arg, &(job->_worker.work_result_size), &job->error_code);
 
     if (job->error_code == GEARMAN_LOST_CONNECTION)
     {
 
=== modified file 'libgearman/interface/worker.hpp'
--- libgearman/interface/worker.hpp	2013-06-05 21:59:31 +0000
+++ libgearman/interface/worker.hpp	2013-07-03 01:51:27 +0000
@@ -127,6 +127,11 @@
     universal.options._ssl= ssl_;
   }
 
+  const char* error() const
+  {
+    return universal.error();
+  }
+
 private:
   gearman_worker_st* _shell;
   gearman_worker_st _owned_shell;
 
=== modified file 'libgearman/job.cc'
--- libgearman/job.cc	2013-06-15 07:55:07 +0000
+++ libgearman/job.cc	2013-07-03 01:51:27 +0000
@@ -170,42 +170,46 @@
 
 gearman_job_st *gearman_job_create(gearman_worker_st *worker, gearman_job_st *job)
 {
-  if (job)
-  {
-    job->options.allocated= false;
-  }
-  else
-  {
-    job= new (std::nothrow) gearman_job_st;
-    if (job == NULL)
-    {
-      gearman_perror(worker->impl()->universal, "new");
-      return NULL;
-    }
-
-    job->options.allocated= true;
-  }
-
-  job->options.assigned_in_use= false;
-  job->options.work_in_use= false;
-  job->options.finished= false;
-
-  job->worker= worker;
-  job->reducer= NULL;
-  job->error_code= GEARMAN_UNKNOWN_STATE;
-
-  if (worker->impl()->job_list)
-  {
-    worker->impl()->job_list->prev= job;
-  }
-  job->next= worker->impl()->job_list;
-  job->prev= NULL;
-  worker->impl()->job_list= job;
-  worker->impl()->job_count++;
-
-  job->con= NULL;
-
-  return job;
+  if (worker)
+  {
+    if (job)
+    {
+      job->options.allocated= false;
+    }
+    else
+    {
+      job= new (std::nothrow) gearman_job_st(*(worker->impl()));
+      if (job == NULL)
+      {
+        gearman_perror(job->_worker.universal, "new");
+        return NULL;
+      }
+
+      job->options.allocated= true;
+    }
+
+    job->options.assigned_in_use= false;
+    job->options.work_in_use= false;
+    job->options.finished= false;
+
+    job->reducer= NULL;
+    job->error_code= GEARMAN_UNKNOWN_STATE;
+
+    if (job->_worker.job_list)
+    {
+      job->_worker.job_list->prev= job;
+    }
+    job->next= job->_worker.job_list;
+    job->prev= NULL;
+    job->_worker.job_list= job;
+    job->_worker.job_count++;
+
+    job->con= NULL;
+
+    return job;
+  }
+
+  return NULL;
 }
 
 bool gearman_job_build_reducer(gearman_job_st *job, gearman_aggregator_fn *aggregator_fn)
@@ -217,7 +221,7 @@
 
   gearman_string_t reducer_func= gearman_job_reducer_string(job);
 
-  job->reducer= new (std::nothrow) gearman_job_reducer_st(job->worker->impl()->universal, reducer_func, aggregator_fn);
+  job->reducer= new (std::nothrow) gearman_job_reducer_st(job->universal(), reducer_func, aggregator_fn);
   if (not job->reducer)
   {
     gearman_job_free(job);
@@ -235,9 +239,9 @@
 
 static inline void gearman_job_reset_error(gearman_job_st *job)
 {
-  if (job->worker)
+  if (job)
   {
-    gearman_worker_reset_error(job->worker);
+    gearman_worker_reset_error(job->_worker);
   }
 }
 
@@ -245,36 +249,41 @@
 {
   if (job)
   {
-    const void *args[2];
-    size_t args_size[2];
-
-    if (job->reducer)
-    {
-      gearman_argument_t value= gearman_argument_make(NULL, 0, static_cast<const char *>(data), data_size);
-      job->reducer->add(value);
-
-      return GEARMAN_SUCCESS;
-    }
-
-    if ((job->options.work_in_use) == false)
-    {
-      args[0]= job->assigned.arg[0];
-      args_size[0]= job->assigned.arg_size[0];
-      args[1]= data;
-      args_size[1]= data_size;
-      gearman_return_t ret= gearman_packet_create_args(job->worker->impl()->universal, job->work,
-                                                       GEARMAN_MAGIC_REQUEST,
-                                                       GEARMAN_COMMAND_WORK_DATA,
-                                                       args, args_size, 2);
-      if (gearman_failed(ret))
-      {
-        return ret;
-      }
-
-      job->options.work_in_use= true;
-    }
-
-    return _job_send(job);
+    if (job->finished() == false)
+    {
+      const void *args[2];
+      size_t args_size[2];
+
+      if (job->reducer)
+      {
+        gearman_argument_t value= gearman_argument_make(NULL, 0, static_cast<const char *>(data), data_size);
+        job->reducer->add(value);
+
+        return GEARMAN_SUCCESS;
+      }
+
+      if ((job->options.work_in_use) == false)
+      {
+        args[0]= job->assigned.arg[0];
+        args_size[0]= job->assigned.arg_size[0];
+        args[1]= data;
+        args_size[1]= data_size;
+        gearman_return_t ret= gearman_packet_create_args(job->universal(), job->work,
+                                                         GEARMAN_MAGIC_REQUEST,
+                                                         GEARMAN_COMMAND_WORK_DATA,
+                                                         args, args_size, 2);
+        if (gearman_failed(ret))
+        {
+          return ret;
+        }
+
+        job->options.work_in_use= true;
+      }
+
+      return _job_send(job);
+    }
+
+    return GEARMAN_SUCCESS;
   }
 
   return GEARMAN_INVALID_ARGUMENT;
@@ -286,30 +295,35 @@
 {
   if (job)
   {
-    const void *args[2];
-    size_t args_size[2];
-
-    if ((job->options.work_in_use) == false)
+    if (job->finished() == false)
     {
-      args[0]= job->assigned.arg[0];
-      args_size[0]= job->assigned.arg_size[0];
-      args[1]= warning;
-      args_size[1]= warning_size;
+      const void *args[2];
+      size_t args_size[2];
 
-      gearman_return_t ret;
-      ret= gearman_packet_create_args(job->worker->impl()->universal, job->work,
-                                      GEARMAN_MAGIC_REQUEST,
-                                      GEARMAN_COMMAND_WORK_WARNING,
-                                      args, args_size, 2);
-      if (gearman_failed(ret))
+      if ((job->options.work_in_use) == false)
       {
-        return ret;
+        args[0]= job->assigned.arg[0];
+        args_size[0]= job->assigned.arg_size[0];
+        args[1]= warning;
+        args_size[1]= warning_size;
+
+        gearman_return_t ret;
+        ret= gearman_packet_create_args(job->universal(), job->work,
+                                        GEARMAN_MAGIC_REQUEST,
+                                        GEARMAN_COMMAND_WORK_WARNING,
+                                        args, args_size, 2);
+        if (gearman_failed(ret))
+        {
+          return ret;
+        }
+
+        job->options.work_in_use= true;
       }
 
-      job->options.work_in_use= true;
+      return _job_send(job);
     }
 
-    return _job_send(job);
+    return GEARMAN_SUCCESS;
   }
 
   return GEARMAN_INVALID_ARGUMENT;
@@ -321,37 +335,42 @@
 {
   if (job)
   {
-    char numerator_string[12];
-    char denominator_string[12];
-    const void *args[3];
-    size_t args_size[3];
-
-    if (not (job->options.work_in_use))
+    if (job->finished() == false)
     {
-      snprintf(numerator_string, 12, "%u", numerator);
-      snprintf(denominator_string, 12, "%u", denominator);
-
-      args[0]= job->assigned.arg[0];
-      args_size[0]= job->assigned.arg_size[0];
-      args[1]= numerator_string;
-      args_size[1]= strlen(numerator_string) + 1;
-      args[2]= denominator_string;
-      args_size[2]= strlen(denominator_string);
-
-      gearman_return_t ret;
-      ret= gearman_packet_create_args(job->worker->impl()->universal, job->work,
-                                      GEARMAN_MAGIC_REQUEST,
-                                      GEARMAN_COMMAND_WORK_STATUS,
-                                      args, args_size, 3);
-      if (gearman_failed(ret))
+      char numerator_string[12];
+      char denominator_string[12];
+      const void *args[3];
+      size_t args_size[3];
+
+      if (not (job->options.work_in_use))
       {
-        return ret;
+        snprintf(numerator_string, 12, "%u", numerator);
+        snprintf(denominator_string, 12, "%u", denominator);
+
+        args[0]= job->assigned.arg[0];
+        args_size[0]= job->assigned.arg_size[0];
+        args[1]= numerator_string;
+        args_size[1]= strlen(numerator_string) + 1;
+        args[2]= denominator_string;
+        args_size[2]= strlen(denominator_string);
+
+        gearman_return_t ret;
+        ret= gearman_packet_create_args(job->universal(), job->work,
+                                        GEARMAN_MAGIC_REQUEST,
+                                        GEARMAN_COMMAND_WORK_STATUS,
+                                        args, args_size, 3);
+        if (gearman_failed(ret))
+        {
+          return ret;
+        }
+
+        job->options.work_in_use= true;
       }
 
-      job->options.work_in_use= true;
+      return _job_send(job);
     }
 
-    return _job_send(job);
+    return GEARMAN_SUCCESS;
   }
 
   return GEARMAN_INVALID_ARGUMENT;
@@ -363,12 +382,17 @@
 {
   if (job)
   {
-    if (job->reducer)
+    if (job->finished() == false)
     {
-      return GEARMAN_INVALID_ARGUMENT;
+      if (job->reducer)
+      {
+        return GEARMAN_INVALID_ARGUMENT;
+      }
+
+      return gearman_job_send_complete_fin(job, result, result_size);
     }
 
-    return gearman_job_send_complete_fin(job, result, result_size);
+    return GEARMAN_SUCCESS;
   }
 
   return GEARMAN_INVALID_ARGUMENT;
@@ -379,66 +403,64 @@
 {
   if (job)
   {
-    if (job->options.finished)
-    {
-      return GEARMAN_SUCCESS;
-    }
-
-    if (job->reducer)
-    {
-      if (result_size)
-      {
-        gearman_argument_t value= gearman_argument_make(NULL, 0, static_cast<const char *>(result), result_size);
-        job->reducer->add(value);
-      }
-
-      gearman_return_t rc= job->reducer->complete();
-      if (gearman_failed(rc))
-      {
-        return gearman_error(job->worker->impl()->universal, rc, "The reducer's complete() returned an error");
-      }
-
-      const gearman_vector_st *reduced_value= job->reducer->result.string();
-      if (reduced_value)
-      {
-        result= gearman_string_value(reduced_value);
-        result_size= gearman_string_length(reduced_value);
-      }
-      else
-      {
-        result= NULL;
-        result_size= 0;
-      }
-    } 
-
-    const void *args[2];
-    size_t args_size[2];
-
-    if (not (job->options.work_in_use))
-    {
-      args[0]= job->assigned.arg[0];
-      args_size[0]= job->assigned.arg_size[0];
-
-      args[1]= result;
-      args_size[1]= result_size;
-      gearman_return_t ret= gearman_packet_create_args(job->worker->impl()->universal, job->work,
-                                                       GEARMAN_MAGIC_REQUEST,
-                                                       GEARMAN_COMMAND_WORK_COMPLETE,
-                                                       args, args_size, 2);
+    if (job->finished() == false)
+    {
+      if (job->reducer)
+      {
+        if (result_size)
+        {
+          gearman_argument_t value= gearman_argument_make(NULL, 0, static_cast<const char *>(result), result_size);
+          job->reducer->add(value);
+        }
+
+        gearman_return_t rc= job->reducer->complete();
+        if (gearman_failed(rc))
+        {
+          return gearman_error(job->universal(), rc, "The reducer's complete() returned an error");
+        }
+
+        const gearman_vector_st *reduced_value= job->reducer->result.string();
+        if (reduced_value)
+        {
+          result= gearman_string_value(reduced_value);
+          result_size= gearman_string_length(reduced_value);
+        }
+        else
+        {
+          result= NULL;
+          result_size= 0;
+        }
+      } 
+
+      const void *args[2];
+      size_t args_size[2];
+
+      if (not (job->options.work_in_use))
+      {
+        args[0]= job->assigned.arg[0];
+        args_size[0]= job->assigned.arg_size[0];
+
+        args[1]= result;
+        args_size[1]= result_size;
+        gearman_return_t ret= gearman_packet_create_args(job->_worker.universal, job->work,
+                                                         GEARMAN_MAGIC_REQUEST,
+                                                         GEARMAN_COMMAND_WORK_COMPLETE,
+                                                         args, args_size, 2);
+        if (gearman_failed(ret))
+        {
+          return ret;
+        }
+        job->options.work_in_use= true;
+      }
+
+      gearman_return_t ret= _job_send(job);
       if (gearman_failed(ret))
       {
         return ret;
       }
-      job->options.work_in_use= true;
-    }
-
-    gearman_return_t ret= _job_send(job);
-    if (gearman_failed(ret))
-    {
-      return ret;
-    }
-
-    job->options.finished= true;
+
+      job->finished(true);
+    }
 
     return GEARMAN_SUCCESS;
   }
@@ -452,21 +474,26 @@
 {
   if (job)
   {
-    if (not (job->options.work_in_use))
+    if (job->finished() == false)
     {
-      gearman_string_t handle_string= { static_cast<const char *>(job->assigned.arg[0]), job->assigned.arg_size[0] };
-      gearman_string_t exception_string= { static_cast<const char *>(exception), exception_size };
-
-      gearman_return_t ret= libgearman::protocol::work_exception(job->worker->impl()->universal, job->work, handle_string, exception_string);
-      if (gearman_failed(ret))
+      if (job->options.work_in_use == false)
       {
-        return ret;
+        gearman_string_t handle_string= { static_cast<const char *>(job->assigned.arg[0]), job->assigned.arg_size[0] };
+        gearman_string_t exception_string= { static_cast<const char *>(exception), exception_size };
+
+        gearman_return_t ret= libgearman::protocol::work_exception(job->_worker.universal, job->work, handle_string, exception_string);
+        if (gearman_failed(ret))
+        {
+          return ret;
+        }
+
+        job->options.work_in_use= true;
       }
 
-      job->options.work_in_use= true;
+      return _job_send(job);
     }
 
-    return _job_send(job);
+    return GEARMAN_SUCCESS;
   }
 
   return GEARMAN_INVALID_ARGUMENT;
@@ -476,12 +503,17 @@
 {
   if (job)
   {
-    if (job->reducer)
+    if (job->finished() == false)
     {
-      return GEARMAN_INVALID_ARGUMENT;
+      if (job->reducer)
+      {
+        return gearman_error(job->universal(), GEARMAN_INVALID_ARGUMENT, "Job has a reducer");
+      }
+
+      return gearman_job_send_fail_fin(job);
     }
 
-    return gearman_job_send_fail_fin(job);
+    return GEARMAN_SUCCESS;
   }
 
   return GEARMAN_INVALID_ARGUMENT;
@@ -494,34 +526,32 @@
     const void *args[1];
     size_t args_size[1];
 
-    if (job->options.finished)
-    {
-      return GEARMAN_SUCCESS;
-    }
-
-    if (not (job->options.work_in_use))
-    {
-      args[0]= job->assigned.arg[0];
-      args_size[0]= job->assigned.arg_size[0] - 1;
-      gearman_return_t ret= gearman_packet_create_args(job->worker->impl()->universal, job->work,
-                                                       GEARMAN_MAGIC_REQUEST,
-                                                       GEARMAN_COMMAND_WORK_FAIL,
-                                                       args, args_size, 1);
+    if (job->finished() == false)
+    {
+      if (not (job->options.work_in_use))
+      {
+        args[0]= job->assigned.arg[0];
+        args_size[0]= job->assigned.arg_size[0] - 1;
+        gearman_return_t ret= gearman_packet_create_args(job->_worker.universal, job->work,
+                                                         GEARMAN_MAGIC_REQUEST,
+                                                         GEARMAN_COMMAND_WORK_FAIL,
+                                                         args, args_size, 1);
+        if (gearman_failed(ret))
+        {
+          return ret;
+        }
+
+        job->options.work_in_use= true;
+      }
+
+      gearman_return_t ret= _job_send(job);
       if (gearman_failed(ret))
       {
         return ret;
       }
 
-      job->options.work_in_use= true;
-    }
-
-    gearman_return_t ret= _job_send(job);
-    if (gearman_failed(ret))
-    {
-      return ret;
-    }
-
-    job->options.finished= true;
+      job->finished(true);
+    }
 
     return GEARMAN_SUCCESS;
   }
@@ -664,9 +694,9 @@
       gearman_packet_free(&(job->work));
     }
 
-    if (job->worker->impl()->job_list == job)
+    if (job->_worker.job_list == job)
     {
-      job->worker->impl()->job_list= job->next;
+      job->_worker.job_list= job->next;
     }
 
     if (job->prev)
@@ -678,7 +708,7 @@
     {
       job->next->prev= job->prev;
     }
-    job->worker->impl()->job_count--;
+    job->_worker.job_count--;
 
     delete job->reducer;
     job->reducer= NULL;
@@ -700,7 +730,7 @@
 
   while ((ret == GEARMAN_IO_WAIT) or (ret == GEARMAN_TIMEOUT))
   {
-    ret= gearman_wait(job->worker->impl()->universal);
+    ret= gearman_wait(job->universal());
     if (ret == GEARMAN_SUCCESS)
     {
       ret= job->con->send_packet(job->work, true);
@@ -720,9 +750,9 @@
 
 const char *gearman_job_error(gearman_job_st *job)
 {
-  if (job and job->worker)
+  if (job)
   {
-    return gearman_worker_error(job->worker);
+    return job->_worker.error();
   }
 
   return NULL;
 
=== modified file 'libgearman/job.hpp'
--- libgearman/job.hpp	2012-08-19 21:00:10 +0000
+++ libgearman/job.hpp	2013-07-03 01:51:27 +0000
@@ -37,15 +37,32 @@
 
 #pragma once
 
+#include "libgearman/interface/worker.hpp"
+
 struct gearman_job_st
 {
+  gearman_job_st(Worker& worker_):
+    _worker(worker_)
+  { }
+
   struct {
     bool allocated;
     bool assigned_in_use;
     bool work_in_use;
     bool finished;
   } options;
-  gearman_worker_st *worker;
+
+  bool finished() const
+  {
+    return options.finished;
+  }
+
+  void finished(const bool finished_)
+  {
+    options.finished= finished_;
+  }
+
+  Worker& _worker;
   gearman_job_st *next;
   gearman_job_st *prev;
   gearman_connection_st *con;
@@ -53,4 +70,9 @@
   gearman_packet_st work;
   struct gearman_job_reducer_st *reducer;
   gearman_return_t error_code;
+
+  gearman_universal_st& universal()
+  {
+    return _worker.universal;
+  }
 };
 
=== modified file 'libgearman/worker.cc'
--- libgearman/worker.cc	2013-06-23 08:23:26 +0000
+++ libgearman/worker.cc	2013-07-03 01:51:27 +0000
@@ -971,6 +971,11 @@
   return GEARMAN_INVALID_ARGUMENT;
 }
 
+void gearman_worker_reset_error(Worker& worker)
+{
+  universal_reset_error(worker.universal);
+}
+
 void gearman_worker_reset_error(gearman_worker_st *worker)
 {
   if (worker and worker->impl())
@@ -1354,7 +1359,7 @@
 {
   if (job)
   {
-    return gearman_worker_clone(NULL, job->worker);
+    return gearman_worker_clone(NULL, job->_worker.shell());
   }
 
   return NULL;