def __init__(self, io_loop, client, request, callback):
        self.start_time = time.time()
        self.io_loop = io_loop
        self.client = client
        self.request = request
        self.callback = callback
        self.code = None
        self.headers = None
        self.chunks = None
        self._decompressor = None
        # Timeout handle returned by IOLoop.add_timeout
        self._timeout = None
        with stack_context.StackContext(self.cleanup):
            parsed = urlparse.urlsplit(_unicode(self.request.url))
            # urlsplit results have hostname and port results, but they
            # didn't support ipv6 literals until python 2.7.
            netloc = parsed.netloc
            if "@" in netloc:
                userpass, _, netloc = netloc.rpartition("@")
            match = re.match(r"^(.+):(\d+)$", netloc)
            if match:
                host = match.group(1)
                port = int(match.group(2))
            else:
                host = netloc
                port = 443 if parsed.scheme == "https" else 80
            if re.match(r"^\[.*\]$", host):
                # raw ipv6 addresses in urls are enclosed in brackets
                host = host[1:-1]
            if self.client.hostname_mapping is not None:
                host = self.client.hostname_mapping.get(host, host)

            if request.allow_ipv6:
                af = socket.AF_UNSPEC
            else:
                # We only try the first IP we get from getaddrinfo,
                # so restrict to ipv4 by default.
                af = socket.AF_INET

            addrinfo = socket.getaddrinfo(host, port, af, socket.SOCK_STREAM, 0, 0)
            af, socktype, proto, canonname, sockaddr = addrinfo[0]

            if parsed.scheme == "https":
                ssl_options = {}
                if request.validate_cert:
                    ssl_options["cert_reqs"] = ssl.CERT_REQUIRED
                if request.ca_certs is not None:
                    ssl_options["ca_certs"] = request.ca_certs
                else:
                    ssl_options["ca_certs"] = _DEFAULT_CA_CERTS
                self.stream = SSLIOStream(
                    socket.socket(af, socktype, proto), io_loop=self.io_loop, ssl_options=ssl_options
                )
            else:
                self.stream = IOStream(socket.socket(af, socktype, proto), io_loop=self.io_loop)
            timeout = min(request.connect_timeout, request.request_timeout)
            if timeout:
                self._connect_timeout = self.io_loop.add_timeout(self.start_time + timeout, self._on_timeout)
            self.stream.set_close_callback(self._on_close)
            self.stream.connect(sockaddr, functools.partial(self._on_connect, parsed))

    def __init__(self, io_loop, request, callback):
        self.start_time = time.time()
        self.io_loop = io_loop
        self.request = request
        self.callback = callback
        self.code = None
        self.headers = None
        self.chunks = None
        self._decompressor = None
        # Timeout handle returned by IOLoop.add_timeout
        self._timeout = None
        with stack_context.StackContext(self.cleanup):
            parsed = urlparse.urlsplit(self.request.url)
            if ":" in parsed.netloc:
                host, _, port = parsed.netloc.partition(":")
                port = int(port)
            else:
                host = parsed.netloc
                port = 443 if parsed.scheme == "https" else 80

            if parsed.scheme == "https":
                # TODO: cert verification, etc
                self.stream = SSLIOStream(socket.socket(),
                                          io_loop=self.io_loop)
            else:
                self.stream = IOStream(socket.socket(),
                                       io_loop=self.io_loop)
            timeout = min(request.connect_timeout, request.request_timeout)
            if timeout:
                self._connect_timeout = self.io_loop.add_timeout(
                    self.start_time + timeout,
                    self._on_timeout)
            self.stream.connect((host, port),
                                functools.partial(self._on_connect, parsed))

