def __connect(self, host, port):
        """ Retries until success """
        Logger.info("Trying to connect {}:{}".format(host, port))
        while True:
            try:
                reader, writer = await asyncio.open_connection(
                    host, port, loop=self.loop
                )
                break
            except Exception as e:
                Logger.info("Failed to connect {} {}: {}".format(host, port, e))
                await asyncio.sleep(
                    self.env.cluster_config.MASTER.MASTER_TO_SLAVE_CONNECT_RETRY_DELAY
                )
        Logger.info("Connected to {}:{}".format(host, port))
        return reader, writer 

def test_bad_request_response(app):
    lines = []

    @app.listener("after_server_start")
    async def _request(sanic, loop):
        connect = asyncio.open_connection("127.0.0.1", 42101)
        reader, writer = await connect
        writer.write(b"not http")
        while True:
            line = await reader.readline()
            if not line:
                break
            lines.append(line)
        app.stop()

    app.run(host="127.0.0.1", port=42101, debug=False)
    assert lines[0] == b"HTTP/1.1 400 Bad Request\r\n"
    assert b"Bad Request" in lines[-1] 

def bus_monitor(receiver,
                      host='localhost',
                      port=6720,
                      decoder=telegram_decoder):
    """ creates a connection to host:port and starts to receive telegrams

    :param receiver: a coroutine or instance of a class that has a `send`
                     method which takes one argument to receive a telegram.
    :param host: hostname to which to connect to
    :param port: port to which to connect to
    :param decoder: optional alternative decoder to transform binary data into
                    telegrams

    received telegrams will be sent to the receiver.
    """
    reader, writer = await open_connection(host, port)
    await listen(reader, receiver, decoder)
    writer.close() 

def tcp_connection(address):
    """Async generator reading from tcp network transport layer"""
    logger = logging.getLogger('asyncio.tcp-connection')
    logger.debug('... connecting to tcp://{}:{}'.format(*address))
    reader, writer = await asyncio.open_connection(*address)
    try:
        while True:
            data = await reader.read(128)
            if data:
                logger.debug('<<< {!r}'.format(data))
                yield data
            else:
                break
    finally:
        logger.debug('... closing')
        writer.close()


# ============================================================================== 

def __connect(self):
        self.__check_closed()
        if self.connected:
            return
        try:
            self.logger.debug("Opening connection to %s:%d", self.host, self.port)
            future = asyncio.open_connection(self.host, self.port, loop=self.__loop)
            self.__reader, self.__writer = await asyncio.wait_for(
                future, timeout=self.connect_timeout, loop=self.__loop
            )

            await asyncio.wait_for(self.__connect_request_response(), timeout=self.request_timeout, loop=self.__loop)

            self.logger.debug("Socket connected successfully. Starting read loop.")
            self.connected = True
            self.__loop.create_task(self.__read_loop())
        except ConnectionError as e:
            self.logger.error("Connection error while connecting to server: %s", e)
            raise 

def connect(self):
        """Make a TCP connection to the alarm system."""
        _LOGGER.debug("Connecting...")

        try:
            self._reader, self._writer = await asyncio.open_connection(
                self._host, self._port, loop=self._loop)
            _LOGGER.debug("sucess connecting...")

        except Exception as e:
            _LOGGER.warning(
                "Exception during connecting: %s.", e)
            self._writer = None
            self._reader = None
            return False

        return True 

def handle_ext_server_connection(
        upstream_host: str,
        upstream_port: int,
        reader: asyncio.StreamReader,
        writer: asyncio.StreamWriter,
        info: adapters.ExtOrPortClientConnection,
) -> None:
    handler_logger.info('Connection received from %r', info)
    async with contexts.log_unhandled_exc(handler_logger), \
               contexts.aclosing_multiple_writers(writer) as writers:
        try:
            ureader, uwriter = await asyncio.open_connection(
                upstream_host, upstream_port)
        except OSError as e:
            handler_logger.warning(
                'Error while connecting to upstream: %r', e)
            return
        writers.add(writer)
        try:
            await relays.relay(reader, writer, ureader, uwriter)
        except OSError as e:
            handler_logger.warning('Connection from %r caught %r', info, e) 

