Python twisted.internet.endpoints 模块，TCP4ClientEndpoint() 实例源码

def _getEndpoint(self, scheme, host, port):
        """
        Get an endpoint for the given host and port, using a transport
        selected based on scheme.

        @param scheme: A string like C{'http'} or C{'https'} (the only two
            supported values) to use to determine how to establish the
            connection.

        @param host: A C{str} giving the hostname which will be connected to in
            order to issue a request.

        @param port: An C{int} giving the port number the connection will be
            on.

        @return: An endpoint which can be used to connect to given address.
        """
        kwargs = {}
        if self._connectTimeout is not None:
            kwargs['timeout'] = self._connectTimeout
        kwargs['bindAddress'] = self._bindAddress
        if scheme == 'http':
            return TCP4ClientEndpoint(self._reactor, host, port, **kwargs)
        elif scheme == 'https':
            return SSL4ClientEndpoint(self._reactor, host, port,
                                      self._wrapContextFactory(host, port),
                                      **kwargs)
        else:
            raise SchemeNotSupported("Unsupported scheme: %r" % (scheme,))

def secureConnection(self):
        """
        Create and return a new SSH connection which has been secured and on
        which authentication has already happened.

        @return: A L{Deferred} which fires with the ready-to-use connection or
            with a failure if something prevents the connection from being
            setup, secured, or authenticated.
        """
        protocol = _CommandTransport(self)
        ready = protocol.connectionReady

        sshClient = TCP4ClientEndpoint(
            self.reactor, nativeString(self.hostname), self.port)

        d = connectProtocol(sshClient, protocol)
        d.addCallback(lambda ignored: ready)
        return d

def createClientEndpoint(self, reactor, clientFactory, **connectArgs):
        """
        Create an L{TCP4ClientEndpoint} and return the values needed to verify
        its behavior.

        @param reactor: A fake L{IReactorTCP} that L{TCP4ClientEndpoint} can
            call L{IReactorTCP.connectTCP} on.
        @param clientFactory: The thing that we expect to be passed to our
            L{IStreamClientEndpoint.connect} implementation.
        @param connectArgs: Optional dictionary of arguments to
            L{IReactorTCP.connectTCP}
        """
        address = IPv4Address("TCP", "localhost", 80)

        return (endpoints.TCP4ClientEndpoint(reactor,
                                             address.host,
                                             address.port,
                                             **connectArgs),
                (address.host, address.port, clientFactory,
                 connectArgs.get('timeout', 30),
                 connectArgs.get('bindAddress', None)),
                address)

def test_tcp(self):
        """
        When passed a TCP strports description, L{endpoints.clientFromString}
        returns a L{TCP4ClientEndpoint} instance initialized with the values
        from the string.
        """
        reactor = object()
        client = endpoints.clientFromString(
            reactor,
            "tcp:host=example.com:port=1234:timeout=7:bindAddress=10.0.0.2")
        self.assertIsInstance(client, endpoints.TCP4ClientEndpoint)
        self.assertIs(client._reactor, reactor)
        self.assertEqual(client._host, "example.com")
        self.assertEqual(client._port, 1234)
        self.assertEqual(client._timeout, 7)
        self.assertEqual(client._bindAddress, ("10.0.0.2", 0))

def test_connectProtocolCreatesFactory(self):
        """
        C{endpoints.connectProtocol} calls the given endpoint's C{connect()}
        method with a factory that will build the given protocol.
        """
        reactor = MemoryReactor()
        endpoint = endpoints.TCP4ClientEndpoint(reactor, "127.0.0.1", 0)
        theProtocol = object()
        endpoints.connectProtocol(endpoint, theProtocol)

        # A TCP connection was made via the given endpoint:
        self.assertEqual(len(reactor.tcpClients), 1)
        # TCP4ClientEndpoint uses a _WrapperFactory around the underlying
        # factory, so we need to unwrap it:
        factory = reactor.tcpClients[0][2]._wrappedFactory
        self.assertIsInstance(factory, protocol.Factory)
        self.assertIs(factory.buildProtocol(None), theProtocol)

