Data Stream Processor

Build a data processing system that combines and processes multiple asynchronous data streams in parallel.

Requirements

You need to implement a data processor that:

1. Accepts multiple asynchronous data streams as input
2. Combines corresponding elements from each stream into tuples
3. Processes each combined tuple and returns the results as a list

The processor should handle streams of different types (user data, transaction amounts, and timestamps) and combine them element-by-element for processing.

Implementation

@generates

API

async def process_streams(users, amounts, timestamps):
    """
    Process multiple async streams by combining corresponding elements.

    Args:
        users: Async iterable of user names (strings)
        amounts: Async iterable of transaction amounts (floats)
        timestamps: Async iterable of timestamps (integers)

    Returns:
        List of tuples containing (user, amount, timestamp) for each position

    Example:
        users = async_generator(['Alice', 'Bob', 'Charlie'])
        amounts = async_generator([100.0, 250.5, 75.25])
        timestamps = async_generator([1609459200, 1609545600, 1609632000])

        result = await process_streams(users, amounts, timestamps)
        # Returns: [('Alice', 100.0, 1609459200), ('Bob', 250.5, 1609545600), ('Charlie', 75.25, 1609632000)]
    """
    pass

Test Cases

• Given three async streams with matching lengths, the function combines corresponding elements into tuples and returns them as a list @test
• Given three async streams where one is empty, the function returns an empty list @test
• Given three single-element async streams, the function returns a list with one tuple containing all three elements @test

Dependencies { .dependencies }

asyncstdlib { .dependency }

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