# 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/errors.py
# class ProtocolError(AZMQError):
#     def __init__(self, msg, fatal=False):
#         super().__init__(msg)
#         self.fatal = fatal
# 
# Path: azmq/log.py



import asyncio
from pyslot import Signal
from ..common import (
    CompositeClosableAsyncObject,
    cancel_on_closing,
)
from ..errors import ProtocolError
from ..log import logger
        self.socket_type = socket_type
        self.identity = identity
        self.mechanism = mechanism
        self.zap_client = zap_client
        self.on_connection_ready = Signal()
        self.on_connection_lost = Signal()
        self.on_connection_failure = Signal()
        self.max_backoff_duration = 300  # 5 minutes.
        self.min_backoff_duration = 0.001
        self.current_backoff_duration = self.min_backoff_duration

    def on_open(self, **kwargs):
        super().on_open(**kwargs)

        self.run_task = asyncio.ensure_future(self.run(), loop=self.loop)

    @cancel_on_closing
    async def run(self):
        while not self.closing:
            try:
                result = await self.open_connection()

                if isinstance(result, ProtocolError) and result.fatal:
                    logger.debug("Fatal error: %s. Not restarting.", result)
                    break

            except asyncio.CancelledError:
                break

            except Exception as ex:

# 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/errors.py
# class ProtocolError(AZMQError):
#     def __init__(self, msg, fatal=False):
#         super().__init__(msg)
#         self.fatal = fatal
# 
# Path: azmq/log.py



# Path: azmq/engines/base.py
import asyncio
from pyslot import Signal
from ..common import (
    CompositeClosableAsyncObject,
    cancel_on_closing,
)
from ..errors import ProtocolError
from ..log import logger

        self.socket_type = socket_type
        self.identity = identity
        self.mechanism = mechanism
        self.zap_client = zap_client
        self.on_connection_ready = Signal()
        self.on_connection_lost = Signal()
        self.on_connection_failure = Signal()
        self.max_backoff_duration = 300  # 5 minutes.
        self.min_backoff_duration = 0.001
        self.current_backoff_duration = self.min_backoff_duration

    def on_open(self, **kwargs):
        super().on_open(**kwargs)

        self.run_task = asyncio.ensure_future(self.run(), loop=self.loop)

    @cancel_on_closing
    async def run(self):
        while not self.closing:
            try:
                result = await self.open_connection()

                if isinstance(result, ProtocolError) and result.fatal:
                    logger.debug("Fatal error: %s. Not restarting.", result)
                    break

            except asyncio.CancelledError:
                break

            except Exception as ex:

                logger.debug("Connection error: %r.", ex)