使用 twisted/asyncio 缓冲应用程序

Question

我有一个简单的应用程序，它接收数据并且必须将其分批推送到另一个服务。

如何使用 twisted 或 asyncio 实现它？

目前，我有下一个使用扭曲的代码：

from twisted.internet import protocol, reactor, endpoints
from twisted.protocols import basic


class FirehoseProtocol(basic.LineReceiver):
    def __init__(self):
        self.data = []

    def lineReceived(self, line):
        self.data.append(line)

    def push_to_firehose(self):
        pass  # TODO


class EchoFactory(protocol.ServerFactory):
    protocol = FirehoseProtocol


endpoints.serverFromString(reactor, "tcp:5001").listen(EchoFactory())

reactor.run()

Answer 1

你没有说什么定义了批次。如果我们假设多行定义一个批次：

def lineReceived(self, line):
    self.data.append(line)
    if len(self.data) == batch_size:
        self.push_to_firehose()