python - 从现有的阻塞代码重构扭曲的 tcp 客户端

Question

我有一个 python 程序，目前使用我编写的 tcp/ip 客户端模块从流式服务器接收数据。服务器输出数据行。

我的 TCP 客户端类相当原始，我想重构以使用扭曲的 ReconnectingClientFactory。

主程序当前从我的 TCP 客户端中的 readLines 函数获取数据，该函数在接收到行时“产生”这些行。

TCP 客户端方法被访问:

for msg_buffer in self.myTcpClient.readLines():
    do some stuff with the data in msg_buffer

在我的 TCP 客户端中，readLines 方法本质上是这样的:

while True:
    newLine = self.sock.recv(self.buffer_size)
    yield newLine

当我实现 twisted 客户端时，我需要一些方法让它像迭代器一样运行并产生数据。我假设我会在协议(protocol) dataReceived 方法中做一些事情。

我迷失了试图弄清楚这是如何工作的。在我看来，扭曲的延迟是为了这种用途，但我不知道如何为我的目的使用延迟(如果我对延迟的假设是正确的)。

在一个完美的世界中，扭曲的客户端会产生接收到的行，因此类似于本方法的调用可以完成这项工作。即

class GetData(protocol):
    def dataReceived(self, data):
        yield data

但我认为这过于简单化了。

总而言之，我想做的是实现一个扭曲的重新连接 TCP 客户端，它的行为类似于我的 readLines 方法，并且可以或多或少地像这样访问:

for msg_buffer in self.twistedTcpClient.readLines():

任何指点将不胜感激

更新: 我只是偶然发现了“钩针编织”的意思。乍一看，Crochet 似乎是为我需要的那种模型设计的……我会在一些测试后报告

Answer 1

执行此操作的 Twisted 方法是编写一个协议(protocol)。而不是做:

for line in self.twistedTcpClient.readLines():
    process_line(line) ...

您将编写您的协议(protocol)(可能通过子类化 twisted.protocols.basic.LineReceiver):

class MyProtocol(LineReceiver):
    ...
    def lineReceived(self, line):
        process_line(line) ...

您想重构代码以使用 lineReceived 回调而不是迭代循环。

你写了什么:

for line in self.twistedTcpClient.readLines():
    process_line(line) ...

是有问题的，因为 Twisted 是异步的。在等待 twistedTcpClient.readLines() 方法时，Twisted 没有办法做任何其他事情。

我建议写一个协议(protocol)，但如果你真的坚持要有这个迭代器模式，那么你也许可以这样做:

@inlineCallbacks
def my_func():
    while True:
        try:
            line = yield self.twistedTcpClient.getNextLine()
        except StopIteration:
            break

        process_line(line) ...

现在，棘手的事情是让 twistedTcpClient 为每次调用 getNextLine() 返回 Deferreds。也许是这样的:

class MyProtocol(LineReceiver):
    ...
    def getNextLine(self):
        self.defer_given_out = Deferred()

    def lineReceived(self, line):
        self.defer_given_out.callback(line)

    def connectionLost(self):
        self.defer_given_out.errback(StopIteration())

(这只是一个说明这个想法的例子，你必须扩展它来处理细节。)