def _run_traffic_jam(nsends, nbytes):
    # This test eats `nsends * nbytes` bytes in RAM
    np = pytest.importorskip('numpy')
    from distributed.protocol import to_serialize
    data = bytes(np.random.randint(0, 255, size=(nbytes,)).astype('u1').data)
    with echo_server() as e:
        client = TCPClient()
        stream = yield client.connect('127.0.0.1', e.port)

        b = BatchedSend(interval=0.01)
        b.start(stream)

        msg = {'x': to_serialize(data)}
        for i in range(nsends):
            b.send(assoc(msg, 'i', i))
            if np.random.random() > 0.5:
                yield gen.sleep(0.001)

        results = []
        count = 0
        while len(results) < nsends:
            # If this times out then I think it's a backpressure issue
            # Somehow we're able to flood the socket so that the receiving end
            # loses some of our messages
            L = yield gen.with_timeout(timedelta(seconds=5), read(stream))
            count += 1
            results.extend(r['i'] for r in L)

        assert count == b.batch_count == e.count
        assert b.message_count == nsends

        assert results == list(range(nsends))

        stream.close()  # external closing
        yield b.close(ignore_closed=True)

def test_stress():
    with echo_server() as e:
        client = TCPClient()
        stream = yield client.connect('127.0.0.1', e.port)
        L = []

        @gen.coroutine
        def send():
            b = BatchedSend(interval=3)
            b.start(stream)
            for i in range(0, 10000, 2):
                b.send(i)
                b.send(i + 1)
                yield gen.sleep(0.00001 * random.randint(1, 10))

        @gen.coroutine
        def recv():
            while True:
                result = yield gen.with_timeout(timedelta(seconds=1), read(stream))
                print(result)
                L.extend(result)
                if result[-1] == 9999:
                    break

        yield All([send(), recv()])

        assert L == list(range(0, 10000, 1))
        stream.close()

    def connect(self, address, deserialize=True, **connection_args):
        self._check_encryption(address, connection_args)
        ip, port = parse_host_port(address)
        kwargs = self._get_connect_args(**connection_args)

        client = TCPClient()
        try:
            stream = yield client.connect(ip, port,
                                          max_buffer_size=MAX_BUFFER_SIZE,
                                          **kwargs)
            # Under certain circumstances tornado will have a closed connnection with an error and not raise
            # a StreamClosedError.
            #
            # This occurs with tornado 5.x and openssl 1.1+
            if stream.closed() and stream.error:
                raise StreamClosedError(stream.error)

        except StreamClosedError as e:
            # The socket connect() call failed
            convert_stream_closed_error(self, e)

        local_address = self.prefix + get_stream_address(stream)
        raise gen.Return(self.comm_class(stream,
                                         local_address,
                                         self.prefix + address,
                                         deserialize))

 def connect(self):
     client = TCPClient(io_loop=self.io_loop)
     self.stream = yield client.connect(self.host, self.port)
     # sock = None
     # try:
     #     if self.unix_socket and self.host in ('localhost', '127.0.0.1'):
     #         sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
     #         t = sock.gettimeout()
     #         sock.settimeout(self.connect_timeout)
     #         sock.connect(self.unix_socket)
     #         sock.settimeout(t)
     #         self.host_info = "Localhost via UNIX socket"
     #         if DEBUG: print('connected using unix_socket')
     #     else:
     #         while True:
     #             try:
     #                 sock = socket.create_connection(
     #                         (self.host, self.port), self.connect_timeout)
     #                 break
     #             except (OSError, IOError) as e:
     #                 if e.errno == errno.EINTR:
     #                     continue
     #                 raise
     #         self.host_info = "socket %s:%d" % (self.host, self.port)
     #         if DEBUG: print('connected using socket')
     #     sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
     #     if self.no_delay:
     #         sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
     #     self.socket = sock
     #     self._rfile = _makefile(sock, 'rb')
     yield self._get_server_information()
     yield self._request_authentication()

    def __init__(self, *args, **kwargs):

        TCPClient.__init__(self, kwargs.pop("resolver", None),
                           kwargs.pop("io_loop", None))

        Connection.__init__(self, parser_class=AsyncParser, *args, **kwargs)

        self._stream = None

def test_close_twice():
    with echo_server() as e:
        client = TCPClient()
        stream = yield client.connect('127.0.0.1', e.port)

        b = BatchedSend(interval=10)
        b.start(stream)
        yield b.close()
        yield b.close()

 def __init__(self, ssl_options=None, gp_module=False, **kwargs):
     self.logger = logging.getLogger(self.__class__.__name__)
     self.gp_module = gp_module
     try:
         TCPClient.__init__(self, ssl_options=ssl_options, **kwargs)
     except:
         etype, evalue, etb = sys.exc_info()
         self.logger.error("Could not create tcp client. Exception: %s, Error: %s." % (etype, evalue))
         self.gp_module.shutDown()

