Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 24 additions & 0 deletions apps/application/workflow/nodes/loop_node/loop_node.py
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,16 @@ class LoopNode(INode):
_workflow_params = None
_iterator = None

def __init__(self, node, workflow_manage, get_node_parameters):
super().__init__(node, workflow_manage, get_node_parameters)
self._current_loop_manage = None

def cancel(self):
super().cancel()
manage = self._current_loop_manage
if manage:
manage.cancel()

def _run(self):
self.execute()

Expand Down Expand Up @@ -96,6 +106,9 @@ def execute(self):
self._run_next()

def _run_next(self):
if self.status == Status.CANCELLED or self.workflow_manage.signal == Signal.CANCELLED:
self.complete(Status.CANCELLED)
return
try:
item, index = next(self._iterator)
except StopIteration:
Expand All @@ -112,6 +125,7 @@ def on_next(wf_manage, content):
self.write(content)

def on_complete(wf_manage, error):
self._current_loop_manage = None
loop_details_list = self.data.setdefault("loop_details_list", [])
loop_details_list.append(wf_manage.get_details())
self.data["message_tokens"] = (self.data.get("message_tokens") or 0) + sum(
Expand All @@ -128,6 +142,10 @@ def on_complete(wf_manage, error):
else:
self.write_context("last_context", wf_manage.context)

if wf_manage.signal == Signal.CANCELLED:
self.complete(Status.CANCELLED)
return

if wf_manage.signal == Signal.BREAK or wf_manage.signal == Signal.FORM:
self.data["run_time"] = time.time() - self.data.get("start_time", time.time())
self.complete(Status.SUCCESS)
Expand Down Expand Up @@ -194,6 +212,12 @@ def get_context():
parent_workflow_manage=self.workflow_manage,
get_context=get_context,
)
if loop_manage:
self._current_loop_manage = loop_manage
# 实例已创建但尚未 run():若取消发生在这期间,这一轮不再启动
if self.status == Status.CANCELLED or self.workflow_manage.signal == Signal.CANCELLED:
self.complete(Status.CANCELLED)
return
loop_manage.run()

def get_details(self, index: int = 0, position: dict = None, old_details: dict = None, **kwargs):
Expand Down
Loading