class TCPClient(object):
    def __init__(self, io_loop=None):
        self.io_loop = self.io_loop = io_loop or IOLoop.current()

        #self.shutdown = False
        self.sock_fd = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
        self.sock_fd.settimeout(0.5)
        self.stream = IOStream(self.sock_fd)
        #self.stream.set_close_callback(self.on_close)

    def connect(self, host, port):
        #self.stream.connect((self.host, self.port), self.send_message)
        self.stream.connect((host, port))
        return self.stream

    @return_future
    def connect_server(self, host, port, callback=None):
        self.stream.connect((host, port), callback=callback)

    def on_close(self):
        if self.shutdown:
            self.io_loop.stop()

    def set_shutdown(self):
        self.shutdown = True

    def __init__(self, io_loop, request, callback):
        self.io_loop = io_loop
        self.request = request
        self.callback = callback
        self.code = None
        self.headers = None
        self.chunks = None
        with stack_context.StackContext(self.cleanup):
            parsed = urlparse.urlsplit(self.request.url)
            if ":" in parsed.netloc:
                host, _, port = parsed.netloc.partition(":")
                port = int(port)
            else:
                host = parsed.netloc
                port = 443 if parsed.scheme == "https" else 80

            if parsed.scheme == "https":
                # TODO: cert verification, etc
                self.stream = SSLIOStream(socket.socket(),
                                          io_loop=self.io_loop)
            else:
                self.stream = IOStream(socket.socket(),
                                       io_loop=self.io_loop)
            self.stream.connect((host, port),
                                functools.partial(self._on_connect, parsed))

 def test_empty_request(self):
     stream = IOStream(socket.socket(), io_loop=self.io_loop)
     stream.connect(("localhost", self.get_http_port()), self.stop)
     self.wait()
     stream.close()
     self.io_loop.add_timeout(datetime.timedelta(seconds=0.001), self.stop)
     self.wait()

    def connect(self):
        self._loop = IOLoop.current()
        try:
            if self.unix_socket and self.host in ("localhost", "127.0.0.1"):
                sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
                self.host_info = "Localhost via UNIX socket"
                address = self.unix_socket
            else:
                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
                self.host_info = "socket %s:%d" % (self.host, self.port)
                address = (self.host, self.port)
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
            if self.no_delay:
                sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
            sock = IOStream(sock)

            child_gr = greenlet.getcurrent()
            main = child_gr.parent
            assert main is not None, "Execut must be running in child greenlet"

            if self.connect_timeout:

                def timeout():
                    if not self.socket:
                        sock.close((None, IOError("connection timeout")))

                self._loop.call_later(self.connect_timeout, timeout)

            def connected(future):
                if future._exc_info is not None:
                    child_gr.throw(future.exception())
                else:
                    self.socket = sock
                    child_gr.switch()

            future = sock.connect(address)
            self._loop.add_future(future, connected)
            main.switch()

            self._rfile = self.socket
            self._get_server_information()
            self._request_authentication()

            if self.sql_mode is not None:
                c = self.cursor()
                c.execute("SET sql_mode=%s", (self.sql_mode,))

            if self.init_command is not None:
                c = self.cursor()
                c.execute(self.init_command)
                self.commit()

            if self.autocommit_mode is not None:
                self.autocommit(self.autocommit_mode)
        except Exception as e:
            if self.socket:
                self._rfile = None
                self.socket.close()
                self.socket = None
            raise err.OperationalError(2003, "Can't connect to MySQL server on %s:%s (%s)" % (self.host, self.port, e))

 def _create_stream(self, max_buffer_size, af, addr, source_ip=None,
                    source_port=None):
     # Always connect in plaintext; we'll convert to ssl if necessary
     # after one connection has completed.
     source_port_bind = source_port if isinstance(source_port, int) else 0
     source_ip_bind = source_ip
     if source_port_bind and not source_ip:
         # User required a specific port, but did not specify
         # a certain source IP, will bind to the default loopback.
         source_ip_bind = '::1' if af == socket.AF_INET6 else '127.0.0.1'
         # Trying to use the same address family as the requested af socket:
         # - 127.0.0.1 for IPv4
         # - ::1 for IPv6
     socket_obj = socket.socket(af)
     set_close_exec(socket_obj.fileno())
     if source_port_bind or source_ip_bind:
         # If the user requires binding also to a specific IP/port.
         try:
             socket_obj.bind((source_ip_bind, source_port_bind))
         except socket.error:
             socket_obj.close()
             # Fail loudly if unable to use the IP/port.
             raise
     try:
         stream = IOStream(socket_obj,
                           max_buffer_size=max_buffer_size)
     except socket.error as e:
         fu = Future()
         fu.set_exception(e)
         return fu
     else:
         return stream, stream.connect(addr)

    def connect(self):
        # Creates and returns a connection object for use.

        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
        stream = IOStream(sock)
        stream.connect((self.host, self.port))
        return self.connection(stream, sock.getsockname(), self.data, self.terminator)

    def test_indexing_line(self):
        client = AsyncHTTPClient(io_loop=self.io_loop)
        ping = yield client.fetch("http://localhost:9200/", raise_error=False)
        if ping.code != 200 or decode(ping)['tagline'] != "You Know, for Search":
            self.fail("This test requires an ES instance running on localhost")

        #delete if existing
        url = "http://localhost:9200/{}/".format(CONFIG['indexer_config']['index_pattern'])
        resp = yield client.fetch(url, method='DELETE', headers=None, raise_error=False)

        app = App(CONFIG)
        app.run()

        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
        stream = IOStream(s)
        yield stream.connect(("localhost", 8888))
        yield stream.write(b"My name is Yuri and I'm 6 years old.\n")

        yield tornado.gen.sleep(MAX_TIMEOUT-2)
        url = "http://localhost:9200/{}/_search?q=*:*".format(CONFIG['indexer_config']['index_pattern'])
        resp = yield client.fetch(url, method='GET', headers=None, raise_error=False)
        resp_hits = json.loads(resp.body.decode('utf-8'))['hits']['hits']
        self.assertEqual(len(FindIn(resp).by(name='Yuri')), 1)
        doc = resp_hits[0]['_source']
        self.assertEqual(doc['@version'], 1)
        self.assertEqual(doc['message'], "My name is Yuri and I'm 6 years old.")

 def test_gaierror(self):
     # Test that IOStream sets its exc_info on getaddrinfo error
     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
     stream = IOStream(s, io_loop=self.io_loop)
     stream.set_close_callback(self.stop)
     stream.connect(('adomainthatdoesntexist.asdf', 54321))
     self.assertTrue(isinstance(stream.error, socket.gaierror), stream.error)

