使用带有 Twisted 回调的 async/await 语法答案

【问题标题】：Using async/await syntax with Twisted callbacks使用带有 Twisted 回调的 async/await 语法
【发布时间】：2017-11-20 05:28:52
【问题描述】：

我想在 Twisted Deferred.addCallback 方法中使用 async/await 语法。但正如文档中所述，addCallback 回调是同步调用的。我已经看到 inlineCallbacks 装饰器用于此目的，但我更喜欢使用 async/await 语法（如果它甚至可能或有意义的话）。

我从pika documentation 获取了原始代码，但我没有运气尝试将其迁移到异步/等待语法：

import pika
from pika import exceptions
from pika.adapters import twisted_connection
from twisted.internet import defer, reactor, protocol, task


async def run_async(connection):
    channel = await connection.channel()
    exchange = await channel.exchange_declare(exchange='topic_link',type='topic')
    queue = await channel.queue_declare(queue='hello', auto_delete=False, exclusive=False)
    await channel.queue_bind(exchange='topic_link', queue='hello', routing_key='hello.world')
    await channel.basic_qos(prefetch_count=1)
    queue_object, consumer_tag = await channel.basic_consume(queue='hello', no_ack=False)
    l = task.LoopingCall(read_async, queue_object)
    l.start(0.01)


async def read_async(queue_object):
    ch,method,properties,body = await queue_object.get()
    if body:
        print(body)
    await ch.basic_ack(delivery_tag=method.delivery_tag)


parameters = pika.ConnectionParameters()
cc = protocol.ClientCreator(reactor, twisted_connection.TwistedProtocolConnection, parameters)
d = cc.connectTCP('rabbitmq', 5672)
d.addCallback(lambda protocol: protocol.ready)
d.addCallback(run_async)
reactor.run()

这显然行不通，因为没有人等待run_async 函数。

【问题讨论】：

ensureDeferred 是你要使用的
@notorious.no 我已经尝试用defer.ensureDeferred 包裹cc.connectTCP('rabbitmq', 5672)，但它没有帮助。那是你的意思吗？谢谢。
@notorious.no 明白了。我必须用 ensureDeferred 包装回调本身。我正在包装显然没有成功的回调结果。谢谢。

标签： python asynchronous twisted python-3.5 python-asyncio

【解决方案1】：

正如 notorious.no 和 Twisted 文档所指出的，ensureDeferred 是要走的路。不过，您必须包装回调结果，而不是我不清楚的回调本身。

这就是它最终的样子：

def ensure_deferred(f):
    @functools.wraps(f)
    def wrapper(*args, **kwargs):
        result = f(*args, **kwargs)
        return defer.ensureDeferred(result)
    return wrapper


@ensure_deferred
async def run(connection):
    channel = await connection.channel()
    exchange = await channel.exchange_declare(exchange='topic_link', type='topic')
    queue = await channel.queue_declare(queue='hello', auto_delete=False, exclusive=False)
    await channel.queue_bind(exchange='topic_link', queue='hello', routing_key='hello.world')
    await channel.basic_qos(prefetch_count=1)
    queue_object, consumer_tag = await channel.basic_consume(queue='hello', no_ack=False)
    l = task.LoopingCall(read, queue_object)
    l.start(0.01)


@ensure_deferred
async def read(queue_object):
    ch, method, properties, body = await queue_object.get()
    if body:
        print(body)
    await ch.basic_ack(delivery_tag=method.delivery_tag)


parameters = pika.ConnectionParameters()
cc = protocol.ClientCreator(reactor, twisted_connection.TwistedProtocolConnection, parameters)
d = cc.connectTCP('rabbitmq', 5672)
d.addCallback(lambda protocol: protocol.ready)
d.addCallback(run)
reactor.run()

谢谢。

【讨论】：