[MQTTv5] Subscription topic callback on_publish_received

[MickaelDelanoe]

Hello,
I'm trying to migrate to this new SDK.
I used the examples available on the repo and it works pretty well
I can connect, subscribe and publish messages which are received by the server.

• python : v3.12.4
• awsiotsdk : v1.22.1

However, I have a problem for reading messages published on a particular topic
Here's my simplified process:

• I create an MQTT client
• I subscribe to the topic id/measures/error.
• I publish my message on the topic id/measures.
• When the server finds an error in the message (payload formatting), it sends a message on the id/measures/error topic.

And so, on the client side, I should have a callback function that reads this new message on id/measures/error (because I'm subscribed to this topic and I've configured a callback).

This works well with MQTT v3.1.1 (even for this SDK) but in MQTT version 5 I don't get my function back.

Callback function

# Callback function for incoming messages
def on_message_received(event_data):
    topic = event_data.topic
    payload = event_data.payload.decode('utf-8')
    print(f"Received message on topic '{topic}': {payload}")
    # Add custom logic here if needed, e.g., stop after a specific count
    # if received_count == <your_target_count>:
    #     received_all_event.set()

Code reset

input_topic = "stage/" + arg_id + "/" + arg_auth + "/measures"
error_topic = "stage/" + arg_id + "/" + arg_auth + "/measures/error"
message = {'data': [{'type': 'TOTO', 'value': 49.0, 'label': 'test-mqtt-v5'}], 'time': '2025-01-09T11:49:44.699507Z'}

client = mqtt5_client_builder.mtls_from_path(
    endpoint=arg_endpoint,
    port=8883,
    cert_filepath=arg_cert,
    pri_key_filepath=arg_key,
    client_id=arg_id,
    ca_filepath=arg_ca,
    on_lifecycle_connection_success=on_lifecycle_connection_success,
    on_lifecycle_connection_failure=on_lifecycle_connection_failure,
    on_publish_received=on_publish_received,
    clean_session=False,
    keep_alive_secs=30)

# client.on_message = on_message_received


client.start()
lifecycle_connect_success_data = future_connection_success.result(TIMEOUT)
connack_packet = lifecycle_connect_success_data.connack_packet
negotiated_settings = lifecycle_connect_success_data.negotiated_settings
print(connack_packet)
print(negotiated_settings)



# print("Subscribing to topic '{}'...".format(message_topic))
subscribe_future = client.subscribe(subscribe_packet=mqtt5.SubscribePacket(
    subscriptions=[mqtt5.Subscription(
        topic_filter=error_topic,
        qos=mqtt5.QoS.AT_LEAST_ONCE)]
))
suback = subscribe_future.result(TIMEOUT)
print("Subscribed with {}".format(suback.reason_codes))
# print("Subscribing to topic '{}'...".format(message_topic))
subscribe_future = client.subscribe(subscribe_packet=mqtt5.SubscribePacket(
    subscriptions=[mqtt5.Subscription(
        topic_filter=input_topic,
        qos=mqtt5.QoS.AT_LEAST_ONCE)]
))
suback = subscribe_future.result(TIMEOUT)
print("Subscribed with {}".format(suback.reason_codes))


message_json = json.dumps(message)
publishStatus = client.publish(mqtt5.PublishPacket(
    topic=input_topic,
    payload=json.dumps(message),
    qos=mqtt5.QoS.AT_LEAST_ONCE
))

publish_completion_data = publishStatus.result(TIMEOUT)
print("PubAck received with {}".format(repr(publish_completion_data.puback.reason_code)))


# client.loop_forever()

print("Stopping Client")
client.stop()

time.sleep(5)

do you know why I can't get any feedback from my subscription to the id/measures/error topic ?

Best Reagds,

Translated with DeepL.com (free version)

[bretambrose]

Unsure if it's a typo, but the actual callback function is on_message_received but you supply on_publish_received to the client constructor which does not exist anywhere.

[MickaelDelanoe]

HI @bretambrose ,
Thanks for your replay.
It's not a typo, according the documentation (https://aws.github.io/aws-iot-device-sdk-python-v2/awsiot/mqtt5_client_builder.html#module-awsiot.mqtt5_client_builder) I understand that on_message_received do not exsiste (but may be I'm wrong)
I tried it anyway, but not betteur :/

Best Regards,

[bretambrose]

I'm not talking about the parameter named on_publish_received, I'm talking about the value on_publish_received. It doesn't exist anywhere in your code snippet. You call your callback on_message_received and then never use it.

[MickaelDelanoe]

yes, sorry, it's a mistake in this code snippet.

I also defined the on_publish_received function as follows

# Callback when any publish is received
def on_publish_received(publish_packet_data):
    publish_packet = publish_packet_data.publish_packet
    assert isinstance(publish_packet, mqtt5.PublishPacket)
    print(" ---------- > >> Received message from topic'{}':{}".format(publish_packet.topic, publish_packet.payload))
    # global received_count
    # received_count += 1
    # if received_count == cmdData.input_count:
    #     received_all_event.set()

[bretambrose]

Are you checking the reason codes in the suback/puback for success? Does the pubsub sample work for you?

[MickaelDelanoe]

Afer retesting, Yes, it's look working

Starting MQTT5 PubSub Sample

MQTT5 Client Created
Connecting to a1x1g7gpi1oruu-ats.iot.eu-west-1.amazonaws.com with client ID 'xxx-clientid-xxxx'...
Lifecycle Connection Success
Connected to endpoint:'a1x1g7gpi1oruu-ats.iot.eu-west-1.amazonaws.com' with Client ID:'xxx-clientid-xxxx' with reason_code:<ConnectReasonCode.SUCCESS: 0>
Subscribing to topic 'stage/xxx-clientid-xxxx/xxx-authtoken-interne-xxxx/measures'...
Subscribed with [<SubackReasonCode.GRANTED_QOS_1: 1>]
Sending 10 message(s)
Publishing message to topic 'stage/xxx-clientid-xxxx/xxx-authtoken-interne-xxxx/measures': {} [1]
PubAck received with <PubackReasonCode.SUCCESS: 0>
Received message from topic'stage/xxx-clientid-xxxx/xxx-authtoken-interne-xxxx/measures':b'"{}"'
Publishing message to topic 'stage/xxx-clientid-xxxx/xxx-authtoken-interne-xxxx/measures': {} [2]
PubAck received with <PubackReasonCode.SUCCESS: 0>
Received message from topic'stage/xxx-clientid-xxxx/xxx-authtoken-interne-xxxx/measures':b'"{}"'
Publishing message to topic 'stage/xxx-clientid-xxxx/xxx-authtoken-interne-xxxx/measures': {} [3]

In fact, in my previous code extraction, it made 2 consecutives subscriptions (one for the error_topc, and another for input_topic)
after removing the subscription of input_topic, I get my feedback from the error_topic topic via the on_publish_received function.

But if I remove the subscription to error_topic to keep only the subscription to input_topic , the on_publish_received function is not called.
This last point is not what I need, but I don't really understand...

[MickaelDelanoe]

I think my problem is simply that I'm closing the client too quickly, so the callback doesn't have time to be executed.
I moved the sleep() before the client.stop() and it works fine.

time.sleep(5)

print("Stopping Client")
client.stop()

sorry for bothering you with this
Thanks for your help

[github-actions[bot]]

Hello! Reopening this discussion to make it searchable.