[help request] How to wrap publishAsync in a Future

[aefn]

Dear Experts,
I have a scenario like this:
1 . connect
2 . publish 10 packet asynchronously
3. disconnect

the signature of publishAsync is publishAsync(topic, payload, QoS, ackCallback=None) link

how can I wrap this function to a async function?
I want to have something like this:

async def publish:
    publishAsync(arg,cb)  # I don't know how should I wait for this, using await does not work
pass


tasks = [publish for i in range(10)]
connect()
loop.run(tasks) # run all in parallel and wait until get their pubAck
disconnect()

I appreciate in advance
ps : I know that maybe it can be implemented in SDK2 but I need to use SDK1

[bretambrose]

Could the callback you supply to publishAsync be a lambda that completes the appropriate future (via lambda closure binding) that you made prior to calling publishAsync?

[aefn]

Sorry, I haven't thought that it makes it unclear. That publishAsync is exactly the function that AWS Sdk provides. I don't know that I can provide a lambda for it or not.

edited pseudo code :

async def publish:
    publishAsync(topic, payload, QoS, ackCallback=None)  # I don't know how should I wait for this, using await does not work
pass


tasks = [publish for i in range(10)]
connect()
loop.run(tasks) # run all in parallel and wait until get their pubAck
disconnect()

there is a sample in SDK samples which uses publishAsync but it uses a while True which is not my case. I want to send some packets and disconnect.
I found a dirty solution for that but I need a professional way.
I can use a counter and check the counter in while True until the counter gets zero and then disconnect. But I know that it is not the best solution.

[bretambrose]

The ackCallback parameter is a function that should get invoked when the client receives the server's acknowledgement of the publish. In this case, you could make it complete the appropriate future related to that particular publish and the easiest way to do that is just make the callback a lambda that captures the future you want to notify. You can then wait on the whole set of futures and when they're all done, your program should continue.

[aefn]

Thanks for your help. I did it by using the futures but I am not sure that I did it in a correct way or not.
this is what I did:

async def publish(self,topic,packet,QoS=1):
        loop = asyncio.get_event_loop()
        fut = loop.create_future()
                
        def publishCb(mid):
            log.info('{} acknowledged'.format(mid))
            self.acknowledgmentsQueue.append(mid)
            # it should be the last thing in the callback
            # after calling this nothing else in this function will be run
            loop.call_soon_threadsafe(fut.set_result,mid)
        pass
        
        
        messageId = self.sensorClient.publishAsync(topic,packet,QoS,publishCb)
        log.info('{} is published'.format(messageId))
        
        acknowledgment =  await fut
        log.info('acknowledgment : {}'.format(acknowledgment))
pass

[bretambrose]

That is the general idea (publish callback completes an associated future). There are ways you could restructure so that you don't wait for each individual publish to complete in order, but that's an implementation detail that is entirely up to you.