create worker pool with worker's queue (#218)

* create worker pool with worker's queue

* add test

Co-authored-by: Samuel Colvin <[email protected]>

Author: erakli

arq/worker.py

@@ -292,7 +292,10 @@ def pool(self) -> ArqRedis:
     async def main(self) -> None:
         if self._pool is None:
             self._pool = await create_pool(
-                self.redis_settings, job_deserializer=self.job_deserializer, job_serializer=self.job_serializer
+                self.redis_settings,
+                job_deserializer=self.job_deserializer,
+                job_serializer=self.job_serializer,
+                default_queue_name=self.queue_name,
             )
 
         logger.info('Starting worker for %d functions: %s', len(self.functions), ', '.join(self.functions))

tests/test_main.py

@@ -90,6 +90,30 @@ async def parent_job(ctx):
     assert inner_result == 42
 
 
+async def test_enqueue_job_custom_queue(arq_redis: ArqRedis, worker):
+    async def foobar(ctx):
+        return 42
+
+    async def parent_job(ctx):
+        inner_job = await ctx['redis'].enqueue_job('foobar')
+        return inner_job.job_id
+
+    job = await arq_redis.enqueue_job('parent_job', _queue_name='spanner')
+
+    worker: Worker = worker(
+        functions=[func(parent_job, name='parent_job'), func(foobar, name='foobar')],
+        arq_redis=None,
+        queue_name='spanner',
+    )
+
+    await worker.main()
+    inner_job_id = await job.result(pole_delay=0)
+    assert inner_job_id is not None
+    inner_job = Job(inner_job_id, arq_redis, _queue_name='spanner')
+    inner_result = await inner_job.result(pole_delay=0)
+    assert inner_result == 42
+
+
 async def test_job_error(arq_redis: ArqRedis, worker):
     async def foobar(ctx):
         raise RuntimeError('foobar error')