test observable properties with synchronous ZMQ client, async already works in test with HTTP protocol binding

Author: VigneshVSV

hololinked/client/factory.py

@@ -55,28 +55,28 @@ def zmq(self, server_id: str, thing_id: str, protocol: str, **kwargs):
                                     execution_timeout=object_proxy.execution_timeout,
                                 )
             self.add_property(object_proxy, consumed_property)
-            # if hasattr(affordance, "observable") and affordance.observable:
-            #     sync_event_client = EventConsumer(
-            #                         id=f"{TD['id']}|{affordance.name}|sync", 
-            #                         event_unique_identifier=affordance.zmq_unique_identifier,
-            #                         socket_address=affordance.zmq_socket_address,
-            #                         logger=object_proxy.logger
-            #                     )
-            #     async_event_client = AsyncEventConsumer(
-            #                         id=f"{TD['id']}|{affordance.name}|async", 
-            #                         event_unique_identifier=affordance.zmq_unique_identifier,
-            #                         socket_address=affordance.zmq_socket_address,
-            #                         logger=object_proxy.logger
-            #                     )
-            #     consumed_observable = ZMQEvent(
-            #                         resource=affordance,
-            #                         sync_zmq_client=sync_event_client,
-            #                         async_zmq_client=async_event_client,
-            #                         owner_inst=object_proxy,
-            #                         invokation_timeout=object_proxy.invokation_timeout,
-            #                         execution_timeout=object_proxy.execution_timeout,
-            #                     )
-            #     self.add_event(object_proxy, consumed_observable)
+            if hasattr(affordance, "observable") and affordance.observable:
+                sync_event_client = EventConsumer(
+                                    id=f"{TD['id']}|{affordance.name}|sync", 
+                                    event_unique_identifier=affordance.zmq_unique_identifier,
+                                    socket_address=affordance.zmq_socket_address,
+                                    logger=object_proxy.logger
+                                )
+                async_event_client = AsyncEventConsumer(
+                                    id=f"{TD['id']}|{affordance.name}|async", 
+                                    event_unique_identifier=affordance.zmq_unique_identifier,
+                                    socket_address=affordance.zmq_socket_address,
+                                    logger=object_proxy.logger
+                                )
+                consumed_observable = ZMQEvent(
+                                    resource=affordance,
+                                    sync_zmq_client=sync_event_client,
+                                    async_zmq_client=async_event_client,
+                                    owner_inst=object_proxy,
+                                    invokation_timeout=object_proxy.invokation_timeout,
+                                    execution_timeout=object_proxy.execution_timeout,
+                                )
+                self.add_event(object_proxy, consumed_observable)
         for action in TD.get("actions", []):
             affordance = ActionAffordance.from_TD(action, TD)
             consumed_action = ZMQAction(

hololinked/td/interaction_affordance.py

@@ -243,6 +243,8 @@ class PropertyAffordance(DataSchema, InteractionAffordance):
     """
     # [Supported Fields]() <br>
     observable: Optional[bool] = None
+    zmq_unique_identifier: Optional[str] = None
+    zmq_socket_address: Optional[str] = None
 
     def __init__(self):
         super().__init__()
@@ -256,7 +258,13 @@ def build(self) -> None:
         self.ds_build_from_property(property)
         if property.observable:
             self.observable = property.observable
-    
+            observable_event_descriptor = property._observable_event_descriptor
+            event_dispatcher = getattr(self.owner, observable_event_descriptor.name, None) # type: EventDispatcher
+            if event_dispatcher:
+                self.zmq_unique_identifier = event_dispatcher._unique_identifier
+                self.zmq_socket_address = event_dispatcher.publisher.socket_address
+                
+
     @classmethod
     def generate(cls, property, owner = None):
         assert isinstance(property, Property), f"property must be instance of Property, given type {type(property)}"

tests/test_12_protocols_zmq_object_proxy.py

@@ -168,82 +168,73 @@ def cb(value):
         self.assertEqual(len(results), 100)
         thing.unsubscribe_event("test_event")
 
-    # def test_05_observe_properties(self):
-    #     thing = ClientFactory.zmq("test-thing", "test-thing", "IPC")
-    #     self.assertIsInstance(thing, ObjectProxy)  
-
-    #     # First check if an attribute is set on the object proxy
-    #     self.assertIsNotNone(thing, "observable_list_prop_change_event")
-    #     self.assertIsNotNone(thing, "observable_readonly_prop_change_event")
-
-    #     event_count = 0
-    #     def callback(value):
-    #         nonlocal event_count
-    #         event_count += 1
-    #         self.assertIsInstance(value, list)
-
-    #     thing.observe_property("observable_list_prop", callback)
-        # thing.write_property("observable_list_prop", fake.pylist(25, value_types=[int, float, str, bool]))
-        # thing.write_property("observable_list_prop", fake.pylist(25, value_types=[int, float, str, bool]))
-        # self.assertEqual(event_count, 2)
-        # self.assertEqual(thing.observable_list_prop, fake.last)
-
-        #  # req 1 - observable events come due to writing a property
-        # propective_values = [
-        #     [1, 2, 3, 4, 5],
-        #     ['a', 'b', 'c', 'd', 'e'],
-        #     [1, 'a', 2, 'b', 3]
-        # ]
-        # result = []
-        # attempt = 0
-        # def cb(value):
-        #     nonlocal attempt, result
-        #     self.assertEqual(value, propective_values[attempt])
-        #     result.append(value)
-        #     attempt += 1
-
-        # self.thing_client.subscribe_event('observable_list_prop_change_event', cb)
-        # time.sleep(3)
-        # # Calm down for event publisher to connect fully as there is no handshake for events
-        # for value in propective_values:
-        #     self.thing_client.observable_list_prop = value
-
-        # for i in range(20):
-        #     if attempt == len(propective_values):
-        #         break
-        #     # wait for the callback to be called
-        #     time.sleep(0.1)
-        # self.thing_client.unsubscribe_event('observable_list_prop_change_event')
-
-        # self.assertEqual(result, propective_values)
+
+    def test_06_observe_properties(self):
+        thing = ClientFactory.zmq("test-thing", "test-thing", "IPC")
+        self.assertIsInstance(thing, ObjectProxy)  
+
+        # First check if an attribute is set on the object proxy
+        self.assertIsNotNone(thing, "observable_list_prop_change_event")
+        self.assertIsNotNone(thing, "observable_readonly_prop_change_event")
+
+        # req 1 - observable events come due to writing a property
+        propective_values = [
+            [1, 2, 3, 4, 5],
+            ['a', 'b', 'c', 'd', 'e'],
+            [1, 'a', 2, 'b', 3]
+        ]
+        result = []
+        attempt = 0
+        def cb(value):
+            nonlocal attempt, result
+            self.assertEqual(value, propective_values[attempt])
+            result.append(value)
+            attempt += 1
+
+        thing.observe_property('observable_list_prop', cb)
+        time.sleep(3)
+        # Calm down for event publisher to connect fully as there is no handshake for events
+        for value in propective_values:
+            thing.observable_list_prop = value
+
+        for i in range(20):
+            if attempt == len(propective_values):
+                break
+            # wait for the callback to be called
+            time.sleep(0.1)
+        thing.unobserve_property('observable_list_prop')
+
+        for res in result:
+            self.assertIn(res, propective_values)
 
         # # req 2 - observable events come due to reading a property
-        # propective_values = [1, 2, 3, 4, 5]
-        # result = []
-        # attempt = 0
-        # def cb(value):
-        #     nonlocal attempt, result
-        #     self.assertEqual(value, propective_values[attempt])
-        #     result.append(value)
-        #     attempt += 1
-
-        # self.thing_client.subscribe_event('observable_readonly_prop_change_event', cb)
-        # time.sleep(3)
-        # # Calm down for event publisher to connect fully as there is no handshake for events
-        # for _ in propective_values:
-        #     self.thing_client.observable_readonly_prop
-
-        # for i in range(20):
-        #     if attempt == len(propective_values):
-        #         break
-        #     # wait for the callback to be called
-        #     time.sleep(0.1)
-
-        # self.thing_client.unsubscribe_event('observable_readonly_prop_change_event')
-        # self.assertEqual(result, propective_values)
-
-
-    def test_06_stop(self):
+        propective_values = [1, 2, 3, 4, 5]
+        result = []
+        attempt = 0
+        def cb(value):
+            nonlocal attempt, result
+            self.assertEqual(value, propective_values[attempt])
+            result.append(value)
+            attempt += 1
+
+        thing.observe_property('observable_readonly_prop', cb)
+        time.sleep(3)
+        # Calm down for event publisher to connect fully as there is no handshake for events
+        for _ in propective_values:
+            thing.observable_readonly_prop # read property through do notation access
+
+        for i in range(20):
+            if attempt == len(propective_values):
+                break
+            # wait for the callback to be called
+            time.sleep(0.1)
+
+        thing.unobserve_property('observable_readonly_prop')
+        for res in result:
+            self.assertIn(res, propective_values)
+      
+
+    def test_07_stop(self):
         """Test the stop of the zmq object proxy client"""
         self.thing.rpc_server.stop()       
 
@@ -304,12 +295,10 @@ async def test_04_stop(self):
 
 
 
-
-
 def load_tests(loader, tests, pattern):
     suite = unittest.TestSuite()
     suite.addTest(unittest.TestLoader().loadTestsFromTestCase(TestZMQObjectProxyClient))
-    # suite.addTest(unittest.TestLoader().loadTestsFromTestCase(TestZMQObjectProxyClientAsync))
+    suite.addTest(unittest.TestLoader().loadTestsFromTestCase(TestZMQObjectProxyClientAsync))
     return suite
 
 if __name__ == '__main__':