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

[Brian Aker]

=== modified file 'libgearman/interface/worker.hpp'
--- libgearman/interface/worker.hpp	2013-07-13 22:59:44 +0000
+++ libgearman/interface/worker.hpp	2013-08-16 08:33:09 +0000
@@ -49,6 +49,7 @@
     bool grab_uniq;
     bool grab_all;
     bool timeout_return;
+    bool _in_work;
 
     Options() :
       non_blocking(false),
@@ -56,7 +57,8 @@
       change(false),
       grab_uniq(true),
       grab_all(true),
-      timeout_return(false)
+      timeout_return(false),
+      _in_work(false)
     { }
   } options;
   enum gearman_worker_state_t state;
@@ -142,6 +144,16 @@
     universal.options._ssl= ssl_;
   }
 
+  bool in_work() const
+  {
+    return options._in_work;
+  }
+
+  void in_work(bool in_work_)
+  {
+    options._in_work= in_work_;
+  }
+
   bool has_identifier() const
   {
     return universal.has_identifier();
 
=== modified file 'libgearman/worker.cc'
--- libgearman/worker.cc	2013-08-05 22:43:13 +0000
+++ libgearman/worker.cc	2013-08-16 08:33:09 +0000
@@ -656,6 +656,7 @@
     Worker* worker= worker_shell->impl();
     struct _worker_function_st *function;
     uint32_t active;
+    bool no_job= false;
 
     gearman_return_t unused;
     if (not ret_ptr)
@@ -842,6 +843,7 @@
                 if (worker->job()->impl()->assigned.command == GEARMAN_COMMAND_NO_JOB or
                     worker->job()->impl()->assigned.command == GEARMAN_COMMAND_OPTION_RES)
                 {
+                  no_job= true;
                   gearman_packet_free(&(worker->job()->impl()->assigned));
                   break;
                 }
@@ -861,6 +863,12 @@
             }
           }
 
+          if (worker->in_work() == false and no_job)
+          {
+            *ret_ptr= GEARMAN_NO_JOBS;
+            break;
+          }
+
         case GEARMAN_WORKER_STATE_PRE_SLEEP:
           for (worker->con= (&worker->universal)->con_list; worker->con;
                worker->con= worker->con->next_connection())
@@ -937,6 +945,11 @@
 
           break;
       }
+
+      if (*ret_ptr == GEARMAN_NO_JOBS)
+      {
+        break;
+      }
     }
   }
 
@@ -1038,8 +1051,10 @@
     {
       case GEARMAN_WORKER_WORK_UNIVERSAL_GRAB_JOB:
         {
+          worker->in_work(true);
           gearman_return_t ret;
           worker->work_job(gearman_worker_grab_job(worker->shell(), NULL, &ret));
+          worker->in_work(false);
 
           if (gearman_failed(ret))
           {
 
=== modified file 'tests/libgearman-1.0/worker_test.cc'
--- tests/libgearman-1.0/worker_test.cc	2013-08-11 05:40:05 +0000
+++ tests/libgearman-1.0/worker_test.cc	2013-08-16 08:33:09 +0000
@@ -148,6 +148,20 @@
   return TEST_SUCCESS;
 }
 
+static test_return_t gearman_worker_grab_job_TEST(void *)
+{
+  libgearman::Worker worker(libtest::default_port());
+
+  ASSERT_EQ(gearman_worker_register(&worker, __func__, 0), GEARMAN_SUCCESS);
+
+  gearman_return_t ret;
+  gearman_job_st* job= gearman_worker_grab_job(&worker, NULL, &ret);
+  ASSERT_NULL(job);
+  ASSERT_EQ(GEARMAN_NO_JOBS, ret);
+
+  return TEST_SUCCESS;
+}
+
 static test_return_t gearman_worker_free_TEST(void *)
 {
   gearman_worker_free(NULL);
@@ -1599,6 +1613,11 @@
   {0, 0, 0}
 };
 
+test_st gearman_worker_grab_job_TESTS[] ={
+  {"gearman_worker_grab_job(NO_JOBS)", 0, gearman_worker_grab_job_TEST },
+  {0, 0, 0}
+};
+
 test_st gearman_worker_st_NULL_invocation_TESTS[] ={
   {"gearman_worker_free()", 0, gearman_worker_free_TEST },
   {"gearman_worker_error()", 0, gearman_worker_error_TEST },
@@ -1619,6 +1638,7 @@
   {"worker", 0, 0, worker_TESTS},
   {"worker defaults", 0, 0, worker_defaults},
   {"null gearman_worker_st invocation", 0, 0, gearman_worker_st_NULL_invocation_TESTS },
+  {"gearman_worker_grab_job()", 0, 0, gearman_worker_grab_job_TESTS },
   {"gearman_worker_set_identifier()", 0, 0, gearman_worker_set_identifier_TESTS},
   {0, 0, 0, 0}
 };