Atlas

Appearance

Connecting to MQTT ​

This document walks through getting connected to Cloud Connect via MQTT or Websocket. Code samples using Paho clients are provided. Additionally Postman configuration is provided for local testing.

Prerequisites ​

  • Host name and port number of the MQTT instance for your Experiencer Manager
  • MQTT software, or destop client such as Postman
  • Experience Manager credentials with assigned Broker permissions

Hostname and Port ​

The hostname and port for you MQTT instance will vary depending on how and where is it deployed.

If your Experience Manager is version 1.3 or greater, select Preferences in the sidebar of your Experience Manager, you will find the information here.

If your Experience Manager is version <= 1.2 contact Upswell (for Cloud instances), or your IT manager for self-hosted instances.

Individual Credentials ​

Your MQTT credentials are the same as the same as your Experience Manager credentials.

TIP

If you sign-in using your Google or Microsoft accounts, you MUST set a password to connect to MQTT. Signin to your Experience Manager and locate your profile in the Sidebar to set your password.

WARNING

Only use your personal MQTT credentials for testing with applications like Postman, never use your personal credentials for scripts, services or other unattended applications. Create a service account (see below).

Node Credentials ​

Node's are responsible for managing their own credentials which are enrypted and stored locally on the Node. Node credentials are scoped to provide limited access to MQTT topics:

  1. ack/<node nanoid>: write permission for Node message acknowledgements
  2. cluster/<cluster nanoid>: subscribe + read permission for the Cluster wide messages
  3. node/<node nanoid>: subscribe + read permission for Node specific messages

TIP

Nodes may only send response messages on their ack topic. The DO NOT have direct control over themselves or others Nodes in their cluster.

For more information see: Topic Scheme.

Additionally, when an Application is launched using the Experience Manager, the Cloud Connect/MQTT connection details are provided as environment variables. This allows your application to connect without additional credentials. See: Experience Agent Environment Variables.

Service Accounts ​

Service Accouts are full Atlas Experience Manager User accounts, but they are not tied to a specific User and will often have the much more limited permissions than an individual user may have.

Example use cases for Service Accounts include Continous Integration/Continouous Deployment systems or customer software workers/agent.

INFO

Service accounts may currently only be issues through the Experice manager Admin at /admin/ by a Superuser, or a User assigned the Administratro role.