def connectableEndpoint(debug=False):
    """
    Create an endpoint that can be fired on demand.

    @param debug: A flag; whether to dump output from the established
        connection to stdout.
    @type debug: L{bool}

    @return: A client endpoint, and an object that will cause one of the
        L{Deferred}s returned by that client endpoint.
    @rtype: 2-L{tuple} of (L{IStreamClientEndpoint}, L{ConnectionCompleter})
    """
    reactor = MemoryReactorClock()
    clientEndpoint = TCP4ClientEndpoint(reactor, "0.0.0.0", 4321)
    serverEndpoint = TCP4ServerEndpoint(reactor, 4321)
    serverEndpoint.listen(Factory.forProtocol(Protocol))
    return clientEndpoint, ConnectionCompleter(reactor)

def test_daphn3(self):
        host = self.localOptions['host']
        port = int(self.localOptions['port'])

        def failure(failure):
            log.msg("Failed to connect")
            self.report['censored'] = True
            self.report['mutation'] = 0
            raise Exception("Error in connection, perhaps the backend is censored")
            return

        def success(protocol):
            log.msg("Successfully connected")
            protocol.sendPayload()
            return protocol.d

        log.msg("Connecting to %s:%s" % (host, port))
        endpoint = endpoints.TCP4ClientEndpoint(reactor, host, port)
        daphn3_factory = Daphn3ClientFactory()
        daphn3_factory.steps = self.input
        daphn3_factory.report = self.report
        d = endpoint.connect(daphn3_factory)
        d.addErrback(failure)
        d.addCallback(success)
        return d

def _test_connect_to_port(self, address, port):
        result = {
            'ip': address,
            'port': port,
            'status': {
                'success': None,
                'failure': None
            }
        }
        point = TCP4ClientEndpoint(reactor, address, port, timeout=10)
        d = point.connect(TCPConnectFactory())
        @d.addCallback
        def cb(p):
            result['status']['success'] = True
            result['status']['failure'] = False
            self.report['tcp_connect'].append(result)

        @d.addErrback
        def eb(failure):
            result['status']['success'] = False
            result['status']['failure'] = failureToString(failure)
            self.report['tcp_connect'].append(result)
            return failure

        return d

def _test_connect_to_port(self, address, port):
        result = {
            'ip': address,
            'port': port,
            'status': {
                'success': None,
                'failure': None
            }
        }
        point = TCP4ClientEndpoint(reactor, address, port, timeout=10)
        d = point.connect(TCPConnectFactory())
        @d.addCallback
        def cb(p):
            result['status']['success'] = True
            result['status']['failure'] = False
            self.report['tcp_connect'].append(result)

        @d.addErrback
        def eb(failure):
            result['status']['success'] = False
            result['status']['failure'] = failureToString(failure)
            self.report['tcp_connect'].append(result)
            return failure

        return d

def _test_connect_to_port(self, address, port):
        result = {
            'ip': address,
            'port': port,
            'status': {
                'success': None,
                'failure': None
            }
        }
        point = TCP4ClientEndpoint(reactor, address, port, timeout=10)
        d = point.connect(TCPConnectFactory())
        @d.addCallback
        def cb(p):
            result['status']['success'] = True
            result['status']['failure'] = False
            self.report['tcp_connect'].append(result)

        @d.addErrback
        def eb(failure):
            result['status']['success'] = False
            result['status']['failure'] = failureToString(failure)
            self.report['tcp_connect'].append(result)
            return failure

        return d

def test_connect(self):
        """
        This test performs a TCP connection to the remote host on the
        specified port.
        The report will contains the string 'success' if the test has
        succeeded, or the reason for the failure if it has failed.
        """
        def connectionSuccess(protocol):
            protocol.transport.loseConnection()
            log.debug("Got a connection to %s" % self.input)
            self.report["connection"] = 'success'

        def connectionFailed(failure):
            self.report['connection'] = handleAllFailures(failure)

        from twisted.internet import reactor
        point = TCP4ClientEndpoint(reactor, self.host, int(self.port))
        d = point.connect(TCPFactory())
        d.addCallback(connectionSuccess)
        d.addErrback(connectionFailed)
        return d

