Google IoT - Правый режим для получения уведомлений (подписка работает)

Прочитав обсуждение в разделе комментариев в ответе @GabeWeiss, я думаю, что есть небольшая путаница в том, чего вы хотите достичь и как (или для чего) вы пытаетесь использовать Pub / Sub.

Учитывая, что я думаю, что проблема здесь более концептуальная, позвольте мне сначала отослать вас на общую страницу документации о Ключевые концепции Cloud IoT Core, где вы фактически найдете некоторую информацию о взаимосвязи между Cloud IoT Core и Pub / Sub. Таким образом, данные телеметрии устройства публикуются в теме Cloud IoT Core, которая позже через Data Broker публикуется в теме Cloud Pub / Sub, которую вы можете использовать извне для других целей: запуск облачных функций, анализ потока данные с потоком данных и т. д.

Теперь, если вы следуете руководству по быстрому запуску Cloud IoT Core, вы увидите, как в определенный момент вы создаете реестр устройств, привязанный к Pub / Подраздел, в котором публикуются события телеметрии устройства. Если вместо того, чтобы писать в тему Pub / Sub по умолчанию, вы хотите писать в несколько тем, вы можете следовать пояснениям в разделе этот другой раздел документации.

Наконец, переходя к вопросу о подписке на темы (темы Cloud IoT Core, а не темы Pub / Sub, поскольку только первые актуальны для устройств), вы можете подписаться на темы MQTT с помощью следующая команда, где (в качестве примера ), устройство подписывается на тему конфигурации, в которой публикуются обновления конфигурации:

# This is the topic that the device will receive configuration updates on.
mqtt_config_topic = '/devices/{}/config'.format(device_id)

# Subscribe to the config topic.
client.subscribe(mqtt_config_topic, qos=1)

Затем с помощью on_message() функция вы можете обрабатывать сообщения, опубликованные по темам, на которые вы подписаны. Обратите внимание, что полезная нагрузка должна быть проанализирована как строка (str(message.payload)).

def on_message(unused_client, unused_userdata, message):
    payload = str(message.payload)
    print('Received message \'{}\' on topic \'{}\' with Qos {}'.format(
            payload, message.topic, str(message.qos)))

Затем в своем вопросе вы указали, что сначала подписались на /devices/{device-id}/config, но это может быть не то, что вам нужно, поскольку именно в этой теме публикуются обновления конфигурации (т.е. не там, где публикуются события телеметрии). Затем, насколько я понимаю, вам следует подписаться на /devices/{device-id}/events, который является актуальной темой MQTT, в которой публикуются показатели телеметрии. Если это не сработает, возможно, возникла другая проблема, поэтому убедитесь, что вы правильно анализируете переменную message, и, возможно, попробуйте используйте Pub / Sub с темой по умолчанию, созданной с помощью реестра, чтобы проверить, правильно ли публикуются показатели телеметрии.

Изменить: уточняется на основе комментария ниже ...

Здесь задействованы два компонента GCP. Есть тема MQTT (которая является темой / events), которая используется устройством для взаимодействия с IoT Core. Затем есть projects/myprojectname/topics/sm1, которого нет в IoT Core, он находится в Pub / Sub. Когда вы отправляете сообщения в тему MQTT / events, IoT Core передает полезные данные с вашего устройства, которые были отправлены в тему MQTT / events, через тему Pub / Sub, которая была создана и прикреплена к реестру IoT Core, где было зарегистрировано ваше устройство. .

Чтобы увидеть эти сообщения, вы должны создать подписку в Pub / Sub по теме projects/myprojectname/topics/sm1. Если зайти в консоль и паб / под-> топики. Щелкните три точки рядом с темой и выберите «Новая подписка». После того, как у вас есть подписка, вы можете отправить некоторые данные со своего устройства, а затем в командной строке вы можете запустить (при условии, что у вас установлены инструменты gcloud):

gcloud beta pubsub subscriptions pull --max-messages=3 <subscription_id>

Чтобы делать что-либо с сообщениями, вы можете подписаться на тему Pub / Sub с помощью сценария (проверьте API-интерфейсы Pub / Sub), чтобы запускать сообщения, добавляемые в тему.

Исходное сообщение:

Вы отправляете конфигурационное сообщение на устройство? Путаница может заключаться в том, что темы MQTT однонаправлены.

Итак: 1) тема / events предназначена для устройства-> IoT Core. 2) тема / config предназначена для IoT Core Admin SDK-> устройство

Где-то в другом скрипте или из интерфейса пользовательского интерфейса IoT Core вам нужно отправить сообщение конфигурации, чтобы правильно увидеть срабатывание on_message.

В пользовательском интерфейсе IoT Core (на console.cloud.google.com) вы можете перейти к отдельному зарегистрированному устройству и в верхней части экрана нажать «Обновить конфигурацию». Появится всплывающее окно, которое позволит вам отправить текстовое сообщение или сообщение в кодировке base64 на это устройство, и оно появится в теме / config.

Мне пришлось сдаться библиотеке Google Pub / Sub, чтобы получать уведомления, связанные с моей указанной темой.

Мой код публикации (только важные части):

mqtt_url = "mqtt.googleapis.com"
mqtt_port = 8883
mqtt_topic = "/devices/sm1/events"

client = mqtt.Client("projects/{}/locations/{}/registries/{}/devices/{}".format(
                     project_id,
                     cloud_region,
                     registry_id,
                     device_id), protocol=mqtt.MQTTv311)

client.username_pw_set(username='unused',
                       password=jwt_maker.create_jwt(project_id,
                                           private_key,
                                           algorithm="RS256"))

client.tls_set(root_ca, 
               certfile = public_crt, 
               keyfile = private_key, 
               cert_reqs = ssl.CERT_REQUIRED, 
               tls_version = ssl.PROTOCOL_TLSv1_2, 
               ciphers = None)

client.connect(mqtt_url, mqtt_port, keepalive=60)
res = client.publish(mqtt_topic, some_data, qos=1)

На портале Google Cloud Platform мне пришлось создать подписку в разделе Pub / Sub, назначив ее моей созданной теме, которая уже была моей темой по умолчанию (возможно, связанной с темой / events). Созданная подписка имеет следующий формат:

projects/project_name/subscriptions/subscription_name

Мой код подписки с использованием библиотеки Google Pub / Sub, поскольку невозможно использовать протокол MQTT:

from google.cloud import pubsub

def callback(message):
    print(message.data)
    message.ack()

project_id = "project_name"
subscription_name = "sm1"

subscriber = pubsub.SubscriberClient()
subscription_name = 'projects/{}/subscriptions/{}'.format(project_id, subscription_name)

subscription = subscriber.subscribe(subscription_name)
future = subscription.open(callback)

try:
    future.result()
except Exception as ex:
    subscription.close()
    raise

Надеюсь, это поможет кому угодно. Более подробную информацию можно найти здесь (github).