def __aenter__(self):
        transports = self._transports
        write_buff = self._write_buff

        class SocksPrimitiveProtocol(asyncio.Protocol):
            _transport = None

            def connection_made(self, transport):
                self._transport = transport
                transports.append(transport)

            def data_received(self, data):
                self._transport.write(write_buff)

        def factory():
            return SocksPrimitiveProtocol()

        self._srv = await self._loop.create_server(
            factory, '127.0.0.1', self.port)

        return self 

def connection_lost(self, exc):
        """Reestablish the connection to the transport.

        Called when asyncio.Protocol loses the network connection.
        """
        if exc is None:
            _LOGGER.warning("End of file received from Insteon Modem")
        else:
            _LOGGER.warning("Lost connection to Insteon Modem: %s", exc)

        self.transport = None
        asyncio.ensure_future(self.pause_writing(), loop=self.loop)

        if self._connection_lost_callback:
            self._connection_lost_callback()

    # Methods used to trigger callbacks for specific events 

def connection_made(self, transport):
        """Start the PLM connection process.

        Called when asyncio.Protocol establishes the network connection.
        """
        _LOGGER.info("Connection established to PLM")
        self.transport = transport

        self._restart_writer = True
        self.restart_writing()

        # Testing to see if this fixes the 2413S issue
        self.transport.serial.timeout = 1
        self.transport.serial.write_timeout = 1
        self.transport.set_write_buffer_limits(128)
        # limit = self.transport.get_write_buffer_size()
        # _LOGGER.debug('Write buffer size is %d', limit)
        if self._aldb.status != ALDBStatus.LOADED:
            asyncio.ensure_future(self._setup_devices(), loop=self._loop) 

def test_make_ssl_transport(self):
        m = mock.Mock()
        self.loop.add_reader = mock.Mock()
        self.loop.add_reader._is_coroutine = False
        self.loop.add_writer = mock.Mock()
        self.loop.remove_reader = mock.Mock()
        self.loop.remove_writer = mock.Mock()
        waiter = asyncio.Future(loop=self.loop)
        with test_utils.disable_logger():
            transport = self.loop._make_ssl_transport(
                m, asyncio.Protocol(), m, waiter)
            # execute the handshake while the logger is disabled
            # to ignore SSL handshake failure
            test_utils.run_briefly(self.loop)

        # Sanity check
        class_name = transport.__class__.__name__
        self.assertIn("ssl", class_name.lower())
        self.assertIn("transport", class_name.lower())

        transport.close()
        # execute pending callbacks to close the socket transport
        test_utils.run_briefly(self.loop) 

def test_empty(self):
        f = mock.Mock()
        p = asyncio.Protocol()
        self.assertIsNone(p.connection_made(f))
        self.assertIsNone(p.connection_lost(f))
        self.assertIsNone(p.data_received(f))
        self.assertIsNone(p.eof_received())

        dp = asyncio.DatagramProtocol()
        self.assertIsNone(dp.connection_made(f))
        self.assertIsNone(dp.connection_lost(f))
        self.assertIsNone(dp.error_received(f))
        self.assertIsNone(dp.datagram_received(f, f))

        sp = asyncio.SubprocessProtocol()
        self.assertIsNone(sp.connection_made(f))
        self.assertIsNone(sp.connection_lost(f))
        self.assertIsNone(sp.pipe_data_received(1, f))
        self.assertIsNone(sp.pipe_connection_lost(1, f))
        self.assertIsNone(sp.process_exited()) 

def make_console(bot):
    class Console(asyncio.Protocol):
        def connection_made(self, transport):
            self.transport = transport
            self.transport.write(b'>>> ')
            self.bot = bot

        def data_received(self, data):
            """
            Called when some data is received.
            The argument is a bytes object.
            """
            if data == b'\xff\xf4\xff\xfd\x06':
                self.transport.close()
                return
            message = data.decode()
            resp = str(eval(message)) + "\n>>> "
            self.transport.write(resp.encode('utf-8'))
    return Console 

def connect(self) -> None:
        """Open a connection to the defined server."""
        def protocol_factory() -> Protocol:
            return Protocol(client=self)

        _, protocol = await self.loop.create_connection(
            protocol_factory,
            host=self.host,
            port=self.port,
            ssl=self.ssl
        )  # type: Tuple[Any, Any]
        if self.protocol:
            self.protocol.close()
        self.protocol = protocol
        # TODO: Delete the following code line. It is currently kept in order
        # to not break the current existing codebase. Removing it requires a
        # heavy change in the test codebase.
        protocol.client = self
        self.trigger("client_connect") 

