Google Cloud IoT Subscribe to Topic Using Python

1
votes

I am trying to collect published data from my topic using an application I have built using Python. So when data gets published, it is processed by the application.

Problem is: subscribing always seemingly succeeds but no events are fetched.

I am able to connect to the broker using the MQTT client.

Here is my current code handling the connecting and subscribing:

broker_address= "mqtt.googleapis.com"
port = 8883
client = mqtt.Client(client_id='projects/<project_id>/locations/<location>/registries/<registry_id>/devices/<device_id>',
        protocol=4)
password=create_jwt(...) #works
client.username_pw_set(username='unused',password=password)
client.tls_set(ca_certs=<route_to_cert>, tls_version=ssl.PROTOCOL_TLSv1_2)
client.on_connect = on_connect
client.on_disconnect = on_disconnect
client.on_message = on_message
client.on_subscribe = on_subscribe
client.connect(broker_address, port)
client.loop_start()
while connected_flag == 0: #changed in "on_connect" method
    print("Wait until connected ", connected_flag)
    time.sleep(1)
print("Connected")
topic = 'projects/<project_id>/topics/<device_id>'
client.subscribe(topic)
while subscribed_flag == 0: #changed in "on_subscribe" method
    time.sleep(1)
time.sleep(5)
client.loop_stop()  
client.disconnect()

The topic I'm trying to subscribe to is: 'projects/_project_id_/topics/_device_id_'

Some logic is in the on_connect and on_subscribe methods. And I know the code is not logical as is, but I just want a reading of the data before proceeding.

So, when I publish an event during the time the client is connected and subscribed, I would assume the on_message method would run, but nothing happens.

I have never utilized MQTT or Google Cloud IoT before, so I might just make some extremely basic mistake.

EDIT

I managed to fetch the messages by changing to Google Cloud Pub Sub libraries (https://cloud.google.com/pubsub/docs/reference/libraries). A subscription had to be created to the Google Cloud IoT and it had to be connected to the topic. The subscription is of pull type.

My solution is not yet perfect, but it is a step forward. Here is my code:

from google.cloud import pubsub;
subscriber = pubsub.SubscriberClient()
subscription_path = subscriber.subscription_path(<project_id>, <subscription_name>)
def callback(message):
    print(message.data)
subscriber.subscribe(subscription_path, callback=callback)

Callback is called when messages arrive.

2
pythongoogle-cloud-platformmqttsubscribegoogle-cloud-iot
I am also dealing this problem... The documentation is poor related to this kind of information... btw, how is your on_message code? – Dalton Cézane
I managed to fetch the messages using Google Pub Sub library (cloud.google.com/pubsub/docs/reference/libraries). I will update the question with my new code, which pulls messages from the topic. – StuckForTooLong
I want to receive the notifications from a subscription using my own code, without the google pub/sub lib... – Dalton Cézane
Put your edited text as an answer and mark it as "the answer". – Dalton Cézane

2 Answers

0
votes

The devices topic is one way only, from device to IoT Core. It won't send messages back. There's a config topic that you can subscribe to as well, and send messages back to the device via the IoT Core Admin SDK.

Check out this guide on configuration messages and reporting state. Note that the code example is on Java by default, but there are tabs for Java, Node.js and Python there.

0
votes

I managed to fetch the messages by changing to Google Cloud Pub Sub libraries (https://cloud.google.com/pubsub/docs/reference/libraries). A subscription had to be created to the Google Cloud IoT and it had to be connected to the topic. The subscription is of pull type.

My solution is not yet perfect, but it is a step forward. Here is my code:

from google.cloud import pubsub;
subscriber = pubsub.SubscriberClient()
subscription_path = subscriber.subscription_path(<project_id>, 
<subscription_name>)
    def callback(message):
print(message.data)
subscriber.subscribe(subscription_path, callback=callback)

Callback is called when messages arrive.