# Path: concert/config.py
# AIODEBUG = 9
# 
# Path: concert/helpers.py
# class PrioItem:
# 
#     """To be used in combination with `queue.PriorityQueue`."""
# 
#     priority: int
#     data: Any = field(compare=False)
# 
# Path: concert/quantities.py



import asyncio
import concert.config
import concurrent.futures
import functools
import queue
import logging
import time
from threading import Thread
from concert.config import AIODEBUG
from concert.helpers import PrioItem
from concert.quantities import q
            used -= 1
            if not used:
                if not consumed.cancelled():
                    consumed.set_result(None)
                    consumed = loop.create_future()
                    next_val = loop.create_future()
                used = num_consumers

    started = [produce()]
    for consumer in consumers:
        started.append(consumer(duplicate()))

    return started


async def feed_queue(producer, func, *args):
    """Feed function *func* with items from *producer* in a separete thread. The signatute must be
    func(queue, \*args) where elements in the queue are instances of
    :class:`concert.helpers.PrioItem`.
    """
    loop = asyncio.get_running_loop()
    pqueue = queue.PriorityQueue()

    try:
        future = loop.run_in_executor(None, func, pqueue, *args)
        prio = 1
        async for item in producer:
            pqueue.put(PrioItem(priority=prio, data=item))
            prio += 1
    except (asyncio.CancelledError, KeyboardInterrupt) as e:

# Path: concert/config.py
# AIODEBUG = 9
# 
# Path: concert/helpers.py
# class PrioItem:
# 
#     """To be used in combination with `queue.PriorityQueue`."""
# 
#     priority: int
#     data: Any = field(compare=False)
# 
# Path: concert/quantities.py



# Path: concert/coroutines/base.py
import asyncio
import concert.config
import concurrent.futures
import functools
import queue
import logging
import time
from threading import Thread
from concert.config import AIODEBUG
from concert.helpers import PrioItem
from concert.quantities import q

            used -= 1
            if not used:
                if not consumed.cancelled():
                    consumed.set_result(None)
                    consumed = loop.create_future()
                    next_val = loop.create_future()
                used = num_consumers

    started = [produce()]
    for consumer in consumers:
        started.append(consumer(duplicate()))

    return started


async def feed_queue(producer, func, *args):
    """Feed function *func* with items from *producer* in a separete thread. The signatute must be
    func(queue, \*args) where elements in the queue are instances of
    :class:`concert.helpers.PrioItem`.
    """
    loop = asyncio.get_running_loop()
    pqueue = queue.PriorityQueue()

    try:
        future = loop.run_in_executor(None, func, pqueue, *args)
        prio = 1
        async for item in producer:
            pqueue.put(PrioItem(priority=prio, data=item))
            prio += 1
    except (asyncio.CancelledError, KeyboardInterrupt) as e:

        LOG.log(concert.config.AIODEBUG, f'feed_queue cancelled by exception {type(e)}')