from tornado.concurrent import Future
from tornado import gen
from tornado import netutil
from tornado.iostream import (
    IOStream,
    SSLIOStream,
    PipeIOStream,
    StreamClosedError,
    _StreamBuffer,
)
from tornado.httputil import HTTPHeaders
from tornado.locks import Condition, Event
from tornado.log import gen_log
from tornado.netutil import ssl_wrap_socket
from tornado.tcpserver import TCPServer
from tornado.testing import (
    AsyncHTTPTestCase,
    AsyncHTTPSTestCase,
    AsyncTestCase,
    bind_unused_port,
    ExpectLog,
    gen_test,
)
from tornado.test.util import skipIfNonUnix, refusing_port, skipPypy3V58
from tornado.web import RequestHandler, Application
import errno
import hashlib
import os
import platform
import random
import socket
import ssl
import sys
from unittest import mock
import unittest


def _server_ssl_options():
    return dict(
        certfile=os.path.join(os.path.dirname(__file__), "test.crt"),
        keyfile=os.path.join(os.path.dirname(__file__), "test.key"),
    )


class HelloHandler(RequestHandler):
    def get(self):
        self.write("Hello")


class TestIOStreamWebMixin(object):
    def _make_client_iostream(self):
        raise NotImplementedError()

    def get_app(self):
        return Application([("/", HelloHandler)])

    def test_connection_closed(self):
        # When a server sends a response and then closes the connection,
        # the client must be allowed to read the data before the IOStream
        # closes itself.  Epoll reports closed connections with a separate
        # EPOLLRDHUP event delivered at the same time as the read event,
        # while kqueue reports them as a second read/write event with an EOF
        # flag.
        response = self.fetch("/", headers={"Connection": "close"})
        response.rethrow()

    @gen_test
    def test_read_until_close(self):
        stream = self._make_client_iostream()
        yield stream.connect(("127.0.0.1", self.get_http_port()))
        stream.write(b"GET / HTTP/1.0\r\n\r\n")

        data = yield stream.read_until_close()
        self.assertTrue(data.startswith(b"HTTP/1.1 200"))
        self.assertTrue(data.endswith(b"Hello"))

    @gen_test
    def test_read_zero_bytes(self):
        self.stream = self._make_client_iostream()
        yield self.stream.connect(("127.0.0.1", self.get_http_port()))
        self.stream.write(b"GET / HTTP/1.0\r\n\r\n")

        # normal read
        data = yield self.stream.read_bytes(9)
        self.assertEqual(data, b"HTTP/1.1 ")

        # zero bytes
        data = yield self.stream.read_bytes(0)
        self.assertEqual(data, b"")

        # another normal read
        data = yield self.stream.read_bytes(3)
        self.assertEqual(data, b"200")

        self.stream.close()

    @gen_test
    def test_write_while_connecting(self):
        stream = self._make_client_iostream()
        connect_fut = stream.connect(("127.0.0.1", self.get_http_port()))
        # unlike the previous tests, try to write before the connection
        # is complete.
        write_fut = stream.write(b"GET / HTTP/1.0\r\nConnection: close\r\n\r\n")
        self.assertFalse(connect_fut.done())

        # connect will always complete before write.
        it = gen.WaitIterator(connect_fut, write_fut)
        resolved_order = []
        while not it.done():
            yield it.next()
            resolved_order.append(it.current_future)
        self.assertEqual(resolved_order, [connect_fut, write_fut])

        data = yield stream.read_until_close()
        self.assertTrue(data.endswith(b"Hello"))

        stream.close()

    @gen_test
    def test_future_interface(self):
        """Basic test of IOStream's ability to return Futures."""
        stream = self._make_client_iostream()
        connect_result = yield stream.connect(("127.0.0.1", self.get_http_port()))
        self.assertIs(connect_result, stream)
        yield stream.write(b"GET / HTTP/1.0\r\n\r\n")
        first_line = yield stream.read_until(b"\r\n")
        self.assertEqual(first_line, b"HTTP/1.1 200 OK\r\n")
        # callback=None is equivalent to no callback.
        header_data = yield stream.read_until(b"\r\n\r\n")
        headers = HTTPHeaders.parse(header_data.decode("latin1"))
        content_length = int(headers["Content-Length"])
        body = yield stream.read_bytes(content_length)
        self.assertEqual(body, b"Hello")
        stream.close()

    @gen_test
    def test_future_close_while_reading(self):
        stream = self._make_client_iostream()
        yield stream.connect(("127.0.0.1", self.get_http_port()))
        yield stream.write(b"GET / HTTP/1.0\r\n\r\n")
        with self.assertRaises(StreamClosedError):
            yield stream.read_bytes(1024 * 1024)
        stream.close()

    @gen_test
    def test_future_read_until_close(self):
        # Ensure that the data comes through before the StreamClosedError.
        stream = self._make_client_iostream()
        yield stream.connect(("127.0.0.1", self.get_http_port()))
        yield stream.write(b"GET / HTTP/1.0\r\nConnection: close\r\n\r\n")
        yield stream.read_until(b"\r\n\r\n")
        body = yield stream.read_until_close()
        self.assertEqual(body, b"Hello")

        # Nothing else to read; the error comes immediately without waiting
        # for yield.
        with self.assertRaises(StreamClosedError):
            stream.read_bytes(1)


