EndFrame: add reason field to indicate why pipeline is being ended

This commit is contained in:
Aleix Conchillo Flaqué
2025-11-02 00:45:27 -07:00
parent e85e93b9b1
commit 62e45f466a
3 changed files with 18 additions and 3 deletions

View File

@@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added
- `EndFrame` and `EndTaskFrame` have an optional `reason` field to indicate why
the pipeline is being ended.
- `CancelFrame` and `CancelTaskFrame` have an optional `reason` field to
indicate why the pipeline is being canceled. This can be also specified when
you cancel a task with `PipelineTask.cancel(reason="cancellation your

View File

@@ -1372,9 +1372,15 @@ class EndTaskFrame(TaskFrame):
This is used to notify the pipeline task that the pipeline should be
closed nicely (flushing all the queued frames) by pushing an EndFrame
downstream. This frame should be pushed upstream.
Parameters:
reason: Optional reason for pushing an end frame.
"""
pass
reason: Optional[str] = None
def __str__(self):
return f"{self.name}(reason: {self.reason})"
@dataclass
@@ -1463,9 +1469,15 @@ class EndFrame(ControlFrame):
sending frames to its output channel(s) and close all its threads. Note,
that this is a control frame, which means it will be received in the order it
was sent.
Parameters:
reason: Optional reason for pushing an end frame.
"""
pass
reason: Optional[str] = None
def __str__(self):
return f"{self.name}(reason: {self.reason})"
@dataclass

View File

@@ -724,7 +724,7 @@ class PipelineTask(BasePipelineTask):
if isinstance(frame, EndTaskFrame):
# Tell the task we should end nicely.
logger.debug(f"{self}: received end task frame {frame}")
await self.queue_frame(EndFrame())
await self.queue_frame(EndFrame(reason=frame.reason))
elif isinstance(frame, CancelTaskFrame):
# Tell the task we should end right away.
logger.debug(f"{self}: received cancel task frame {frame}")