add object proxy subscribe event test

Author: VigneshVSV

hololinked/client/factory.py

@@ -3,8 +3,8 @@
 from tornado.httpclient import HTTPClient, HTTPRequest, HTTPResponse
 from tornado.simple_httpclient import HTTPTimeoutError
 
-from ..core.zmq import SyncZMQClient, AsyncZMQClient
 from ..core import Thing, Action
+from ..core.zmq import SyncZMQClient, AsyncZMQClient, EventConsumer, AsyncEventConsumer
 from ..td.interaction_affordance import PropertyAffordance, ActionAffordance, EventAffordance
 from ..serializers import Serializers
 from .abstractions import ConsumedThingAction, ConsumedThingProperty, ConsumedThingEvent
@@ -55,17 +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:
-                pass
-            #     consumed_property_event = ZMQEvent(
+            # 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_zmq_client,
-            #                         async_zmq_client=async_zmq_client,
+            #                         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_property_event)
+            #                     )
+            #     self.add_event(object_proxy, consumed_observable)
         for action in TD.get("actions", []):
             affordance = ActionAffordance.from_TD(action, TD)
             consumed_action = ZMQAction(
@@ -79,13 +90,27 @@ def zmq(self, server_id: str, thing_id: str, protocol: str, **kwargs):
             self.add_action(object_proxy, consumed_action)
         for event in TD.get("events", []):
             affordance = EventAffordance.from_TD(event, TD)
+            zmq_socket_address = affordance.zmq_socket_address
+            # if protocol.lower() == "ipc":
+            #     zmq_socket_address = affordance.zmq_socket_address.replace("inproc://", "ipc://")
+            #     zmq_socket_address = f'{zmq_socket_address}.ipc'
+            sync_event_client = EventConsumer(
+                                    id=f"{TD['id']}|{affordance.name}|sync", 
+                                    event_unique_identifier=affordance.zmq_unique_identifier,
+                                    socket_address=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=zmq_socket_address,
+                                    logger=object_proxy.logger
+                                )
             consumed_event = ZMQEvent(
                                     resource=affordance, 
-                                    sync_zmq_client=sync_zmq_client,
-                                    async_zmq_client=async_zmq_client,
                                     owner_inst=object_proxy,
-                                    invokation_timeout=object_proxy.invokation_timeout,
-                                    execution_timeout=object_proxy.execution_timeout,
+                                    sync_zmq_client=sync_event_client,
+                                    async_zmq_client=async_event_client,
                                 )
             self.add_event(object_proxy, consumed_event)
         for opname, ophandler in zip(['_get_properties', '_set_properties'], [ReadMultipleProperties, WriteMultipleProperties]):
@@ -173,7 +198,7 @@ def add_property(self, client, property: ConsumedThingProperty) -> None:
 
     @classmethod
     def add_event(cls, client, event: ConsumedThingEvent) -> None:
-        setattr(client, event._resource.name, event)
-    
-
-
+        if hasattr(event._resource, "observable") and event._resource.observable:
+            setattr(client, f"{event._resource.name}_change_event", event)
+        else:
+            setattr(client, event._resource.name, event)

hololinked/config.py

@@ -91,7 +91,7 @@ class Configuration:
         # Schema
         'VALIDATE_SCHEMA_ON_CLIENT', 'VALIDATE_SCHEMAS',
         # ZMQ
-        "ZMQ_ASYNC_CONTEXT", "ZMQ_SYNC_CONTEXT",
+        "ZMQ_CONTEXT",
         # make debugging easier
         "DEBUG",    
         # serializers
@@ -115,8 +115,7 @@ def load_variables(self, use_environment : bool = False):
         self.TRACE_MALLOC = False
         self.VALIDATE_SCHEMA_ON_CLIENT = False
         self.VALIDATE_SCHEMAS = True
-        self.ZMQ_ASYNC_CONTEXT = zmq.asyncio.Context()
-        self.ZMQ_SYNC_CONTEXT = zmq.Context()
+        self.ZMQ_CONTEXT = zmq.asyncio.Context()
         self.DEBUG = False
         self.ALLOW_PICKLE = False
         self.ALLOW_UNKNOWN_SERIALIZATION = False
@@ -152,15 +151,13 @@ def asdict(self):
         "returns this config as a regular dictionary"
         return {item: getattr(self, item) for item in self.__slots__}
 
-    def zmq_context(self, asynch: bool = True) -> zmq.Context | zmq.asyncio.Context:
+    def zmq_context(self) -> zmq.asyncio.Context:
         """
-        Returns a global ZMQ context based on the asynch flag. 
-        Intended to share the same context across an application.
-        """
-        if asynch:
-            return self.ZMQ_ASYNC_CONTEXT
-        return self.ZMQ_SYNC_CONTEXT
-    
+        Returns a global ZMQ async context. Use socket_class argument to retrieve 
+        a synchronous socket if necessary.
+        """        
+        return self.ZMQ_CONTEXT
+
     def set_default_server_execution_context(self, 
         invokation_timeout: typing.Optional[int] = None,
         execution_timeout: typing.Optional[int] = None,

hololinked/core/zmq/brokers.py

@@ -52,9 +52,14 @@ def __del__(self) -> None:
 
 
     @classmethod
-    def get_socket(cls, *, id: str, node_type: str, context: zmq.asyncio.Context | zmq.Context, 
-                    transport: ZMQ_TRANSPORTS = ZMQ_TRANSPORTS.IPC, socket_type: zmq.SocketType = zmq.ROUTER, 
-                    **kwargs) -> typing.Tuple[zmq.Socket, str]:
+    def get_socket(cls, *, 
+                id: str, 
+                node_type: str, 
+                context: zmq.asyncio.Context | zmq.Context, 
+                transport: ZMQ_TRANSPORTS = ZMQ_TRANSPORTS.IPC, 
+                socket_type: zmq.SocketType = zmq.ROUTER, 
+                **kwargs
+            ) -> typing.Tuple[zmq.Socket, str]:
         """
         Create a socket with certain specifications. Supported ZeroMQ transports are TCP, IPC & INPROC. 
         For IPC sockets, a file is created under TEMP_DIR of global configuration.
@@ -164,7 +169,7 @@ def create_socket(self, *, id: str, node_type: str = 'server', context: zmq.asyn
         """
         if context and not isinstance(context, zmq.asyncio.Context):
             raise TypeError("async ZMQ message broker accepts only async ZMQ context. supplied type {}".format(type(context)))
-        self.context = context or global_config.zmq_context(asynch=True)
+        self.context = context or global_config.zmq_context()
         self.socket, self.socket_address = BaseZMQ.get_socket(id=id, node_type=node_type, context=self.context, 
                                                 transport=transport, socket_type=socket_type, **kwargs)
         self.logger.info("created socket {} with address {} & identity {} and {}".format(get_socket_type_name(socket_type), 
@@ -183,17 +188,9 @@ def create_socket(self, *, id: str, node_type: str = 'server', context: zmq.Cont
         Overloads ``create_socket()`` to create, bind/connect a synchronous socket. A synchronous context is created 
         if none is supplied. 
         """
-        socket_class = None
-        if context: 
-            if not isinstance(context, zmq.Context):
-                raise TypeError("sync ZMQ message broker accepts only sync ZMQ context. supplied type {}".format(type(context)))
-            if isinstance(context, zmq.asyncio.Context):
-                # create sync socket when async context is supplied for sync brokers.
-                # especially useful for INPROC sync client where teh context needs to be shared and the server is async. 
-                socket_class = zmq.Socket
-        self.context = context or global_config.zmq_context(asynch=False)
+        self.context = context or global_config.zmq_context()
         self.socket, self.socket_address = BaseZMQ.get_socket(id=id, node_type=node_type, context=self.context, 
-                                                transport=transport, socket_type=socket_type, socket_class=socket_class, 
+                                                transport=transport, socket_type=socket_type, socket_class=zmq.Socket, 
                                                 **kwargs)
         self.logger.info("created socket {} with address {} & identity {} and {}".format(get_socket_type_name(socket_type), 
                                                         self.socket_address, id, "bound" if node_type == 'server' else "connected"))
@@ -593,7 +590,7 @@ class ZMQServerPool(BaseZMQServer):
     """
 
     def __init__(self, *, ids: typing.List[str] | None = None, **kwargs) -> None:
-        self.context = global_config.zmq_context(asynch=True)
+        self.context = global_config.zmq_context()
         self.poller = zmq.asyncio.Poller()
         self.pool = dict() # type: typing.Dict[str, AsyncZMQServer]
         if ids:
@@ -791,8 +788,13 @@ def exit(self) -> None:
                 self.poller.unregister(self._monitor_socket)
                 # print("poller exception did not occur 3")
         except Exception as ex:
-            self.logger.warning(f"unable to deregister from poller - {str(ex)}")
-
+            # raises a weird key error for some reason
+            # unable to deregister from poller - <zmq.asyncio.Socket(zmq.PAIR) at 0x1c9e5028830> - KeyError
+            # unable to deregister from poller - <zmq.asyncio.Socket(zmq.PAIR) at 0x1c9e502a350> - KeyError
+            # unable to deregister from poller - <zmq.asyncio.Socket(zmq.PAIR) at 0x1c9e5080750> - KeyError
+            # unable to deregister from poller - <zmq.asyncio.Socket(zmq.PAIR) at 0x1c9e5082430> - KeyError
+            # self.logger.warning(f"unable to deregister from poller - {str(ex)} - {type(ex).__name__}")
+            pass
         try:
             if self._monitor_socket is not None:
                 self._monitor_socket.close(0)
@@ -1341,7 +1343,7 @@ def __init__(self,
         if len(client_ids) != len(server_ids):
             raise ValueError("client_ids and server_ids must have same length")
         # this class does not call create_socket method
-        self.context = context or global_config.zmq_context(asynch=True)
+        self.context = context or global_config.zmq_context()
         self.pool = dict() # type: typing.Dict[str, AsyncZMQClient]
         self.poller = zmq.asyncio.Poller()
         for client_id, server_id in zip(client_ids, server_ids):
@@ -1941,11 +1943,13 @@ def __init__(self,
                 **kwargs
             ) -> None:
         if isinstance(self, BaseSyncZMQ):
-            self.context = context or global_config.zmq_context(asynch=False)
+            self.context = context or global_config.zmq_context()
             self.poller = zmq.Poller()
+            socket_class = zmq.Socket
         elif isinstance(self, BaseAsyncZMQ):
-            self.context = context or global_config.zmq_context(asynch=True)
+            self.context = context or global_config.zmq_context()
             self.poller = zmq.asyncio.Poller()
+            socket_class = zmq.asyncio.Socket
         else:
             raise TypeError("BaseEventConsumer must be subclassed by either BaseSyncZMQ or BaseAsyncZMQ")
         super().__init__(id=id, server_id=kwargs.get('server_id', None), **kwargs)
@@ -1960,13 +1964,13 @@ def __init__(self,
                     )
         self.event_unique_identifier = bytes(event_unique_identifier, encoding='utf-8')
         short_uuid = uuid4().hex[:8]
-        self.interruptor = self.context.socket(zmq.PAIR)
+        self.interruptor = self.context.socket(zmq.PAIR, socket_class=socket_class)
         self.interruptor.setsockopt_string(zmq.IDENTITY, f'interrupting-server-{short_uuid}')
-        self.interrupting_peer = self.context.socket(zmq.PAIR)
+        self.interrupting_peer = self.context.socket(zmq.PAIR, socket_class=socket_class)
         self.interrupting_peer.setsockopt_string(zmq.IDENTITY, f'interrupting-client-{short_uuid}')
         self.interruptor.bind(f'inproc://{self.id}-{short_uuid}/interruption')    
         self.interrupting_peer.connect(f'inproc://{self.id}-{short_uuid}/interruption')
-
+        
 
     def subscribe(self) -> None:
         self.socket.setsockopt(zmq.SUBSCRIBE, self.event_unique_identifier) 
@@ -1993,8 +1997,10 @@ def exit(self):
             self.poller.unregister(self.socket)
             self.poller.unregister(self.interruptor)
         except Exception as E:
-            self.logger.warning("could not properly terminate socket or attempted to terminate an already terminated socket of event consuming socket at address '{}'. Exception message: {}".format(
-                self.socket_address, str(E)))  
+            # self.logger.warning("could not properly terminate socket or attempted to terminate an already terminated socket of event consuming socket at address '{}'. Exception message: {}".format(
+            #     self.socket_address, str(E)))  
+            # above line prints too many warnings 
+            pass
         try:
             self.socket.close(0)
             self.interruptor.close(0)

hololinked/core/zmq/rpc_server.py

@@ -113,7 +113,7 @@ def __init__(self, *,
             kwargs['logger'] = self.logger
         # contexts and poller
         self._run = False # flag to stop all the
-        self.context = context or global_config.zmq_context(asynch=True)
+        self.context = context or global_config.zmq_context()
 
         self.req_rep_server = AsyncZMQServer(
                                 id=self.id, 
@@ -809,7 +809,7 @@ def prepare_rpc_server(
     # dont specify http server as a kwarg, as the other method run_with_http_server has to be used
     if context is not None and not isinstance(context, zmq.asyncio.Context):
         raise TypeError("context must be an instance of zmq.asyncio.Context")
-    context = context or global_config.zmq_context(asynch=True)
+    context = context or global_config.zmq_context()
 
     if transports == 'INPROC' or transports == ZMQ_TRANSPORTS.INPROC:
         RPCServer(

hololinked/server/http/handlers.py

@@ -516,7 +516,7 @@ async def handle_datastream(self) -> None:
                 id=f"{self.resource.name}|HTTPEvent|{uuid.uuid4().hex[:8]}",
                 event_unique_identifier=self.resource.zmq_unique_identifier,
                 socket_address=self.resource.zmq_socket_address,
-                context=global_config.zmq_context(asynch=True),
+                context=global_config.zmq_context(),
                 logger=self.logger,
             )
             event_consumer.subscribe()

hololinked/td/interaction_affordance.py

@@ -1,5 +1,6 @@
-from enum import Enum
 import typing
+import socket
+from enum import Enum
 from typing import ClassVar, Optional
 from pydantic import ConfigDict
 
@@ -373,6 +374,11 @@ def build_non_compliant_metadata(self):
         bound_event_dispatcher = getattr(self.owner, self.objekt.name, None) # type: EventDispatcher
         if not isinstance(bound_event_dispatcher, EventDispatcher) or not bound_event_dispatcher.publisher:
             return
-        self.zmq_socket_address = bound_event_dispatcher.publisher.socket_address
+        socket_address = bound_event_dispatcher.publisher.socket_address
+        if socket_address.startswith("tcp://"):
+            socket_address = socket_address.replace("tcp://*", f"tcp://{socket.gethostname()}")
+            socket_address = socket_address.replace("tcp://localhost", f"tcp://{socket.gethostname()}")
+            socket_address = socket_address.replace("tcp://0.0.0.0", f"tcp://{socket.gethostname()}")
+        self.zmq_socket_address = socket_address
         self.zmq_unique_identifier = bound_event_dispatcher._unique_identifier
 

tests/test_09_rpc_broker.py

@@ -2,7 +2,6 @@
 import threading
 import typing
 import unittest
-import zmq.asyncio
 import jsonschema
 import logging
 import random
@@ -18,6 +17,7 @@
 from hololinked.server.zmq import ZMQServer
 from hololinked.td.utils import get_zmq_unique_identifier_from_event_affordance
 from hololinked.utils import get_all_sub_things_recusively, get_current_async_loop
+from hololinked.config import global_config
 from hololinked.td import ActionAffordance, PropertyAffordance, EventAffordance
 from hololinked.client.zmq.consumed_interactions import ZMQAction, ZMQProperty, ZMQEvent
 
@@ -189,7 +189,7 @@ def startServer(self):
 
     @classmethod
     def setUpClass(self):
-        self.context = zmq.asyncio.Context()
+        self.context = global_config.zmq_context()
         super().setUpClass()
         print(f"test ZMQ RPC Server {self.__name__}")
 
@@ -1022,15 +1022,13 @@ def startServer(self):
                                 server_id=self.thing_id, 
                                 logger=self.logger,
                                 handshake=False,
-                                context=self.thing.rpc_server.context,
                                 transport='INPROC'
                             )
         self.async_client = AsyncZMQClient(
                                 id=self.client_id+'async',
                                 server_id=self.thing_id, 
                                 logger=self.logger,
                                 handshake=False,
-                                context=self.thing.rpc_server.context,
                                 transport='INPROC'
                             )
         time.sleep(2)
@@ -1055,7 +1053,7 @@ def load_tests(loader, tests, pattern):
     suite.addTest(unittest.TestLoader().loadTestsFromTestCase(TestRPCServer))
     suite.addTest(unittest.TestLoader().loadTestsFromTestCase(TestExposedActions))
     suite.addTest(unittest.TestLoader().loadTestsFromTestCase(TestExposedProperties))
-    # suite.addTest(unittest.TestLoader().loadTestsFromTestCase(TestExposedEvents))
+    suite.addTest(unittest.TestLoader().loadTestsFromTestCase(TestExposedEvents))
     suite.addTest(unittest.TestLoader().loadTestsFromTestCase(TestThingRunRPCServer))
     return suite