init project

Author: grongierisc

demo/python/generator/bo.py

@@ -0,0 +1,24 @@
+from typing import Any
+from iop import BusinessOperation
+
+from msg import MyGeneratorResponse, MyGenerator  # Import the generator message class
+
+class MyGeneratorOperation(BusinessOperation):
+
+    def on_other_request(self, request: MyGeneratorResponse) -> Any:
+        self.log_info("Received other request")
+        return MyGeneratorResponse(my_other_string="Hello, World!")
+
+    def on_private_session_started(self, request: MyGenerator) -> Any:
+        self.log_info("Private session started")
+        return self.my_generator(request)
+
+    def my_generator(self, request: Any) -> Any:
+        self.log_info(f"Processing request: {request}")
+        # Simulate some processing and yield responses
+        for i in range(5):
+            response = f"Response {i} from MyGeneratorOperation"
+            self.log_info(response)
+            yield MyGeneratorResponse(my_other_string=response)
+
+    

demo/python/generator/bp.py

@@ -0,0 +1,20 @@
+from typing import Any
+from iop import BusinessProcess
+
+from msg import MyGenerator, MyGeneratorResponse 
+
+class MyGeneratorProcess(BusinessProcess):
+
+    def on_request(self, request: Any) -> Any:
+        rsp = self.send_request_sync(
+            target="User.MyGeneratorOperation",
+            request=MyGeneratorResponse(my_other_string="Hello, World!"),
+            timeout=10,
+            description="My generator request")
+        gen = self.send_generator_request(
+            target="User.MyGeneratorOperation",
+            request=MyGenerator(my_string="Hello, World!"),
+            timeout=10,
+            description="My generator request")
+        for response in gen:
+            self.log_info(f"Received response: {response}")

demo/python/generator/msg.py

@@ -0,0 +1,12 @@
+from iop import GeneratorMessage,Message
+from dataclasses import dataclass
+
+@dataclass
+class MyGenerator(GeneratorMessage):
+    """Base message to initialize generator function"""
+    my_string: str
+
+@dataclass
+class MyGeneratorResponse(Message):
+    """Base message to return generator function response"""
+    my_other_string: str

demo/python/generator/settings.py

@@ -0,0 +1,7 @@
+from bo import MyGeneratorOperation
+from bp import MyGeneratorProcess
+
+CLASSES = {
+    "MyGeneratorOperation": MyGeneratorOperation,
+    "MyGeneratorProcess": MyGeneratorProcess,
+}

src/iop/__init__.py

@@ -3,7 +3,7 @@
 from iop._business_service import _BusinessService
 from iop._director import _Director
 from iop._inbound_adapter import _InboundAdapter
-from iop._message import _Message, _PickleMessage, _PydanticMessage, _PydanticPickleMessage
+from iop._message import _Message, _PickleMessage, _PydanticMessage, _PydanticPickleMessage, _GeneratorMessage
 from iop._outbound_adapter import _OutboundAdapter
 from iop._private_session_duplex import _PrivateSessionDuplex
 from iop._private_session_process import _PrivateSessionProcess
@@ -22,4 +22,5 @@ class Message(_Message): pass
 class PickleMessage(_PickleMessage): pass
 class PydanticMessage(_PydanticMessage): pass
 class PydanticPickleMessage(_PydanticPickleMessage): pass
+class GeneratorMessage(_GeneratorMessage): pass
 class Director(_Director): pass

src/iop/_business_host.py

@@ -7,6 +7,7 @@
 from ._decorators import input_serializer_param, output_deserializer
 from ._dispatch import dispatch_serializer, dispatch_deserializer
 from ._async_request import AsyncRequest
+from ._generator_request import _GeneratorRequest
 
 class _BusinessHost(_Common):
     """Base class for business components that defines common methods.
@@ -67,6 +68,22 @@ async def send_request_async_ng(self, target: str, request: Union[Message, Any],
             Response from target component
         """
         return await AsyncRequest(target, request, timeout, description, self)
