Upgrade from deprecated trio.Queue

Author: Zac-HD

newsfragments/49.bugfix.rst

@@ -0,0 +1 @@
+Replace deprecated ``trio.Queue`` with new channels, requiring Trio 0.9 or later.

setup.py

@@ -53,7 +53,7 @@
 """
 
 install_requires = [
-    "trio >= 0.5.0",
+    "trio >= 0.9.0",
     "async_generator >= 1.6",
     "outcome",
 ]

trio_asyncio/async_.py

@@ -16,7 +16,7 @@ class TrioEventLoop(BaseTrioEventLoop):
 
     def _queue_handle(self, handle):
         self._check_closed()
-        self._q.put_nowait(handle)
+        self._q_send.send_nowait(handle)
         return handle
 
     def default_exception_handler(self, context):

trio_asyncio/base.py

@@ -150,7 +150,7 @@ def __init__(self, queue_len=None):
             queue_len = 10000
 
         # Processing queue
-        self._q = trio.Queue(queue_len)
+        self._q_send, self._q_recv = trio.open_memory_channel(queue_len)
 
         # which files to close?
         self._close_files = set()
@@ -406,7 +406,7 @@ def call_soon_threadsafe(self, callback, *args, context=None):
         self._check_callback(callback, 'call_soon_threadsafe')
         self._check_closed()
         h = Handle(callback, args, self, context=context, is_sync=True)
-        self._token.run_sync_soon(self._q.put_nowait, h)
+        self._token.run_sync_soon(self._q_send.send_nowait, h)
 
     # drop all timers
 
@@ -512,7 +512,7 @@ async def synchronize(self):
     def _handle_sig(self, sig, _):
         """Helper to safely enqueue a signal handler."""
         h = self._signal_handlers[sig]
-        self._token.run_sync_soon(self._q.put_nowait, h)
+        self._token.run_sync_soon(self._q_send.send_nowait, h)
 
     def add_signal_handler(self, sig, callback, *args):
         """asyncio's method to add a signal handler.
@@ -741,10 +741,10 @@ async def _main_loop_one(self, no_wait=False):
 
         if obj is None:
             if no_wait:
-                obj = self._q.get_nowait()
+                obj = self._q_recv.receive_nowait()
             else:
                 with trio.move_on_after(timeout):
-                    obj = await self._q.get()
+                    obj = await self._q_recv.receive()
                 if obj is None:
                     # Timeout reached. Presumably now a timer is ready,
                     # so restart from the beginning.

trio_asyncio/sync.py

@@ -68,7 +68,7 @@ def _queue_handle(self, handle):
 
         def put(self, handle):
             self._some_deferred -= 1
-            self._q.put_nowait(handle)
+            self._q_send.send_nowait(handle)
 
         # If we don't have a token, the main loop is not yet running
         # thus we can't have a race condition.
@@ -81,7 +81,7 @@ def put(self, handle):
             self._some_deferred += 1
             self._token.run_sync_soon(put, self, handle)
         else:
-            self._q.put_nowait(handle)
+            self._q_send.send_nowait(handle)
         return handle
 
     def run_forever(self):