# Path: azmq/common.py
# class CompositeClosableAsyncObject(ClosableAsyncObject):
#     """
#     Base class for objects that can be asynchronously closed and awaited and
#     have ownership of other closable objects.
#     """
#     def on_open(self):
#         self._children = set()
# 
#     async def on_close(self):
#         await asyncio.gather(
#             *[
#                 self.on_close_child(child)
#                 for child in self._children
#             ],
#             loop=self.loop,
#         )
#         del self._children
# 
#     async def on_close_child(self, child):
#         """
#         Called whenever a child instance must be closed.
# 
#         :returns: An awaitable that must only complete when the specified child
#             instance is effectively closed.
# 
#         May be redefined by child-classes. The default implementation calls
#             `close` followed by `wait_closed` on the specified child instance.
#         """
#         child.close()
#         await child.wait_closed()
# 
#     def register_child(self, child):
#         """
#         Register a new child that will be closed whenever the current instance
#         closes.
# 
#         :param child: The child instance.
#         """
#         if self.closing:
#             child.close()
#         else:
#             self._children.add(child)
#             child.on_closed.connect(self.unregister_child)
# 
#     def unregister_child(self, child):
#         """
#         Unregister an existing child that is no longer to be owned by the
#         current instance.
# 
#         :param child: The child instance.
#         """
#         self._children.remove(child)
#         child.on_closed.disconnect(self.unregister_child)
# 
# 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/constants.py
# LEGAL_COMBINATIONS = {
#     (REQ, REP),
#     (REP, REQ),
#     (REQ, ROUTER),
#     (REQ, XREP),
#     (ROUTER, REQ),
#     (XREP, REQ),
#     (REP, DEALER),
#     (REP, XREQ),
#     (DEALER, REP),
#     (XREQ, REP),
#     (DEALER, DEALER),
#     (XREQ, DEALER),
#     (DEALER, XREQ),
#     (DEALER, ROUTER),
#     (DEALER, XREP),
#     (XREQ, ROUTER),
#     (XREQ, XREP),
#     (ROUTER, DEALER),
#     (XREP, DEALER),
#     (ROUTER, XREQ),
#     (XREP, XREQ),
#     (ROUTER, ROUTER),
#     (XREP, ROUTER),
#     (ROUTER, XREP),
#     (PUB, SUB),
#     (SUB, PUB),
#     (PUB, XSUB),
#     (XSUB, PUB),
#     (XPUB, SUB),
#     (SUB, XPUB),
#     (XPUB, XSUB),
#     (XSUB, XPUB),
#     (PUSH, PULL),
#     (PULL, PUSH),
#     (PAIR, PAIR),
# }
# 
# XPUB = b'XPUB'
# 
# Path: azmq/errors.py
# class ProtocolError(AZMQError):
#     def __init__(self, msg, fatal=False):
#         super().__init__(msg)
#         self.fatal = fatal
# 
# Path: azmq/log.py



import asyncio
from contextlib import contextmanager
from ..common import (
    CompositeClosableAsyncObject,
    cancel_on_closing,
)
from ..constants import (
    LEGAL_COMBINATIONS,
    XPUB,
)
from ..errors import ProtocolError
from ..log import logger
    def discard_incoming_messages(self):
        """
        Discard all incoming messages for the time of the context manager.
        """
        # Flush any received message so far.
        self.inbox.clear()

        # This allows nesting of discard_incoming_messages() calls.
        previous = self._discard_incoming_messages
        self._discard_incoming_messages = True

        try:
            yield
        finally:
            self._discard_incoming_messages = previous

    @cancel_on_closing
    async def local_subscribe(self, topic):
        logger.debug("Subscribed to topic %r.", topic)

        await self.outbox.write([b'\x01' + topic])

    @cancel_on_closing
    async def local_unsubscribe(self, topic):
        logger.debug("Unsubscribed from topic %r.", topic)

        await self.outbox.write([b'\x00' + topic])

    async def subscribe(self, topic):
        self.subscriptions.append(topic)