+    
+    @input_serializer_param(1, 'request')
+    def send_generator_request(self, target: str, request: Union[Message, Any], 
+                              timeout: int = -1, description: Optional[str] = None) -> _GeneratorRequest:
+        """Send message as a generator request to target component.
+        Args:
+            target: Name of target component
+            request: Message to send
+            timeout: Timeout in seconds, -1 means wait forever
+            description: Optional description for logging
+        Returns:
+            _GeneratorRequest: An instance of _GeneratorRequest to iterate over responses
+        Raises:
+            TypeError: If request is not of type Message or _GeneratorMessage
+        """
+        return _GeneratorRequest(self, target, request, timeout, description)
 
     def send_multi_request_sync(self, target_request: List[Tuple[str, Union[Message, Any]]], 
                                timeout: int = -1, description: Optional[str] = None) -> List[Tuple[str, Union[Message, Any], Any, int]]:

src/iop/_business_operation.py

@@ -72,3 +72,34 @@ def OnMessage(self, request: Any) -> Any:
         The response object
         """
         return
+    
+    @input_deserializer
+    @output_serializer
+    def _dispatch_private_session_started(self, request: Any) -> Any:
+        """For internal use only."""
+        self._gen = self.on_private_session_started(request)
+        from . import _iris
+        return _iris.get_iris().IOP.PrivateSession.Message.Ack._New()
+    
+    def on_private_session_started(self, request: Any) -> Any:
+        """Called when a private session is started.
+        
+        This method can be overridden to handle the start of a private session.
+        
+        Args:
+            request: The request object containing session details.
+            
+        Returns:
+            Any response object or None.
+        """
+        return None
+    
+    @input_serializer
+    @output_deserializer
+    def _dispatch_private_session_poll(self, request: Any) -> Any:
+        """For internal use only."""
+        try:
+            return next(self._gen)
+        except StopIteration:
+            import _iris
+            return _iris.get_iris().IOP.PrivateSession.Message.Stop._New()

src/iop/_dispatch.py

@@ -4,6 +4,8 @@
 from ._serialization import serialize_message, serialize_pickle_message, deserialize_message, deserialize_pickle_message
 from ._message_validator import is_message_instance, is_pickle_message_instance, is_iris_object_instance
 
+from ._message import _GeneratorMessage
+
 def dispatch_serializer(message: Any) -> Any:
     """Serializes the message based on its type.
     
@@ -73,6 +75,9 @@ def dispach_message(host: Any, request: Any) -> Any:
     module = request.__class__.__module__
     classname = request.__class__.__name__
 
+    if isinstance(request, _GeneratorMessage):
+        return getattr(host, '_dispatch_private_session_started')(request)
+
     for msg, method in host.DISPATCH:
         if msg == module + "." + classname:
             call = method

src/iop/_generator_request.py

@@ -0,0 +1,36 @@
+from typing import Any, Optional, Union
+
+from . import _iris
+from ._message import _GeneratorMessage
+
+
+class _GeneratorRequest:
+    """Generator class to interetate over responses from a request.
+    This class is used to handle the responses from a request in a generator-like manner."""
+
+    def __init__(self, host: Any, target: str, request: Union[_GeneratorMessage, Any], 
+                 timeout: int = -1, description: Optional[str] = None) -> None:
+        self.host = host
+        self.target = target
+        self.request = request
+        self.timeout = timeout
+        self.description = description
+        self._response = None
+
+        # if not isinstance(self.request, _GeneratorMessage):
+        #     raise TypeError("request must be of type Message or _GeneratorMessage")
+        
+        ack_rsponse = self.host.send_request_sync(self.target, self.request)
+        
+        if ack_rsponse is None:
+            raise RuntimeError("Failed to send request, no acknowledgment received.")
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        poll = _iris.get_iris().IOP.PrivateSession.Message.Poll._New()
+        rsp = self.host.send_request_sync(self.target, poll)
+        if rsp is None:
+            raise StopIteration("No more responses available.")
+        return rsp

src/iop/_message.py

@@ -26,4 +26,7 @@ class _PydanticPickleMessage(BaseModel):
     """Base class for Pydantic-based messages that can be serialized to IRIS."""
 
     def __init__(self, **data: Any):
-        super().__init__(**data)
+        super().__init__(**data)
+
+class _GeneratorMessage(_Message):
+    """"Base message to initialize generator function"""