
借助 PollingBusinessService 提升开发者体验
4.0.0 版本中最引人注目的新增功能之一是 PollingBusinessService。
在此版本之前,创建定时 Python 业务服务意味着需要 直接处理低级入站适配器模式。在 4.0.0 版本中,定时 Python 入口点可以直接编写为 PollingBusinessService, 框架会在后台自动使用默认的 IRIS 入站适配器。
以下是一个完整的小型示例:
from dataclasses import dataclass
from iop import BusinessOperation, Message, PollingBusinessService, Production, target
@dataclass
class HelloRequest(Message):
text: str = "Hello World"
class HelloService(PollingBusinessService):
Output = target()
def on_poll(self):
self.send_request_async(self.Output, HelloRequest())
class HelloOperation(BusinessOperation):
def on_message(self, request: HelloRequest):
self.log_info(request.text)
return request
prod = Production("HelloWorld.Production", testing_enabled=True)
service = prod.service("HelloService", HelloService)
operation = prod.operation("HelloOperation", HelloOperation)
service.connect(HelloService.Output, operation)
PRODUCTIONS = [prod]
.png)







