async def base_combine(source, switch=False, ordered=False, task_limit=None):
    """Base operator for managing an asynchronous sequence of sequences.

    The sequences are awaited concurrently, although it's possible to limit
    the amount of running sequences using the `task_limit` argument.

    The ``switch`` argument enables the switch mecanism, which cause the
    previous subsequence to be discarded when a new one is created.

    The items can either be generated in order or as soon as they're received,
    depending on the ``ordered`` argument.
    """

    # Task limit
    if task_limit is not None and not task_limit > 0:
        raise ValueError('The task limit must be None or greater than 0')

    # Safe context
    async with StreamerManager() as manager:

        main_streamer = await manager.enter_and_create_task(source)

        # Loop over events
        while manager.tasks:

            # Extract streamer groups
            substreamers = manager.streamers[1:]
            mainstreamers = [main_streamer] if main_streamer in manager.tasks else []

            # Switch - use the main streamer then the substreamer
            if switch:
                filters = mainstreamers + substreamers
            # Concat - use the first substreamer then the main streamer
            elif ordered:
                filters = substreamers[:1] + mainstreamers
            # Flat - use the substreamers then the main streamer
            else:
                filters = substreamers + mainstreamers

            # Wait for next event
            streamer, task = await manager.wait_single_event(filters)

            # Get result
            try:
                result = task.result()

            # End of stream
            except StopAsyncIteration:

                # Main streamer is finished
                if streamer is main_streamer:
                    main_streamer = None

                # A substreamer is finished
                else:
                    await manager.clean_streamer(streamer)

                    # Re-schedule the main streamer if necessary
                    if main_streamer is not None and main_streamer not in manager.tasks:
                        manager.create_task(main_streamer)

            # Process result
            else:

                # Switch mecanism
                if switch and streamer is main_streamer:
                    await manager.clean_streamers(substreamers)

                # Setup a new source
                if streamer is main_streamer:
                    await manager.enter_and_create_task(result)

                    # Re-schedule the main streamer if task limit allows it
                    if task_limit is None or task_limit > len(manager.tasks):
                        manager.create_task(streamer)

                # Yield the result
                else:
                    yield result

                    # Re-schedule the streamer
                    manager.create_task(streamer)
Base operator for managing an asynchronous sequence of sequences.

    The sequences are awaited concurrently, although it's possible to limit
    the amount of running sequences using the `task_limit` argument.

    The ``switch`` argument enables the switch mecanism, which cause the
    previous subsequence to be discarded when a new one is created.

    The items can either be generated in order or as soon as they're received,
    depending on the ``ordered`` argument.