class TestReadWriteMixin(object):
    # Tests where one stream reads and the other writes.
    # These should work for BaseIOStream implementations.

    def make_iostream_pair(self, **kwargs):
        raise NotImplementedError

    @gen_test
    def test_write_zero_bytes(self):
        # Attempting to write zero bytes should run the callback without
        # going into an infinite loop.
        rs, ws = yield self.make_iostream_pair()
        yield ws.write(b"")
        ws.close()
        rs.close()

    @gen_test
    def test_future_delayed_close_callback(self):
        # Same as test_delayed_close_callback, but with the future interface.
        rs, ws = yield self.make_iostream_pair()

        try:
            ws.write(b"12")
            chunks = []
            chunks.append((yield rs.read_bytes(1)))
            ws.close()
            chunks.append((yield rs.read_bytes(1)))
            self.assertEqual(chunks, [b"1", b"2"])
        finally:
            ws.close()
            rs.close()

    @gen_test
    def test_close_buffered_data(self):
        # Similar to the previous test, but with data stored in the OS's
        # socket buffers instead of the IOStream's read buffer.  Out-of-band
        # close notifications must be delayed until all data has been
        # drained into the IOStream buffer. (epoll used to use out-of-band
        # close events with EPOLLRDHUP, but no longer)
        #
        # This depends on the read_chunk_size being smaller than the
        # OS socket buffer, so make it small.
        rs, ws = yield self.make_iostream_pair(read_chunk_size=256)
        try:
            ws.write(b"A" * 512)
            data = yield rs.read_bytes(256)
            self.assertEqual(b"A" * 256, data)
            ws.close()
            # Allow the close to propagate to the `rs` side of the
            # connection.  Using add_callback instead of add_timeout
            # doesn't seem to work, even with multiple iterations
            yield gen.sleep(0.01)
            data = yield rs.read_bytes(256)
            self.assertEqual(b"A" * 256, data)
        finally:
            ws.close()
            rs.close()

    @gen_test
    def test_read_until_close_after_close(self):
        # Similar to test_delayed_close_callback, but read_until_close takes
        # a separate code path so test it separately.
        rs, ws = yield self.make_iostream_pair()
        try:
            ws.write(b"1234")
            ws.close()
            # Read one byte to make sure the client has received the data.
            # It won't run the close callback as long as there is more buffered
            # data that could satisfy a later read.
            data = yield rs.read_bytes(1)
            self.assertEqual(data, b"1")
            data = yield rs.read_until_close()
            self.assertEqual(data, b"234")
        finally:
            ws.close()
            rs.close()

    @gen_test
    def test_large_read_until(self):
        # Performance test: read_until used to have a quadratic component
        # so a read_until of 4MB would take 8 seconds; now it takes 0.25
        # seconds.
        rs, ws = yield self.make_iostream_pair()
        try:
            # This test fails on pypy with ssl.  I think it's because
            # pypy's gc defeats moves objects, breaking the
            # "frozen write buffer" assumption.
            if (
                isinstance(rs, SSLIOStream)
                and platform.python_implementation() == "PyPy"
            ):
                raise unittest.SkipTest("pypy gc causes problems with openssl")
            NUM_KB = 4096
            for i in range(NUM_KB):
                ws.write(b"A" * 1024)
            ws.write(b"\r\n")
            data = yield rs.read_until(b"\r\n")
            self.assertEqual(len(data), NUM_KB * 1024 + 2)
        finally:
            ws.close()
            rs.close()

    @gen_test
    def test_close_callback_with_pending_read(self):
        # Regression test for a bug that was introduced in 2.3
        # where the IOStream._close_callback would never be called
        # if there were pending reads.
        OK = b"OK\r\n"
        rs, ws = yield self.make_iostream_pair()
        event = Event()
        rs.set_close_callback(event.set)
        try:
            ws.write(OK)
            res = yield rs.read_until(b"\r\n")
            self.assertEqual(res, OK)

            ws.close()
            rs.read_until(b"\r\n")
            # If _close_callback (self.stop) is not called,
            # an AssertionError: Async operation timed out after 5 seconds
            # will be raised.
            yield event.wait()
        finally:
            ws.close()
            rs.close()

    @gen_test
    def test_future_close_callback(self):
        # Regression test for interaction between the Future read interfaces
        # and IOStream._maybe_add_error_listener.
        rs, ws = yield self.make_iostream_pair()
        closed = [False]
        cond = Condition()

        def close_callback():
            closed[0] = True
            cond.notify()

        rs.set_close_callback(close_callback)
        try:
            ws.write(b"a")
            res = yield rs.read_bytes(1)
            self.assertEqual(res, b"a")
            self.assertFalse(closed[0])
            ws.close()
            yield cond.wait()
            self.assertTrue(closed[0])
        finally:
            rs.close()
            ws.close()

    @gen_test
    def test_write_memoryview(self):
        rs, ws = yield self.make_iostream_pair()
        try:
            fut = rs.read_bytes(4)
            ws.write(memoryview(b"hello"))
            data = yield fut
            self.assertEqual(data, b"hell")
        finally:
            ws.close()
            rs.close()

    @gen_test
    def test_read_bytes_partial(self):
        rs, ws = yield self.make_iostream_pair()
        try:
            # Ask for more than is available with partial=True
            fut = rs.read_bytes(50, partial=True)
            ws.write(b"hello")
            data = yield fut
            self.assertEqual(data, b"hello")

            # Ask for less than what is available; num_bytes is still
            # respected.
            fut = rs.read_bytes(3, partial=True)
            ws.write(b"world")
            data = yield fut
            self.assertEqual(data, b"wor")

            # Partial reads won't return an empty string, but read_bytes(0)
            # will.
            data = yield rs.read_bytes(0, partial=True)
            self.assertEqual(data, b"")
        finally:
            ws.close()
            rs.close()

    @gen_test
    def test_read_until_max_bytes(self):
        rs, ws = yield self.make_iostream_pair()
        closed = Event()
        rs.set_close_callback(closed.set)
        try:
            # Extra room under the limit
            fut = rs.read_until(b"def", max_bytes=50)
            ws.write(b"abcdef")
            data = yield fut
            self.assertEqual(data, b"abcdef")

            # Just enough space
            fut = rs.read_until(b"def", max_bytes=6)
            ws.write(b"abcdef")
            data = yield fut
            self.assertEqual(data, b"abcdef")

            # Not enough space, but we don't know it until all we can do is
            # log a warning and close the connection.
            with ExpectLog(gen_log, "Unsatisfiable read"):
                fut = rs.read_until(b"def", max_bytes=5)
                ws.write(b"123456")
                yield closed.wait()
        finally:
            ws.close()
            rs.close()

    @gen_test
    def test_read_until_max_bytes_inline(self):
        rs, ws = yield self.make_iostream_pair()
        closed = Event()
        rs.set_close_callback(closed.set)
        try:
            # Similar to the error case in the previous test, but the
            # ws writes first so rs reads are satisfied
            # inline.  For consistency with the out-of-line case, we
            # do not raise the error synchronously.
            ws.write(b"123456")
            with ExpectLog(gen_log, "Unsatisfiable read"):
                with self.assertRaises(StreamClosedError):
                    yield rs.read_until(b"def", max_bytes=5)
            yield closed.wait()
        finally:
            ws.close()
            rs.close()

    @gen_test
    def test_read_until_max_bytes_ignores_extra(self):
        rs, ws = yield self.make_iostream_pair()
        closed = Event()
        rs.set_close_callback(closed.set)
        try:
            # Even though data that matches arrives the same packet that
            # puts us over the limit, we fail the request because it was not
            # found within the limit.
            ws.write(b"abcdef")
            with ExpectLog(gen_log, "Unsatisfiable read"):
                rs.read_until(b"def", max_bytes=5)
                yield closed.wait()
        finally:
            ws.close()
            rs.close()

    @gen_test
    def test_read_until_regex_max_bytes(self):
        rs, ws = yield self.make_iostream_pair()
        closed = Event()
        rs.set_close_callback(closed.set)
        try:
            # Extra room under the limit
            fut = rs.read_until_regex(b"def", max_bytes=50)
            ws.write(b"abcdef")
            data = yield fut
            self.assertEqual(data, b"abcdef")

            # Just enough space
            fut = rs.read_until_regex(b"def", max_bytes=6)
            ws.write(b"abcdef")
            data = yield fut
            self.assertEqual(data, b"abcdef")

            # Not enough space, but we don't know it until all we can do is
            # log a warning and close the connection.
            with ExpectLog(gen_log, "Unsatisfiable read"):
                rs.read_until_regex(b"def", max_bytes=5)
                ws.write(b"123456")
                yield closed.wait()
        finally:
            ws.close()
            rs.close()

    @gen_test
    def test_read_until_regex_max_bytes_inline(self):
        rs, ws = yield self.make_iostream_pair()
        closed = Event()
        rs.set_close_callback(closed.set)
        try:
            # Similar to the error case in the previous test, but the
            # ws writes first so rs reads are satisfied
            # inline.  For consistency with the out-of-line case, we
            # do not raise the error synchronously.
            ws.write(b"123456")
            with ExpectLog(gen_log, "Unsatisfiable read"):
                rs.read_until_regex(b"def", max_bytes=5)
                yield closed.wait()
        finally:
            ws.close()
            rs.close()

    @gen_test
    def test_read_until_regex_max_bytes_ignores_extra(self):
        rs, ws = yield self.make_iostream_pair()
        closed = Event()
        rs.set_close_callback(closed.set)
        try:
            # Even though data that matches arrives the same packet that
            # puts us over the limit, we fail the request because it was not
            # found within the limit.
            ws.write(b"abcdef")
            with ExpectLog(gen_log, "Unsatisfiable read"):
                rs.read_until_regex(b"def", max_bytes=5)
                yield closed.wait()
        finally:
            ws.close()
            rs.close()

    @gen_test
    def test_small_reads_from_large_buffer(self):
        # 10KB buffer size, 100KB available to read.
        # Read 1KB at a time and make sure that the buffer is not eagerly
        # filled.
        rs, ws = yield self.make_iostream_pair(max_buffer_size=10 * 1024)
        try:
            ws.write(b"a" * 1024 * 100)
            for i in range(100):
                data = yield rs.read_bytes(1024)
                self.assertEqual(data, b"a" * 1024)
        finally:
            ws.close()
            rs.close()

    @gen_test
    def test_small_read_untils_from_large_buffer(self):
        # 10KB buffer size, 100KB available to read.
        # Read 1KB at a time and make sure that the buffer is not eagerly
        # filled.
        rs, ws = yield self.make_iostream_pair(max_buffer_size=10 * 1024)
        try:
            ws.write((b"a" * 1023 + b"\n") * 100)
            for i in range(100):
                data = yield rs.read_until(b"\n", max_bytes=4096)
                self.assertEqual(data, b"a" * 1023 + b"\n")
        finally:
            ws.close()
            rs.close()

    @gen_test
    def test_flow_control(self):
        MB = 1024 * 1024
        rs, ws = yield self.make_iostream_pair(max_buffer_size=5 * MB)
        try:
            # Client writes more than the rs will accept.
            ws.write(b"a" * 10 * MB)
            # The rs pauses while reading.
            yield rs.read_bytes(MB)
            yield gen.sleep(0.1)
            # The ws's writes have been blocked; the rs can
            # continue to read gradually.
            for i in range(9):
                yield rs.read_bytes(MB)
        finally:
            rs.close()
            ws.close()

    @gen_test
    def test_read_into(self):
        rs, ws = yield self.make_iostream_pair()

        def sleep_some():
            self.io_loop.run_sync(lambda: gen.sleep(0.05))

        try:
            buf = bytearray(10)
            fut = rs.read_into(buf)
            ws.write(b"hello")
            yield gen.sleep(0.05)
            self.assertTrue(rs.reading())
            ws.write(b"world!!")
            data = yield fut
            self.assertFalse(rs.reading())
            self.assertEqual(data, 10)
            self.assertEqual(bytes(buf), b"helloworld")

            # Existing buffer is fed into user buffer
            fut = rs.read_into(buf)
            yield gen.sleep(0.05)
            self.assertTrue(rs.reading())
            ws.write(b"1234567890")
            data = yield fut
            self.assertFalse(rs.reading())
            self.assertEqual(data, 10)
            self.assertEqual(bytes(buf), b"!!12345678")

            # Existing buffer can satisfy read immediately
            buf = bytearray(4)
            ws.write(b"abcdefghi")
            data = yield rs.read_into(buf)
            self.assertEqual(data, 4)
            self.assertEqual(bytes(buf), b"90ab")

            data = yield rs.read_bytes(7)
            self.assertEqual(data, b"cdefghi")
        finally:
            ws.close()
            rs.close()

    @gen_test
    def test_read_into_partial(self):
        rs, ws = yield self.make_iostream_pair()

        try:
            # Partial read
            buf = bytearray(10)
            fut = rs.read_into(buf, partial=True)
            ws.write(b"hello")
            data = yield fut
            self.assertFalse(rs.reading())
            self.assertEqual(data, 5)
            self.assertEqual(bytes(buf), b"hello\0\0\0\0\0")

            # Full read despite partial=True
            ws.write(b"world!1234567890")
            data = yield rs.read_into(buf, partial=True)
            self.assertEqual(data, 10)
            self.assertEqual(bytes(buf), b"world!1234")

            # Existing buffer can satisfy read immediately
            data = yield rs.read_into(buf, partial=True)
            self.assertEqual(data, 6)
            self.assertEqual(bytes(buf), b"5678901234")

        finally:
            ws.close()
            rs.close()

    @gen_test
    def test_read_into_zero_bytes(self):
        rs, ws = yield self.make_iostream_pair()
        try:
            buf = bytearray()
            fut = rs.read_into(buf)
            self.assertEqual(fut.result(), 0)
        finally:
            ws.close()
            rs.close()

    @gen_test
    def test_many_mixed_reads(self):
        # Stress buffer handling when going back and forth between
        # read_bytes() (using an internal buffer) and read_into()
        # (using a user-allocated buffer).
        r = random.Random(42)
        nbytes = 1000000
        rs, ws = yield self.make_iostream_pair()

        produce_hash = hashlib.sha1()
        consume_hash = hashlib.sha1()

        @gen.coroutine
        def produce():
            remaining = nbytes
            while remaining > 0:
                size = r.randint(1, min(1000, remaining))
                data = os.urandom(size)
                produce_hash.update(data)
                yield ws.write(data)
                remaining -= size
            assert remaining == 0

        @gen.coroutine
        def consume():
            remaining = nbytes
            while remaining > 0:
                if r.random() > 0.5:
                    # read_bytes()
                    size = r.randint(1, min(1000, remaining))
                    data = yield rs.read_bytes(size)
                    consume_hash.update(data)
                    remaining -= size
                else:
                    # read_into()
                    size = r.randint(1, min(1000, remaining))
                    buf = bytearray(size)
                    n = yield rs.read_into(buf)
                    assert n == size
                    consume_hash.update(buf)
                    remaining -= size
            assert remaining == 0

        try:
            yield [produce(), consume()]
            assert produce_hash.hexdigest() == consume_hash.hexdigest()
        finally:
            ws.close()
            rs.close()


