# Path: paco/observer.py
# class Observer(object):
#     """
#     Observer implements a simple observer pub/sub pattern with a minimal
#     interface and built-in coroutines support for asynchronous-first approach,
#     desiged as abstract class to be inherited or embed by observable classes.
#     """
# 
#     def __init__(self):
#         self._pool = {}
# 
#     def observe(self, event, fn):
#         """
#         Arguments:
#             event (str): event to subscribe.
#             fn (function|coroutinefunction): function to trigger.
# 
#         Raises:
#             TypeError: if fn argument is not valid
#         """
#         iscoroutine = asyncio.iscoroutinefunction(fn)
#         if not iscoroutine and not isfunction(fn):
#             raise TypeError('paco: fn param must be a callable '
#                             'object or coroutine function')
# 
#         observers = self._pool.get(event)
#         if not observers:
#             observers = self._pool[event] = []
# 
#         # Register the observer
#         observers.append(fn if iscoroutine else coroutine_wrapper(fn))
# 
#     def remove(self, event=None):
#         """
#         Remove all the registered observers for the given event name.
# 
#         Arguments:
#             event (str): event name to remove.
#         """
#         observers = self._pool.get(event)
#         if observers:
#             self._pool[event] = []
# 
#     def clear(self):
#         """
#         Clear all the registered observers.
#         """
#         self._pool = {}
# 
#     # Shortcut methods
#     on = observe
#     off = remove
# 
#     @asyncio.coroutine
#     def trigger(self, event, *args, **kw):
#         """
#         Triggers event observers for the given event name,
#         passing custom variadic arguments.
#         """
#         observers = self._pool.get(event)
# 
#         # If no observers registered for the event, do no-op
#         if not observers or len(observers) == 0:
#             return None
# 
#         # Trigger observers coroutines in FIFO sequentially
#         for fn in observers:
#             # Review: perhaps this should not wait
#             yield from fn(*args, **kw)
# 
# Path: paco/assertions.py
# def isiter(x):
#     """
#     Returns `True` if the given value implements an valid iterable
#     interface.
# 
#     Arguments:
#         x (mixed): value to check if it is an iterable.
# 
#     Returns:
#         bool
#     """
#     return hasattr(x, '__iter__') and not isinstance(x, (str, bytes))


import asyncio
from collections import deque, namedtuple
from .observer import Observer
from .assertions import isiter

    Returns:
        ConcurrentExecutor

    Usage::

        async def sum(x, y):
            return x + y

        pool = paco.ConcurrentExecutor(limit=2)
        pool.add(sum, 1, 2)
        pool.add(sum, None, 'str')

        done, pending = await pool.run(return_exceptions=True)
        [task.result() for task in done]
        # => [3, TypeError("unsupported operand type(s) for +: 'NoneType' and 'str'")]  # noqa
    """

    def __init__(self, limit=10, loop=None, coros=None, ignore_empty=False):
        self.errors = []
        self.running = False
        self.return_exceptions = False
        self.limit = max(int(limit), 0)
        self.pool = deque()
        self.observer = Observer()
        self.ignore_empty = ignore_empty
        self.loop = loop or asyncio.get_event_loop()
        self.semaphore = asyncio.Semaphore(self.limit, loop=self.loop)

        # Register coroutines in the pool

# Path: paco/observer.py
# class Observer(object):
#     """
#     Observer implements a simple observer pub/sub pattern with a minimal
#     interface and built-in coroutines support for asynchronous-first approach,
#     desiged as abstract class to be inherited or embed by observable classes.
#     """
# 
#     def __init__(self):
#         self._pool = {}
# 
#     def observe(self, event, fn):
#         """
#         Arguments:
#             event (str): event to subscribe.
#             fn (function|coroutinefunction): function to trigger.
# 
#         Raises:
#             TypeError: if fn argument is not valid
#         """
#         iscoroutine = asyncio.iscoroutinefunction(fn)
#         if not iscoroutine and not isfunction(fn):
#             raise TypeError('paco: fn param must be a callable '
#                             'object or coroutine function')
# 
#         observers = self._pool.get(event)
#         if not observers:
#             observers = self._pool[event] = []
# 
#         # Register the observer
#         observers.append(fn if iscoroutine else coroutine_wrapper(fn))
# 
#     def remove(self, event=None):
#         """
#         Remove all the registered observers for the given event name.
# 
#         Arguments:
#             event (str): event name to remove.
#         """
#         observers = self._pool.get(event)
#         if observers:
#             self._pool[event] = []
# 
#     def clear(self):
#         """
#         Clear all the registered observers.
#         """
#         self._pool = {}
# 
#     # Shortcut methods
#     on = observe
#     off = remove
# 
#     @asyncio.coroutine
#     def trigger(self, event, *args, **kw):
#         """
#         Triggers event observers for the given event name,
#         passing custom variadic arguments.
#         """
#         observers = self._pool.get(event)
# 
#         # If no observers registered for the event, do no-op
#         if not observers or len(observers) == 0:
#             return None
# 
#         # Trigger observers coroutines in FIFO sequentially
#         for fn in observers:
#             # Review: perhaps this should not wait
#             yield from fn(*args, **kw)
# 
# Path: paco/assertions.py
# def isiter(x):
#     """
#     Returns `True` if the given value implements an valid iterable
#     interface.
# 
#     Arguments:
#         x (mixed): value to check if it is an iterable.
# 
#     Returns:
#         bool
#     """
#     return hasattr(x, '__iter__') and not isinstance(x, (str, bytes))


# Path: paco/concurrent.py
import asyncio
from collections import deque, namedtuple
from .observer import Observer
from .assertions import isiter


    Returns:
        ConcurrentExecutor

    Usage::

        async def sum(x, y):
            return x + y

        pool = paco.ConcurrentExecutor(limit=2)
        pool.add(sum, 1, 2)
        pool.add(sum, None, 'str')

        done, pending = await pool.run(return_exceptions=True)
        [task.result() for task in done]
        # => [3, TypeError("unsupported operand type(s) for +: 'NoneType' and 'str'")]  # noqa
    """

    def __init__(self, limit=10, loop=None, coros=None, ignore_empty=False):
        self.errors = []
        self.running = False
        self.return_exceptions = False
        self.limit = max(int(limit), 0)
        self.pool = deque()
        self.observer = Observer()
        self.ignore_empty = ignore_empty
        self.loop = loop or asyncio.get_event_loop()
        self.semaphore = asyncio.Semaphore(self.limit, loop=self.loop)

        # Register coroutines in the pool

        if isiter(coros):