Twisted 中的异步客户端不发送/接收请求（使用 NetStringReceiver）

Question

我正在尝试使用 Twisted 进行异步编程。我想要做的是创建一个客户端，它将传递查询参数（在我的例子中，哈希类型和哈希值）并等待服务器的响应。我目前正在使用 NetstringReceiver 作为参数。但是，我遇到了这些问题：

1. 客户端无法将其请求发送到服务器，并且
2. 当我运行它时，客户端永远挂起。似乎有一个没有返回的回调。

以下是客户端和服务器的代码。 这段代码其实就是基于这个tutorial by Dave Peticolas。

客户端代码

import os, sys, argparse
from twisted.internet import defer
from twisted.internet.protocol import Protocol, ClientFactory
from twisted.protocols.basic import NetstringReceiver

class QueryProtocol(Protocol):
    response = ''

    def dataReceived(self, data):
        self.response = data

    def connectionLost(self, reason):
        self.responseReceived(self.response)

    def responseReceived(self, response):
        self.factory.response_finished(response)

class QueryFactory(ClientFactory):
    protocol = QueryProtocol

    def __init__(self, deferred):
        self.deferred = deferred

    def response_finished(self, response):
        if self.deferred is not None:
            d, self.deferred = self.deferred, None
            d.callback(response)

    def clientConnectionFailed(self, connector, reason):
        if self.deferred is not None:
            d, self.deferred = self.deferred, None
            d.errback(reason)

class QueryNetProtocol(NetstringReceiver):
    def connectionMade(self):
        self.sendRequest(self.factory.hash_type, self.factory.hash_value)

    def sendRequest(self, hash_type, hash_value):
        self.sendString(hash_type + '.' + hash_value)

    def stringReceived(self, s):
        self.transport.loseConnection()
        self.responseReceived(s)

    def responseReceived(self, response):
        self.factory.handleResponse(response)

class QueryNetFactory(ClientFactory):
    protocol = QueryNetProtocol

    def __init__(self, hash_type, hash_value):
        self.hash_type = hash_type
        self.hash_value = hash_value
        self.deferred = defer.Deferred()

    def handleResponse(self, response):
        d, self.deferred = self.deferred, None
        d.callback(response)

    def clientConnectionLost(self, _, reason):
        if self.deferred is not None:
            d, self.deferred = self.deferred, None
            d.errback(reason)

    clientConnectionFailed = clientConnectionLost

class QueryProxy(object):
    def __init__(self, host, port):
        self.host = host
        self.port = port

    def query(self, hash_type, hash_value):
        factory = QueryNetFactory(hash_type, hash_value)
        from twisted.internet import reactor
        reactor.connectTCP(self.host, self.port, factory)
        return factory.deferred

def perform_query(host, port):
    d = defer.Deferred()
    from twisted.internet import reactor
    factory = QueryFactory(d)
    reactor.connectTCP(host, port, factory)
    return d

def main(options):
    done = False
    query_result = ""
    host = options.host
    port = int(options.port)
    sha1 = options.sha1
    proxy = QueryProxy(host, port)
    from twisted.internet import reactor

    def process_query_result(response):
        d = proxy.query('sha1', sha1)

        def fail(err):
            print "Problem in processing response : %s" % err
            return response

        return d.addErrback(fail)

    def query_ok(response):
        query_result = response
        done = True

    def query_failed(err):
        print  "Problem in query : %s" % err
        done = True

    def query_done(_):
        if done == True: reactor.stop()

    d = perform_query(host, port)
    d.addCallback(process_query_result)
    d.addCallbacks(query_ok, query_failed)
    d.addBoth(query_done)
    reactor.run()
    print "The result of the query is : %s" % query_result

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("host", help="server host/ip")
    parser.add_argument("port", help="server port number to listen to")
    parser.add_argument("-sha1", help="sha1 value to be queried")
    options = parser.parse_args()
    main(options)

服务器代码

import os, sys, argparse
from twisted.internet.protocol import ServerFactory
from twisted.protocols.basic import NetstringReceiver


class GridQueryService(object):
    def query(self, hash_type, hash_value):
        print "this is the query service. Type is %s and value is %s" % (hash_type, hash_value)
        return hash_value

class GridQueryProtocol(NetstringReceiver):
    def stringReceived(self, request):
        print >>sys.stderr, request
        if '.' not in request: 
            self.transport.loseConnection() 
            return
        hash_type, hash_value = request.split('.')
        self.formRequestReceived(hash_type, hash_value)

    def formRequestReceived(self, hash_type, hash_value):
        found_flag = self.factory.query(hash_type, hash_value)
        if found_flag: self.sendString(str(found_flag))
        self.transport.loseConnection()

class GridQueryFactory(ServerFactory):
    protocol = GridQueryProtocol

    def __init__(self, service):
        self.service = service

    def query(self, hash_type, hash_value):
        return self.service.query(hash_type, hash_value)