def test_send_after_stream_finish():
    with echo_server() as e:
        client = TCPClient()
        stream = yield client.connect('127.0.0.1', e.port)

        b = BatchedSend(interval=10)
        b.start(stream)
        yield b.last_send

        b.send('hello')
        result = yield read(stream); assert result == ['hello']

def test_send_before_start():
    with echo_server() as e:
        client = TCPClient()
        stream = yield client.connect('127.0.0.1', e.port)

        b = BatchedSend(interval=10)

        b.send('hello')
        b.send('world')

        b.start(stream)
        result = yield read(stream); assert result == ['hello', 'world']

    def connect(self, host, port):
        self.host = host
        self.port = port

        client = TCPClient()
        try:
            self.stream = yield client.connect(self.host, self.port)
        except IOError as e:
            log.error("%s", repr(e))
            raise gen.Return((False, 'Failed to connect'))
        self.trigger(Event.CONNECT, self)
        raise gen.Return((True, "OK"))

def test_close_closed():
    with echo_server() as e:
        client = TCPClient()
        stream = yield client.connect('127.0.0.1', e.port)

        b = BatchedSend(interval=10)
        b.start(stream)

        b.send(123)
        stream.close()  # external closing

        yield b.close(ignore_closed=True)

def connect(ip, port, timeout=1):
    client = TCPClient()
    start = time()
    while True:
        try:
            stream = yield client.connect(ip, port)
            raise Return(stream)
        except StreamClosedError:
            if time() - start < timeout:
                yield gen.sleep(0.01)
                logger.debug("sleeping on connect")
            else:
                raise

def start_app():
  tcpClient = TCPClient()
  try:
    stream = yield tcpClient.connect('127.0.0.1', 9999)
    print 'Connection started'
    app = LaternController(LanternDriver())
    client = TLVClient(stream)
    executer = ThreadPoolExecutor(max_workers=5)
    while True:
      command = yield client.getCommand()
      executer.submit(app.handle, command)
  except Exception as e:
    print 'Caught Error: %s' % e
    IOLoop.instance().add_callback(IOLoop.instance().stop)

def connect(ip, port, timeout=3):
    client = TCPClient()
    start = time()
    while True:
        try:
            future = client.connect(ip, port, max_buffer_size=MAX_BUFFER_SIZE)
            stream = yield gen.with_timeout(timedelta(seconds=timeout), future)
            raise Return(stream)
        except StreamClosedError:
            if time() - start < timeout:
                yield gen.sleep(0.01)
                logger.debug("sleeping on connect")
            else:
                raise
        except gen.TimeoutError:
            raise IOError("Timed out while connecting to %s:%d" % (ip, port))

    def __init__(self):
        logger.debug('Starting Envisalink Client')

        # Register events for alarmserver requests -> envisalink
        events.register('alarm_update', self.request_action)

        # Register events for envisalink proxy
        events.register('envisalink', self.envisalink_proxy)

        # Create TCP Client
        self.tcpclient = TCPClient()

        # Connection
        self._connection = None

        # Set our terminator to \r\n
        self._terminator = b"\r\n"

        # Reconnect delay
        self._retrydelay = 10

        # Connect to Envisalink
        self.do_connect()

        # Setup timer to refresh envisalink
        tornado.ioloop.PeriodicCallback(self.check_connection, 1000).start()

        # Last activity
        self._last_activity = time.time()

    def __init__(self, routes, node, pipe):
        """
        Application instantiates and registers handlers for each message type,
        and routes messages to the pre-instantiated instances of each message handler

        :param routes: list of tuples in the form of (<message type str>, <MessageHandler class>)
        :param node: Node instance of the local node
        :param pipe: Instance of multiprocessing.Pipe for communicating with the parent process
        """
        # We don't really have to worry about synchronization
        # so long as we're careful about explicit context switching
        self.nodes = {node.node_id: node}

        self.local_node = node
        self.handlers = {}

        self.tcpclient = TCPClient()

        self.gossip_inbox = Queue()
        self.gossip_outbox = Queue()

        self.sequence_number = 0

        if routes:
            self.add_handlers(routes)

        self.pipe = pipe
        self.ioloop = IOLoop.current()

        self.add_node_event = Event()

    def __init__(self, io_loop, request, compression_options=None):
        self.compression_options = compression_options
        self.connect_future = TracebackFuture()
        self.read_future = None
        self.read_queue = collections.deque()
        self.key = base64.b64encode(os.urandom(16))

        scheme, sep, rest = request.url.partition(":")
        scheme = {"ws": "http", "wss": "https"}[scheme]
        request.url = scheme + sep + rest
        request.headers.update(
            {
                "Upgrade": "websocket",
                "Connection": "Upgrade",
                "Sec-WebSocket-Key": self.key,
                "Sec-WebSocket-Version": "13",
            }
        )
        if self.compression_options is not None:
            # Always offer to let the server set our max_wbits (and even though
            # we don't offer it, we will accept a client_no_context_takeover
            # from the server).
            # TODO: set server parameters for deflate extension
            # if requested in self.compression_options.
            request.headers["Sec-WebSocket-Extensions"] = "permessage-deflate; client_max_window_bits"

        self.tcp_client = TCPClient(io_loop=io_loop)
        super(WebSocketClientConnection, self).__init__(
            io_loop, None, request, lambda: None, self._on_http_response, 104857600, self.tcp_client, 65536
        )