def test_reader_writer_echo(loop, sock_pair):
    """Verify readers and writers can send data to each other."""
    c_sock, s_sock = sock_pair

    @asyncio.coroutine
    def mycoro():
        c_reader, c_writer = yield from asyncio.open_connection(sock=c_sock)
        s_reader, s_writer = yield from asyncio.open_connection(sock=s_sock)

        data = b'Echo... Echo... Echo...'
        s_writer.write(data)
        yield from s_writer.drain()
        read_data = yield from c_reader.readexactly(len(data))
        assert data == read_data
        s_writer.close()

    loop.run_until_complete(asyncio.wait_for(mycoro(), timeout=1.0)) 

def connectTarget(self, addr, port, data):
        logging.info('requested %s <--> %s:%s' % (self.peer, addr, port))
        try:
            reader, writer = yield from open_connection(addr, port)
        except (ConnectionError, OSError, TimeoutError) as e:
            logging.info("can't connect to %s:%s (from %s)" % (addr, port, self.peer))
            return self.resetTunnel("can't connect to %s:%s" % (addr, port), str(e))
        self.setProxy(reader, writer)
        if data:
            writer.write(data)
        if self._dataToTarget:
            writer.write(self._dataToTarget)
            self._dataToTarget.clear()
        self.connectTargetTask = None

    # next 2 overrides deal with a implicit state which exists only in wstan server: CONNECTING
    # data received during CONNECTING will be sent after connected
    # IDLE --onConnect--> CONNECTING --connectTarget--> USING
    # CONNECTING --RST-received-and-RST-sent--> IDLE
    # CONNECTING --RST-sent--> RESETTING --RST-received--> IDLE 

def remote_sort():
    reader, writer = await asyncio.open_connection("127.0.0.1", 2015)
    print("Generating random list...")
    numbers = [random.randrange(10000) for r in range(10000)]
    data = json.dumps(numbers).encode()
    print("List Generated, Sending data")
    writer.write(len(data).to_bytes(8, "big"))
    writer.write(data)

    print("Waiting for data...")
    data = await reader.readexactly(len(data))
    print("Received data")
    sorted_values = json.loads(data.decode())
    print(sorted_values)
    print("\n")
    writer.close() 

def create_connection(repetitions):
    reader, writer = await asyncio.open_connection(
        host=HOST, port=PORT)

    start_time = float((await reader.readline()))

    writer.write(repetitions.encode() + b'\n')
    await writer.drain()

    async for line in reader:
        # Sleeping a little to emulate processing time and make
        # it easier to add more simultaneous clients
        await asyncio.sleep(1)

        printer(start_time, 'Got line: ', line.decode(),
                end='')

    writer.close() 

def remote_sort():
    reader, writer = yield from asyncio.open_connection('127.0.0.1', 2015)
    print("Generating random list...")
    numbers = [random.randrange(10000) for r in range(10000)]
    data = json.dumps(numbers).encode()
    print("List Generated, Sending data")
    writer.write(len(data).to_bytes(8, 'big'))
    writer.write(data)

    print("Waiting for data...")
    data = yield from reader.readexactly(len(data))
    print("Received data")
    sorted_values = json.loads(data.decode())
    print(sorted_values)
    print('\n')
    writer.close() 

def __init__(self, *, socket_timeout=None,
                 read_speed_limit=None, write_speed_limit=None,
                 path_timeout=None, path_io_factory=pathio.PathIO,
                 encoding="utf-8", ssl=None, parse_list_line_custom=None,
                 **siosocks_asyncio_kwargs):
        self.socket_timeout = socket_timeout
        self.throttle = StreamThrottle.from_limits(
            read_speed_limit,
            write_speed_limit,
        )
        self.path_timeout = path_timeout
        self.path_io = path_io_factory(timeout=path_timeout)
        self.encoding = encoding
        self.stream = None
        self.ssl = ssl
        self.parse_list_line_custom = parse_list_line_custom
        self._open_connection = partial(open_connection, ssl=self.ssl,
                                        **siosocks_asyncio_kwargs) 

