Fix behaviour of route_until with timeout and queued messages (#419)

traits_futures/multiprocessing_router.py

@@ -481,10 +481,9 @@ def _route_message(self, *, block=False, timeout=None):
         queue.Empty
             If no message arrives within the given timeout.
         """
-        if block and timeout is not None and timeout <= 0.0:
-            raise queue.Empty
         connection_id, message = self._local_message_queue.get(
-            block=block, timeout=timeout
+            block=block,
+            timeout=None if timeout is None else max(timeout, 0.0),
         )
         try:
             receiver = self._receivers[connection_id]

traits_futures/multithreading_router.py

@@ -425,10 +425,9 @@ def _route_message(self, *, block=False, timeout=None):
         queue.Empty
             If no message arrives within the given timeout.
         """
-        if block and timeout is not None and timeout <= 0.0:
-            raise queue.Empty
         connection_id, message = self._message_queue.get(
-            block=block, timeout=timeout
+            block=block,
+            timeout=None if timeout is None else max(timeout, 0.0),
         )
         try:
             receiver = self._receivers[connection_id]

traits_futures/tests/i_message_router_tests.py

@@ -374,6 +374,32 @@ def test_route_until_second_call_after_timeout(self):
             # an unusable state.
             router.route_until(lambda: True, timeout=SAFETY_TIMEOUT)
 
+    def test_route_until_timeout_with_queued_messages(self):
+        # Even with a timeout of 0.0, route_until shouldn't fail if
+        # there are sufficient messages already queued to satisfy
+        # the condition.
+        messages = list(range(10))
+
+        with self.started_router() as router:
+            sender, receiver = router.pipe()
+            listener = ReceiverListener(receiver=receiver)
+            try:
+                with self.context.worker_pool() as worker_pool:
+                    worker_pool.submit(send_messages, sender, messages)
+
+                # At this point we've closed the worker pool, so all
+                # background jobs have completed and all messages are
+                # already queued. route_until should be able to
+                # process all of them, regardless of timeout.
+                router.route_until(
+                    lambda: len(listener.messages) >= len(messages),
+                    timeout=0.0,
+                )
+            finally:
+                router.close_pipe(receiver)
+
+        self.assertEqual(listener.messages, messages)
+
     def test_event_loop_after_route_until(self):
         # This tests a potentially problematic situation:
         # - route_until processes at least one message manually