fix parallel pipelines cleanup

src/pipecat/pipeline/parallel_pipeline.py

@@ -23,7 +23,7 @@ from pipecat.pipeline.pipeline import Pipeline
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 
 
-class Source(FrameProcessor):
+class ParallelPipelineSource(FrameProcessor):
     def __init__(
         self,
         upstream_queue: asyncio.Queue,
@@ -46,7 +46,7 @@ class Source(FrameProcessor):
                 await self.push_frame(frame, direction)
 
 
-class Sink(FrameProcessor):
+class ParallelPipelineSink(FrameProcessor):
     def __init__(
         self,
         downstream_queue: asyncio.Queue,
@@ -92,8 +92,8 @@ class ParallelPipeline(BasePipeline):
                 raise TypeError(f"ParallelPipeline argument {processors} is not a list")
 
             # We will add a source before the pipeline and a sink after.
-            source = Source(self._up_queue, self._parallel_push_frame)
-            sink = Sink(self._down_queue, self._parallel_push_frame)
+            source = ParallelPipelineSource(self._up_queue, self._parallel_push_frame)
+            sink = ParallelPipelineSink(self._down_queue, self._parallel_push_frame)
             self._sources.append(source)
             self._sinks.append(sink)
 
@@ -117,6 +117,7 @@ class ParallelPipeline(BasePipeline):
     #
 
     async def cleanup(self):
+        await super().cleanup()
         await asyncio.gather(*[s.cleanup() for s in self._sources])
         await asyncio.gather(*[p.cleanup() for p in self._pipelines])
         await asyncio.gather(*[s.cleanup() for s in self._sinks])

src/pipecat/pipeline/sync_parallel_pipeline.py

@@ -24,7 +24,7 @@ class SyncFrame(ControlFrame):
     pass
 
 
-class Source(FrameProcessor):
+class SyncParallelPipelineSource(FrameProcessor):
     def __init__(self, upstream_queue: asyncio.Queue):
         super().__init__()
         self._up_queue = upstream_queue
@@ -39,7 +39,7 @@ class Source(FrameProcessor):
                 await self.push_frame(frame, direction)
 
 
-class Sink(FrameProcessor):
+class SyncParallelPipelineSink(FrameProcessor):
     def __init__(self, downstream_queue: asyncio.Queue):
         super().__init__()
         self._down_queue = downstream_queue
@@ -76,8 +76,8 @@ class SyncParallelPipeline(BasePipeline):
             # We add a source at the beginning of the pipeline and a sink at the end.
             up_queue = asyncio.Queue()
             down_queue = asyncio.Queue()
-            source = Source(up_queue)
-            sink = Sink(down_queue)
+            source = SyncParallelPipelineSource(up_queue)
+            sink = SyncParallelPipelineSink(down_queue)
             processors: List[FrameProcessor] = [source] + processors + [sink]
 
             # Keep track of sources and sinks. We also keep the output queue of
@@ -101,6 +101,10 @@ class SyncParallelPipeline(BasePipeline):
     # Frame processor
     #
 
+    async def cleanup(self):
+        await super().cleanup()
+        await asyncio.gather(*[p.cleanup() for p in self._pipelines])
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)