第三方插件轮询支持阶梯时间间隔

pipeline_plugins/components/collections/remote_plugin/v1_0_0.py

@@ -12,13 +12,14 @@
 """
 import logging
 
+from django.utils.translation import ugettext_lazy as _
 from pipeline.component_framework.component import Component
-from pipeline.core.flow import Service, StaticIntervalGenerator
+from pipeline.core.flow import AbstractIntervalGenerator, Service
 from pipeline.core.flow.io import StringItemSchema
+
 from plugin_service.conf import PLUGIN_LOGGER
 from plugin_service.exceptions import PluginServiceException
 from plugin_service.plugin_client import PluginServiceApiClient
-from django.utils.translation import ugettext_lazy as _
 
 logger = logging.getLogger(PLUGIN_LOGGER)
 
@@ -34,8 +35,15 @@ class State:
 UNFINISHED_STATES = {State.POLL, State.CALLBACK}
 
 
+class StepIntervalGenerator(AbstractIntervalGenerator):
+    def next(self):
+        super(StepIntervalGenerator, self).next()
+        # 最小 10s，最大 3600s 一次
+        return 10 if self.count < 30 else min((self.count - 25) ** 2, 3600)
+
+
 class RemotePluginService(Service):
-    interval = StaticIntervalGenerator(5)
+    interval = StepIntervalGenerator()
 
     def outputs_format(self):
         return [
@@ -72,9 +80,7 @@ def execute(self, data, parent_data):
         )
         ok, result_data = plugin_client.invoke(plugin_version, {"inputs": data.inputs, "context": plugin_context})
         if not ok:
-            message = _(
-                f"调用第三方插件invoke接口错误, 错误内容: {result_data['message']}, trace_id: {result_data.get('trace_id')}"
-            )
+            message = _(f"调用第三方插件invoke接口错误, 错误内容: {result_data['message']}, trace_id: {result_data.get('trace_id')}")
             logger.error(message)
             data.set_outputs("ex_data", message)
             return False