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

Question

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

客户端无法向服务器发送请求，
当我运行客户端永远挂起。好像有回调没有返回

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

Client Code

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)

Server Code

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，它从不发送任何数据并且没有任何与处理 netstrings 相关的逻辑，即使它发送了也是如此。

Answer 2

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

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

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

Client Code

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

再次感谢您的帮助！谢谢！

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

Asynchronous client in Twisted not sending / receiving request (using NetStringReceiver)

python

asynchronous

twisted