class ForwardConnection(object):

    def __init__(self, remote_address, stream, address, headers):
        self.remote_address = remote_address
        self.stream = stream
        self.address = address
        self.headers = headers
        sock = socket.socket()
        self.remote_stream = IOStream(sock)
        self.remote_stream.connect(self.remote_address, self._on_remote_connected)    
        self.remote_stream.set_close_callback(self._on_close)    

    def _on_remote_write_complete(self):
        logging.info('send request to %s', self.remote_address)
        self.remote_stream.read_until_close(self._on_remote_read_close)

    def _on_remote_connected(self):
        logging.info('forward %r to %r', self.address, self.remote_address)
        self.remote_stream.write(self.headers, self._on_remote_write_complete)

    def _on_remote_read_close(self, data):
        self.stream.write(data, self.stream.close)

    def _on_close(self):
        logging.info('remote quit %s', self.remote_address)
        self.remote_stream.close()

    def test_handle_stream_coroutine_logging(self):
        # handle_stream may be a coroutine and any exception in its
        # Future will be logged.
        class TestServer(TCPServer):
            @gen.coroutine
            def handle_stream(self, stream, address):
                yield gen.moment
                stream.close()
                1 / 0

        server = client = None
        try:
            sock, port = bind_unused_port()
            with NullContext():
                server = TestServer()
                server.add_socket(sock)
            client = IOStream(socket.socket())
            with ExpectLog(app_log, "Exception in callback"):
                yield client.connect(('localhost', port))
                yield client.read_until_close()
                yield gen.moment
        finally:
            if server is not None:
                server.stop()
            if client is not None:
                client.close()

class Connection(object):
    def __init__(self, host, port, event_handler,
                 stop_after=None, io_loop=None):
        self.host = host
        self.port = port
        self._event_handler = weakref.proxy(event_handler)
        self.timeout = stop_after
        self._stream = None
        self._io_loop = io_loop
        self.try_left = 2

        self.in_progress = False
        self.read_queue = []
        self.read_callbacks = []

    def __del__(self):
        self.disconnect()

    def connect(self):
        if not self._stream:
            try:
                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
                sock.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1)
                sock.settimeout(self.timeout)
                sock.connect((self.host, self.port))
                self._stream = IOStream(sock, io_loop=self._io_loop)
                self._stream.set_close_callback(self.on_stream_close)
                self.connected()
            except socket.error, e:
                raise ConnectionError(str(e))
            self.fire_event('on_connect')

    def execute(self, cmd):
        """ Executes `cmd` on host and returns results

        Creates socket and tries to execute command against zookeeper. Socket
        is limited by quasi-Tornado's timeout. It doesn't check validity of response.

        Note:
            Timeout should be implemented using tornado.concurrent.chain_future:
            https://github.com/tornadoweb/tornado/blob/master/tornado/concurrent.py#L316

            such a wrapper exists in Tornado 4.0+ - with_timeout
            https://github.com/tornadoweb/tornado/blob/master/tornado/gen.py#L507

        Args:
            cmd: Four-letter string containing command to execute
        Returns:
            Raw response - bytes.
        Raises:
            HostConnectionTimeout: If sum times of connection, request, respons exceeds timeout
            Socket Errors: like ECONNNECTIONREFUSED,...
        """

        ioloop = IOLoop.current()
        address_family, addr = yield self._resolve(ioloop)
        stream = IOStream(socket.socket(address_family), io_loop=ioloop)
        stream.connect(addr)
        cmd = '{}\n'.format(cmd.strip())
        yield gen.Task(stream.write, cmd.encode('utf-8'))
        data = yield gen.Task(stream.read_until_close)
        raise gen.Return(data)

