Add script for removing data from queue.

kt/scripts/kt.lua

@@ -812,6 +812,54 @@ function queue_add(inmap, outmap)
   return kt.RVSUCCESS
 end
 
+
+-- remove data from a queue
+-- accepts: { queue, data, db }
+-- returns { num }
+function queue_remove(inmap, outmap)
+  local db = _select_db(inmap)
+  local queue = inmap.queue
+  local data = inmap.data
+  if not queue or not data then
+    kt.log("info", "missing queue or data parameter in queue_remove call")
+    return kt.RVEINVALID
+  end
+
+  local cursor = db:cursor()
+  local key = string.format("%s\t", queue)
+  local pattern = string.format("^%s\t", queue)
+
+  -- No data, we're done.
+  if not cursor:jump(key) then
+    cursor:disable()
+    outmap['num'] = '0'
+    return kt.RVSUCCESS
+  end
+
+  local k, v, xt
+  local num = 0
+
+  while true do
+    -- Retrieve the key, value and xt from the cursor. If the cursor is
+    -- invalidated (e.g., during the remove()), then nil is returned.
+    k, v, xt = cursor:get(false)
+    if not k then break end
+
+    -- If this is not a queue item key, we are done.
+    if not k:match(pattern) then break end
+
+    -- Data matches value, remove this item from the queue.
+    if data == v and cursor:remove() then
+      num = num + 1
+    elseif not cursor:step() then
+      break
+    end
+  end
+  cursor:disable()
+  outmap['num'] = tostring(num)
+  return kt.RVSUCCESS
+end
+
 -- pop/dequeue data from queue
 -- accepts: { queue, n, db }
 -- returns { idx: data, ... }

tests.py

@@ -759,6 +759,18 @@ def test_queue_methods(self):
         self.assertEqual(L.queue_rpop(queue='tq'), {})
         self.assertEqual(L.queue_pop(queue='tq'), {})
 
+        # Verify we can remove data by value.
+        for i in range(5):
+            L.queue_add(queue='tq', data='i%s' % (i % 2))
+
+        self.assertEqual(L.queue_remove(queue='tq', data='i1'), {'num': '2'})
+        self.assertEqual(L.queue_remove(queue='tq', data='x'), {'num': '0'})
+        self.assertEqual(L.queue_size(queue='tq'), {'num': '3'})
+        self.assertEqual(L.queue_pop(queue='tq'), {'0': 'i0'})
+        self.assertEqual(L.queue_remove(queue='tq', data='i0'), {'num': '2'})
+        self.assertEqual(L.queue_size(queue='tq'), {'num': '0'})
+        self.assertEqual(L.queue_pop(queue='tq'), {})
+
     def test_hexastore(self):
         L = self.db.lua
         data = (