Restore intended interface of RmqThreadCommunicator (#124)

In commit 3806a9e the requirement of the
`aio-pika` dependency was updated. In adjusting the code to the changes
in their API, certain methods of the `RmqThreadCommunicator` were made
asynchronous. However, the whole point of the `RmqThreadCommunicator` is
to shield the consumer from any async code.

The changes in the interface and the tests are reverted and the
implementation of `RmqThreadCommunicator` is updated to match the inter-
face of the newer `aio-pika` version.

Doing so actually revealed a bug in `pytray` causing the unit test
`test_queue_get_next` to hang forever while waiting for the future to be
set. The `pytray.LoopScheduler` was propagating a future the wrong way
around: instead of the result from a thread future being passed to the
asyncio future, it was doing the opposite causing the future to never be
resolved. The bug is fixed in `pytray==0.3.4`.

pyproject.toml

@@ -27,10 +27,9 @@ keywords = ['ommunication', 'messaging', 'rpc', 'broadcast']
 requires-python = '>=3.7'
 dependencies = [
     'aio-pika~=8.0',
-    'async_generator',
     'deprecation',
     'nest_asyncio~=1.5,>=1.5.1',
-    'pytray>=0.2.2,<0.4.0',
+    'pytray>=0.3.4,<0.4.0',
     'pyyaml~=5.4',
 ]
 

src/kiwipy/rmq/tasks.py

@@ -3,12 +3,11 @@
 import collections
 from contextlib import asynccontextmanager
 import logging
-from typing import Optional
+from typing import Generator, Optional
 import uuid
 import weakref
 
 import aio_pika
-from async_generator import async_generator, yield_
 import shortuuid
 
 import kiwipy
@@ -94,14 +93,13 @@ async def connect(self):
 
         await self._create_task_queue()
 
-    @async_generator
     async def __aiter__(self):
         tasks = []
         try:
             while True:
                 task = RmqIncomingTask(self, await self._task_queue.get(timeout=1.))
                 tasks.append(task)
-                await yield_(task)
+                yield task
         except aio_pika.exceptions.QueueEmpty:
             return
         finally:
@@ -111,8 +109,10 @@ async def __aiter__(self):
                     await task.requeue()
 
     @asynccontextmanager
-    @async_generator
-    async def next_task(self, no_ack=False, fail=True, timeout=defaults.TASK_FETCH_TIMEOUT):
+    async def next_task(self,
+                        no_ack=False,
+                        fail=True,
+                        timeout=defaults.TASK_FETCH_TIMEOUT) -> Generator['RmqIncomingTask', None, None]:
         """
         Get the next task from the queue.
 
@@ -126,7 +126,7 @@ async def next_task(self, no_ack=False, fail=True, timeout=defaults.TASK_FETCH_T
         else:
             task = RmqIncomingTask(self, message)
             try:
-                await yield_(task)
+                yield task
             finally:
                 if task.state == TASK_PENDING:
                     await task.requeue()
@@ -236,7 +236,7 @@ def process(self) -> asyncio.Future:
         outcome = self._loop.create_future()
         # Rely on the done callback to signal the end of processing
         outcome.add_done_callback(self._on_task_done)
-        # Or the user let's the future get destroyed
+        # Or the user lets the future get destroyed
         self._outcome_ref = weakref.ref(outcome, self._outcome_destroyed)
 
         return outcome
@@ -250,9 +250,10 @@ async def requeue(self):
         self._finalise()
 
     @asynccontextmanager
-    async def processing(self):
+    async def processing(self) -> Generator[asyncio.Future, None, None]:
         """Processing context.  The task should be done at the end otherwise it's assumed the
-        caller doesn't want to process it and it's sent back to the queue"""
+        caller doesn't want to process it, and it's sent back to the queue"""
+
         if self._state != TASK_PENDING:
             raise asyncio.InvalidStateError(f'The task is {self._state}')
 
@@ -310,7 +311,7 @@ def _outcome_destroyed(self, outcome_ref):
         assert outcome_ref is self._outcome_ref
         # This task will not be processed
         self._outcome_ref = None
-        self._loop.create_task(self.requeue())
+        asyncio.run_coroutine_threadsafe(self.requeue(), loop=self._loop)
 
     def _finalise(self):
         self._outcome_ref = None
@@ -421,13 +422,12 @@ def __init__(
             testing_mode=testing_mode
         )
 
-    @async_generator
     async def __aiter__(self):
         # Have to do it this way rather than the more convenient yield from style because
         # python doesn't support it for coroutines.  See:
         # https://stackoverflow.com/questions/47376408/why-cant-i-yield-from-inside-an-async-function
         async for task in self._subscriber:
-            await yield_(task)
+            yield task
 
     async def task_send(self, task, no_reply: bool = False):
         """Send a task to the queue"""
@@ -440,10 +440,9 @@ async def remove_task_subscriber(self, identifier):
         return await self._subscriber.remove_task_subscriber(identifier)
 
     @asynccontextmanager
-    @async_generator
     async def next_task(self, no_ack=False, fail=True, timeout=defaults.TASK_FETCH_TIMEOUT):
         async with self._subscriber.next_task(no_ack=no_ack, fail=fail, timeout=timeout) as task:  # pylint: disable=not-async-context-manager
-            await yield_(task)
+            yield task
 
     async def connect(self):
         await self._subscriber.connect()

src/kiwipy/rmq/threadcomms.py

@@ -351,11 +351,15 @@ def process(self) -> ThreadFuture:
         return aiothreads.aio_future_to_thread(self._task.process())
 
     def requeue(self):
-        self._task.requeue()
+        """Requeue the task.
+
+        This call is blocking and by the time function returns the task will be back in the queue.
+        """
+        self._loop_scheduler.await_(self._task.requeue())
 
     @contextmanager
     def processing(self):
-        with self._loop_scheduler.ctx(self._task.processing()) as outcome:
+        with self._loop_scheduler.async_ctx(self._task.processing()) as outcome:
             yield outcome
 
 

test/rmq/bench/support.py

@@ -32,10 +32,10 @@ def callback(_comm, _task):
             queue.remove_task_subscriber(identifier)
 
 
-async def clear_all_tasks(queue: rmq.RmqThreadTaskQueue):
+def clear_all_tasks(queue: rmq.RmqThreadTaskQueue):
     """Just go through all tasks picking them up so the queue is cleared"""
     for task in queue:
-        async with task.processing() as outcome:
+        with task.processing() as outcome:
             outcome.set_result(True)
 
 

test/rmq/test_coroutine_communicator.py

@@ -1,7 +1,9 @@
 # -*- coding: utf-8 -*-
 import asyncio
+import gc
 
 import pytest
+import shortuuid
 
 import kiwipy
 import kiwipy.rmq
@@ -208,3 +210,36 @@ def test_server_properties(communicator: kiwipy.rmq.RmqCommunicator):
     assert props['product'] == 'RabbitMQ'
     assert 'version' in props
     assert props['platform'].startswith('Erlang')
+
+
+@pytest.mark.asyncio
+async def test_queue_task_forget(communicator: kiwipy.rmq.RmqCommunicator):
+    """
+    Check what happens when we forget to process a task we said we would
+    WARNING: This test may fail when running with a debugger as it relies on the 'outcome'
+    reference count dropping to zero but the debugger may be preventing this.
+    """
+    task_queue_name = f'{__file__}.{shortuuid.uuid()}'
+    task_queue = await communicator.task_queue(task_queue_name)
+
+    outcomes = [await task_queue.task_send(1)]
+
+    # Get the first task and say that we will process it
+    outcome = None
+    async with task_queue.next_task() as task:
+        outcome = task.process()
+
+    with pytest.raises(kiwipy.exceptions.QueueEmpty):
+        async with task_queue.next_task():
+            pass
+
+    # Now let's 'forget' i.e. lose the outcome
+    del outcome
+    gc.collect()
+
+    # Now the task should be back in the queue
+    async with task_queue.next_task() as task:
+        task.process().set_result(10)
+
+    await asyncio.wait(outcomes)
+    assert outcomes[0].result() == 10

test/rmq/test_rmq_thread_communicator.py

@@ -1,6 +1,11 @@
 # -*- coding: utf-8 -*-
+"""
+WARNING: This file should not contain any coroutines code (i.e. no use of async/await) as the intended
+function of RmqThreadCommunicator is that users should be completely shielded from the asyncio code.
+"""
 # pylint: disable=invalid-name, redefined-outer-name
 import concurrent.futures
+import gc
 import pathlib
 import unittest
 
@@ -38,9 +43,7 @@ def thread_communicator():
 @pytest.fixture
 def thread_task_queue(thread_communicator: rmq.RmqThreadCommunicator):
     task_queue_name = f'{__file__}.{shortuuid.uuid()}'
-
     task_queue = thread_communicator.task_queue(task_queue_name)
-
     yield task_queue
 
 
@@ -103,7 +106,7 @@ def on_task(_comm, task):
         self.assertEqual(TASK, tasks[0])
         self.assertEqual(RESULT, result)
 
-    async def test_task_queue_next(self):
+    def test_task_queue_next(self):
         """Test creating a custom task queue"""
         TASK = 'The meaning?'
         RESULT = 42
@@ -114,24 +117,24 @@ async def test_task_queue_next(self):
 
         # Get the task and carry it out
         with task_queue.next_task() as task:
-            await task.process().set_result(RESULT)
+            task.process().set_result(RESULT)
 
         # Now wait for the result
         result = task_future.result(timeout=self.WAIT_TIMEOUT)
         self.assertEqual(RESULT, result)
 
 
-async def test_queue_get_next(thread_task_queue: rmq.RmqThreadTaskQueue):
+def test_queue_get_next(thread_task_queue: rmq.RmqThreadTaskQueue):
     """Test getting the next task from the queue"""
     result = thread_task_queue.task_send('Hello!')
     with thread_task_queue.next_task(timeout=1.) as task:
-        async with task.processing() as outcome:
+        with task.processing() as outcome:
             assert task.body == 'Hello!'
             outcome.set_result('Goodbye')
     assert result.result() == 'Goodbye'
 
 
-async def test_queue_iter(thread_task_queue: rmq.RmqThreadTaskQueue):
+def test_queue_iter(thread_task_queue: rmq.RmqThreadTaskQueue):
     """Test iterating through a task queue"""
     results = []
 
@@ -140,7 +143,7 @@ async def test_queue_iter(thread_task_queue: rmq.RmqThreadTaskQueue):
         results.append(thread_task_queue.task_send(i))
 
     for task in thread_task_queue:
-        async with task.processing() as outcome:
+        with task.processing() as outcome:
             outcome.set_result(task.body * 10)
 
     concurrent.futures.wait(results)
@@ -151,7 +154,7 @@ async def test_queue_iter(thread_task_queue: rmq.RmqThreadTaskQueue):
         assert False, "Shouldn't get here"
 
 
-async def test_queue_iter_not_process(thread_task_queue: rmq.RmqThreadTaskQueue):
+def test_queue_iter_not_process(thread_task_queue: rmq.RmqThreadTaskQueue):
     """Check what happens when we iterate a queue but don't process all tasks"""
     outcomes = []
 
@@ -162,44 +165,45 @@ async def test_queue_iter_not_process(thread_task_queue: rmq.RmqThreadTaskQueue)
     # Now let's see what happens when we have tasks but don't process some of them
     for task in thread_task_queue:
         if task.body < 5:
-            await task.process().set_result(task.body * 10)
+            task.process().set_result(task.body * 10)
 
     concurrent.futures.wait(outcomes[:5])
     for i, outcome in enumerate(outcomes[:5]):
         assert outcome.result() == i * 10
 
     # Now, to through and process the rest
     for task in thread_task_queue:
-        await task.process().set_result(task.body * 10)
+        task.process().set_result(task.body * 10)
 
     concurrent.futures.wait(outcomes)
     for i, outcome in enumerate(outcomes):
         assert outcome.result() == i * 10
 
 
-async def test_queue_task_forget(thread_task_queue: rmq.RmqThreadTaskQueue):
+def test_queue_task_forget(thread_task_queue: rmq.RmqThreadTaskQueue):
     """
     Check what happens when we forget to process a task we said we would
-    WARNING: This test mail fail when running with a debugger as it relies on the 'outcome'
+    WARNING: This test may fail when running with a debugger as it relies on the 'outcome'
     reference count dropping to zero but the debugger may be preventing this.
     """
     outcomes = [thread_task_queue.task_send(1)]
 
     # Get the first task and say that we will process it
     outcome = None
     with thread_task_queue.next_task() as task:
-        outcome = await task.process()
+        outcome = task.process()
 
     with pytest.raises(kiwipy.exceptions.QueueEmpty):
         with thread_task_queue.next_task():
             pass
 
     # Now let's 'forget' i.e. lose the outcome
     del outcome
+    gc.collect()
 
     # Now the task should be back in the queue
     with thread_task_queue.next_task() as task:
-        await task.process().set_result(10)
+        task.process().set_result(10)
 
     concurrent.futures.wait(outcomes)
     assert outcomes[0].result() == 10
@@ -211,14 +215,14 @@ def test_empty_queue(thread_task_queue: rmq.RmqThreadTaskQueue):
             pass
 
 
-async def test_task_processing_exception(thread_task_queue: rmq.RmqThreadTaskQueue):
+def test_task_processing_exception(thread_task_queue: rmq.RmqThreadTaskQueue):
     """Check that if there is an exception processing a task that it is removed from the queue"""
     task_future = thread_task_queue.task_send('Do this')
 
     # The error should still get propageted in the 'worker'
     with pytest.raises(RuntimeError):
         with thread_task_queue.next_task(timeout=WAIT_TIMEOUT) as task:
-            async with task.processing():
+            with task.processing():
                 raise RuntimeError('Cannea do it captain!')
 
     # And the task sender should get a remote exception to inform them of the problem

test/rmq/test_tasks.py

@@ -251,7 +251,7 @@ async def test_queue_iter_not_process(task_queue: rmq.RmqTaskQueue):
 async def test_queue_task_forget(task_queue: rmq.RmqTaskQueue):
     """
     Check what happens when we forget to process a task we said we would
-    WARNING: This test mail fail when running with a debugger as it relies on the 'outcome'
+    WARNING: This test may fail when running with a debugger as it relies on the 'outcome'
     reference count dropping to zero but the debugger may be preventing this.
     """
     outcomes = [await task_queue.task_send(1)]