svc2svc - fix deadlock error when making multiple requests, update callback function API

Signed-off-by: Lance Drane <dranelt@ornl.gov>

Author: Lance-Drane

examples/4_service_to_service/example_1_service.py

@@ -40,10 +40,27 @@ def pass_text_to_service_2(self, text: str) -> None:
         self.intersect_sdk_call_service(msg_to_send, self.service_2_handler)
 
     @intersect_event(events={'response_event': IntersectEventDefinition(event_type=str)})
-    def service_2_handler(self, msg: str) -> None:
-        """Handles response from service 2 and emits the response as an event for the client."""
+    def service_2_handler(self, _source: str, _operation: str, _has_error: bool, msg: str) -> None:
+        """Handles first response from service 2, emits the response as an event for the client, and sends a hardcoded message to service 2."""
         self.intersect_sdk_emit_event('response_event', f'Received Response from Service 2: {msg}')
 
+        # verify that we can call the service multiple
+        msg_to_send = IntersectDirectMessageParams(
+            destination='example-organization.example-facility.example-system.example-subsystem.service-two',
+            operation='ServiceTwo.test_service',
+            payload='Final Verification',
+        )
+        self.intersect_sdk_call_service(msg_to_send, self.additional_service_handler)
+
+    @intersect_event(events={'response_event': IntersectEventDefinition(event_type=str)})
+    def additional_service_handler(
+        self, _source: str, _operation: str, _has_error: bool, msg: str
+    ) -> None:
+        """Handles second response from service 2 and emits the response as an event for the client."""
+        self.intersect_sdk_emit_event(
+            'response_event', f'Received Second Response from Service 2: {msg}'
+        )
+
 
 if __name__ == '__main__':
     from_config_file = {

examples/4_service_to_service/example_client.py

@@ -23,6 +23,10 @@
 class SampleOrchestrator:
     """Simply contains an event callback for events from Service 1."""
 
+    def __init__(self) -> None:
+        """Straightforward constructor, just initializes global variable which counts events."""
+        self.got_first_event = False
+
     def event_callback(
         self, _source: str, _operation: str, _event_name: str, payload: INTERSECT_JSON_VALUE
     ) -> None:
@@ -35,8 +39,11 @@ def event_callback(
           payload: Value of the response from the Service.
         """
         print(payload)
-        # break out of pubsub loop
-        raise Exception
+        if self.got_first_event:
+            # break out of pubsub loop
+            raise Exception
+        self.got_first_event = True
+        # empty return, don't send any additional messages or modify the events listened to
 
 
 if __name__ == '__main__':

pdm.lock

@@ -54,7 +54,7 @@ docs = [
 [tool.pdm.dev-dependencies]
 lint = [
     "pre-commit>=3.3.1",
-    "ruff>=0.4.2",
+    "ruff>=0.5.7",
     "mypy>=1.10.0",
     "types-paho-mqtt>=1.6.0.20240106",
 ]

pyproject.toml

@@ -19,6 +19,7 @@
 from .config.service import IntersectServiceConfig
 from .config.shared import (
     ControlPlaneConfig,
+    ControlProvider,
     DataStoreConfig,
     DataStoreConfigMap,
     HierarchyConfig,
@@ -63,6 +64,7 @@
     'IntersectServiceConfig',
     'HierarchyConfig',
     'ControlPlaneConfig',
+    'ControlProvider',
     'DataStoreConfig',
     'DataStoreConfigMap',
     '__version__',

src/intersect_sdk/__init__.py

@@ -36,12 +36,14 @@ def create_external_request(
         self,
         request: IntersectDirectMessageParams,
         response_handler: INTERSECT_SERVICE_RESPONSE_CALLBACK_TYPE | None = None,
+        timeout: float = 300.0,
     ) -> UUID:
         """Observed entity (capabilitiy) tells observer (i.e. service) to send an external request.
 
         Params:
           - request: the request we want to send out, encapsulated as an IntersectClientMessageParams object
           - response_handler: optional callback for how we want to handle the response from this request.
+          - timeout: optional value for how long we should wait on the request, in seconds (default: 300 seconds)
 
         Returns:
           - generated RequestID associated with your request

src/intersect_sdk/_internal/interfaces.py

@@ -133,12 +133,14 @@ def intersect_sdk_call_service(
         self,
         request: IntersectDirectMessageParams,
         response_handler: INTERSECT_SERVICE_RESPONSE_CALLBACK_TYPE | None = None,
+        timeout: float = 300,
     ) -> list[UUID]:
         """Create an external request that we'll send to a different Service.
 
         Params:
           - request: the request we want to send out, encapsulated as an IntersectClientMessageParams object
           - response_handler: optional callback for how we want to handle the response from this request.
+          - timeout: optional value for how long we should wait on the request, in seconds (default: 300 seconds)
 
         Returns:
           - list of generated RequestIDs associated with your request. Note that for almost all use cases,
@@ -148,6 +150,6 @@ def intersect_sdk_call_service(
           - pydantic.ValidationError - if the request parameter isn't valid
         """
         return [
-            observer.create_external_request(request, response_handler)
+            observer.create_external_request(request, response_handler, timeout)
             for observer in self.__intersect_sdk_observers__
         ]