Fluksio/app
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Let teardown and activate be asked for one flow

Browse Source 
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01StpRc2C6au1WJ1EUU7fsfu
This commit is contained in:
stroblme
2026-08-23 17:44:52 +02:00
co-authored by Claude Opus 5
parent 245c386517
commit e22b4795fd
1 changed files with 11 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/fluksio/flow/controller.py
+11 -4
View File
@@ -510,12 +510,14 @@ class FlowController:
} }
) )
async def _teardown(self) -> None: async def _teardown(self, flow: str | None = None) -> None:
"""Stop everything the previous pipeline started.""" """Stop everything the previous pipeline started, or one flow's share."""
for entry in self.loaded.values(): for entry in self.loaded.values():
node = entry.node node = entry.node
if node is None: if node is None:
continue continue
if flow is not None and entry.flow != flow:
continue
try: try:
await asyncio.wait_for(node.stop(self.app), NODE_STOP_TIMEOUT) await asyncio.wait_for(node.stop(self.app), NODE_STOP_TIMEOUT)
except asyncio.TimeoutError: except asyncio.TimeoutError:
@@ -529,9 +531,12 @@ class FlowController:
except Exception: except Exception:
logger.exception("Error stopping node '%s'", entry.id) logger.exception("Error stopping node '%s'", entry.id)
# After the nodes, so a loop still winding down is not restarted. # After the nodes, so a loop still winding down is not restarted.
await self.supervisor.cancel_all() if flow is None:
await self.supervisor.cancel_all()
else:
await self.supervisor.cancel_flow(flow)
async def _activate(self) -> None: async def _activate(self, flow: str | None = None) -> None:
"""Start subscriptions, schedules and webhooks of the new pipeline. """Start subscriptions, schedules and webhooks of the new pipeline.
What that means per node type is the node's own business — a connector What that means per node type is the node's own business — a connector
@@ -541,6 +546,8 @@ class FlowController:
node = entry.node node = entry.node
if node is None: if node is None:
continue continue
if flow is not None and entry.flow != flow:
continue
# A stopped flow gets no subscriptions, schedules or webhooks — # A stopped flow gets no subscriptions, schedules or webhooks —
# that is what stopping it means. Nor does a batch flow, which has # that is what stopping it means. Nor does a batch flow, which has
# no outside to listen to. # no outside to listen to.