To issues a Service Accounts:

  1. Access the Admin from the sidebar (if you don't see Admin you do not have permission to access it)
  2. Locate Users under Users and Authentication in the middle column
  3. Select + Add User
  4. Enter user details, the prefered username format is to prefix usernames with their use, (for example: CI/CD roles would be ci:job-name and applications would be app:app-short-name)
  5. Create a secure password and same in the password manager of your choice, click Save
  6. Add the appropriate Role to the user on the following screen and select Save again

TIP

To quickly generate a secure, random password use: openssl rand -hex 16.

Once the Service Account is issued, if the Service Account needs access to MQTT, topic assignments will need to be manually created:

  1. From the Admin landing screen, select Cluster Management and Broker Authrorized Topics
  2. Select + Add Broker Authorized Topic
  3. Select the user, add the Topic Name and select the permission
  4. Save and add additional permissions as needed

For more information on Topics and Topic assignments see: Cloud Connect Topic Scheme.

Short-lived Tokens ​

Like direct MQTT connections, browser based MQTT connections require a username/password. This presents both UX and security issues requiring users to re-enter there password, or holding it in a store until MQTT is ready to connect. This is additionally problematic when users login with SSO and the user does not have a set password.

To work around this, the Experience Manager can issue short lived MQTT credentials for a user. This API is available to both Users logged in with session authentication, as well as clients using Bearer API authentication.

When using MQTT token authentication, the username will have :::token appended to the end of the username, this indicates to MQTT that the short-lived token should be used instead of the users password.

The obtain a token, perform a get request to: GET /api/v2/accounts/users/mine/mqtt-token/. This will return the full connection details, including the username and password for the requesting user.

TIP

Short lived token expire 1-minute from the time of issuance. If you are disconnected for any reason, you will need to first call the mqtt-token endpoint again to get valid credentials to reconnect.

json
{
    "username": "<username>:::token",
    "password": "",
    "expires": "0000-00-00T00:00:00.000000Z",
    "broker": {
        "mqtt": {
            "host": "<client id>-connect.upswell.cloud",
            "port": 000000,
            "tls": true,
            "allow_insecure": false
        },
        "ws": {
            "host": "upswell-connect-ws.upswell.cloud",
            "port": 443,
            "tls": true,
            "allow_insecure": false
        }
    }
}

For more information on working with the API's directly see: Experience Manager API’s

Software Clients ​

Atlas Cloud Connect uses Mosquitto MQTT under the hood. While MQTT is standards based, and many MQTT clients exist, Paho from the Eclipse Foundation has provided reliable connectivity between Atlas services.

Additionally qtmqtt has proven a reliable C++ client when working in Qt/C++.

The following clients have undergone extensive testing:

Example: Python (paho-mqtt) ​

The paho-mqtt for Python has been extremely reliable for Python based software clients. This minimal examples shows how to conect to MQTT with a gradual backoff if the client cannot connect.

INFO

If you receive Authentication errors while connecting, or subscribing to a MQTT topic, ensure the connecting user has the appropriate permission. See above for details.

python
import random
import string
import time

import ssl
import paho.mqtt.client as mqtt

client: mqtt.Client

host: str = ''
port: int = 0
username: str = ''
password: str = ''
rand = ''.join(random.choice(string.ascii_letters + string.digits) for i in range(10))
client_id: str = f'${username}-${rand}'

attempt: int = 0
retry_interval: int = 5

def connect():
    try:
        client.connect(self.store.node.broker.mqtt.host, self.store.node.broker.mqtt.port, 60)  # noqa: E501
    except Exception as e:
        print(f'MQTT :: connect => Connection error, will attempt reconnect : {e}')
        reconnect()

def reconnect():
    attempt += 1
    retry_in: int = self.attempt * self.retry_interval
    retry_in = retry_in if retry_in <= retry_max else retry_max
    time.sleep(retry_in)
    connect()

def on_connect(self, client: mqtt.Client, userdata: Any, flags: mqtt.ConnectFlags, rc: int) -> None:
    if rc == 0:
        on_connection_success()
    elif rc == 2:
        print('MQTT :: on_connect => Connection failed with invalid client identifier, '
            'likely from reusing the connection, generating a new client id and reconnecting')
        rand = ''.join(random.choice(string.ascii_letters + string.digits) for i in range(10))
        client_id = f'{username}-{rand}'
        reconnect()
    elif rc == 3:
        print('MQTT :: on_connect => Connection failed RC3, Server Unavailable will attempt reconnect.')
        reconnect()
    else:
        print(f'MQTT :: on_connect :: {rc}, will not attempt to reconnect')

def on_connect_fail(self, client: mqtt.Client, userdata: Any) -> None:
    reconnect()

def on_connection_success(self):
    client.subscribe('system')

def on_disconnect(self, client, userdata, rc):
    if rc != 0:
        print("MQTT :: on_disconnect => Unexpected disconnect, attempt to reconnect ...")
        reconnect()
    else:
        print("MQTT :: Sucesfully disconnected from broker")

def on_message(self, client, userdata, msg) -> None:
    '''Handle incoming message, process and convert to RequestMessage. If
    message contains `RESPONSE_MESSAGE` it will be ignored.
    '''
    try:
        payload: dict = json.loads(msg.payload)
    except json.JSONDecodeError as e:
        print(f'MQTT :: On Message Error => Cannot process message, JSONDecodeError: {e}')
        return

def on_subscribe(self, client, userdata, mid, granted_qos):
    print(f"MQTT :: on_subscribe => Sucesfully subscribed to topic with qos `{granted_qos}`")

client: mqtt.Client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION1, client_id)
client.username_pw_set(username, password=password)
client.tls_set(cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLS_CLIENT)

client.on_connect = on_connecct
client.on_connect_fail = on_connect_fail
client.on_disconnect = on_disconnect
client.on_message = on_message
client.on_subscribe = on_subscribe

if __name__ == "__main__":
    connect()

Example: Javascript (paho.mqtt.javascript) ​

The Paho Javascript library (paho.mqtt.javascript) may be used in-browser with a Websocket. paho.mqtt.javascript plays well with frameworks such as React and Vue as well as Vanilla Javascript.

TIP

The MQTT websocket is typically deployed with TLS on port 443 and useSSL will be required to connect except in rate instances. If you have difficulty connecting, or have questions Check with Upswell for managed Cloud instance, or your IT team for self-hosted instances to confirm.

The following provides a basic working example using paho.mqtt.javascript to establish a connection with MQTT. Consult the MQTT Topic Scheme for more information about topic subscriptions.

INFO

If you receive Authentication errors while connecting, or subscribing to a MQTT topic, ensure the connecting user has the appropriate permission. See above for details.

javascript
const host = ''
const port = 443 
const password = ''
const use_tls = true

const client = new Paho.Client(host, port, '/ws', username);

client.onConnectionLost = function(responseObject) {
    console.log("MQTT :: Connection Lost => " + responseObject.errorMessage);
}

client.onMessageArrived = function (message) {
    try {
        const msg = JSON.parse(message.payloadString)
        console.log(`MQTT :: Message on ${message.destinationName} => ${msg}`)
    } catch(e) {
        console.error(`MQTT :: Client Error => ${e}`)
    }
}

function connect() {
    client.connect({
        onSuccess: onConnect,
        onFailure: onFailure,
        userName : username,
        password : password,
        useSSL: use_tls
    })
}

function onConnect() {
    connected = false
    client.subscribe('system')
}

function onFailure(message) {
    console.error(`MQTT :: Connection Failed => ${message}`)
    connect()
}

connect()

Atlas, Hybrid cloud, on-premesis platform for large scale media program development, delivery and operation.

Copyright © 2018-present Upswell