# Path: azmq/common.py
# class CompositeClosableAsyncObject(ClosableAsyncObject):
#     """
#     Base class for objects that can be asynchronously closed and awaited and
#     have ownership of other closable objects.
#     """
#     def on_open(self):
#         self._children = set()
# 
#     async def on_close(self):
#         await asyncio.gather(
#             *[
#                 self.on_close_child(child)
#                 for child in self._children
#             ],
#             loop=self.loop,
#         )
#         del self._children
# 
#     async def on_close_child(self, child):
#         """
#         Called whenever a child instance must be closed.
# 
#         :returns: An awaitable that must only complete when the specified child
#             instance is effectively closed.
# 
#         May be redefined by child-classes. The default implementation calls
#             `close` followed by `wait_closed` on the specified child instance.
#         """
#         child.close()
#         await child.wait_closed()
# 
#     def register_child(self, child):
#         """
#         Register a new child that will be closed whenever the current instance
#         closes.
# 
#         :param child: The child instance.
#         """
#         if self.closing:
#             child.close()
#         else:
#             self._children.add(child)
#             child.on_closed.connect(self.unregister_child)
# 
#     def unregister_child(self, child):
#         """
#         Unregister an existing child that is no longer to be owned by the
#         current instance.
# 
#         :param child: The child instance.
#         """
#         self._children.remove(child)
#         child.on_closed.disconnect(self.unregister_child)
# 
# 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/constants.py
# LEGAL_COMBINATIONS = {
#     (REQ, REP),
#     (REP, REQ),
#     (REQ, ROUTER),
#     (REQ, XREP),
#     (ROUTER, REQ),
#     (XREP, REQ),
#     (REP, DEALER),
#     (REP, XREQ),
#     (DEALER, REP),
#     (XREQ, REP),
#     (DEALER, DEALER),
#     (XREQ, DEALER),
#     (DEALER, XREQ),
#     (DEALER, ROUTER),
#     (DEALER, XREP),
#     (XREQ, ROUTER),
#     (XREQ, XREP),
#     (ROUTER, DEALER),
#     (XREP, DEALER),
#     (ROUTER, XREQ),
#     (XREP, XREQ),
#     (ROUTER, ROUTER),
#     (XREP, ROUTER),
#     (ROUTER, XREP),
#     (PUB, SUB),
#     (SUB, PUB),
#     (PUB, XSUB),
#     (XSUB, PUB),
#     (XPUB, SUB),
#     (SUB, XPUB),
#     (XPUB, XSUB),
#     (XSUB, XPUB),
#     (PUSH, PULL),
#     (PULL, PUSH),
#     (PAIR, PAIR),
# }
# 
# XPUB = b'XPUB'
# 
# Path: azmq/errors.py
# class ProtocolError(AZMQError):
#     def __init__(self, msg, fatal=False):
#         super().__init__(msg)
#         self.fatal = fatal
# 
# Path: azmq/log.py



# Path: azmq/connections/base.py
import asyncio
from contextlib import contextmanager
from ..common import (
    CompositeClosableAsyncObject,
    cancel_on_closing,
)
from ..constants import (
    LEGAL_COMBINATIONS,
    XPUB,
)
from ..errors import ProtocolError
from ..log import logger

    def discard_incoming_messages(self):
        """
        Discard all incoming messages for the time of the context manager.
        """
        # Flush any received message so far.
        self.inbox.clear()

        # This allows nesting of discard_incoming_messages() calls.
        previous = self._discard_incoming_messages
        self._discard_incoming_messages = True

        try:
            yield
        finally:
            self._discard_incoming_messages = previous

    @cancel_on_closing
    async def local_subscribe(self, topic):
        logger.debug("Subscribed to topic %r.", topic)

        await self.outbox.write([b'\x01' + topic])

    @cancel_on_closing
    async def local_unsubscribe(self, topic):
        logger.debug("Unsubscribed from topic %r.", topic)

        await self.outbox.write([b'\x00' + topic])

    async def subscribe(self, topic):
        self.subscriptions.append(topic)

        logger.debug("Peer subscribed to topic %r.", topic)