Komunikacja z centrum IoT przy użyciu protokołu AMQP

Artykuł
03/20/2024

Usługa Azure IoT Hub obsługuje protokół 1.0 Advanced Message Queuing Protocol (AMQP) o nazwie OASIS Advanced Message Queuing Protocol w wersji 1.0 w celu dostarczania różnych funkcji za pośrednictwem punktów końcowych dostępnych dla urządzeń i usług. W tym dokumencie opisano używanie klientów amQP do nawiązywania połączenia z centrum IoT w celu korzystania z funkcji usługi IoT Hub.

Klient usługi

Połączenie i uwierzytelnianie w centrum IoT (klient usługi)

Aby nawiązać połączenie z centrum IoT przy użyciu protokołu AMQP, klient może użyć uwierzytelniania opartego na oświadczeniach (CBS) lub prostego uwierzytelniania i warstwy zabezpieczeń (SASL).

Następujące informacje są wymagane dla klienta usługi:

Informacja	Wartość
Nazwa hosta centrum IoT	`<iot-hub-name>.azure-devices.net`
Nazwa klucza	`service`
Klucz dostępu	Klucz podstawowy lub pomocniczy skojarzony z usługą
Sygnatura dostępu współdzielonego	Krótkotrwały sygnatura dostępu współdzielonego w następującym formacie: `SharedAccessSignature sig={signature-string}&se={expiry}&skn={policyName}&sr={URL-encoded-resourceURI}`. Aby uzyskać kod generowania tego podpisu, zobacz Kontrola dostępu do usługi IoT Hub.

Poniższy fragment kodu używa biblioteki uAMQP w języku Python do nawiązywania połączenia z centrum IoT Hub za pośrednictwem linku nadawcy.

import uamqp
import urllib
import time

# Use generate_sas_token implementation available here:
# https://learn.microsoft.com/azure/iot-hub/iot-hub-devguide-security#sas-token-structure
from helper import generate_sas_token

iot_hub_name = '<iot-hub-name>'
hostname = '{iot_hub_name}.azure-devices.net'.format(iot_hub_name=iot_hub_name)
policy_name = 'service'
access_key = '<primary-or-secondary-key>'
operation = '<operation-link-name>'  # example: '/messages/devicebound'

username = '{policy_name}@sas.root.{iot_hub_name}'.format(
    iot_hub_name=iot_hub_name, policy_name=policy_name)
sas_token = generate_sas_token(hostname, access_key, policy_name)
uri = 'amqps://{}:{}@{}{}'.format(urllib.quote_plus(username),
                                  urllib.quote_plus(sas_token), hostname, operation)

# Create a send or receive client
send_client = uamqp.SendClient(uri, debug=True)
receive_client = uamqp.ReceiveClient(uri, debug=True)

Wywoływanie komunikatów chmura-urządzenie (klient usługi)

Aby dowiedzieć się więcej na temat wymiany komunikatów z chmury na urządzenie między usługą a centrum IoT i między urządzeniem a centrum IoT, zobacz Wysyłanie komunikatów z chmury do urządzenia z centrum IoT. Klient usługi używa dwóch linków do wysyłania komunikatów i odbierania opinii dotyczących wcześniej wysyłanych komunikatów z urządzeń, zgodnie z opisem w poniższej tabeli:

Utworzone przez	Typ linku	Ścieżka łącza	opis
Usługa	Link nadawcy	`/messages/devicebound`	Komunikaty z chmury do urządzenia przeznaczone dla urządzeń są wysyłane do tego linku przez usługę. Komunikaty wysyłane za pośrednictwem tego linku mają właściwość `To` ustawioną na ścieżkę łącza odbiorcy urządzenia docelowego, `/devices/<deviceID>/messages/devicebound`.
Usługa	Link odbiorcy	`/messages/serviceBound/feedback`	Komunikaty zwrotne dotyczące uzupełniania, odrzucania i porzucania, które pochodzą z urządzeń odebranych na tym linku przez usługę. Aby uzyskać więcej informacji na temat komunikatów opinii, zobacz Wysyłanie komunikatów z chmury do urządzenia z centrum IoT Hub.