def _connect(self, address, callback, errback):
        if self._local_mode:
            point = TCP4ClientEndpoint(self._twisted_reactor,
                                       host=HOST, port=address.port)

        else:
            point = TorClientEndpoint(address.host, address.port,
                                      socks_hostname=HOST,
                                      socks_port=self._port_tor_socks)

        def connect_from_thread():
            d = connectProtocol(point,
                                _ConversationProtocol(self._twisted_factory,
                                                      callback))
            d.addErrback(errback)

        self._twisted_reactor.callFromThread(connect_from_thread)

def connect(self):
        torEndpoint = TCP4ClientEndpoint(reactor, TOR_HOST, TOR_PORT)
        proxiedEndpoint = SOCKS5ClientEndpoint(self.active_host.hostname.encode("ascii"), self.current_port, torEndpoint)
        d = proxiedEndpoint.connect(PortScannerClientFactory(self))
        d.addCallback(gotProtocol, self)
        d.addErrback(gotErr, self)
        #reactor.callLater(60, d.cancel)

def connect(self, adapter, ip, port):
        """ Establish a new TCP connection and link it with this protocol. """

        endpoint = TCP4ClientEndpoint(adapter.reactor, ip, port)
        d = connectProtocol(endpoint, self)

        def bad_connection(failure):
            message = "Could not connect to {}:{}\n {}\n".format(ip, port, failure.getErrorMessage())
            raise IOError(message)

        d.addErrback(bad_connection)
        return d

def build_irc(self):
        """The main starting method that creates a protocol object
        according to the config variables, ready for whenever
        the reactor starts running.
        """
        wlog('building irc')
        if self.tx_irc_client:
            raise Exception('irc already built')
        if self.usessl.lower() == 'true' and not self.socks5.lower() == 'true':
            factory = TxIRCFactory(self)
            ctx = ClientContextFactory()
            reactor.connectSSL(self.serverport[0], self.serverport[1],
                               factory, ctx)
        elif self.socks5.lower() == 'true':
            factory = TxIRCFactory(self)
            #str() casts needed else unicode error
            torEndpoint = TCP4ClientEndpoint(reactor, str(self.socks5_host),
                                             self.socks5_port)
            ircEndpoint = SOCKS5ClientEndpoint(str(self.serverport[0]),
                                               self.serverport[1], torEndpoint)
            if self.usessl.lower() == 'true':
                ctx = ClientContextFactory()
                tlsEndpoint = TLSWrapClientEndpoint(ctx, ircEndpoint)
                myRS = ClientService(tlsEndpoint, factory)
                myRS.startService()
            else:
                myRS = ClientService(ircEndpoint, factory)
                myRS.startService()
        else:
            try:
                factory = TxIRCFactory(self)
                wlog('build_irc: ', self.serverport[0], self.serverport[1],
                     self.channel)
                self.tcp_connector = reactor.connectTCP(
                        self.serverport[0], self.serverport[1], factory)
            except Exception as e:
                wlog('error in buildirc: ' + repr(e))

def get_orport_endpoint(tor_state):
    proxy_endpoint = tor_state.protocol.get_conf("SocksPort")
    def extract_port_value(result):
        port = result['SocksPort'].split()[0]
        return int(port) if port != 'DEFAULT' else 9050
    proxy_endpoint.addCallback(extract_port_value)
    proxy_endpoint.addCallback(
        lambda port: TCP4ClientEndpoint(reactor, '127.0.0.1', port))
    return proxy_endpoint

def connectServer(self, hostname, port):
        """Tell the proxy what the end server is and start the connection.

        The messages in `self.client_queue` will automatically be consumed.

        This method should only be called once.

        :param str hostname:
        :param int port:
        """
        endpoint = TCP4ClientEndpoint(reactor, hostname, port)
        protocol = ServerProtocol(
            self.server_queue, self.client_queue)
        connectProtocol(endpoint, protocol)

