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

def _connect_to_management(self, retries=30):
        if retries == 0:
            self.log.error('Timeout while connecting to management')
            self.failed = True
            return

        def retry(retries):
            ctr = retries - 1
            self.log.warn(
                'Error connecting to management, retrying. '
                'Retries left:  %s' % ctr)
            reactor.callLater(
                0.1, self._connect_to_management, ctr)

        self._d = connectProtocol(
            self._management_endpoint,
            ManagementProtocol(verbose=True))
        self._d.addCallbacks(
            self._got_management_protocol,
            lambda f: retry(retries))

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 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 _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 test_onion_datagram_proxy():
    received_buffer = []
    received_d = defer.Deferred()

    def received(data):
        received_buffer.append(data)
        received_d.callback(None)

    received_size = 10
    proxy_factory = OnionDatagramProxyFactory(received)
    protocol = proxy_factory.buildProtocol(123)
    packet = b"A" * received_size
    protocol.stringReceived(packet)
    assert received_buffer[0] == packet

    service_port = yield txtorcon.util.available_tcp_port(reactor)
    service_endpoint_desc = "tcp:interface=127.0.0.1:%s" % service_port
    service_endpoint = endpoints.serverFromString(reactor, service_endpoint_desc)
    yield service_endpoint.listen(proxy_factory)

    client_endpoint_desc = "tcp:127.0.0.1:%s" % service_port
    client_endpoint = endpoints.clientFromString(reactor, client_endpoint_desc)
    client_protocol = Int32StringReceiver()
    yield endpoints.connectProtocol(client_endpoint, client_protocol)
    client_protocol.sendString(packet)
    print "BEFORE CLOSE"
    client_protocol.transport.loseConnection()
    yield received_d
    assert received_buffer[0] == packet

def test_onion_transport():
    """
    integration test for onion transport
    """
    chutney_control_port = os.environ.get('CHUTNEY_CONTROL_PORT')
    if chutney_control_port is None:
        print "CHUTNEY_CONTROL_PORT not set, aborting test"
        return

    params = SphinxParams(max_hops=5, payload_size=1024)
    sphinx_packet_size = params.get_sphinx_forward_size()
    transport_factory = create_transport_factory(sphinx_packet_size, chutney_control_port)
    transport = yield transport_factory.build_transport()
    received_d = defer.Deferred()
    received_buffer = []

    def packet_received(packet):
        print "packet received of len %s" % len(packet)
        received_buffer.append(packet)
        received_d.callback(None)

    protocol = FakeMixProtocol(packet_received)
    yield protocol.make_connection(transport)
    onion_host, onion_port = transport.addr
    tor_endpoint = transport.tor.stream_via(onion_host, onion_port)
    send_message_protocol = Int32StringReceiver()
    remote_mix_protocol = yield endpoints.connectProtocol(tor_endpoint, send_message_protocol)
    message = b"A" * sphinx_packet_size
    remote_mix_protocol.sendString(message)
    remote_mix_protocol.transport.loseConnection()
    yield received_d
    assert received_buffer[0] == message

def do_send(self, addr, message):
        """
        send message to addr
        where addr is a 2-tuple of type: (onion host, onion port)
        """
        onion_host, onion_port = addr
        tor_endpoint = self.tor.stream_via(onion_host, onion_port)
        send_message_protocol = Int32StringReceiver()
        self.remote_mix_protocol = yield endpoints.connectProtocol(tor_endpoint, send_message_protocol)
        self.remote_mix_protocol.sendString(message)
        self.remote_mix_protocol.transport.loseConnection()

    # Protocol parent method overwriting

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 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 channelOpen(self, specificData):
        """
        See: L{channel.SSHChannel}
        """
        log.msg("connecting to %s:%i" % self.hostport)
        ep = HostnameEndpoint(
            self._reactor, self.hostport[0], self.hostport[1])
        d = connectProtocol(ep, SSHForwardingClient(self))
        d.addCallbacks(self._setClient, self._close)
        self._channelOpenDeferred = d

def test_connectProtocolReturnsConnectResult(self):
        """
        C{endpoints.connectProtocol} returns the result of calling the given
        endpoint's C{connect()} method.
        """
        result = defer.Deferred()
        class Endpoint:
            def connect(self, factory):
                """
                Return a marker object for use in our assertion.
                """
                return result

        endpoint = Endpoint()
        self.assertIs(result, endpoints.connectProtocol(endpoint, object()))

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