class HTTPServerRawTest(AsyncHTTPTestCase):
    def get_app(self):
        return Application([("/echo", EchoHandler)])

    def setUp(self):
        super(HTTPServerRawTest, self).setUp()
        self.stream = IOStream(socket.socket())
        self.stream.connect(("localhost", self.get_http_port()), self.stop)
        self.wait()

    def tearDown(self):
        self.stream.close()
        super(HTTPServerRawTest, self).tearDown()

    def test_empty_request(self):
        self.stream.close()
        self.io_loop.add_timeout(datetime.timedelta(seconds=0.001), self.stop)
        self.wait()

    def test_malformed_first_line(self):
        with ExpectLog(gen_log, ".*Malformed HTTP request line"):
            self.stream.write(b"asdf\r\n\r\n")
            # TODO: need an async version of ExpectLog so we don't need
            # hard-coded timeouts here.
            self.io_loop.add_timeout(datetime.timedelta(seconds=0.01), self.stop)
            self.wait()

    def test_malformed_headers(self):
        with ExpectLog(gen_log, ".*Malformed HTTP headers"):
            self.stream.write(b"GET / HTTP/1.0\r\nasdf\r\n\r\n")
            self.io_loop.add_timeout(datetime.timedelta(seconds=0.01), self.stop)
            self.wait()

    def test_connection_refused(self):
        # When a connection is refused, the connect callback should not
        # be run.  (The kqueue IOLoop used to behave differently from the
        # epoll IOLoop in this respect)
        cleanup_func, port = refusing_port()
        self.addCleanup(cleanup_func)
        stream = IOStream(socket.socket(), self.io_loop)
        self.connect_called = False

        def connect_callback():
            self.connect_called = True
            self.stop()

        stream.set_close_callback(self.stop)
        # log messages vary by platform and ioloop implementation
        with ExpectLog(gen_log, ".*", required=False):
            stream.connect(("127.0.0.1", port), connect_callback)
            self.wait()
        self.assertFalse(self.connect_called)
        self.assertTrue(isinstance(stream.error, socket.error), stream.error)
        if sys.platform != "cygwin":
            _ERRNO_CONNREFUSED = (errno.ECONNREFUSED,)
            if hasattr(errno, "WSAECONNREFUSED"):
                _ERRNO_CONNREFUSED += (errno.WSAECONNREFUSED,)
            # cygwin's errnos don't match those used on native windows python
            self.assertTrue(stream.error.args[0] in _ERRNO_CONNREFUSED)

 def _create_stream(self, max_buffer_size, af, addr):
     # Always connect in plaintext; we'll convert to ssl if necessary
     # after one connection has completed.
     stream = IOStream(socket.socket(af),
                       io_loop=self.io_loop,
                       max_buffer_size=max_buffer_size)
     return stream.connect(addr)

class IRCStream(object):
    """
    A connection to an IRC server utilizing IOStream
    """
    def __init__(self, nick, url, io_loop=None):
        self.nick = nick
        self.url = url
        self.io_loop = io_loop or IOLoop.instance()

        parsed = urlparse.urlsplit(self.url)
        assert parsed.scheme == 'irc'
        if ':' in parsed.netloc:
            host, _, port = parsed.netloc.partition(':')
            port = int(port)
        else:
            host = parsed.netloc
            port = 6667
        self.host = host
        self.port = port

    def connect(self, callback):
        self.stream = IOStream(socket.socket(), io_loop=self.io_loop)
        self.stream.connect((self.host, self.port),
                            functools.partial(self._on_connect, callback))


    def _on_connect(self, callback):
        self.stream.write('NICK %s\r\n' % self.nick)
        callback(True)

def handle_connection(connection, address):
    log.info("Connection received from %s" % str(address))
    stream = IOStream(connection, ioloop, max_buffer_size=1024 * 1024 * 1024)
    # Getting uuid
    try:
        stream.read_bytes(4, partial(read_uuid_size, stream))
    except StreamClosedError:
        log.warn("Closed stream for getting uuid length")

 def initiate(cls, host, port, infohash):
     af = socket.AF_INET
     addrinfo = socket.getaddrinfo(host, port, af, socket.SOCK_STREAM,
                                       0, 0)
     af, socktype, proto, canonname, sockaddr = addrinfo[0]
     stream = IOStream(socket.socket(af, socktype, proto),
                            io_loop=cls.io_loop)
     stream.connect(sockaddr, functools.partial(cls.initiate_connected, stream, sockaddr, infohash))