Any way to acks message consuming by hand?

[YAGregor]

consider a situation:

async for message in messages:
    # message acknowledged here
    # do something...

it works when assume program never stop, but when consider program may stop to update or by accident:

async for message in messages:
    # message acknowledged here
    # process may stop here, message *acknowledged* but *nothing has be done*
    # do something...

is there any way to confirm message consumed after do something like

async for message in messages:
    # do something...
    await message.acks()

or it's just my misunderstanding?

[frederikaalund]

Hi YAGregor, thanks for opening this discussion. :) Let me have a look.

Try to describe what you mean by "message acknowledged".

---

If you mean "client sends ACK message back to server when QoS>0" then that's already done internally by asyncio-mqtt. As soon as you enter the async for message in messages loop, the message is acknowledged. Simple as that. :)

---

If you're afraid that an early return or Exception inside the async for message in messages loop somehow breaks things, don't worry. You're safe as long as you do:

async with client.messages() as messages:  # This with statement does the proper cleanup
    async for message in messages:
        ...

Note that the with statement automatically cleans things up behind the scenes regardless of return, Exception, sys.exit() or whatever else that you may do inside the loop.

---

If you mean something else by "message acknowledged" then try to elaborate. 👍

[YAGregor]

Yes, "message acknowledged" means send ACK to server.

As soon as you enter the async for message in messages loop, the message is acknowledged.

that is the question, what I mean "process stop" is not Exception or early return, I mean the machine's power might down accidentally, or process force killed by OS, in this case, there is no chance to clear up.

I mean we might should not send ACK when enter for loop, we should send ACK at the end of for loop, if send ACK before business code, there is a risk that it might be equal to the message is lost.

here is a example of aio-pika

async with queue.iterator() as queue_iter:
    async for message in queue_iter:
        async with message.process():
            print(message.body)
            if queue.name in message.body.decode():
                break

and it's implementation
we clearly see that message acknowledged at the end of business code, not the begin, I think that is the right way to avoid message lost.

[frederikaalund]

that is the question, what I mean "process stop" is not Exception or early return, I mean the machine's power might down accidentally, or process force killed by OS, in this case, there is no chance to clear up.

Thanks for the elaboration, now your point makes sense to me. 👍

---

Let's disambiguate the terminology a bit here first. I'll invent two concepts, MSG_ACK and MSG_PROCESSED, defined as follows:

• MSG_ACK semantics: Client tells the server that it received the message and proceeds right away.
• MSG_PROCESSED semantics: Client tells the server that not only did it receive the message, it also processed it completely.

asyncio-mqtt supports MSG_ACK semantics (there is currently no support for MSG_PROCESSED).
aio-pika supports MSG_PROCESSED semantics.

---

The asyncio-mqtt implementation of the MSG_ACK semantics is to send PUBACK packet when QoS > 0 as soon as we receive the corresponding PUBLISH packet. This is what paho-mqtt does under the hood. Currently, the only way to stop paho-mqtt from sending PUBACK is to raise an exception inside the on_message callback (and use suppress_exceptions=False). Otherwise, paho simply logs the exception and sends PUBACK to the server anyhow.

Theoretically, we could support MSG_PROCESSED semantics in asyncio-mqtt but it requires a careful rewrite if the internals. E.g., if we opt to raise the exception and not suppress it, we have to handle said exception in loop_read instead. Possible but cumbersome. PRs are welcome! 👍

In turn, we also have to add something like the aio-pika's message.process() context manager to send PUBACK when we're "done" with the message. You'll have to take a lot of care to avoid breaking API changes.

---

Lastly, I'm concerned that MSG_PROCESSED semantics are not the right "fit" for the MQTT protocol. When the MQTT specification talks about QoS and PUBACK, it's always about message delivery. Not about message processing. In that light, MSG_ACK semantics seem proper.
However, I do get how MSG_PROCESSED semantics are useful in practice. Guess it's good to have the discussion! 👍

[YAGregor]

You totally understand me.

Lastly, I'm concerned that MSG_PROCESSED semantics are not the right "fit" for the MQTT protocol.

I am not sure about this.

        elif message.qos == 1:
            self._handle_on_message(message)
            return self._send_puback(message.mid)

I cheked the implementation of paho, it sends PUBACK when call back (business code) finish just like I describe, I think there is a simple way to implement MSG_PROCESSED semantics, just block on_message, let on_message wait, and let it finish when all message.process() context manager exit.

I'd like to make a PR for this if you do not against implement MSG_PROCESSED, just add a method like messages_with_process on Client, it won't break API. Anyway asyncio-mqtt have not reach 1.0, it's OK to break API.

considering #95 , there is also a simple way to fix it and no need to buffer: just block the on_message until a subscriber exists.

[frederikaalund]

Thanks for the quick feedback. 👍

[...] paho, it sends PUBACK when call back (business code) finish just like I describe [...]

That's a good point! That trumps my argument against MSG_PROCESSED semantics for MQTT. In the spirit of asyncio-mqtt being a "small wrapper around paho-mqtt" I think that we should follow their (paho's) established semantics. Let's do this!

just add a method like messages_with_process on Client, it won't break API

Sounds good to me. I'd gladly to review a PR on this. 😄 👍

It's not going to be easy. I think that the devil is in the details (e.g., the implementation) here. For instance, you have to the following code example into consideration:

async with client.messages_with_process() as messages:  # Or whatever you want to call it
    async for message in messages:
        message.payload  # ERROR! Should raise an exception since we didn't __enter__ message yet
        try:
            async with message:
                raise RuntimeError("Something happened here")  # Just for the sake of this code example
        except:
            pass  # We opt to ignore exceptions in this example
        # Because we raise `RuntimeError` within the message processing, we never send PUBACK to the server.
        # However, the `async for` loop should go on even though we raised `RuntimeError` within the message
        # processing because we chose to suppress said exception outside the `async with message` context.

Does it make sense? That's just one code example of the subtleties with this API.

For these reasons, it's paramount that the PR comes with a test suite that verifies the new behaviour.

Let me know if you're still up for it. :)