使用 Twisted 阻止 Thrift 调用

Question

我有一个使用 TTwisted 协议的 Twisted/Thrift 服务器。我希望保持与客户端的连接打开，直到发生给定事件（Twisted 反应器通过回调通知我此事件）。

class ServiceHandler(object):
    interface.implements(Service.Iface)

    def listenForEvent(self):
        """
        A method defined in my Thrift interface that should block the Thrift
        response until my event of interest occurs.
        """
        # I need to somehow return immediately to free up the reactor thread,
        # but I don't want the Thrift call to be considered completed. The current
        # request needs to be marked as "waiting" somehow.

    def handleEvent(self):
        """
        A method called when the event of interest occurs. It is a callback method
        registered with the Twisted reactor.
        """
        # Find all the "waiting" requests and notify them that the event occurred.
        # This will cause all of the Thrift requests to complete.

如何快速从处理程序对象的方法返回，同时保持阻塞 Thrift 调用的错觉？

---

我从 Twisted 启动触发器初始化 Thrift 处理程序：

def on_startup():
    handler = ServiceHandler()                      
    processor = Service.Processor(handler)                                   
    server = TTwisted.ThriftServerFactory(processor=processor,                  
        iprot_factory=TBinaryProtocol.TBinaryProtocolFactory())                 
    reactor.listenTCP(9160, server)

我的 PHP 客户端连接：

  $socket = new TSocket('localhost', 9160);
  $transport = new TFramedTransport($socket);
  $protocol = new TBinaryProtocol($transport);
  $client = new ServiceClient($protocol);
  $transport->open();
  $client->listenForEvent();

最后一次调用 ($client->listenForEvent()) 成功将其转移到服务器并运行 ServiceHandler.listenForEvent，但即使该服务器方法返回一个 twisted.internet.defer.Deferred() 实例，客户端也会立即收到一个空数组，并且出现异常：

异常“TTransportException”，消息“TSocket：超时” 从 localhost:9160 读取 4 个字节到本地端口 38395'

Answer 1

您应该能够从 listenForEvent 返回 Deferred。稍后 handleEvent 应该触发返回的 Deferred（或返回的 Deferreds）以实际生成响应。

Answer 2

您看到的错误似乎表明传输未成帧（Twisted 需要它，以便它可以事先知道每条消息的长度）。另外，Thrift 服务器支持从处理程序返回延迟，所以这更奇怪。您是否尝试过返回 defer.succeed("some value") 并查看 deferreds 是否真正有效？然后您可以移至此处检查它是否完全正常工作：

    d = defer.Deferred()
    reactor.callLater(0, d.callback, results)
    return d