def main(options):
    grid_query_service = GridQueryService()
    grid_query_factory = GridQueryFactory(grid_query_service)
    from twisted.internet import reactor
    port = reactor.listenTCP(int(options.port), grid_query_factory, interface=options.host)
    print "Serving GRID query service on %s" % str(port.getHost())
    reactor.run()

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("host", help="server host/ip")
    parser.add_argument("port", help="server port number to listen to")
    options = parser.parse_args()
    main(options)

关于如何解决这个问题的任何想法？提前致谢。非常感谢您的帮助！

Answer 1

您的服务器可能使用网络字符串，您的客户端可能包含使用网络字符串的代码，但您的客户端不使用网络字符串。

main 调用 perform_query。 perform_query 产生一个 QueryFactory 连接一个 QueryProtocol，它从不发送任何数据，并且即使发送了也没有任何与处理网络字符串相关的逻辑。

Answer 2

我已经更新了我的客户端。修改后的客户端代码如下。 我现在可以发送请求了。以下是输出：

Sent request! 
String received! 
Handling response! 
I'm called!
Connection lost! - QueryNetFactory

如您所见，回调 process_query_result 回调已被触发。但是，其他回调没有，我仍然无法使用 QueryFactory/QueryProtocol 接收数据/结果。我怀疑它与延迟和回调有关，但我很困惑。 在初始化 QueryFactory 类时，我真的应该创建一个新的 deferred 吗？如果是，它将如何实现/知道添加到原始延迟的回调（由 perform_query 方法返回的回调）？现在，如果我不应该创建一个新的 deferred，我该如何触发在 QueryProtocol 的 'responseReceived' 方法中添加到原始 deferred 中的回调？

客户端代码

class QueryProtocol(Protocol):
    response = ''

    def dataReceived(self, data):
        print "Data received!"
        self.response = data

    def connectionLost(self, reason):
        print "Connection lost!"
        self.responseReceived(self.response)

    def responseReceived(self, response):
        print "Response received!"
        self.factory.response_finished(response)

class QueryFactory(ClientFactory):
    protocol = QueryProtocol

    def __init__(self):
        self.deferred = defer.Deferred()

    def response_finished(self, response):
        print "Response finished!"
        if self.deferred is not None:
            d, self.deferred = self.deferred, None
            d.callback(response)

    def clientConnectionFailed(self, connector, reason):
        print "Client connection failed! - QueryFactory"
        if self.deferred is not None:
            d, self.deferred = self.deferred, None
            d.errback(reason)

class QueryNetProtocol(NetstringReceiver):
    def connectionMade(self):
        self.sendRequest(self.factory.hash_type, self.factory.hash_value)

    def sendRequest(self, hash_type, hash_value):
        print "Sent request!"
        self.sendString(hash_type + '.' + hash_value)

    def stringReceived(self, s):
        print "String received!"
        self.transport.loseConnection()
        self.responseReceived(s)

    def responseReceived(self, response):
        self.factory.handleResponse(response)

class QueryNetFactory(ClientFactory):
    protocol = QueryNetProtocol

    def __init__(self, deferred, hash_type, hash_value):
        self.hash_type = hash_type
        self.hash_value = hash_value
        self.deferred = deferred

    def handleResponse(self, response):
        print "Handling response!"
        d, self.deferred = self.deferred, None
        d.callback(response)

    def clientConnectionLost(self, _, reason):
        print "Connection lost! - QueryNetFactory"
        if self.deferred is not None:
            d, self.deferred = self.deferred, None
            d.errback(reason)

    clientConnectionFailed = clientConnectionLost

class QueryProxy(object):
    def __init__(self, host, port):
        self.host = host
        self.port = port

    def query(self):
        factory = QueryFactory()
        from twisted.internet import reactor
        reactor.connectTCP(self.host, self.port, factory)
        return factory.deferred

def perform_query(host, port, hash_type, hash_value):
    d = defer.Deferred()
    from twisted.internet import reactor
    factory = QueryNetFactory(d, hash_type, hash_value)
    reactor.connectTCP(host, port, factory)
    return d

def main(options):
    done = False
    query_result = ""
    host = options.host
    port = int(options.port)
    sha1 = options.sha1
    proxy = QueryProxy(host, port)
    from twisted.internet import reactor

    def process_query_result(response):
        print "I'm called!"
        d = proxy.query()

        def fail(err):
            print "Process query result failure : %s" % err

        return d.addErrback(fail)

    def query_ok(response):
        print "query ok!"
        query_result = response
        done = True

    def query_failed(err):
        print  "Problem in query : %s" % err
        done = True

    def query_done(_):
        if done == True: reactor.stop()

    d = perform_query(host, port, "sha1", sha1)
    d.addCallback(process_query_result)
    d.addCallbacks(query_ok, query_failed)
    d.addBoth(query_done)
    reactor.run()
    print "The result of the query is : %s" % query_result

再次感谢任何帮助！谢谢！