class TestIOStreamMixin(TestReadWriteMixin):
    def _make_server_iostream(self, connection, **kwargs):
        raise NotImplementedError()

    def _make_client_iostream(self, connection, **kwargs):
        raise NotImplementedError()

    @gen.coroutine
    def make_iostream_pair(self, **kwargs):
        listener, port = bind_unused_port()
        server_stream_fut = Future()  # type: Future[IOStream]

        def accept_callback(connection, address):
            server_stream_fut.set_result(
                self._make_server_iostream(connection, **kwargs)
            )

        netutil.add_accept_handler(listener, accept_callback)
        client_stream = self._make_client_iostream(socket.socket(), **kwargs)
        connect_fut = client_stream.connect(("127.0.0.1", port))
        server_stream, client_stream = yield [server_stream_fut, connect_fut]
        self.io_loop.remove_handler(listener.fileno())
        listener.close()
        raise gen.Return((server_stream, client_stream))

    @gen_test
    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())

        stream.set_close_callback(self.stop)
        # log messages vary by platform and ioloop implementation
        with ExpectLog(gen_log, ".*", required=False):
            with self.assertRaises(StreamClosedError):
                yield stream.connect(("127.0.0.1", port))

        self.assertTrue(isinstance(stream.error, socket.error), stream.error)
        if sys.platform != "cygwin":
            _ERRNO_CONNREFUSED = [errno.ECONNREFUSED]
            if hasattr(errno, "WSAECONNREFUSED"):
                _ERRNO_CONNREFUSED.append(errno.WSAECONNREFUSED)  # type: ignore
            # cygwin's errnos don't match those used on native windows python
            self.assertTrue(stream.error.args[0] in _ERRNO_CONNREFUSED)  # type: ignore

    @gen_test
    def test_gaierror(self):
        # Test that IOStream sets its exc_info on getaddrinfo error.
        # It's difficult to reliably trigger a getaddrinfo error;
        # some resolvers own't even return errors for malformed names,
        # so we mock it instead. If IOStream changes to call a Resolver
        # before sock.connect, the mock target will need to change too.
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
        stream = IOStream(s)
        stream.set_close_callback(self.stop)
        with mock.patch(
            "socket.socket.connect", side_effect=socket.gaierror(errno.EIO, "boom")
        ):
            with self.assertRaises(StreamClosedError):
                yield stream.connect(("localhost", 80))
            self.assertTrue(isinstance(stream.error, socket.gaierror))

    @gen_test
    def test_read_until_close_with_error(self):
        server, client = yield self.make_iostream_pair()
        try:
            with mock.patch(
                "tornado.iostream.BaseIOStream._try_inline_read",
                side_effect=IOError("boom"),
            ):
                with self.assertRaisesRegexp(IOError, "boom"):
                    client.read_until_close()
        finally:
            server.close()
            client.close()

    @skipIfNonUnix
    @skipPypy3V58
    @gen_test
    def test_inline_read_error(self):
        # An error on an inline read is raised without logging (on the
        # assumption that it will eventually be noticed or logged further
        # up the stack).
        #
        # This test is posix-only because windows os.close() doesn't work
        # on socket FDs, but we can't close the socket object normally
        # because we won't get the error we want if the socket knows
        # it's closed.
        server, client = yield self.make_iostream_pair()
        try:
            os.close(server.socket.fileno())
            with self.assertRaises(socket.error):
                server.read_bytes(1)
        finally:
            server.close()
            client.close()

    @skipPypy3V58
    @gen_test
    def test_async_read_error_logging(self):
        # Socket errors on asynchronous reads should be logged (but only
        # once).
        server, client = yield self.make_iostream_pair()
        closed = Event()
        server.set_close_callback(closed.set)
        try:
            # Start a read that will be fulfilled asynchronously.
            server.read_bytes(1)
            client.write(b"a")
            # Stub out read_from_fd to make it fail.

            def fake_read_from_fd():
                os.close(server.socket.fileno())
                server.__class__.read_from_fd(server)

            server.read_from_fd = fake_read_from_fd
            # This log message is from _handle_read (not read_from_fd).
            with ExpectLog(gen_log, "error on read"):
                yield closed.wait()
        finally:
            server.close()
            client.close()

    @gen_test
    def test_future_write(self):
        """
        Test that write() Futures are never orphaned.
        """
        # Run concurrent writers that will write enough bytes so as to
        # clog the socket buffer and accumulate bytes in our write buffer.
        m, n = 5000, 1000
        nproducers = 10
        total_bytes = m * n * nproducers
        server, client = yield self.make_iostream_pair(max_buffer_size=total_bytes)

        @gen.coroutine
        def produce():
            data = b"x" * m
            for i in range(n):
                yield server.write(data)

        @gen.coroutine
        def consume():
            nread = 0
            while nread < total_bytes:
                res = yield client.read_bytes(m)
                nread += len(res)

        try:
            yield [produce() for i in range(nproducers)] + [consume()]
        finally:
            server.close()
            client.close()