def run(self):
		try:
			self.in_queue = asyncio.Queue()
			self.out_queue = asyncio.Queue()
			self.reader, self.writer = await asyncio.wait_for(
				asyncio.open_connection(
					self.target.serverip if self.target.serverip is not None else self.target.host, 
					self.target.port, 
					ssl=self.target.get_ssl_context()
					),
				timeout = self.target.timeout
			)

			self.handle_in_task = asyncio.create_task(self.handle_in_q())
			self.handle_out_task = asyncio.create_task(self.handle_out_q())
			return True, None
		except Exception as e:
			return False, e 

def connect(self, service, port, use_ssl):
        for _ in range(3):
            try:
                self.reader, self.writer = await \
                    asyncio.open_connection(service, port, ssl=use_ssl)
                return
            except ConnectionRefusedError:
                """ connection refused. Try again """
        raise ConnectionRefusedError(
            f'Connection refused to "{service}" on port {port}') 

def open(self):
        """Open TCP connection."""
        self._debug('connecting to {}'.format(self))
        # If a task is canceled while it is waiting for another concurrent operation,
        # the task is notified of its cancellation by having a CancelledError exception
        # raised at the point where it is waiting
        try:
            self._stream_reader, self._stream_writer = await asyncio.open_connection(host=self.host, port=self.port)
            # self._stream_reader, self._stream_writer = await asyncio.wait_for(asyncio.open_connection(host=self.host, port=self.port), timeout=10)
        except asyncio.CancelledError as err:
            self._debug("CancelledError while awaiting for open_connection({}), err: {}".format(self, err))
            # TODO: stop child task of asyncio.open_connection
            raise
        else:
            self.connection_lost = asyncio.Future()  # delayed to be created in same loop as open()
            asyncio.ensure_future(self.forward_connection_read_data())
        self._debug('connection {} is open'.format(self)) 

def connect(self):
        log.debug("Initial connection to server %s:%d", self.host, self.port)
        self.reader, self.writer = await asyncio.open_connection(self.host, self.port,
                                                                 loop=self.loop)

        try:
            await self._make_handshake()
        finally:
            self.writer.close()

        self.reader, self.writer = await asyncio.open_connection(self.host, self.port,
                                                                 loop=self.loop) 

def connect(self) -> Tuple[asyncio.StreamReader, asyncio.StreamWriter]:
        return await self.cancel_token.cancellable_wait(
            asyncio.open_connection(
                host=self.remote.address.ip, port=self.remote.address.tcp_port
            ),
            timeout=REPLY_TIMEOUT,
        ) 

def connect_to_slave(self, slave_info: SlaveInfo) -> str:
        """ Create a connection to a slave server.
        Returns empty str on success otherwise return the error message."""
        if slave_info.id == self.slave_server.id or slave_info.id in self.slave_ids:
            return ""

        host = slave_info.host.decode("ascii")
        port = slave_info.port
        try:
            reader, writer = await asyncio.open_connection(host, port, loop=self.loop)
        except Exception as e:
            err_msg = "Failed to connect {}:{} with exception {}".format(host, port, e)
            Logger.info(err_msg)
            return err_msg

        conn_name = "{}<->{}".format(
            self.slave_server.id.decode("ascii"), slave_info.id.decode("ascii")
        )
        slave = SlaveConnection(
            self.env,
            reader,
            writer,
            self.slave_server,
            slave_info.id,
            slave_info.full_shard_id_list,
            conn_name,
        )
        await slave.wait_until_active()
        # Tell the remote slave who I am
        id, full_shard_id_list = await slave.send_ping()
        # Verify that remote slave indeed has the id and shard mask list advertised by the master
        if id != slave.id:
            return "id does not match. expect {} got {}".format(slave.id, id)
        if full_shard_id_list != slave.full_shard_id_list:
            return "shard list does not match. expect {} got {}".format(
                slave.full_shard_id_list, full_shard_id_list
            )

        self._add_slave_connection(slave)
        return "" 

def connect(self, ip, port):
        Logger.info("connecting {} {}".format(ip, port))
        try:
            reader, writer = await asyncio.open_connection(ip, port, loop=self.loop)
        except Exception as e:
            Logger.info("failed to connect {} {}: {}".format(ip, port, e))
            return None
        peer = Peer(
            self.env,
            reader,
            writer,
            self,
            self.master_server,
            self.__get_next_cluster_peer_id(),
        )
        peer.send_hello()
        result = await peer.start(is_server=False)
        if result is not None:
            return None
        return peer 

