Python twisted.protocols.basic 模块，Int32StringReceiver() 实例源码

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