Poniższy fragment kodu przedstawia sposób tworzenia komunikatu z chmury do urządzenia i wysyłania go do urządzenia przy użyciu biblioteki uAMQP w języku Python.

import uuid
# Create a message and set message property 'To' to the device-bound link on device
msg_id = str(uuid.uuid4())
msg_content = b"Message content goes here!"
device_id = '<device-id>'
to = '/devices/{device_id}/messages/devicebound'.format(device_id=device_id)
ack = 'full'  # Alternative values are 'positive', 'negative', and 'none'
app_props = {'iothub-ack': ack}
msg_props = uamqp.message.MessageProperties(message_id=msg_id, to=to)
msg = uamqp.Message(msg_content, properties=msg_props,
                    application_properties=app_props)

# Send the message by using the send client that you created and connected to the IoT hub earlier
send_client.queue_message(msg)
results = send_client.send_all_messages()

# Close the client if it's not needed
send_client.close()

Aby otrzymywać opinie, klient usługi tworzy link odbiorcy. Poniższy fragment kodu przedstawia sposób tworzenia linku przy użyciu biblioteki uAMQP w języku Python:

import json

operation = '/messages/serviceBound/feedback'

# ...
# Re-create the URI by using the preceding feedback path and authenticate it
uri = 'amqps://{}:{}@{}{}'.format(urllib.quote_plus(username),
                                  urllib.quote_plus(sas_token), hostname, operation)

receive_client = uamqp.ReceiveClient(uri, debug=True)
batch = receive_client.receive_message_batch(max_batch_size=10)
for msg in batch:
    print('received a message')
    # Check content_type in message property to identify feedback messages coming from device
    if msg.properties.content_type == 'application/vnd.microsoft.iothub.feedback.json':
        msg_body_raw = msg.get_data()
        msg_body_str = ''.join(msg_body_raw)
        msg_body = json.loads(msg_body_str)
        print(json.dumps(msg_body, indent=2))
        print('******************')
        for feedback in msg_body:
            print('feedback received')
            print('\tstatusCode: ' + str(feedback['statusCode']))
            print('\toriginalMessageId: ' + str(feedback['originalMessageId']))
            print('\tdeviceId: ' + str(feedback['deviceId']))
            print
    else:
        print('unknown message:', msg.properties.content_type)

Jak pokazano w poprzednim kodzie, komunikat opinii z chmury do urządzenia ma typ zawartości aplikacji/vnd.microsoft.iothub.feedback.json. Właściwości w treści JSON wiadomości można użyć, aby wywnioskować stan dostarczania oryginalnej wiadomości:

Klucz statusCode w treści opinii ma jedną z następujących wartości: Powodzenie, Wygaśnięcie, DeliveryCountExceeded, Odrzucone lub Przeczyszczone.
Klucz deviceId w treści opinii ma identyfikator urządzenia docelowego.
Klucz originalMessageId w treści opinii ma identyfikator oryginalnego komunikatu z chmury do urządzenia, który został wysłany przez usługę. Ten stan dostarczania umożliwia skorelowanie opinii z komunikatami z chmury do urządzenia.

Odbieranie komunikatów telemetrycznych (klient usługi)

Domyślnie centrum IoT przechowuje pozyskane komunikaty telemetryczne urządzenia w wbudowanym centrum zdarzeń. Klient usługi może odbierać przechowywane zdarzenia przy użyciu protokołu AMQP.

W tym celu klient usługi musi najpierw nawiązać połączenie z punktem końcowym centrum IoT Hub i otrzymać adres przekierowania do wbudowanych centrów zdarzeń. Następnie klient usługi używa podanego adresu, aby nawiązać połączenie z wbudowanym centrum zdarzeń.

W każdym kroku klient musi przedstawić następujące informacje:

Prawidłowe poświadczenia usługi (token sygnatury dostępu współdzielonego usługi).
Dobrze sformatowana ścieżka do partycji grupy odbiorców, z którą zamierza pobierać komunikaty. W przypadku danej grupy odbiorców i identyfikatora partycji ścieżka ma następujący format: /messages/events/ConsumerGroups/<consumer_group>/Partitions/<partition_id> (domyślna grupa odbiorców to $Default).
Opcjonalny predykat filtrowania w celu wyznaczenia punktu początkowego w partycji. Ten predykat może być w postaci numeru sekwencji, przesunięcia lub w kolejce znacznika czasu.

Poniższy fragment kodu używa biblioteki uAMQP w języku Python do zademonstrowania powyższych kroków:

import json
import uamqp
import urllib
import time

# Use the generate_sas_token implementation that's available here: https://learn.microsoft.com/azure/iot-hub/iot-hub-devguide-security#sas-token-structure
from helper import generate_sas_token

iot_hub_name = '<iot-hub-name>'
hostname = '{iot_hub_name}.azure-devices.net'.format(iot_hub_name=iot_hub_name)
policy_name = 'service'
access_key = '<primary-or-secondary-key>'
operation = '/messages/events/ConsumerGroups/{consumer_group}/Partitions/{p_id}'.format(
    consumer_group='$Default', p_id=0)

username = '{policy_name}@sas.root.{iot_hub_name}'.format(
    policy_name=policy_name, iot_hub_name=iot_hub_name)
sas_token = generate_sas_token(hostname, access_key, policy_name)
uri = 'amqps://{}:{}@{}{}'.format(urllib.quote_plus(username),
                                  urllib.quote_plus(sas_token), hostname, operation)

# Optional filtering predicates can be specified by using endpoint_filter
# Valid predicates include:
# - amqp.annotation.x-opt-sequence-number
# - amqp.annotation.x-opt-offset
# - amqp.annotation.x-opt-enqueued-time
# Set endpoint_filter variable to None if no filter is needed
endpoint_filter = b'amqp.annotation.x-opt-sequence-number > 2995'

# Helper function to set the filtering predicate on the source URI


def set_endpoint_filter(uri, endpoint_filter=''):
    source_uri = uamqp.address.Source(uri)
    source_uri.set_filter(endpoint_filter)
    return source_uri


receive_client = uamqp.ReceiveClient(
    set_endpoint_filter(uri, endpoint_filter), debug=True)
try:
    batch = receive_client.receive_message_batch(max_batch_size=5)
except uamqp.errors.LinkRedirect as redirect:
    # Once a redirect error is received, close the original client and recreate a new one to the re-directed address
    receive_client.close()

    sas_auth = uamqp.authentication.SASTokenAuth.from_shared_access_key(
        redirect.address, policy_name, access_key)
    receive_client = uamqp.ReceiveClient(set_endpoint_filter(
        redirect.address, endpoint_filter), auth=sas_auth, debug=True)

# Start receiving messages in batches
batch = receive_client.receive_message_batch(max_batch_size=5)
for msg in batch:
    print('*** received a message ***')
    print(''.join(msg.get_data()))
    print('\t: ' + str(msg.annotations['x-opt-sequence-number']))
    print('\t: ' + str(msg.annotations['x-opt-offset']))
    print('\t: ' + str(msg.annotations['x-opt-enqueued-time']))

W przypadku danego identyfikatora urządzenia centrum IoT używa skrótu identyfikatora urządzenia, aby określić, w której partycji mają być przechowywane komunikaty. Powyższy fragment kodu pokazuje, jak zdarzenia są odbierane z jednej takiej partycji. Należy jednak pamiętać, że typowa aplikacja często musi pobierać zdarzenia przechowywane we wszystkich partycjach centrum zdarzeń.

Klient urządzenia

Połączenie i uwierzytelnianie w centrum IoT (klient urządzenia)

Aby nawiązać połączenie z centrum IoT przy użyciu protokołu AMQP, urządzenie może używać oświadczeń opartych na zabezpieczeniach (CBS) lub uwierzytelniania Simple Authentication and Security Layer (SASL).

Następujące informacje są wymagane dla klienta urządzenia:

