feat: add has_data_available() method to BaseConnection

Author: AndyEveritt

README.md

@@ -13,3 +13,19 @@ This package contains a `setup.py` so it can be installed with `python3 setup.py
 
 ## Usage
 See included `examples/` folder for various use cases.
+
+For patch subscriptions, `BaseConnection.has_data_available()` can be used to poll a
+subscription socket without blocking on `receive_json()` or `get_object_model_patch()`.
+This is useful when object model updates arrive less frequently than another data source.
+
+```python
+from dsf.connections import SubscribeConnection, SubscriptionMode
+
+subscription = SubscribeConnection(SubscriptionMode.PATCH)
+subscription.connect()
+object_model = subscription.get_object_model()
+
+while True:
+	if subscription.has_data_available():
+		object_model.update_from_json(subscription.get_object_model_patch())
+```

docs/source/index.rst

@@ -13,6 +13,24 @@ Welcome to dsf's documentation!
 API
 ===
 
+Subscription Polling
+====================
+
+Patch subscriptions can be polled without blocking by checking
+``BaseConnection.has_data_available()`` before reading the next patch.
+
+.. code-block:: python
+
+   from dsf.connections import SubscribeConnection, SubscriptionMode
+
+   subscription = SubscribeConnection(SubscriptionMode.PATCH)
+   subscription.connect()
+   object_model = subscription.get_object_model()
+
+   while True:
+      if subscription.has_data_available():
+         object_model.update_from_json(subscription.get_object_model_patch())
+
 .. automodule:: dsf
    :members:
 .. automodule:: dsf.connections

src/dsf/connections/base_connection.py

@@ -1,4 +1,5 @@
 import json
+import select
 import socket
 import time
 from typing import Optional
@@ -84,6 +85,17 @@ def receive_response(self):
         json_string = self.receive_json()
         return responses.decode_response(json.loads(json_string))
 
+    def has_data_available(self) -> bool:
+        """Return whether buffered or socket data is available to be read without blocking."""
+        if self.get_json_object_end_index(self.input) > 1:
+            return True
+
+        if not self.socket:
+            return False
+
+        readable, _, _ = select.select([self.socket], [], [], 0)
+        return bool(readable)
+
     def receive_json(self) -> str:
         """Receive the JSON response from the server"""
         if not self.socket:

tests/test_base_connection.py

@@ -0,0 +1,34 @@
+import unittest
+from unittest.mock import patch
+
+from src.dsf.connections.base_connection import BaseConnection
+
+
+class TestBaseConnection(unittest.TestCase):
+    def test_has_data_available_returns_true_for_complete_buffered_json(self):
+        connection = BaseConnection()
+        connection.input = '{"key":1}'
+
+        self.assertTrue(connection.has_data_available())
+
+    def test_has_data_available_returns_false_for_partial_buffer_without_socket_data(self):
+        connection = BaseConnection()
+        connection.input = '{"key"'
+        connection.socket = object()
+
+        with patch('src.dsf.connections.base_connection.select.select', return_value=([], [], [])):
+            self.assertFalse(connection.has_data_available())
+
+    def test_has_data_available_returns_true_when_socket_is_readable(self):
+        connection = BaseConnection()
+        connection.socket = object()
+
+        with patch(
+            'src.dsf.connections.base_connection.select.select',
+            return_value=([connection.socket], [], []),
+        ):
+            self.assertTrue(connection.has_data_available())
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_subscribe_object_model.py

@@ -17,6 +17,15 @@
 class TestSubscribeObjectModel(unittest.TestCase):
     """Test suite for the object model subscription example."""
 
+    @staticmethod
+    def _wait_for_data_available(subscribe_connection: SubscribeConnection, timeout: float = 1.0) -> bool:
+        deadline = time.time() + timeout
+        while time.time() < deadline:
+            if subscribe_connection.has_data_available():
+                return True
+            time.sleep(0.01)
+        return False
+
     def setUp(self):
         """Set up test environment before each test."""
         self.tmp_dir = tempfile.TemporaryDirectory()
@@ -131,6 +140,32 @@ def test_subscribe_object_model(self):
         # Verify the test completed successfully
         self.assertTrue(self.dcs_passed.is_set(), "The mock DCS did not complete successfully")
 
+    def test_has_data_available_during_subscription_flow(self):
+        """Test that has_data_available reflects queued model and patch updates."""
+        subscribe_connection = SubscribeConnection(SubscriptionMode.PATCH)
+        subscribe_connection.connect(self.mock_dcs_socket_file)
+
+        self.assertTrue(
+            self._wait_for_data_available(subscribe_connection),
+            "Expected the initial object model to be readable",
+        )
+
+        subscribe_connection.get_object_model()
+
+        self.assertTrue(
+            self._wait_for_data_available(subscribe_connection),
+            "Expected the next object model patch to be readable after acknowledge",
+        )
+
+        subscribe_connection.get_object_model_patch()
+
+        self.assertFalse(subscribe_connection.has_data_available())
+
+        subscribe_connection.close()
+
+        self.server_thread.join(timeout=5)
+        self.assertTrue(self.dcs_passed.is_set(), "The mock DCS did not complete successfully")
+
 
 if __name__ == "__main__":
     unittest.main()