Async Data Stream Merger

Build a data processing utility that merges multiple async data streams of potentially different lengths into a unified output format.

Requirements

You need to implement a function that processes multiple async data streams and combines them into a single list of records. Each stream may have a different number of items, and you must handle this gracefully by filling in missing values.

Input

The function should accept:

• Multiple async iterables (streams) containing data items
• A default value to use when a stream runs out of items before others

Output

Return a list where each element contains the corresponding items from all streams at that position. When a stream is exhausted, use the provided default value for that stream's position in subsequent records.

Behavior

• Process all streams concurrently until ALL streams are exhausted
• When shorter streams run out, fill their positions with the default value
• Maintain the order of items from each stream
• Handle empty streams appropriately

Test Cases

• Given three async streams [1, 2], ['a', 'b', 'c'], and [True, False] with fill value None, the result should be [(1, 'a', True), (2, 'b', False), (None, 'c', None)]. @test

• Given two async streams [] and [10, 20] with fill value "N/A", the result should be [("N/A", 10), ("N/A", 20)]. @test

• Given two async streams ['x', 'y', 'z'] and [1, 2, 3] with fill value 0, the result should be [('x', 1), ('y', 2), ('z', 3)]. @test

Implementation

@generates

API

async def merge_streams(*streams, fill_value):
    """
    Merge multiple async iterables into a list of tuples.

    Args:
        *streams: Variable number of async iterables to merge
        fill_value: Value to use when a stream is exhausted

    Returns:
        List of tuples containing items from all streams
    """
    pass

Dependencies { .dependencies }

asyncstdlib { .dependency }

Provides async-compatible versions of standard library functions for working with async iterables.