class TestIOStreamWebHTTP(TestIOStreamWebMixin, AsyncHTTPTestCase):
    def _make_client_iostream(self):
        return IOStream(socket.socket())


class TestIOStreamWebHTTPS(TestIOStreamWebMixin, AsyncHTTPSTestCase):
    def _make_client_iostream(self):
        return SSLIOStream(socket.socket(), ssl_options=dict(cert_reqs=ssl.CERT_NONE))


class TestIOStream(TestIOStreamMixin, AsyncTestCase):
    def _make_server_iostream(self, connection, **kwargs):
        return IOStream(connection, **kwargs)

    def _make_client_iostream(self, connection, **kwargs):
        return IOStream(connection, **kwargs)


class TestIOStreamSSL(TestIOStreamMixin, AsyncTestCase):
    def _make_server_iostream(self, connection, **kwargs):
        connection = ssl.wrap_socket(
            connection,
            server_side=True,
            do_handshake_on_connect=False,
            **_server_ssl_options()
        )
        return SSLIOStream(connection, **kwargs)

    def _make_client_iostream(self, connection, **kwargs):
        return SSLIOStream(
            connection, ssl_options=dict(cert_reqs=ssl.CERT_NONE), **kwargs
        )


# This will run some tests that are basically redundant but it's the
# simplest way to make sure that it works to pass an SSLContext
# instead of an ssl_options dict to the SSLIOStream constructor.
class TestIOStreamSSLContext(TestIOStreamMixin, AsyncTestCase):
    def _make_server_iostream(self, connection, **kwargs):
        context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
        context.load_cert_chain(
            os.path.join(os.path.dirname(__file__), "test.crt"),
            os.path.join(os.path.dirname(__file__), "test.key"),
        )
        connection = ssl_wrap_socket(
            connection, context, server_side=True, do_handshake_on_connect=False
        )
        return SSLIOStream(connection, **kwargs)

    def _make_client_iostream(self, connection, **kwargs):
        context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
        return SSLIOStream(connection, ssl_options=context, **kwargs)