def test_make_ssl_transport(self):
        m = mock.Mock()
        self.loop.add_reader = mock.Mock()
        self.loop.add_reader._is_coroutine = False
        self.loop.add_writer = mock.Mock()
        self.loop.remove_reader = mock.Mock()
        self.loop.remove_writer = mock.Mock()
        waiter = asyncio.Future(loop=self.loop)
        with test_utils.disable_logger():
            transport = self.loop._make_ssl_transport(
                m, asyncio.Protocol(), m, waiter)
            # execute the handshake while the logger is disabled
            # to ignore SSL handshake failure
            test_utils.run_briefly(self.loop)

        # Sanity check
        class_name = transport.__class__.__name__
        self.assertIn("ssl", class_name.lower())
        self.assertIn("transport", class_name.lower())

        transport.close()
        # execute pending callbacks to close the socket transport
        test_utils.run_briefly(self.loop) 

def test_empty(self):
        f = mock.Mock()
        p = asyncio.Protocol()
        self.assertIsNone(p.connection_made(f))
        self.assertIsNone(p.connection_lost(f))
        self.assertIsNone(p.data_received(f))
        self.assertIsNone(p.eof_received())

        dp = asyncio.DatagramProtocol()
        self.assertIsNone(dp.connection_made(f))
        self.assertIsNone(dp.connection_lost(f))
        self.assertIsNone(dp.error_received(f))
        self.assertIsNone(dp.datagram_received(f, f))

        sp = asyncio.SubprocessProtocol()
        self.assertIsNone(sp.connection_made(f))
        self.assertIsNone(sp.connection_lost(f))
        self.assertIsNone(sp.pipe_data_received(1, f))
        self.assertIsNone(sp.pipe_connection_lost(1, f))
        self.assertIsNone(sp.process_exited()) 

def __init__(self, protocol_cls, loop=None, max_workers=2,
                 sync_kind=TextDocumentSyncKind.INCREMENTAL):
        if not issubclass(protocol_cls, asyncio.Protocol):
            raise TypeError('Protocol class should be subclass of asyncio.Protocol')

        self._max_workers = max_workers
        self._server = None
        self._stop_event = None
        self._thread_pool = None
        self._thread_pool_executor = None
        self.sync_kind = sync_kind

        if IS_WIN:
            asyncio.set_event_loop(asyncio.ProactorEventLoop())
        else:
            asyncio.set_event_loop(asyncio.SelectorEventLoop())

        self.loop = loop or asyncio.get_event_loop()

        try:
            asyncio.get_child_watcher().attach_loop(self.loop)
        except NotImplementedError:
            pass

        self.lsp = protocol_cls(self) 

def datagram_received(self, data, addr):
        """
        Opens a read or write connection to remote host by scheduling
        an asyncio.Protocol.
        """
        logger.debug('received: {}'.format(data.decode()))

        first_packet = self.packet_factory.from_bytes(data)
        protocol = self.select_protocol(first_packet)
        file_handler_cls = self.select_file_handler(first_packet)

        connect = self.loop.create_datagram_endpoint(
            lambda: protocol(data, file_handler_cls, addr, self.extra_opts),
            local_addr=(self.host_interface,
                        0, ))

        self.loop.create_task(connect) 

def task():
    rfd, wfd = os.pipe()
    args = [sys.executable, '-c', code, str(rfd)]
    proc = yield from asyncio.create_subprocess_exec(
                          *args,
                          pass_fds={rfd},
                          stdout=subprocess.PIPE)

    pipe = open(wfd, 'wb', 0)
    transport, _ = yield from loop.connect_write_pipe(asyncio.Protocol,
                                                      pipe)
    transport.write(b'data')

    stdout, stderr = yield from proc.communicate()
    print("stdout = %r" % stdout.decode())
    transport.close() 

