raw subqueues should use their root queue's config

Author: Florian Perucki

mrq/queue.py

@@ -15,6 +15,11 @@ class Queue(object):
     is_set = False
     is_reverse = False
 
+    # root_id will contain the root queue id without any trailing subqueue delimiter
+    # e.g. if self.id is "some_queue/" then self.root_id will contain "some_queue"
+    # and if self.id is "some_queue/some_subqueue" then self.root_id will contain "some_queue"
+    root_id = None
+
     use_large_ids = False
 
     # This is a mutable type so it is shared by all instances
@@ -47,6 +52,13 @@ def __init__(self, queue_id, add_to_known_queues=False):
         if "_sorted" in self.id:
             self.is_sorted = True
 
+        self.root_id = self.id
+
+        delimiter = context.get_current_config().get("subqueues_delimiter")
+        if delimiter is not None and delimiter in self.id:
+            # Get the root queue id with no trailing delimiter
+            self.root_id = self.id.split(delimiter)[0]
+
         self.use_large_ids = context.get_current_config()["use_large_job_ids"]
 
         # If this is the first time this process sees this queue, try to add it
@@ -117,7 +129,7 @@ def redis_known_subqueues(self):
     def get_config(self):
         """ Returns the specific configuration for this queue """
 
-        return context.get_current_config().get("raw_queues", {}).get(self.id) or {}
+        return context.get_current_config().get("raw_queues", {}).get(self.root_id) or {}
 
     def serialize_job_ids(self, job_ids):
         """ Returns job_ids serialized for storage in Redis """

tests/test_raw.py

@@ -79,14 +79,23 @@ def test_raw_sorted(worker, p_queue, p_pushback, p_timed, p_flags):
     assert test_collection.count() == 3
 
 
+@pytest.mark.parametrize("has_subqueue", [False, True])
 @pytest.mark.parametrize(["p_queue", "p_set"], [
     ["test_raw", False],
     ["test_set", True]
 ])
-def test_raw_set(worker, p_queue, p_set):
-
-    worker.start(
-        flags="--greenlets 10 --config tests/fixtures/config-raw1.py", queues=p_queue)
+def test_raw_set(worker, has_subqueue, p_queue, p_set):
+    flags = "--greenlets 10 --config tests/fixtures/config-raw1.py"
+    if has_subqueue:
+      flags = "%s --subqueues_refresh_interval=0.1" % flags
+      # worker should dequeue all subqueues
+      p_queue = "%s/" % p_queue
+
+    worker.start(flags=flags, queues=p_queue)
+
+    if has_subqueue:
+      # queue tasks in p_queue/subqueue
+      p_queue = "%ssubqueue" % p_queue
 
     test_collection = worker.mongodb_logs.tests_inserts
     jobs_collection = worker.mongodb_jobs.mrq_jobs