class TestIOStreamStartTLS(AsyncTestCase):
    def setUp(self):
        try:
            super(TestIOStreamStartTLS, self).setUp()
            self.listener, self.port = bind_unused_port()
            self.server_stream = None
            self.server_accepted = Future()  # type: Future[None]
            netutil.add_accept_handler(self.listener, self.accept)
            self.client_stream = IOStream(socket.socket())
            self.io_loop.add_future(
                self.client_stream.connect(("127.0.0.1", self.port)), self.stop
            )
            self.wait()
            self.io_loop.add_future(self.server_accepted, self.stop)
            self.wait()
        except Exception as e:
            print(e)
            raise

    def tearDown(self):
        if self.server_stream is not None:
            self.server_stream.close()
        if self.client_stream is not None:
            self.client_stream.close()
        self.listener.close()
        super(TestIOStreamStartTLS, self).tearDown()

    def accept(self, connection, address):
        if self.server_stream is not None:
            self.fail("should only get one connection")
        self.server_stream = IOStream(connection)
        self.server_accepted.set_result(None)

    @gen.coroutine
    def client_send_line(self, line):
        self.client_stream.write(line)
        recv_line = yield self.server_stream.read_until(b"\r\n")
        self.assertEqual(line, recv_line)

    @gen.coroutine
    def server_send_line(self, line):
        self.server_stream.write(line)
        recv_line = yield self.client_stream.read_until(b"\r\n")
        self.assertEqual(line, recv_line)

    def client_start_tls(self, ssl_options=None, server_hostname=None):
        client_stream = self.client_stream
        self.client_stream = None
        return client_stream.start_tls(False, ssl_options, server_hostname)

    def server_start_tls(self, ssl_options=None):
        server_stream = self.server_stream
        self.server_stream = None
        return server_stream.start_tls(True, ssl_options)

    @gen_test
    def test_start_tls_smtp(self):
        # This flow is simplified from RFC 3207 section 5.
        # We don't really need all of this, but it helps to make sure
        # that after realistic back-and-forth traffic the buffers end up
        # in a sane state.
        yield self.server_send_line(b"220 mail.example.com ready\r\n")
        yield self.client_send_line(b"EHLO mail.example.com\r\n")
        yield self.server_send_line(b"250-mail.example.com welcome\r\n")
        yield self.server_send_line(b"250 STARTTLS\r\n")
        yield self.client_send_line(b"STARTTLS\r\n")
        yield self.server_send_line(b"220 Go ahead\r\n")
        client_future = self.client_start_tls(dict(cert_reqs=ssl.CERT_NONE))
        server_future = self.server_start_tls(_server_ssl_options())
        self.client_stream = yield client_future
        self.server_stream = yield server_future
        self.assertTrue(isinstance(self.client_stream, SSLIOStream))
        self.assertTrue(isinstance(self.server_stream, SSLIOStream))
        yield self.client_send_line(b"EHLO mail.example.com\r\n")
        yield self.server_send_line(b"250 mail.example.com welcome\r\n")

    @gen_test
    def test_handshake_fail(self):
        server_future = self.server_start_tls(_server_ssl_options())
        # Certificates are verified with the default configuration.
        with ExpectLog(gen_log, "SSL Error"):
            client_future = self.client_start_tls(server_hostname="localhost")
            with self.assertRaises(ssl.SSLError):
                yield client_future
            with self.assertRaises((ssl.SSLError, socket.error)):
                yield server_future

    @gen_test
    def test_check_hostname(self):
        # Test that server_hostname parameter to start_tls is being used.
        # The check_hostname functionality is only available in python 2.7 and
        # up and in python 3.4 and up.
        server_future = self.server_start_tls(_server_ssl_options())
        with ExpectLog(gen_log, "SSL Error"):
            client_future = self.client_start_tls(
                ssl.create_default_context(), server_hostname="127.0.0.1"
            )
            with self.assertRaises(ssl.SSLError):
                # The client fails to connect with an SSL error.
                yield client_future
            with self.assertRaises(Exception):
                # The server fails to connect, but the exact error is unspecified.
                yield server_future