def test_empty(self):
        f = mock.Mock()
        p = asyncio.Protocol()
        self.assertIsNone(p.connection_made(f))
        self.assertIsNone(p.connection_lost(f))
        self.assertIsNone(p.data_received(f))
        self.assertIsNone(p.eof_received())

        dp = asyncio.DatagramProtocol()
        self.assertIsNone(dp.connection_made(f))
        self.assertIsNone(dp.connection_lost(f))
        self.assertIsNone(dp.error_received(f))
        self.assertIsNone(dp.datagram_received(f, f))

        sp = asyncio.SubprocessProtocol()
        self.assertIsNone(sp.connection_made(f))
        self.assertIsNone(sp.connection_lost(f))
        self.assertIsNone(sp.pipe_data_received(1, f))
        self.assertIsNone(sp.pipe_connection_lost(1, f))
        self.assertIsNone(sp.process_exited()) 

def setUp(self):
        self.loop = self.new_test_loop()
        self.protocol = test_utils.make_test_protocol(asyncio.Protocol)
        self.pipe = mock.Mock(spec_set=io.RawIOBase)
        self.pipe.fileno.return_value = 5

        blocking_patcher = mock.patch('asyncio.unix_events._set_nonblocking')
        blocking_patcher.start()
        self.addCleanup(blocking_patcher.stop)

        fstat_patcher = mock.patch('os.fstat')
        m_fstat = fstat_patcher.start()
        st = mock.Mock()
        st.st_mode = stat.S_IFIFO
        m_fstat.return_value = st
        self.addCleanup(fstat_patcher.stop) 

def start_reading_pty(protocol, pty_fd):
    """
    Make asyncio to read file descriptor of Pty

    :param protocol: protocol of subprocess speaking via Pty
    :param pty_fd: file descriptor of Pty (dialog with subprocess goes that way)
    :return:
    """
    loop, its_new = thread_secure_get_event_loop()

    # Create Protocol classes
    class PtyFdProtocol(asyncio.Protocol):
        def connection_made(self, transport):
            if hasattr(protocol, 'on_pty_open'):
                protocol.on_pty_open()

        def data_received(self, data, recv_time):
            if hasattr(protocol, 'data_received'):
                protocol.data_received(data)

        def connection_lost(self, exc):
            if hasattr(protocol, 'on_pty_close'):
                protocol.on_pty_close(exc)

    # Add the pty's to the read loop
    # Also store the transport, protocol tuple for each call to
    # connect_read_pipe, to prevent the destruction of the protocol
    # class instance, otherwise no data is received.
    fd_transport, fd_protocol = await loop.connect_read_pipe(PtyFdProtocol, os.fdopen(pty_fd, 'rb', 0))
    protocol.pty_fd_transport = fd_transport
    protocol.pty_fd_protocol = fd_protocol 

def test_base_negotiate_with_app_proto(loop):
    waiter = asyncio.Future(loop=loop)
    proto = make_base(loop, waiter=waiter,
                      ap_factory=lambda: asyncio.Protocol())
    proto.socks_request = make_mocked_coro((None, None))

    await proto.negotiate(None, None)
    await waiter
    assert waiter.done() 

def __init__(self, node_name: str):
        """ Create connection handler object. """
        super().__init__()

        self.node_name_ = node_name
        """ Name of the running Erlang node. """

        self.packet_len_size_ = 2
        """ Packet size header is variable, 2 bytes before handshake is finished
            and 4 bytes afterwards. """

        self.addr_ = None  # type: [None, Tuple[str, int]]

        self.inbox_ = asyncio.Queue()
        """ Inbox is used to ask the connection to do something. """

        self.peer_distr_version_ = (None, None)  # type: (int, int)
        """ Protocol version range supported by the remote peer. Erlang/OTP 
            versions 19-20 supports protocol version 7, older Erlangs down to 
            R6B support version 5. """

        self.peer_flags_ = 0
        self.peer_name_ = None  # type: Union[None, str]
        self.my_challenge_ = None

        self.state_ = self.DISCONNECTED
        """ FSM state for the protocol state-machine. """

        self.transport_ = None  # type: [None, asyncio.Transport]
        self.unconsumed_data_ = b''

        self._last_interaction = time.time() 

def __init__(
        self,
        protocol: asyncio.Protocol,
    ) -> None:
        super().__init__()
        self._loop = asyncio.get_event_loop()
        self._protocol = protocol 

def process_stream_reset(self, event: StreamReset) -> None:
        stream = self.streams.get(event.stream_id)
        if stream is not None:
            if event.remote_reset:
                msg = ('Stream reset by remote party, error_code: {}'
                       .format(event.error_code))
            else:
                msg = 'Protocol error'
            stream.__terminated__(msg)
            self.handler.cancel(stream)

        self.connection.streams_failed += 1 

