# Path: azmq/common.py
# class ClosableAsyncObject(AsyncObject):
#     """
#     Base class for objects that can be asynchronously closed and awaited.
#     """
#     def __init__(self, *args, loop=None, **kwargs):
#         super().__init__(loop=loop)
#         self._closed_future = asyncio.Future(loop=self.loop)
#         self._closing = asyncio.Event(loop=self.loop)
#         self.on_closed = Signal()
#         logger.debug("%s[%s] opened.", self.__class__.__name__, id(self))
#         self.on_open(*args, **kwargs)
# 
#     @property
#     def closing(self):
#         return self._closing.is_set()
# 
#     async def wait_closing(self):
#         """
#         Wait for the instance to be closing.
#         """
#         await self._closing.wait()
# 
#     @property
#     def closed(self):
#         return self._closed_future.done()
# 
#     async def wait_closed(self):
#         """
#         Wait for the instance to be closed.
#         """
#         # Prevents the future from being cancelled in case the wait itself gets
#         # cancelled.
#         await asyncio.shield(self._closed_future)
#         return self._closed_future.result()
# 
#     async def await_until_closing(self, coro):
#         """
#         Wait for some task to complete but aborts as soon asthe instance is
#         being closed.
# 
#         :param coro: The coroutine or future-like object to wait for.
#         """
#         wait_task = asyncio.ensure_future(self.wait_closing(), loop=self.loop)
#         coro_task = asyncio.ensure_future(coro, loop=self.loop)
# 
#         try:
#             done, pending = await asyncio.wait(
#                 [wait_task, coro_task],
#                 return_when=asyncio.FIRST_COMPLETED,
#                 loop=self.loop,
#             )
# 
#         finally:
#             wait_task.cancel()
#             coro_task.cancel()
# 
#         # It could be that the previous instructions cancelled coro_task if it
#         # wasn't done yet.
#         return await coro_task
# 
#     def on_open(self):
#         """
#         Called whenever the instance is opened.
# 
#         Should be redefined by child-classes.
#         """
# 
#     async def on_close(self):
#         """
#         Called whenever a close of the instance was requested.
# 
#         :returns: An awaitable that must only complete when the instance is
#             effectively closed.
# 
#         Should be redefined by child-classes.
#         """
# 
#     def __enter__(self):
#         """
#         Enter the context manager.
# 
#         :returns: The instance.
#         """
#         return self
# 
#     def __exit__(self, exc_type, exc, tb):
#         """
#         Exit the context manager.
# 
#         Closes the instance.
#         """
#         self.close()
# 
#     async def __aenter__(self):
#         """
#         Enter the context manager.
# 
#         :returns: The instance.
#         """
#         return self
# 
#     async def __aexit__(self, exc_type, exc, tb):
#         """
#         Exit the context manager.
# 
#         Closes the instance and waits for it to be closed.
#         """
#         self.close()
#         await self.wait_closed()
# 
#     def _set_closed(self, future):
#         """
#         Indicate that the instance is effectively closed.
# 
#         :param future: The close future.
#         """
#         logger.debug("%s[%s] closed.", self.__class__.__name__, id(self))
#         self.on_closed.emit(self)
#         self._closed_future.set_result(future.result())
# 
#     def close(self):
#         """
#         Close the instance.
#         """
#         if not self.closed and not self.closing:
#             logger.debug(
#                 "%s[%s] closing...",
#                 self.__class__.__name__,
#                 id(self),
#             )
#             self._closing.set()
#             future = asyncio.ensure_future(self.on_close(), loop=self.loop)
#             future.add_done_callback(self._set_closed)
# 
# def cancel_on_closing(func):
#     """
#     Automatically cancels a function or coroutine when the defining instance
#     gets closed.
# 
#     :param func: The function to cancel on closing.
#     :returns: A decorated function or coroutine.
#     """
#     @wraps(func)
#     async def wrapper(self, *args, **kwargs):
#         return await self.await_until_closing(func(self, *args, **kwargs))
# 
#     return wrapper


import asyncio
from .common import (
    ClosableAsyncObject,
    cancel_on_closing,
)



class Multiplexer(ClosableAsyncObject):
    def on_open(self):
        super().on_open()
        self._sockets = set()

    def add_socket(self, socket):
        """
        Add a socket to the multiplexer.

        :param socket: The socket. If it was added already, it won't be added a
            second time.
        """
        if socket not in self._sockets:
            self._sockets.add(socket)
            socket.on_closed.connect(self.remove_socket)

    def remove_socket(self, socket):
        """
        Remove a socket from the multiplexer.

        :param socket: The socket. If it was removed already or if it wasn't
            added, the call does nothing.
        """
        if socket in self._sockets:
            socket.on_closed.disconnect(self.remove_socket)
            self._sockets.remove(socket)