class WaitForHandshakeTest(AsyncTestCase):
    @gen.coroutine
    def connect_to_server(self, server_cls):
        server = client = None
        try:
            sock, port = bind_unused_port()
            server = server_cls(ssl_options=_server_ssl_options())
            server.add_socket(sock)

            client = SSLIOStream(
                socket.socket(), ssl_options=dict(cert_reqs=ssl.CERT_NONE)
            )
            yield client.connect(("127.0.0.1", port))
            self.assertIsNotNone(client.socket.cipher())
        finally:
            if server is not None:
                server.stop()
            if client is not None:
                client.close()

    @gen_test
    def test_wait_for_handshake_future(self):
        test = self
        handshake_future = Future()  # type: Future[None]

        class TestServer(TCPServer):
            def handle_stream(self, stream, address):
                test.assertIsNone(stream.socket.cipher())
                test.io_loop.spawn_callback(self.handle_connection, stream)

            @gen.coroutine
            def handle_connection(self, stream):
                yield stream.wait_for_handshake()
                handshake_future.set_result(None)

        yield self.connect_to_server(TestServer)
        yield handshake_future

    @gen_test
    def test_wait_for_handshake_already_waiting_error(self):
        test = self
        handshake_future = Future()  # type: Future[None]

        class TestServer(TCPServer):
            @gen.coroutine
            def handle_stream(self, stream, address):
                fut = stream.wait_for_handshake()
                test.assertRaises(RuntimeError, stream.wait_for_handshake)
                yield fut

                handshake_future.set_result(None)

        yield self.connect_to_server(TestServer)
        yield handshake_future

    @gen_test
    def test_wait_for_handshake_already_connected(self):
        handshake_future = Future()  # type: Future[None]

        class TestServer(TCPServer):
            @gen.coroutine
            def handle_stream(self, stream, address):
                yield stream.wait_for_handshake()
                yield stream.wait_for_handshake()
                handshake_future.set_result(None)

        yield self.connect_to_server(TestServer)
        yield handshake_future