def data_received(self, data: bytes) -> None:
        try:
            events = self.connection.feed(data)
        except ProtocolError:
            log.debug('Protocol error', exc_info=True)
            self.processor.close('Protocol error')
        else:
            self.connection.flush()
            for event in events:
                self.processor.process(event)
            self.connection.flush() 

def __init__(
        self,
        loop=None,
        connection_lost_callback=None,
        workdir=None,
        poll_devices=True,
        load_aldb=True,
    ):
        """Protocol handler that handles all status and changes on PLM."""
        self._loop = loop
        self._connection_lost_callback = connection_lost_callback

        self._buffer = asyncio.Queue(loop=self._loop)
        self._recv_queue = deque([])
        self._send_queue = asyncio.Queue(loop=self._loop)
        self._acknak_queue = asyncio.Queue(loop=self._loop)
        self._next_all_link_rec_nak_retries = 0
        self._aldb_devices = {}
        self._devices = LinkedDevices(loop, workdir)
        self._poll_devices = poll_devices
        self._load_aldb = load_aldb
        self._write_transport_lock = asyncio.Lock(loop=self._loop)
        self._message_callbacks = MessageCallback()
        self._x10_address = None

        # Callback lists
        self._cb_load_all_link_db_done = []
        self._cb_device_not_active = []

        super().__init__(self, "000000", 0x03, None, None, "", "")

        self.transport = None

        self._register_message_handlers()
        self._writer_task = None
        self._restart_writer = False
        self.restart_writing()

    # public properties 

def connection_made(self, transport):
        """Complete the network connection.

        Called when asyncio.Protocol establishes the network connection.
        """
        raise NotImplementedError 

def data_received(self, data):
        """Receive data from the protocol.

        Called when asyncio.Protocol detects received data from network.
        """
        _LOGGER.debug("Starting: data_received")
        _LOGGER.debug(
            "Received %d bytes from PLM: %s", len(data), binascii.hexlify(data)
        )
        self._buffer.put_nowait(data)
        asyncio.ensure_future(self._peel_messages_from_buffer(), loop=self._loop)

        _LOGGER.debug("Finishing: data_received") 

def sendrecv(self, data, resp_type='command/reply', fut=None):
        """Send raw data to the transport and return a future representing
        a response.
        """
        if not self.connected():
            raise ConnectionError("Protocol is not connected")
        fut = self.reg_fut(resp_type, fut=fut)
        self.send(data)
        return fut 

def handle_PoisonPillFrame(self, frame):
        """ Is sent in case protocol lost connection to server."""
        # Will be delivered after Close or CloseOK handlers. It's for channels,
        # so ignore it.
        if self.connection.closed.done():
            return
        # If connection was not closed already - we lost connection.
        # Protocol should already be closed
        self._close_all(frame.exception) 

def setUp(self):
        self.loop = self.new_test_loop()
        self.addCleanup(self.loop.close)
        self.proactor = mock.Mock()
        self.loop._proactor = self.proactor
        self.protocol = test_utils.make_test_protocol(asyncio.Protocol)
        self.sock = mock.Mock(socket.socket) 

def test_make_socket_transport(self):
        tr = self.loop._make_socket_transport(self.sock, asyncio.Protocol())
        self.assertIsInstance(tr, _ProactorSocketTransport)
        close_transport(tr) 

def test_make_socket_transport(self):
        m = mock.Mock()
        self.loop.add_reader = mock.Mock()
        self.loop.add_reader._is_coroutine = False
        transport = self.loop._make_socket_transport(m, asyncio.Protocol())
        self.assertIsInstance(transport, _SelectorSocketTransport)

        # Calling repr() must not fail when the event loop is closed
        self.loop.close()
        repr(transport)

        close_transport(transport) 

def setUp(self):
        self.loop = self.new_test_loop()
        self.protocol = test_utils.make_test_protocol(asyncio.Protocol)
        self.sock = mock.Mock(socket.socket)
        self.sock.fileno.return_value = 7 

def setUp(self):
        self.loop = self.new_test_loop()
        self.protocol = test_utils.make_test_protocol(asyncio.Protocol)
        self.sock = mock.Mock(socket.socket)
        self.sock.fileno.return_value = 7
        self.sslsock = mock.Mock()
        self.sslsock.fileno.return_value = 1
        self.sslcontext = mock.Mock()
        self.sslcontext.wrap_socket.return_value = self.sslsock