ParallelPipeline

class ParallelPipeline(BasePipeline):
    def __init__(self, *args):
        super().__init__()
        if len(args) == 0:
            raise Exception("ParallelPipeline needs at least one argument")

        self._sources = []
        self._sinks = []
        self._pipelines = []

        self._up_queue = asyncio.Queue()
        self._down_queue = asyncio.Queue()

class Source(FrameProcessor):
    def __init__(self, upstream_queue: asyncio.Queue):
        """Handles input for each parallel path"""

class Sink(FrameProcessor):
    def __init__(self, downstream_queue: asyncio.Queue):
        """Handles output for each parallel path"""

async def process_frame(self, frame: Frame, direction: FrameDirection):
    """
    Distributes frame to all parallel paths and collects results
    """

async def _process_up_queue(self):
    """Handles upstream frame collection"""

async def _process_down_queue(self):
    """Handles downstream frame collection"""

# Create parallel processing paths
pipeline = ParallelPipeline(
    [audio_processor, audio_handler],
    [video_processor, video_handler],
    [text_processor, text_handler]
)

# Process different frame types in parallel
pipeline = Pipeline([
    input_processor,
    ParallelPipeline(
        [audio_frames_processor],
        [video_frames_processor],
        [text_frames_processor]
    ),
    output_aggregator
])

# Task creation
self._up_task = loop.create_task(self._process_up_queue())
self._down_task = loop.create_task(self._process_down_queue())

# Task cleanup
if isinstance(frame, (CancelFrame, EndFrame)):
    await self._up_queue.put(None)
    await self._down_queue.put(None)
    await self._up_task
    await self._down_task