def _connectServer(self, hostname, port, server_queue, client_queue):
        """A helper function for connecting to (hostname, port)
        with the given server and client queues.

        :param str hostname:
        :param int port:
        :param DeferredQueue server_queue:
        :param DeferredQueue client_queue:
        """
        endpoint = TCP4ClientEndpoint(reactor, hostname, port)
        protocol = ServerProtocol(
            server_queue, client_queue)
        connectProtocol(endpoint, protocol)

def subEndpoint(self, reactor, host, port, contextFactory):
        """
        Create an endpoint to connect to based on a single address result from
        L{getaddrinfo}.

        @param reactor: the reactor to connect to
        @type reactor: L{IReactorTCP}

        @param host: The IP address of the host to connect to, in presentation
            format.
        @type host: L{str}

        @param port: The numeric port number to connect to.
        @type port: L{int}

        @param contextFactory: If not L{None}, the OpenSSL context factory to
            use to produce client connections.

        @return: a stream client endpoint that will connect to the given host
            and port via the given reactor.
        @rtype: L{IStreamClientEndpoint}
        """
        if contextFactory is None:
            return TCP4ClientEndpoint(reactor, host, port)
        else:
            return SSL4ClientEndpoint(reactor, host, port, contextFactory)

def endpoint(self, reactor):
        """
        Create an L{IStreamServerEndpoint} that will talk to the node process
        that is described by this L{NodeInfo}.

        @return: an endpoint that will connect to this host.
        @rtype: L{IStreamServerEndpoint}
        """
        return TCP4ClientEndpoint(reactor, self.hostname, self.port)

def make_new_connection(self, host, port):
        logging.debug("NODE: making client connection {}:{}".format(host, port))
        point = TCP4ClientEndpoint(reactor, host, port, timeout=90)
        proto = MyProto(self)
        d = connectProtocol(point, proto)
        d.addCallback(got_protocol).addErrback(my_err_back)

def connect(self, on_connect, on_disconnect=None, on_event=None):
        """Connect to QTM

        :param on_connect: Called on successful connection to QTM. Callback receives an :class:`QRTConnection` object.
        :param on_disconnect: Called if connection fails or when connection is lost.
        :param on_event: Called when there's an event from QTM.

        """
        point = TCP4ClientEndpoint(reactor, self.host, self.port)
        factory = QRTFactory(self.version, on_disconnect, on_event, self.logger)
        try:
            p = yield point.connect(factory)
        except ConnectionRefusedError as reason:
            if on_disconnect:
                on_disconnect(QRTCommandException(str(reason)))
            return
        except Exception as reason:
            if on_disconnect:
                on_disconnect(reason)
            return

        try:
            version = yield p.connected_d
        except Exception as reason:
            if on_disconnect:
                p.on_disconnect = None
                p.transport.loseConnection()
                on_disconnect(reason)
            return

        on_connect(QRTConnection(p), version)

def connect(self):
        log.info('Connecting')
        factory = MQTTFactory(profile=MQTTFactory.PUBLISHER | MQTTFactory.SUBSCRIBER)
        point   = TCP4ClientEndpoint(reactor, self.broker_host, self.broker_port)
        d = point.connect(factory).addCallback(self.gotProtocol)
        d.addErrback(self.on_error)

def client(self, reactor, serverAddress):
        """
        Create a client end point that will connect to the given address.

        @type serverAddress: L{IPv4Address}
        """
        return TCP4ClientEndpoint(reactor, self.interface, serverAddress.port)

def test_tcpHostPositionalArg(self):
        """
        When passed a TCP strports description specifying host as a positional
        argument, L{endpoints.clientFromString} returns a L{TCP4ClientEndpoint}
        instance initialized with the values from the string.
        """
        reactor = object()

        client = endpoints.clientFromString(
            reactor,
            "tcp:example.com:port=1234:timeout=7:bindAddress=10.0.0.2")
        self.assertEqual(client._host, "example.com")
        self.assertEqual(client._port, 1234)

