refactoring: move task check to "taken" into a separate function

Needed for #85

Author: LeonidVas

queue/abstract.lua

@@ -125,6 +125,16 @@ function tube.take(self, timeout)
     end
 end
 
+--- Checks is the task has been taken in a session with
+-- connection id == "conn_id".
+-- Throw an error on failure.
+local function check_task_on_taken(tube_id, task_id, conn_id)
+    local _taken = box.space._queue_taken:get{conn_id, tube_id, task_id}
+    if _taken == nil then
+        error("Task was not taken in the session")
+    end
+end
+
 function tube.touch(self, id, delta)
     if delta == nil then
         return
@@ -140,10 +150,7 @@ function tube.touch(self, id, delta)
         return
     end
 
-    local _taken = box.space._queue_taken:get{connection.id(), self.tube_id, id}
-    if _taken == nil then
-        error("Task was not taken in the session")
-    end
+    check_task_on_taken(self.tube_id, id, connection.id())
 
     local space_name = box.space._queue:get{self.name}[3]
     queue.stat[space_name]:inc('touch')
@@ -152,16 +159,14 @@ function tube.touch(self, id, delta)
 end
 
 function tube.ack(self, id)
-    local _taken = box.space._queue_taken:get{connection.id(), self.tube_id, id}
-    if _taken == nil then
-        error("Task was not taken in the session")
-    end
+    local conn_id = connection.id()
+    check_task_on_taken(self.tube_id, id, conn_id)
     local tube = box.space._queue:get{self.name}
     local space_name = tube[3]
 
     self:peek(id)
     -- delete task
-    box.space._queue_taken:delete{connection.id(), self.tube_id, id}
+    box.space._queue_taken:delete{conn_id, self.tube_id, id}
     local result = self.raw:normalize_task(
         self.raw:delete(id):transform(2, 1, state.DONE)
     )
@@ -173,10 +178,7 @@ end
 
 local function tube_release_internal(self, id, opts, connection_id)
     opts = opts or {}
-    local _taken = box.space._queue_taken:get{connection_id, self.tube_id, id}
-    if _taken == nil then
-        error("Task was not taken in the session")
-    end
+    check_task_on_taken(self.tube_id, id, connection_id)
 
     box.space._queue_taken:delete{connection_id, self.tube_id, id}
     self:peek(id)
@@ -197,9 +199,10 @@ end
 
 function tube.bury(self, id)
     local task = self:peek(id)
-    local _taken = box.space._queue_taken:get{connection.id(), self.tube_id, id}
-    if _taken ~= nil then
-        box.space._queue_taken:delete{connection.id(), self.tube_id, id}
+    local conn_id = connection.id()
+    local is_taken, _ = pcall(check_task_on_taken, self.tube_id, id, conn_id)
+    if is_taken then
+        box.space._queue_taken:delete{conn_id, self.tube_id, id}
     end
     if task[2] == state.BURIED then
         return task