python 3：使用异步任务时如何等待异步函数中的回调答案

【问题标题】：python 3 : how to wait for the callback in async function when use asyncio taskspython 3：使用异步任务时如何等待异步函数中的回调
【发布时间】：2020-05-15 14:07:02
【问题描述】：

AWS IoT python Sdk有一个函数publishAsync
函数的签名是publishAsync(topic, payload, QoS, ackCallback=None)

我想把它变成一个协程并使用 asyncio。

我想做的是：

连接到 AWS
异步发布 10 条消息
断开连接

我不知道应该如何将该函数包装为异步函数。

async def asyncPublish(self,
                     msg:str,
                     topic:str,
                     QoS=1):

        # the publishAckFn
        def internalPubAckHandler(mid):
            print(json.dumps({
                'messageID':mid,
                'acknowledged':True
            }))
            return True
        pass

        # publish to the topic asynchronously
        messageID = self.awsIoTClient.publishAsync(topic,msg,QoS,ackCallback=internalPubAckHandler)
        print(json.dumps({
           'messageID':messageID,
           'topic':topic,
           'payload':msg,
           'QoS':QoS,
           'async':True
        }))

    pass


---- in my main file
tasks = []
for i in range(10):
        tasks += asyncPublish('test','test')
pass

loop = asyncio.get_event_loop()
loop.set_debug(True)  # debug mode
gw1.connect()
loop.run_until_complete(asyncio.wait(tasks))
loop.close()
gw1.disconnect()

但是当我这样做时，断开连接将比 internalPubAckHandler 更快地调用而且我不会得到任何确认。

【问题讨论】：

标签： python aws-iot

【解决方案1】：

最后，我找到了解决方案。我们可以使用期货。

Future 对象用于将基于回调的低级代码与高级异步/等待代码连接起来。 Ref

如果函数的签名是这样的：

def functionWithCallback(args,callback)

我们需要将它封装到一个等待回调的异步函数中，我们可以执行如下操作：

async def asyncVersion(args):

    # one way of creating a future
    loop = asyncio.get_running_loop()
    future = loop.create_future()

    def futureCallback():
        # do something here and then signal the future
        ...
        loop.call_soon_threadsafe(future .set_result,time.time())
    pass

    # call your function
    functionWithCallback(args,futureCallback)

    # wait for the callback
    await future


    # do other stuff after the callback is done
    .....

pass

【讨论】：