# Path: azmq/connections/inproc.py
# class InprocConnection(BaseConnection):
#     """
#     Implements a ZMTP connection that works in-process.
#     """
#     def __init__(
#         self,
#         channel,
#         **kwargs
#     ):
#         super().__init__(**kwargs)
#         self.channel = channel
#         self.register_child(self.channel)
# 
#     async def on_close(self):
#         self.channel.close()
#         return await super().on_close()
# 
#     async def on_run(self):
#         await self.channel.write(self.get_metadata())
#         metadata = await self.channel.read()
#         self.set_remote_metadata(metadata)
# 
#         logger.debug("Connection is now ready to read and write.")
#         self.on_ready(self)
# 
#         assert self.inbox or self.outbox, (
#             "on_ready callback must either set an inbox or an outbox."
#         )
# 
#         try:
#             tasks = []
#             read_task = asyncio.ensure_future(self.read(), loop=self.loop)
#             tasks.append(read_task)
#             write_task = asyncio.ensure_future(
#                 self.write(),
#                 loop=self.loop,
#             )
#             tasks.append(write_task)
# 
#             try:
#                 await self.await_until_closing(
#                     asyncio.gather(*tasks, loop=self.loop),
#                 )
#             except asyncio.CancelledError:
#                 logger.debug(
#                     "Read/write cycle interrupted. Connection will die soon.",
#                 )
#             finally:
#                 for task in tasks:
#                     task.cancel()
# 
#                 await asyncio.wait(tasks, loop=self.loop)
#         finally:
#             self.on_lost(self)
# 
#         # Flush out the unset outgoing messages before we exit.
#         while not self.outbox.empty():
#             await self.channel.write(self.outbox.read_nowait())
# 
#     async def read(self):
#         while True:
#             frames = await self.channel.read()
# 
#             if self.socket_type in {PUB, XPUB}:
#                 type_ = frames[0][0]
# 
#                 if type_ == 1:
#                     await self.subscribe(frames[0][1:])
#                 elif type_ == 0:
#                     await self.unsubscribe(frames[0][1:])
#                 else:
#                     raise ProtocolError(
#                         "Unexpected subscription message type (%s)." % type_,
#                     )
#             elif not self._discard_incoming_messages:
#                 await self.inbox.write(frames)
# 
#     async def write(self):
#         outbox = self.outbox
# 
#         while True:
#             frames = await outbox.read()
#             await self.channel.write(frames)
# 
# Path: azmq/log.py

# 
# Path: azmq/engines/base.py
# class BaseEngine(CompositeClosableAsyncObject):
#     def __init__(
#         self,
#         *,
#         socket_type,
#         identity,
#         mechanism,
#         zap_client,
#         **kwargs
#     ):
#         super().__init__(**kwargs)
#         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)
#             else:
#                 self.current_backoff_duration = self.min_backoff_duration
# 
#             await asyncio.sleep(self.current_backoff_duration, loop=self.loop)
# 
#             self.current_backoff_duration = min(
#                 self.max_backoff_duration,
#                 self.current_backoff_duration * 2,
#             )


import asyncio
from ..connections.inproc import InprocConnection
from ..log import logger
from .base import BaseEngine
        try:
            channel = await self.await_until_closing(
                self.context._open_inproc_connection(
                    path=self.path,
                ),
            )
        except Exception as ex:
            logger.debug("Connection to %s failed: %s", self.path, ex)
        else:
            logger.debug(
                "Connection to %s established.",
                self.path,
            )

            try:
                async with channel:
                    async with InprocConnection(
                        channel=channel,
                        socket_type=self.socket_type,
                        identity=self.identity,
                        mechanism=self.mechanism,
                        on_ready=self.on_connection_ready.emit,
                        on_lost=self.on_connection_lost.emit,
                        on_failure=self.on_connection_failure,
                    ) as connection:
                        self.register_child(connection)
                        await connection.wait_closed()

                    return connection.get_error()
            finally:

# Path: azmq/connections/inproc.py
# class InprocConnection(BaseConnection):
#     """
#     Implements a ZMTP connection that works in-process.
#     """
#     def __init__(
#         self,
#         channel,
#         **kwargs
#     ):
#         super().__init__(**kwargs)
#         self.channel = channel
#         self.register_child(self.channel)
# 
#     async def on_close(self):
#         self.channel.close()
#         return await super().on_close()
# 
#     async def on_run(self):
#         await self.channel.write(self.get_metadata())
#         metadata = await self.channel.read()
#         self.set_remote_metadata(metadata)
# 
#         logger.debug("Connection is now ready to read and write.")
#         self.on_ready(self)
# 
#         assert self.inbox or self.outbox, (
#             "on_ready callback must either set an inbox or an outbox."
#         )
# 
#         try:
#             tasks = []
#             read_task = asyncio.ensure_future(self.read(), loop=self.loop)
#             tasks.append(read_task)
#             write_task = asyncio.ensure_future(
#                 self.write(),
#                 loop=self.loop,
#             )
#             tasks.append(write_task)
# 
#             try:
#                 await self.await_until_closing(
#                     asyncio.gather(*tasks, loop=self.loop),
#                 )
#             except asyncio.CancelledError:
#                 logger.debug(
#                     "Read/write cycle interrupted. Connection will die soon.",
#                 )
#             finally:
#                 for task in tasks:
#                     task.cancel()
# 
#                 await asyncio.wait(tasks, loop=self.loop)
#         finally:
#             self.on_lost(self)
# 
#         # Flush out the unset outgoing messages before we exit.
#         while not self.outbox.empty():
#             await self.channel.write(self.outbox.read_nowait())
# 
#     async def read(self):
#         while True:
#             frames = await self.channel.read()
# 
#             if self.socket_type in {PUB, XPUB}:
#                 type_ = frames[0][0]
# 
#                 if type_ == 1:
#                     await self.subscribe(frames[0][1:])
#                 elif type_ == 0:
#                     await self.unsubscribe(frames[0][1:])
#                 else:
#                     raise ProtocolError(
#                         "Unexpected subscription message type (%s)." % type_,
#                     )
#             elif not self._discard_incoming_messages:
#                 await self.inbox.write(frames)
# 
#     async def write(self):
#         outbox = self.outbox
# 
#         while True:
#             frames = await outbox.read()
#             await self.channel.write(frames)
# 
# Path: azmq/log.py

# 
# Path: azmq/engines/base.py
# class BaseEngine(CompositeClosableAsyncObject):
#     def __init__(
#         self,
#         *,
#         socket_type,
#         identity,
#         mechanism,
#         zap_client,
#         **kwargs
#     ):
#         super().__init__(**kwargs)
#         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)
#             else:
#                 self.current_backoff_duration = self.min_backoff_duration
# 
#             await asyncio.sleep(self.current_backoff_duration, loop=self.loop)
# 
#             self.current_backoff_duration = min(
#                 self.max_backoff_duration,
#                 self.current_backoff_duration * 2,
#             )


# Path: azmq/engines/inproc.py
import asyncio
from ..connections.inproc import InprocConnection
from ..log import logger
from .base import BaseEngine

        try:
            channel = await self.await_until_closing(
                self.context._open_inproc_connection(
                    path=self.path,
                ),
            )
        except Exception as ex:
            logger.debug("Connection to %s failed: %s", self.path, ex)
        else:
            logger.debug(
                "Connection to %s established.",
                self.path,
            )

            try:
                async with channel:
                    async with InprocConnection(
                        channel=channel,
                        socket_type=self.socket_type,
                        identity=self.identity,
                        mechanism=self.mechanism,
                        on_ready=self.on_connection_ready.emit,
                        on_lost=self.on_connection_lost.emit,
                        on_failure=self.on_connection_failure,
                    ) as connection:
                        self.register_child(connection)
                        await connection.wait_closed()

                    return connection.get_error()
            finally:

                logger.debug(