# 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 get_metadata(self):
        metadata = {
            b'Socket-Type': self.socket_type,
        }

        if self.identity:
            metadata[b'Identity'] = self.identity

        return metadata

    def set_remote_metadata(self, metadata):
        logger.debug("Peer metadata: %s", metadata)

        for key, value in metadata.items():
            key = key.lower()

            if key == b'identity':
                # Peer-specified identities can't start with b'\x00'.
                if value and not value[0]:
                    raise ProtocolError(
                        "Peer specified an invalid identity (%r)." % value,
                        fatal=True,
                    )

                self.remote_identity = value
            elif key == b'socket-type':
                self.remote_socket_type = value

        pair = (self.socket_type, self.remote_socket_type)


# 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 get_metadata(self):
        metadata = {
            b'Socket-Type': self.socket_type,
        }

        if self.identity:
            metadata[b'Identity'] = self.identity

        return metadata

    def set_remote_metadata(self, metadata):
        logger.debug("Peer metadata: %s", metadata)

        for key, value in metadata.items():
            key = key.lower()

            if key == b'identity':
                # Peer-specified identities can't start with b'\x00'.
                if value and not value[0]:
                    raise ProtocolError(
                        "Peer specified an invalid identity (%r)." % value,
                        fatal=True,
                    )

                self.remote_identity = value
            elif key == b'socket-type':
                self.remote_socket_type = value

        pair = (self.socket_type, self.remote_socket_type)


        if pair not in LEGAL_COMBINATIONS: