Consume messages from topics

A downstream application can consume the lifecycle, shadow, error and device messages from topics.

Prerequisites:

1. Create a group in profiles/IAM service and activate groups in IoT with topics as delivery endpoint. Lifecycle, Shadow and Error topic endpoints will be returned in the response:

   "deliveryEndpoints": {
           "lifecycle": "iot-lifecycle-00637b20-c4a1-4475-aeb5-adc3435fcce8",
           "shadow": "iot-shadow-00637b20-c4a1-4475-aeb5-adc3435fcce8",
           "error": "iot-error-00637b20-c4a1-4475-aeb5-adc3435fcce8"
       }

2. Create a device in Profiles/ IAM and activate a device in IoT.

3. Create rules with ‘gps’ as topic suffix. A topic endpoint will be returned in the response.

   "distributionTopic": "iot-telemetry-00637b20-c4a1-4475-aeb5-adc3435fcce8-{rulename}"

4. Create consumer groups. The name of the consumer group will be returned in the response, which can be used while subscribing to topics.

Steps:

1. Connect an MQTT client using the following credentials to consume messages:

   • Broker endpoint: Broker endpoint returned in the activate group response
   • Port: 443
   • Username: username?group_id=
   • Password: TID Trimble Identity - see identity token authorized to access the group
   • ALPN Protocol: Set ALPN Protocol as MQTTSample code

   from __future__ import print_function
   import sys
   import ssl
   import logging, traceback
   import paho.mqtt.client as mqtt
   
   IoT_protocol_name = "mqtt"
   aws_iot_endpoint = "<Broker Endpoint>"
   
   
   logger = logging.getLogger()
   logger.setLevel(logging.DEBUG)
   handler = logging.StreamHandler(sys.stdout)
   log_format = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
   handler.setFormatter(log_format)
   logger.addHandler(handler)
   mqttc = mqtt.Client(client_id="<ClientId>")
   
   def ssl_alpn():
       try:
           #debug print openssl version
           logger.info("open ssl version:{}".format(ssl.OPENSSL_VERSION))
           ssl_context = ssl.create_default_context()
           ssl_context.set_alpn_protocols([IoT_protocol_name])
           # ssl_context.load_verify_locations(cafile=caPath)
           # ssl_context.load_cert_chain(certfile=certPath, keyfile=keyPath)
   
           return  ssl_context
       except Exception as e:
           print("exception ssl_alpn()")
           raise e
   
   def on_connect(client, userdata, flags, rc):
      print("Connected With Result Code: {}".format(rc))
      mqttc.subscribe(topic='$share/<consumer group name>/iot-lifecycle-<groupId>', qos=1)
      mqttc.subscribe(topic='$share/<consumer group name>/iot-shadow-<groupId>', qos=1)
      mqttc.subscribe(topic='$share/<consumer group name>/iot-error-<groupId>', qos=1)
      mqttc.subscribe(topic='$share/<consumer group name>/iot-telemetry-<groupId>-<rulename>', qos=1)
   
   def on_disconnect(client, userdata, rc):
      print("Client Got Disconnected".format(rc))
   
   def on_subscribe(client, userdata, mid, granted_qos):
        print("Client Subscribed")
   
   def on_message(client, userdata, msg):  # The callback for when a PUBLISH
       print("Message received-> " + msg.topic + " " + str(msg.payload))
   
   if __name__ == '__main__':
       try:
           ssl_context= ssl_alpn()
           mqttc.username_pw_set(username='username?group_id=<groupId>', password='<tid token>')
           mqttc.tls_set_context(context=ssl_context)
           mqttc.on_connect = on_connect
           mqttc.on_disconnect = on_disconnect
           mqttc.on_subscribe = on_subscribe
           mqttc.on_message=on_message
           logger.info("start connect")
           mqttc.connect(aws_iot_endpoint, port=443)
           logger.info("connect success")
           mqttc.loop_forever()
   
       except Exception as e:
           logger.error("exception main()")
           logger.error("e obj:{}".format(vars(e)))
           logger.error("message:{}".format(e.message))
           traceback.print_exc(file=sys.stdout)

Consume Replay Events from Topics

Telemetry events between a particular time period can be re-published to the consumer applications by scheduling a replay job.

Steps to replay events

1. Establish an MQTT connection to the IoT endpoint using the above sample snippet.
2. Subscribe to the replay topic. Format (iot-replay-telemetry-{groupId}).
3. Create a replay job. For more information, refer to the API documentation.
4. Events will be published to the replay topic.

Note: To pause, resume or cancel a replay job refer to the API documentation.