如何包装 grpc 返回的 _Rendezvous 未来答案

【问题标题】：How to wrap _Rendezvous future returned by grpc如何包装 grpc 返回的 _Rendezvous 未来
【发布时间】：2020-04-26 16:08:50
【问题描述】：

我想用python中的拦截器为grpc客户端实现重试逻辑。

如果我们想为 grpc 的异步调用实现它，它会变得有点棘手。我们知道我们可以像这样得到一个异步 grpc 调用：

future_response = stub.SayHello.future(SayHelloRequest())

所以如果我想实现重试逻辑，我需要检查未来是否有异常：

future_response.exception()

实际上我们不能这样做，因为在响应到来之前它会被阻塞。所以我们需要使用add_done_callback返回future的方法。所以要做到这一点，我需要一种方法来通过 grpc 调用来包装返回的未来，但我不知道任何安全的方法来做到这一点。异步 grpc 调用返回的对象类型是_Rendezvous，我需要用另一个_Rendezvous 未来包装这个对象，这样我就可以在主未来的回调中手动调用set_result。但我在 grpc 代码中找不到合适的方法或任何辅助函数来让我这样做。

【问题讨论】：

_Rendezvous 是RpcError 和RpcContext 的具体实现。这些接口都没有为您提供提取失败的请求有效负载的方法，因此“包装”此对象不是您想要重试 RPC 的操作。相反，您应该以某种方式跟踪需要重新发送的请求并启动新的 RPC。

标签： python grpc future

【解决方案1】：

您可以使用以下函数对其进行包装：

import asyncio

def wrap_grpc_future(grpc_future, loop=None):
    """Wrap grpc.Future into awaitable asyncio Future."""

    if asyncio.isfuture(grpc_future):
        return grpc_future
    if loop is None:
        loop = asyncio.get_running_loop()
    future = loop.create_future()

    def done(grpc_future):
        try:
            if grpc_future.cancelled():
                future.cancel()

            if not future.cancelled():
                e = grpc_future.exception()
                if e:
                    future.set_exception(e)
                else:
                    future.set_result(grpc_future.result())

        except Exception as e:
            future.set_exception(e)

    grpc_future.add_done_callback(done)

    return future

还有一个新的gRPC AsyncIO API，但文档很少。

【讨论】：