def test_receive_tcp_message(self):
        # configure a TCPListener
        listener_addr = self._get_available_local_address()
        listener = self._create_tcp_listener(*listener_addr, self._handle_tcp_stream)
        await listener.start()

        # create a TCP connection
        reader, writer = await asyncio.open_connection(*listener_addr)

        # send message to listener
        writer.write(b'hello world')

        await asyncio.sleep(0.1)

        self.assertMessageReceived(b'hello world') 

def test_vnc_authentication(self):

    async def vnc_auth():
      reader, writer = await asyncio.open_connection(
          '127.0.0.1', 8888, loop=self.loop)
      # server rfb version
      _ = await reader.readline()
      writer.write(RFB_VERSION)

      # available auth methods
      _ = await reader.read(1024)
      writer.write(VNC_AUTH)

      # challenge
      _ = await reader.read(1024)
      # Pretending, that we encrypt received challenge with DES and send back the result.
      client_response = os.urandom(16)
      writer.write(client_response)

      # security result
      _ = await reader.read(1024)

    options = {'enabled': 'True', 'port': 8888, 'timeout': 30}
    capability = Vnc(options, self.loop)

    server_coro = asyncio.start_server(
        capability.handle_session, '0.0.0.0', 8888, loop=self.loop)
    self.server = self.loop.run_until_complete(server_coro)

    self.loop.run_until_complete(vnc_auth()) 

def test_socks_authentication(self):

    async def socks_auth():
      reader, writer = await asyncio.open_connection(
          '127.0.0.1', 8888, loop=self.loop)

      # Greeting to the server. version+authmethod number+authmethod
      client_greeting = socks.SOCKS_VERSION + b"\x01" + socks.AUTH_METHOD
      writer.write(client_greeting)

      # Receive version+chosen authmethod
      _ = await reader.read(2)

      # Send credentials.
      # version+username len+username+password len+password
      credentials = b"\x05\x08username\x08password"
      writer.write(credentials)

      # Receive authmethod+\xff
      res = await reader.read(2)
      self.assertEqual(res, socks.AUTH_METHOD + socks.SOCKS_FAIL)

    options = {'enabled': 'True', 'port': 8888, 'timeout': 30}
    capability = socks.Socks5(options, self.loop)

    server_coro = asyncio.start_server(
        capability.handle_session, '127.0.0.1', 8888, loop=self.loop)
    self.server = self.loop.run_until_complete(server_coro)
    self.loop.run_until_complete(socks_auth()) 

def create_soc(self):
		if self.soc_type == KerberosSocketType.TCP:
			self.reader, self.writer = await asyncio.open_connection(self.dst_ip, self.dst_port)
		
		elif self.soc_type == KerberosSocketType.UDP:
			raise Exception('UDP not implemented!')
			
		else:
			raise Exception('Unknown socket type!') 

def open_tg_connection(self, host, port, init_func=None):
        task = asyncio.open_connection(host, port, limit=get_to_clt_bufsize())
        reader_tgt, writer_tgt = await asyncio.wait_for(task, timeout=config.TG_CONNECT_TIMEOUT)

        set_keepalive(writer_tgt.get_extra_info("socket"))
        set_bufsizes(writer_tgt.get_extra_info("socket"), get_to_clt_bufsize(), get_to_tg_bufsize())

        if init_func:
            return await asyncio.wait_for(init_func(host, port, reader_tgt, writer_tgt),
                                          timeout=config.TG_CONNECT_TIMEOUT)
        return reader_tgt, writer_tgt 

def make_https_req(url, host="core.telegram.org"):
    """ Make request, return resp body and headers. """
    SSL_PORT = 443
    url_data = urllib.parse.urlparse(url)

    HTTP_REQ_TEMPLATE = "\r\n".join(["GET %s HTTP/1.1", "Host: %s",
                                     "Connection: close"]) + "\r\n\r\n"
    reader, writer = await asyncio.open_connection(url_data.netloc, SSL_PORT, ssl=True)
    req = HTTP_REQ_TEMPLATE % (urllib.parse.quote(url_data.path), host)
    writer.write(req.encode("utf8"))
    data = await reader.read()
    writer.close()

    headers, body = data.split(b"\r\n\r\n", 1)
    return headers, body 