@skipIfNonUnix
class TestPipeIOStream(TestReadWriteMixin, AsyncTestCase):
    @gen.coroutine
    def make_iostream_pair(self, **kwargs):
        r, w = os.pipe()

        return PipeIOStream(r, **kwargs), PipeIOStream(w, **kwargs)

    @gen_test
    def test_pipe_iostream(self):
        rs, ws = yield self.make_iostream_pair()

        ws.write(b"hel")
        ws.write(b"lo world")

        data = yield rs.read_until(b" ")
        self.assertEqual(data, b"hello ")

        data = yield rs.read_bytes(3)
        self.assertEqual(data, b"wor")

        ws.close()

        data = yield rs.read_until_close()
        self.assertEqual(data, b"ld")

        rs.close()

    @gen_test
    def test_pipe_iostream_big_write(self):
        rs, ws = yield self.make_iostream_pair()

        NUM_BYTES = 1048576

        # Write 1MB of data, which should fill the buffer
        ws.write(b"1" * NUM_BYTES)

        data = yield rs.read_bytes(NUM_BYTES)
        self.assertEqual(data, b"1" * NUM_BYTES)

        ws.close()
        rs.close()


class TestStreamBuffer(unittest.TestCase):
    """
    Unit tests for the private _StreamBuffer class.
    """

    def setUp(self):
        self.random = random.Random(42)

    def to_bytes(self, b):
        if isinstance(b, (bytes, bytearray)):
            return bytes(b)
        elif isinstance(b, memoryview):
            return b.tobytes()  # For py2
        else:
            raise TypeError(b)

    def make_streambuffer(self, large_buf_threshold=10):
        buf = _StreamBuffer()
        assert buf._large_buf_threshold
        buf._large_buf_threshold = large_buf_threshold
        return buf

    def check_peek(self, buf, expected):
        size = 1
        while size < 2 * len(expected):
            got = self.to_bytes(buf.peek(size))
            self.assertTrue(got)  # Not empty
            self.assertLessEqual(len(got), size)
            self.assertTrue(expected.startswith(got), (expected, got))
            size = (size * 3 + 1) // 2

    def check_append_all_then_skip_all(self, buf, objs, input_type):
        self.assertEqual(len(buf), 0)

        expected = b""

        for o in objs:
            expected += o
            buf.append(input_type(o))
            self.assertEqual(len(buf), len(expected))
            self.check_peek(buf, expected)

        while expected:
            n = self.random.randrange(1, len(expected) + 1)
            expected = expected[n:]
            buf.advance(n)
            self.assertEqual(len(buf), len(expected))
            self.check_peek(buf, expected)

        self.assertEqual(len(buf), 0)

    def test_small(self):
        objs = [b"12", b"345", b"67", b"89a", b"bcde", b"fgh", b"ijklmn"]

        buf = self.make_streambuffer()
        self.check_append_all_then_skip_all(buf, objs, bytes)

        buf = self.make_streambuffer()
        self.check_append_all_then_skip_all(buf, objs, bytearray)

        buf = self.make_streambuffer()
        self.check_append_all_then_skip_all(buf, objs, memoryview)

        # Test internal algorithm
        buf = self.make_streambuffer(10)
        for i in range(9):
            buf.append(b"x")
        self.assertEqual(len(buf._buffers), 1)
        for i in range(9):
            buf.append(b"x")
        self.assertEqual(len(buf._buffers), 2)
        buf.advance(10)
        self.assertEqual(len(buf._buffers), 1)
        buf.advance(8)
        self.assertEqual(len(buf._buffers), 0)
        self.assertEqual(len(buf), 0)

    def test_large(self):
        objs = [
            b"12" * 5,
            b"345" * 2,
            b"67" * 20,
            b"89a" * 12,
            b"bcde" * 1,
            b"fgh" * 7,
            b"ijklmn" * 2,
        ]

        buf = self.make_streambuffer()
        self.check_append_all_then_skip_all(buf, objs, bytes)

        buf = self.make_streambuffer()
        self.check_append_all_then_skip_all(buf, objs, bytearray)

        buf = self.make_streambuffer()
        self.check_append_all_then_skip_all(buf, objs, memoryview)

        # Test internal algorithm
        buf = self.make_streambuffer(10)
        for i in range(3):
            buf.append(b"x" * 11)
        self.assertEqual(len(buf._buffers), 3)
        buf.append(b"y")
        self.assertEqual(len(buf._buffers), 4)
        buf.append(b"z")
        self.assertEqual(len(buf._buffers), 4)
        buf.advance(33)
        self.assertEqual(len(buf._buffers), 1)
        buf.advance(2)
        self.assertEqual(len(buf._buffers), 0)
        self.assertEqual(len(buf), 0)