# Path: azmq/common.py
# class ClosableAsyncObject(AsyncObject):
#     """
#     Base class for objects that can be asynchronously closed and awaited.
#     """
#     def __init__(self, *args, loop=None, **kwargs):
#         super().__init__(loop=loop)
#         self._closed_future = asyncio.Future(loop=self.loop)
#         self._closing = asyncio.Event(loop=self.loop)
#         self.on_closed = Signal()
#         logger.debug("%s[%s] opened.", self.__class__.__name__, id(self))
#         self.on_open(*args, **kwargs)
# 
#     @property
#     def closing(self):
#         return self._closing.is_set()
# 
#     async def wait_closing(self):
#         """
#         Wait for the instance to be closing.
#         """
#         await self._closing.wait()
# 
#     @property
#     def closed(self):
#         return self._closed_future.done()
# 
#     async def wait_closed(self):
#         """
#         Wait for the instance to be closed.
#         """
#         # Prevents the future from being cancelled in case the wait itself gets
#         # cancelled.
#         await asyncio.shield(self._closed_future)
#         return self._closed_future.result()
# 
#     async def await_until_closing(self, coro):
#         """
#         Wait for some task to complete but aborts as soon asthe instance is
#         being closed.
# 
#         :param coro: The coroutine or future-like object to wait for.
#         """
#         wait_task = asyncio.ensure_future(self.wait_closing(), loop=self.loop)
#         coro_task = asyncio.ensure_future(coro, loop=self.loop)
# 
#         try:
#             done, pending = await asyncio.wait(
#                 [wait_task, coro_task],
#                 return_when=asyncio.FIRST_COMPLETED,
#                 loop=self.loop,
#             )
# 
#         finally:
#             wait_task.cancel()
#             coro_task.cancel()
# 
#         # It could be that the previous instructions cancelled coro_task if it
#         # wasn't done yet.
#         return await coro_task
# 
#     def on_open(self):
#         """
#         Called whenever the instance is opened.
# 
#         Should be redefined by child-classes.
#         """
# 
#     async def on_close(self):
#         """
#         Called whenever a close of the instance was requested.
# 
#         :returns: An awaitable that must only complete when the instance is
#             effectively closed.
# 
#         Should be redefined by child-classes.
#         """
# 
#     def __enter__(self):
#         """
#         Enter the context manager.
# 
#         :returns: The instance.
#         """
#         return self
# 
#     def __exit__(self, exc_type, exc, tb):
#         """
#         Exit the context manager.
# 
#         Closes the instance.
#         """
#         self.close()
# 
#     async def __aenter__(self):
#         """
#         Enter the context manager.
# 
#         :returns: The instance.
#         """
#         return self
# 
#     async def __aexit__(self, exc_type, exc, tb):
#         """
#         Exit the context manager.
# 
#         Closes the instance and waits for it to be closed.
#         """
#         self.close()
#         await self.wait_closed()
# 
#     def _set_closed(self, future):
#         """
#         Indicate that the instance is effectively closed.
# 
#         :param future: The close future.
#         """
#         logger.debug("%s[%s] closed.", self.__class__.__name__, id(self))
#         self.on_closed.emit(self)
#         self._closed_future.set_result(future.result())
# 
#     def close(self):
#         """
#         Close the instance.
#         """
#         if not self.closed and not self.closing:
#             logger.debug(
#                 "%s[%s] closing...",
#                 self.__class__.__name__,
#                 id(self),
#             )
#             self._closing.set()
#             future = asyncio.ensure_future(self.on_close(), loop=self.loop)
#             future.add_done_callback(self._set_closed)
# 
# def cancel_on_closing(func):
#     """
#     Automatically cancels a function or coroutine when the defining instance
#     gets closed.
# 
#     :param func: The function to cancel on closing.
#     :returns: A decorated function or coroutine.
#     """
#     @wraps(func)
#     async def wrapper(self, *args, **kwargs):
#         return await self.await_until_closing(func(self, *args, **kwargs))
# 
#     return wrapper


# Path: azmq/multiplexer.py
import asyncio
from .common import (
    ClosableAsyncObject,
    cancel_on_closing,
)




class Multiplexer(ClosableAsyncObject):
    def on_open(self):
        super().on_open()
        self._sockets = set()

    def add_socket(self, socket):
        """
        Add a socket to the multiplexer.

        :param socket: The socket. If it was added already, it won't be added a
            second time.
        """
        if socket not in self._sockets:
            self._sockets.add(socket)
            socket.on_closed.connect(self.remove_socket)

    def remove_socket(self, socket):
        """
        Remove a socket from the multiplexer.

        :param socket: The socket. If it was removed already or if it wasn't
            added, the call does nothing.
        """
        if socket in self._sockets:
            socket.on_closed.disconnect(self.remove_socket)
            self._sockets.remove(socket)


    @cancel_on_closing