Informacja	Wartość
Nazwa hosta centrum IoT	`<iot-hub-name>.azure-devices.net`
Klucz dostępu	Klucz podstawowy lub pomocniczy skojarzony z urządzeniem
Sygnatura dostępu współdzielonego	Krótkotrwały sygnatura dostępu współdzielonego w następującym formacie: `SharedAccessSignature sig={signature-string}&se={expiry}&skn={policyName}&sr={URL-encoded-resourceURI}`. Aby uzyskać kod generowania tego podpisu, zobacz Kontrola dostępu do usługi IoT Hub.

Poniższy fragment kodu używa biblioteki uAMQP w języku Python do nawiązywania połączenia z centrum IoT Hub za pośrednictwem linku nadawcy.

import uamqp
import urllib
import uuid

# Use generate_sas_token implementation available here:
# https://learn.microsoft.com/azure/iot-hub/iot-hub-devguide-security#sas-token-structure
from helper import generate_sas_token

iot_hub_name = '<iot-hub-name>'
hostname = '{iot_hub_name}.azure-devices.net'.format(iot_hub_name=iot_hub_name)
device_id = '<device-id>'
access_key = '<primary-or-secondary-key>'
username = '{device_id}@sas.{iot_hub_name}'.format(
    device_id=device_id, iot_hub_name=iot_hub_name)
sas_token = generate_sas_token('{hostname}/devices/{device_id}'.format(
    hostname=hostname, device_id=device_id), access_key, None)

# e.g., '/devices/{device_id}/messages/devicebound'
operation = '<operation-link-name>'
uri = 'amqps://{}:{}@{}{}'.format(urllib.quote_plus(username),
                                  urllib.quote_plus(sas_token), hostname, operation)

receive_client = uamqp.ReceiveClient(uri, debug=True)
send_client = uamqp.SendClient(uri, debug=True)

Następujące ścieżki linków są obsługiwane jako operacje urządzeń:

Utworzone przez	Typ linku	Ścieżka łącza	opis
Urządzenia	Link odbiorcy	`/devices/<deviceID>/messages/devicebound`	Komunikaty z chmury do urządzenia przeznaczone dla urządzeń są odbierane na tym linku przez każde urządzenie docelowe.
Urządzenia	Link nadawcy	`/devices/<deviceID>/messages/events`	Komunikaty przesyłane z urządzenia do chmury wysyłane z urządzenia są wysyłane za pośrednictwem tego linku.
Urządzenia	Link nadawcy	`/messages/serviceBound/feedback`	Opinie dotyczące komunikatów z chmury do urządzenia wysyłane do usługi za pośrednictwem tego linku przez urządzenia.

Odbieranie poleceń z chmury do urządzenia (klient urządzenia)

Polecenia z chmury do urządzenia wysyłane do urządzeń są dostarczane za pomocą linku /devices/<deviceID>/messages/devicebound . Urządzenia mogą odbierać te komunikaty w partiach i używać ładunku danych komunikatu, właściwości komunikatów, adnotacji lub właściwości aplikacji w wiadomości zgodnie z potrzebami.

Poniższy fragment kodu używa biblioteki uAMQP w języku Python do odbierania komunikatów z chmury do urządzenia.

# ...
# Create a receive client for the cloud-to-device receive link on the device
operation = '/devices/{device_id}/messages/devicebound'.format(
    device_id=device_id)
uri = 'amqps://{}:{}@{}{}'.format(urllib.quote_plus(username),
                                  urllib.quote_plus(sas_token), hostname, operation)

