BusABC.recv: keep calling _recv_internal until it returns None

Even if recv() is called with timeout=0, the caller's intention is
probably for recv() to check all of the messages that have already
arrived at the interface until one of them matches the filters.

This is already the way recv() behaves for interface drivers that take
advantage of hardware or OS-level filtering, but those that use BusABC's
default software-based filtering might return None even if a matching
message has already arrived.

Author: malsyned

can/bus.py

@@ -120,24 +120,18 @@ def recv(self, timeout: Optional[float] = None) -> Optional[Message]:
             # try to get a message
             msg, already_filtered = self._recv_internal(timeout=time_left)
 
+            # propagate timeouts from _recv_internal()
+            if not msg:
+                return None
+
             # return it, if it matches
-            if msg and (already_filtered or self._matches_filters(msg)):
+            if already_filtered or self._matches_filters(msg):
                 LOG.log(self.RECV_LOGGING_LEVEL, "Received: %s", msg)
                 return msg
 
-            # if not, and timeout is None, try indefinitely
-            elif timeout is None:
-                continue
-
-            # try next one only if there still is time, and with
-            # reduced timeout
-            else:
-                time_left = timeout - (time() - start)
-
-                if time_left > 0:
-                    continue
-
-                return None
+            # try again with reduced timeout
+            if timeout is not None:
+                time_left = max(0, timeout - (time() - start))
 
     def _recv_internal(
         self, timeout: Optional[float]

test/test_message_filtering.py

@@ -5,6 +5,7 @@
 """
 
 import unittest
+from unittest.mock import MagicMock, patch
 
 from can import Bus, Message
 
@@ -46,6 +47,57 @@ def test_match_example_message(self):
         self.assertFalse(self.bus._matches_filters(EXAMPLE_MSG))
         self.assertTrue(self.bus._matches_filters(HIGHEST_MSG))
 
+    def test_empty_queue_up_to_match(self):
+        bus2 = Bus(interface="virtual", channel="testy")
+        self.bus.set_filters(MATCH_EXAMPLE)
+        bus2.send(HIGHEST_MSG)
+        bus2.send(EXAMPLE_MSG)
+        actual = self.bus.recv(timeout=0)
+        bus2.shutdown()
+        self.assertTrue(
+            EXAMPLE_MSG.equals(
+                actual, timestamp_delta=None, check_direction=False, check_channel=False
+            )
+        )
+
+
+@patch("can.bus.time")
+class TestMessageFilterRetryTiming(unittest.TestCase):
+    def setUp(self):
+        self.bus = Bus(interface="virtual", channel="testy")
+        self.bus._recv_internal = MagicMock(name="_recv_internal")
+
+    def tearDown(self):
+        self.bus.shutdown()
+
+    def test_propagate_recv_internal_timeout(self, time: MagicMock) -> None:
+        self.bus._recv_internal.side_effect = [
+            (None, False),
+        ]
+        time.side_effect = [0]
+        self.bus.set_filters(MATCH_EXAMPLE)
+        self.assertIsNone(self.bus.recv(timeout=3))
+
+    def test_retry_with_adjusted_timeout(self, time: MagicMock) -> None:
+        self.bus._recv_internal.side_effect = [
+            (HIGHEST_MSG, False),
+            (EXAMPLE_MSG, False),
+        ]
+        time.side_effect = [0, 1]
+        self.bus.set_filters(MATCH_EXAMPLE)
+        self.bus.recv(timeout=3)
+        self.bus._recv_internal.assert_called_with(timeout=2)
+
+    def test_keep_timeout_non_negative(self, time: MagicMock) -> None:
+        self.bus._recv_internal.side_effect = [
+            (HIGHEST_MSG, False),
+            (EXAMPLE_MSG, False),
+        ]
+        time.side_effect = [0, 1]
+        self.bus.set_filters(MATCH_EXAMPLE)
+        self.bus.recv(timeout=0.5)
+        self.bus._recv_internal.assert_called_with(timeout=0)
+
 
 if __name__ == "__main__":
     unittest.main()