def test_tcpPortPositionalArg(self):
        """
        When passed a TCP strports description specifying port as a positional
        argument, L{endpoints.clientFromString} returns a L{TCP4ClientEndpoint}
        instance initialized with the values from the string.
        """
        reactor = object()
        client = endpoints.clientFromString(
            reactor,
            "tcp:host=example.com:1234:timeout=7:bindAddress=10.0.0.2")
        self.assertEqual(client._host, "example.com")
        self.assertEqual(client._port, 1234)

def test_tcp_riemann(self):

        event = Event('ok', 'sky', 'Sky has not fallen', 1.0, 60.0)

        end = TCP4ClientEndpoint(reactor, "localhost", 5555)

        p = yield connectProtocol(end, riemann.RiemannProtocol())

        yield p.sendEvents([event])

        p.transport.loseConnection()

def test_lantern_circumvent(self):
        def addResultToReport(result):
            self.report['body'] = result
            if result.startswith(self.localOptions['expected-body']):
                log.msg("Got the HTTP response body I expected!")
                self.report['success'] = True
            else:
                self.report['success'] = False

        def addFailureToReport(failure):
            log.err("Failed to connect to lantern")
            log.failure(failure)
            self.report['failure'] = handleAllFailures(failure)
            self.report['success'] = False

        def doRequest(noreason):
            proxyEndpoint = TCP4ClientEndpoint(reactor, '127.0.0.1', 8787)
            agent = ProxyAgent(proxyEndpoint, reactor)
            log.msg("Doing HTTP request via Lantern (127.0.0.1:8787) for %s" % self.url)
            request = agent.request("GET", self.url)
            request.addCallback(readBody)
            request.addCallback(addResultToReport)
            request.addCallback(self.processDirector.close)
            return request

        self.bootstrapped.addCallback(doRequest)
        self.bootstrapped.addErrback(addFailureToReport)
        self.bootstrapped.addBoth(self.stop)
        self.d = self.run(self.command, env=os.environ, usePTY=1)
        return self.d

def test_create_agent(self):
        proxyEndpoint = TCP4ClientEndpoint(reactor, '127.0.0.1', 9050)
        agent = TrueHeadersSOCKS5Agent(reactor, proxyEndpoint=proxyEndpoint)

def connect_to_control_port():
    connection = TCP4ClientEndpoint(reactor, '127.0.0.1',
                                    config.tor.control_port)
    config.tor_state = yield build_tor_connection(connection)

def check_tor(self):
        """
        Called only when we must start tor by director.start
        """
        from ooni.utils.net import ConnectAndCloseProtocol, connectProtocol
        incoherent = []
        if not self.advanced.start_tor:
            if self.tor.socks_port is None:
                incoherent.append('tor:socks_port')
            else:
                socks_port_ep = TCP4ClientEndpoint(reactor,
                                                   "localhost",
                                                   self.tor.socks_port)
                try:
                    yield connectProtocol(socks_port_ep, ConnectAndCloseProtocol())
                except Exception:
                    incoherent.append('tor:socks_port')

            if self.tor.control_port is not None:
                if isinstance(self.tor.control_port, int):
                    control_port_ep = TCP4ClientEndpoint(reactor,
                                                         "localhost",
                                                         self.tor.control_port)
                    try:
                        yield connectProtocol(control_port_ep, ConnectAndCloseProtocol())
                    except Exception:
                        incoherent.append('tor:control_port')
                else:
                    conf_unix_socket_path = self.tor.control_port.lstrip()
                    if conf_unix_socket_path.startswith("unix:"):
                        if os.path.exists(conf_unix_socket_path.lstrip("unix:")):
                            unix_socket_path = conf_unix_socket_path.lstrip("unix:")
                        else:
                            incoherent.append('tor:control_port')
                    else:
                        incoherent.append('tor:control_port')

            self.log_incoherences(incoherent)

