MQTT Stack Question

RichSurprenant

I am using AWS and MQTT. I have not been able to figure out why i keep getting stale, repeat MQTT messages. for instance, I changed my device shadow on AWS, and I get the new MQTT message, version 513, which is correct, but then I get message 510.

Question: Is there something in my handling of the MQTT message that I need to do to say I received the MQTT message so it doesn’t get sent again?

This is the order I receive them. You can see that timestamps for earlier versions are older.

Received: {“version”:515,“timestamp”:1730299139,“state”:{“whitelist”:[]},“metadata”:{“whitelist”:[]}}
topic: $aws/things/RichGateway/shadow/name/G2000-022512/update/delta

Received: {“version”:509,“timestamp”:1730298928,“state”:{“whitelist”:[]},“metadata”:{“whitelist”:[]}}
topic: $aws/things/RichGateway/shadow/name/G2000-022512/update/delta

Received: {“version”:512,“timestamp”:1730299045,“state”:{“whitelist”:[“PS4550-04143”]},“metadata”:{“whitelist”:[{“timestamp”:1730299045}]}}
topic: $aws/things/RichGateway/shadow/name/G2000-022512/update/delta

Received: {“version”:514,“timestamp”:1730299097,“state”:{“whitelist”:[“PS4550-04143”]},“metadata”:{“whitelist”:[{“timestamp”:1730299097}]}}
topic: $aws/things/RichGateway/shadow/name/G2000-022512/update/delta

Received: {“version”:515,“timestamp”:1730299139,“state”:{“whitelist”:[]},“metadata”:{“whitelist”:[]}}
topic: $aws/things/RichGateway/shadow/name/G2000-022512/update/delta

Received: {“version”:513,“timestamp”:1730299069,“state”:{“whitelist”:[]},“metadata”:{“whitelist”:[]}}
topic: $aws/things/RichGateway/shadow/name/G2000-022512/update/delta

Received: {“version”:510,“timestamp”:1730298953,“state”:{“whitelist”:[“PS4550-04143”]},“metadata”:{“whitelist”:[{“timestamp”:1730298953}]}}
topic: $aws/things/RichGateway/shadow/name/G2000-022512/update/delta
Get Whitelist, Code, …

jaredwolff

Hi @RichSurprenant

If you’re using the AWS specific client, it should ack messages that require one. Can you confirm you’re using the AWS specific MQTT implementation?

RichSurprenant

I AM using the AWS specific MQTT implementation. I don’t see the ACK. Do you have any pointers to look to LOG it?

RichSurprenant

RichSurprenant AND THANKS!

jaredwolff

It’s in the on_publish function in mqtt_helper.c (See send_ack) There’s no logging there so you may not see anything. It only sends an ACK if QOS==1

MQTT_HELPER_STATIC void on_publish(const struct mqtt_evt *mqtt_evt)
{
	int err;
	const struct mqtt_publish_param *p = &mqtt_evt->param.publish;
	struct mqtt_helper_buf topic = {
		.ptr = (char *)p->message.topic.topic.utf8,
		.size = p->message.topic.topic.size,
	};
	struct mqtt_helper_buf payload = {
		.ptr = payload_buf,
	};

	err = publish_get_payload(&mqtt_client, p->message.payload.len);
	if (err) {
		LOG_ERR("publish_get_payload, error: %d", err);

		if (current_cfg.cb.on_error) {
			current_cfg.cb.on_error(MQTT_HELPER_ERROR_MSG_SIZE);
		}

		return;
	}

	if (p->message.topic.qos == MQTT_QOS_1_AT_LEAST_ONCE) {
		send_ack(&mqtt_client, p->message_id);
	}

	payload.size = p->message.payload.len;

	if (current_cfg.cb.on_publish) {
		current_cfg.cb.on_publish(topic, payload);
	}
}

RichSurprenant

Jared, thank you so much. This problem has been plaguing me for months. Your response prompted me to look at the aws_iot implemenation, which I branch off of so long ago. I discovered that I was missing this ack:

	case MQTT_EVT_PUBLISH: {
		const struct mqtt_publish_param *p = &mqtt_evt->param.publish;

		LOG_DBG("MQTT_EVT_PUBLISH: id = %d len = %d ",
			p->message_id,
			p->message.payload.len);

		err = publish_get_payload(c, p->message.payload.len);
		if (err) {
			LOG_ERR("publish_get_payload, error: %d", err);
			break;
		}

		if (p->message.topic.qos == MQTT_QOS_1_AT_LEAST_ONCE) {
			const struct mqtt_puback_param ack = {
				.message_id = p->message_id
			};

			mqtt_publish_qos1_ack(c, &ack);
		}

When i added the ack, it fixed the problem. Again, thanks… :-)