def get_encrypted_cert(host, port, server_name):
    async def get_tls_record(reader):
        try:
            record_type = (await reader.readexactly(1))[0]
            tls_version = await reader.readexactly(2)
            if tls_version != b"\x03\x03":
                return 0, b""
            record_len = int.from_bytes(await reader.readexactly(2), "big")
            record = await reader.readexactly(record_len)

            return record_type, record
        except asyncio.IncompleteReadError:
            return 0, b""

    reader, writer = await asyncio.open_connection(host, port)
    writer.write(gen_tls_client_hello_msg(server_name))
    await writer.drain()

    record1_type, record1 = await get_tls_record(reader)
    if record1_type != 22:
        return b""

    record2_type, record2 = await get_tls_record(reader)
    if record2_type != 20:
        return b""

    record3_type, record3 = await get_tls_record(reader)
    if record3_type != 23:
        return b""

    return record3 

def _open_connection(self) -> None:
        """
        Opens a connection to the GPSD server and configures the TCP socket.
        """
        self.logger.info(
            f"Connecting to gpsd at {self.connection_args['host']}" +
            (f":{self.connection_args['port']}"
             if self.connection_args['port'] else ''))
        self.reader, self.writer = await asyncio.wait_for(
            asyncio.open_connection(**self.connection_args),
            self.connection_timeout,
            loop=self.loop)
        # Set socket options
        sock = self.writer.get_extra_info('socket')
        if sock is not None:
            if 'SO_KEEPALIVE' in self.alive_opts:
                sock.setsockopt(socket.SOL_SOCKET,
                                socket.SO_KEEPALIVE,
                                self.alive_opts['SO_KEEPALIVE'])
            if hasattr(
                    sock,
                    'TCP_KEEPIDLE') and 'TCP_KEEPIDLE' in self.alive_opts:
                sock.setsockopt(socket.IPPROTO_TCP,
                                socket.TCP_KEEPIDLE,    # pylint: disable=E1101
                                self.alive_opts['TCP_KEEPIDLE'])
            if hasattr(
                    sock,
                    'TCP_KEEPINTVL') and 'TCP_KEEPINTVL' in self.alive_opts:
                sock.setsockopt(socket.IPPROTO_TCP,
                                socket.TCP_KEEPINTVL,   # pylint: disable=E1101
                                self.alive_opts['TCP_KEEPINTVL'])
            if hasattr(
                    sock,
                    'TCP_KEEPCNT') and 'TCP_KEEPCNT' in self.alive_opts:
                sock.setsockopt(socket.IPPROTO_TCP,
                                socket.TCP_KEEPCNT,
                                self.alive_opts['TCP_KEEPCNT']) 

def connect(self) -> bool:
        """ Establish a long running connection to EPMD, will not return until
            the connection has been established.

            :rtype: True if connection succeeded within 5 tries, else False
        """
        # Try to connect for N tries then fail
        for n_try in range(self.n_connection_attempts_):
            try:
                LOG.info("Connecting to EPMD %s:%d", self.host_, self.port_)

                self.reader_, self.writer_ = await asyncio.open_connection(
                    host=self.host_,
                    port=self.port_
                )
                LOG.info("EPMD socket connected")
                return True

            except Exception as err:
                LOG.error("EPMD connection error %s. Is local EPMD running? "
                          "Try `epmd -daemon`", err)
                await asyncio.sleep(5.0)

        LOG.error("Could not connect to EPMD in %d tries" %
                  self.n_connection_attempts_)
        return False 

def _fire_forget_query(ip: str, query: bytes) -> bytes:
        """ Connect to node, fire the query, read and disconnect. """
        try:
            reader, writer = await asyncio.open_connection(
                host=ip,
                port=EPMD_DEFAULT_PORT,
                # timeout=EPMD_REMOTE_DEFAULT_TIMEOUT
            )
        except Exception as err:
            LOG.error(str(err))
            raise err

        query1 = util.to_u16(len(query)) + query
        writer.write(query1)

        # Expect that after everything is received, the peer will close
        # the socket automatically, so we will too
        result = b''
        while True:
            incoming = await reader.read(4096)
            if incoming == b'':
                break

            result += incoming

        writer.close()
        return result