def sendPayload(self, payload):
        d1 = defer.Deferred()

        def closeConnection(proto):
            self.report['sent'].append(proto.sent_data)
            self.report['received'].append(proto.received_data)
            proto.transport.loseConnection()
            log.debug("Closing connection")
            d1.callback(proto.received_data)

        def timedOut(proto):
            self.report['failure'] = 'tcp_timed_out_error'
            proto.transport.loseConnection()

        def errback(failure):
            self.report['failure'] = failureToString(failure)
            d1.errback(failure)

        def connected(proto):
            log.debug("Connected to %s:%s" % (self.address, self.port))
            proto.report = self.report
            proto.deferred = d1
            proto.sendPayload(payload)
            if self.timeout:
                # XXX-Twisted this logic should probably go inside of the protocol
                reactor.callLater(self.timeout, closeConnection, proto)

        point = TCP4ClientEndpoint(reactor, self.address, self.port)
        log.debug("Connecting to %s:%s" % (self.address, self.port))
        d2 = point.connect(TCPSenderFactory())
        d2.addCallback(connected)
        d2.addErrback(errback)
        return d1

def _test_proxy_alive(self, host, port, protocol, proxy_type,
                          url=b'http://www.baidu.com', timeout=10):
        endpoint = TCP4ClientEndpoint(reactor, host, int(port))
        agent = ProxyAgent(endpoint)
        d = agent.request(b'GET', url)
        self.currents += 1
        proxy = '{}:{}'.format(host, port)
        key = 'proxy_info:' + proxy

        if proxy_type == 'rookies_checking':
            def _callback(ignored):
                pipe = self.conn.pipeline(False)
                pipe.zrem('rookies_checking', proxy)
                pipe.hset(key, 'failed_times', 0)
                # Move proxy from rookies to availables
                pipe.smove('rookie_proxies', 'available_proxies',
                           '{}://{}'.format(protocol, proxy))
                pipe.zadd('availables_checking', proxy, time.time() + 30)
                pipe.execute()

            def _errback(err):
                if self.conn.hincrby(key, 'failed_times', 1) < 3:
                    # If not reach the maximum of failed_times
                    # Since it is not important so re-check it after 10 seconds
                    self.conn.zadd('rookies_checking', proxy, time.time() + 10)
                else:
                    pipe = self.conn.pipeline(False)
                    pipe.zrem('rookies_checking', proxy)
                    pipe.smove('rookie_proxies', 'dead_proxies',
                               '{}://{}'.format(protocol, proxy))
                    pipe.execute()
        else:
            def _callback(ignored):
                pipe = self.conn.pipeline(False)
                pipe.hset(key, 'failed_times', 0)
                pipe.zadd('availables_checking', proxy, time.time() + 30)
                pipe.smove('lost_proxies', 'available_proxies',
                           '{}://{}'.format(protocol, proxy))
                pipe.execute()

            def _errback(err):
                pipe = self.conn.pipeline(False)
                if self.conn.hincrby(key, 'failed_times', 1) < 3:
                    pipe.zadd('availables_checking', proxy, time.time() + 10)
                    pipe.smove('available_proxies', 'lost_proxies',
                               '{}://{}'.format(protocol, proxy))
                else:
                    pipe.zrem('availables_checking', proxy)
                    pipe.smove('lost_proxies', 'dead_proxies',
                               '{}://{}'.format(protocol, proxy))
                    pipe.delete(key)
                pipe.execute()

        d.addCallbacks(_callback, _errback)
        reactor.callLater(timeout, d.cancel)

        def _clean(ignored):
            self.currents -= 1

        d.addBoth(_clean)