def test_BatchedStream_raises():
    port = 3435
    server = MyServer()
    server.listen(port)

    client = TCPClient()
    stream = yield client.connect('127.0.0.1', port)
    b = BatchedStream(stream, interval=20)

    stream.close()

    with pytest.raises(StreamClosedError):
        yield b.recv()

    with pytest.raises(StreamClosedError):
        yield b.send('123')

    def __init__(self, io_loop, request, on_message_callback=None,
                 compression_options=None):
        self.compression_options = compression_options
        self.connect_future = TracebackFuture()
        self.protocol = None
        self.read_future = None
        self.read_queue = collections.deque()
        self.key = base64.b64encode(os.urandom(16))
        self._on_message_callback = on_message_callback
        self.close_code = self.close_reason = None

        scheme, sep, rest = request.url.partition(':')
        scheme = {'ws': 'http', 'wss': 'https'}[scheme]
        request.url = scheme + sep + rest
        request.headers.update({
            'Upgrade': 'websocket',
            'Connection': 'Upgrade',
            'Sec-WebSocket-Key': self.key,
            'Sec-WebSocket-Version': '13',
        })
        if self.compression_options is not None:
            # Always offer to let the server set our max_wbits (and even though
            # we don't offer it, we will accept a client_no_context_takeover
            # from the server).
            # TODO: set server parameters for deflate extension
            # if requested in self.compression_options.
            request.headers['Sec-WebSocket-Extensions'] = (
                'permessage-deflate; client_max_window_bits')

        self.tcp_client = TCPClient(io_loop=io_loop)
        super(WebSocketClientConnection, self).__init__(
            io_loop, None, request, lambda: None, self._on_http_response,
            104857600, self.tcp_client, 65536, 104857600)

    def initialize(self, io_loop=None, max_clients=10,
                   max_simultaneous_connections=None,
                   hostname_mapping=None, max_buffer_size=104857600,
                   resolver=None, defaults=None, max_header_size=None,
                   max_body_size=None):
        """Creates a AsyncHTTPClient.

        Only a single AsyncHTTPClient instance exists per IOLoop
        in order to provide limitations on the number of pending connections.
        ``force_instance=True`` may be used to suppress this behavior.

        Note that because of this implicit reuse, unless ``force_instance``
        is used, only the first call to the constructor actually uses
        its arguments. It is recommended to use the ``configure`` method
        instead of the constructor to ensure that arguments take effect.

        ``max_clients`` is the number of concurrent requests that can be
        in progress; when this limit is reached additional requests will be
        queued. Note that time spent waiting in this queue still counts
        against the ``request_timeout``.

        ``hostname_mapping`` is a dictionary mapping hostnames to IP addresses.
        It can be used to make local DNS changes when modifying system-wide
        settings like ``/etc/hosts`` is not possible or desirable (e.g. in
        unittests).

        ``max_buffer_size`` (default 100MB) is the number of bytes
        that can be read into memory at once. ``max_body_size``
        (defaults to ``max_buffer_size``) is the largest response body
        that the client will accept.  Without a
        ``streaming_callback``, the smaller of these two limits
        applies; with a ``streaming_callback`` only ``max_body_size``
        does.

        .. versionchanged:: 4.2
           Added the ``max_body_size`` argument.
        """
        super(SimpleAsyncHTTPClient, self).initialize(io_loop,
                                                      defaults=defaults)
        self.max_clients = max_clients
        self.queue = collections.deque()
        self.active = {}
        self.waiting = {}
        self.max_buffer_size = max_buffer_size
        self.max_header_size = max_header_size
        self.max_body_size = max_body_size
        # TCPClient could create a Resolver for us, but we have to do it
        # ourselves to support hostname_mapping.
        if resolver:
            self.resolver = resolver
            self.own_resolver = False
        else:
            self.resolver = Resolver(io_loop=io_loop)
            self.own_resolver = True
        if hostname_mapping is not None:
            self.resolver = OverrideResolver(resolver=self.resolver,
                                             mapping=hostname_mapping)
        self.tcp_client = TCPClient(resolver=self.resolver, io_loop=io_loop)