Блог

Очень новичок в ядре интернета вещей. Поэтому я в основном просто перепрофилирую образец AWS SDK pubsub.py код (https://github.com/aws/aws-iot-device-sdk-python-v2/blob/main/samples/pubsub.py) но по какой-то причине, даже несмотря на то, что я могу установить соединение с AWS, я не могу:

• Подпишитесь на тему

При iot_connection.subscribe() выполнении терминал покажет «Подписка на тест темы/тему…» и зависнет на неопределенный срок. Обратите внимание, что если я не включу subscribe_result = subscribe_future.result() полный приведенный ниже код, он будет успешно выполнен, но, опять же, в консоли не будет получено никакого сообщения. Побочный вопрос: вам действительно нужно подписаться на тему, чтобы иметь возможность публиковаться в ней?

• Смотрите сообщения в консоли

При iot_connection.publish() выполнении все, похоже, выполняется успешно, однако «Привет, мир» не отображается в консоли тестирования AWS, хотя я подписан как на»#», так и на «тест/тема».

Любая помощь будет очень признательна!

Код:

 import argparse
from uuid import uuid4
import json
import time
from awscrt import io, mqtt, auth, http
from awsiot import mqtt_connection_builder


def make_parser():
    parser = argparse.ArgumentParser(description="Send and receive messages through and MQTT connection.")
    parser.add_argument('endpoint', help="Your AWS IoT custom endpoint, not including a port.")
    parser.add_argument('--port', type=int, help="Specify port. AWS IoT supports 443 and 8883.", metavar='')
    parser.add_argument('--cert', help="File path to your client certificate, in PEM format.", metavar='')
    parser.add_argument('--key', help="File path to your private key, in PEM format.", metavar='')
    parser.add_argument('--root-ca', help="File path to root certificate authority, in PEM format.", metavar='')
    parser.add_argument('--client-id', default="test-"   str(uuid4()), help="Client ID for MQTT connection.",
                        metavar='')
    parser.add_argument('--topic', default="test/topic", help="Topic to subscribe to, and publish messages to.",
                        metavar='')
    parser.add_argument('--message', default="Hello World!", help="Message to publish. ", metavar='')
    parser.add_argument('--count', default=10, type=int, help="Number of messages to publish.", metavar='')
    return parser


class IoT:

    def __init__(self, args):
        self.endpoint = args.endpoint
        self.port = args.port
        self.cert = args.cert
        self.key = args.key
        self.root_ca = args.root_ca
        self.client_id = args.client_id
        self.topic = args.topic
        self.message = args.message
        self.count = args.count
        print("Initializing parameters...")

    def __enter__(self):
        print("Spinning up resources...")
        self.event_loop_group = io.EventLoopGroup(1)
        self.host_resolver = io.DefaultHostResolver(self.event_loop_group)
        self.client_bootstrap = io.ClientBootstrap(self.event_loop_group, self.host_resolver)
        print("Establishing connection to AWS...")
        self.mqtt_connection = mqtt_connection_builder.mtls_from_path(
            endpoint=self.endpoint,
            port=self.port,
            cert_filepath=self.cert,
            pri_key_filepath=self.key,
            ca_filepath=self.root_ca,
            client_id=self.client_id,
            client_bootstrap=self.client_bootstrap,
            clean_session=False,
            keep_alive_secs=30
        )
        connect_future = self.mqtt_connection.connect()
        connect_future.result()
        print("Connected!")

    def __exit__(self, exception_type, exception_value, traceback):
        print("Disconnecting...")
        disconnect_future = self.mqtt_connection.disconnect()
        disconnect_future.result()
        print("Disconnected!")

    def subscribe(self):
        print("Subscribing to topic {}...".format(self.topic))
        subscribe_future, packet_id = self.mqtt_connection.subscribe(
            topic=self.topic,
            qos=mqtt.QoS.AT_LEAST_ONCE,
            callback=self.on_message_recieved
        )
        subscribe_result = subscribe_future.result()
        print("Result: {}".format(str(subscribe_result['qos'])))
        print("Subscribed!")

    def on_message_recieved(self, topic, paylod, dup, qos, retain, **kwargs):
        print("Recieved message")

    def publish(self):
        if self.message:
            print("Publishing message to topic '{}': {}".format(self.topic, self.message))
            message = "{} [{}]".format(self.message, self.count)
            message_json = json.dumps(message)
            self.mqtt_connection.publish(
                topic=self.topic,
                payload=message_json,
                qos=mqtt.QoS.AT_LEAST_ONCE
            )
            time.sleep(1)  # thought this might fix things, it did not


if __name__ == '__main__':
    print("Gathering user inputs...")
    parser = make_parser()
    args = parser.parse_args()
    # Starting IoT Core connection
    iot_connection = IoT(args)
    with iot_connection:
        iot_connection.subscribe()  # subscribing to topic
        iot_connection.publish()  # publishing message hopefully
 

Тем временем все решится. Должно быть, это была проблема с моим ресурсом ARN — когда я создал новую политику и установил для ресурса ARN значение»*», все работало так, как ожидалось. Ранее я пытался ограничить ARN определенным идентификатором клиента. Не знаю, почему это не сработало, так как, насколько я могу судить, я правильно обработал идентификатор клиента.