def run(config, bcast, discovery_addr):
    f = MyFactory(config)

    try:
        port = reactor.listenTCP(config.port, f)
        config.port = port.getHost().port
    except error.CannotListenError:
        logging.error("cannot listen on {}".format(config.port))
        sys.exit(1)

    # connect to discovery server
    point = TCP4ClientEndpoint(reactor, discovery_addr, 8123, timeout=90)
    d = connectProtocol(point, Discovery({}, f))
    d.addCallback(got_discovery, b64encode(f.vk), config.port).addErrback(my_err_back)

    # connect to myself
    point = TCP4ClientEndpoint(reactor, "localhost", config.port, timeout=90)
    d = connectProtocol(point, MyProto(f))
    d.addCallback(got_protocol).addErrback(my_err_back)

    if bcast:
        call_later(5, f.overwrite_promoters)

    # optionally run tests, args.test == None implies reactive node
    # we use call later to wait until the nodes are registered
    if config.test == 'dummy':
        call_later(5, f.bcast, pb.Dummy(m='z'))
    elif config.test == 'bracha':
        call_later(6, f.bracha.bcast_init)
    elif config.test == 'mo14':
        call_later(6, f.mo14.start, config.value)
    elif config.test == 'acs':
        # use port number (unique on local network) as test message
        call_later(6, f.acs.start, str(config.port), 1)
    elif config.test == 'tc':
        call_later(5, f.tc_runner.make_tx, 1.0 / config.tx_rate, True)
        # optionally use validate
        if config.validate:
            call_later(10, f.tc_runner.make_validation)
    elif config.test == 'bootstrap':
        call_later(5, f.tc_runner.bootstrap_promoters)

    logging.info("NODE: reactor starting on port {}".format(config.port))
    reactor.run()

def test_stopStartReading(self):
        """
        This test verifies transport socket read state after multiple
        pause/resumeProducing calls.
        """
        sf = ServerFactory()
        reactor = sf.reactor = self.buildReactor()

        skippedReactors = ["Glib2Reactor", "Gtk2Reactor"]
        reactorClassName = reactor.__class__.__name__
        if reactorClassName in skippedReactors and platform.isWindows():
            raise SkipTest(
                "This test is broken on gtk/glib under Windows.")

        sf.protocol = StopStartReadingProtocol
        sf.ready = Deferred()
        sf.stop = Deferred()
        p = reactor.listenTCP(0, sf)
        port = p.getHost().port
        def proceed(protos, port):
            """
            Send several IOCPReactor's buffers' worth of data.
            """
            self.assertTrue(protos[0])
            self.assertTrue(protos[1])
            protos = protos[0][1], protos[1][1]
            protos[0].transport.write(b'x' * (2 * 4096) + b'y' * (2 * 4096))
            return (sf.stop.addCallback(cleanup, protos, port)
                           .addCallback(lambda ign: reactor.stop()))

        def cleanup(data, protos, port):
            """
            Make sure IOCPReactor didn't start several WSARecv operations
            that clobbered each other's results.
            """
            self.assertEqual(data, b'x'*(2*4096) + b'y'*(2*4096),
                                 'did not get the right data')
            return DeferredList([
                    maybeDeferred(protos[0].transport.loseConnection),
                    maybeDeferred(protos[1].transport.loseConnection),
                    maybeDeferred(port.stopListening)])

        cc = TCP4ClientEndpoint(reactor, '127.0.0.1', port)
        cf = ClientFactory()
        cf.protocol = Protocol
        d = DeferredList([cc.connect(cf), sf.ready]).addCallback(proceed, p)
        d.addErrback(log.err)
        self.runReactor(reactor)

def _request(self, method, urn, genReceiver, bodyProducer=None, retries=3):
        if self.backend_type == 'onion':
            agent = TrueHeadersSOCKS5Agent(reactor,
                                           proxyEndpoint=TCP4ClientEndpoint(reactor,
                                                                            '127.0.0.1',
                                                                            config.tor.socks_port))
        else:
            agent = Agent(reactor)

        attempts = 0

        finished = defer.Deferred()

        def perform_request(attempts):
            uri = urljoin(self.base_address, urn)
            d = agent.request(method, uri, bodyProducer=bodyProducer,
                              headers=Headers(self.base_headers))

            @d.addCallback
            def callback(response):
                try:
                    content_length = int(response.headers.getRawHeaders('content-length')[0])
                except:
                    content_length = None
                response.deliverBody(genReceiver(finished, content_length))

            def errback(err, attempts):
                # We we will recursively keep trying to perform a request until
                # we have reached the retry count.
                if attempts < retries:
                    log.err("Lookup {} failed. Retrying.".format(uri))
                    attempts += 1
                    perform_request(attempts)
                else:
                    log.err("Failed. Giving up.")
                    finished.errback(err)

            d.addErrback(errback, attempts)

        perform_request(attempts)

        return finished