receive_client = uamqp.ReceiveClient(uri, debug=True)
while True:
    batch = receive_client.receive_message_batch(max_batch_size=5)
    for msg in batch:
        print('*** received a message ***')
        print(''.join(msg.get_data()))

        # Property 'to' is set to: '/devices/device1/messages/devicebound',
        print('\tto:                     ' + str(msg.properties.to))

        # Property 'message_id' is set to value provided by the service
        print('\tmessage_id:             ' + str(msg.properties.message_id))

        # Other properties are present if they were provided by the service
        print('\tcreation_time:          ' + str(msg.properties.creation_time))
        print('\tcorrelation_id:         ' +
              str(msg.properties.correlation_id))
        print('\tcontent_type:           ' + str(msg.properties.content_type))
        print('\treply_to_group_id:      ' +
              str(msg.properties.reply_to_group_id))
        print('\tsubject:                ' + str(msg.properties.subject))
        print('\tuser_id:                ' + str(msg.properties.user_id))
        print('\tgroup_sequence:         ' +
              str(msg.properties.group_sequence))
        print('\tcontent_encoding:       ' +
              str(msg.properties.content_encoding))
        print('\treply_to:               ' + str(msg.properties.reply_to))
        print('\tabsolute_expiry_time:   ' +
              str(msg.properties.absolute_expiry_time))
        print('\tgroup_id:               ' + str(msg.properties.group_id))

        # Message sequence number in the built-in event hub
        print('\tx-opt-sequence-number:  ' +
              str(msg.annotations['x-opt-sequence-number']))

Wysyłanie komunikatów telemetrycznych (klient urządzenia)

Możesz również wysyłać komunikaty telemetryczne z urządzenia przy użyciu protokołu AMQP. Urządzenie może opcjonalnie podać słownik właściwości aplikacji lub różne właściwości komunikatu, takie jak identyfikator komunikatu.

Aby kierować komunikaty na podstawie treści komunikatu content_type , należy ustawić właściwość na application/json;charset=utf-8. Aby dowiedzieć się więcej na temat routingu komunikatów na podstawie właściwości komunikatów lub treści komunikatów, zobacz dokumentację składni zapytania routingu komunikatów usługi IoT Hub.

Poniższy fragment kodu używa biblioteki uAMQP w języku Python do wysyłania komunikatów z urządzenia do chmury z urządzenia.

# ...
# Create a send client for the device-to-cloud send link on the device
operation = '/devices/{device_id}/messages/events'.format(device_id=device_id)
uri = 'amqps://{}:{}@{}{}'.format(urllib.quote_plus(username), urllib.quote_plus(sas_token), hostname, operation)

send_client = uamqp.SendClient(uri, debug=True)

# Set any of the applicable message properties
msg_props = uamqp.message.MessageProperties()
msg_props.message_id = str(uuid.uuid4())
msg_props.creation_time = None
msg_props.correlation_id = None
msg_props.content_type = 'application/json;charset=utf-8'
msg_props.reply_to_group_id = None
msg_props.subject = None
msg_props.user_id = None
msg_props.group_sequence = None
msg_props.to = None
msg_props.content_encoding = None
msg_props.reply_to = None
msg_props.absolute_expiry_time = None
msg_props.group_id = None

# Application properties in the message (if any)
application_properties = { "app_property_key": "app_property_value" }

# Create message
msg_data = b"Your message payload goes here"
message = uamqp.Message(msg_data, properties=msg_props, application_properties=application_properties)

send_client.queue_message(message)
results = send_client.send_all_messages()

for result in results:
    if result == uamqp.constants.MessageState.SendFailed:
        print result

Dodatkowe uwagi

Połączenia AMQP mogą być zakłócane z powodu usterki sieci lub wygaśnięcia tokenu uwierzytelniania (wygenerowanego w kodzie). Klient usługi musi obsługiwać te okoliczności i w razie potrzeby ponownie opublikować połączenie i łącza. Jeśli token uwierzytelniania wygaśnie, klient może uniknąć porzucania połączenia, proaktywnie odnawiając token przed jego wygaśnięciem.
Klient musi od czasu do czasu mieć możliwość poprawnego obsługi przekierowań linków. Aby zrozumieć taką operację, zapoznaj się z dokumentacją klienta protokołu AMQP.

Następne kroki

Aby dowiedzieć się więcej o protokole AMQP, zobacz specyfikację protokołu AMQP w wersji 1.0.

Aby dowiedzieć się więcej o komunikatach usługi IoT Hub, zobacz:

Udostępnij za pośrednictwem