MQTT Service

Overview

Architecture

The MQTT Service works together with the Messaging Service to provide a framework for highly customizable and flexible MQTT message processing solutions. The diagram below illustrates how a message flows, starting from the device, through the Messaging Service, then to a user-provided microservice where it is converted to the Cumulocity JSON format and delivered to Cumulocity using the standard REST API.

All MQTT messages published to the MQTT Service are forwarded to the Messaging Service, where they are persisted, waiting to be consumed. A custom microservice that understands the topic and payload structure can, with the help of the Java Client, consume the MQTT messages, translate them to the Cumulocity format, and then use the Microservice SDK to push them into Cumulocity.

Similarly, messages can be sent to devices, as shown in the diagram below. In this case, the user-provided microservice receives messages from Cumulocity through a Notifications 2.0 subscription. These messages are mapped to the payload structure used by the MQTT devices, then published to MQTT topics using the Java Client.

As with MQTT messages published by devices, messages published from a microservice will be forwarded to the Messaging Service, where they can be consumed by MQTT devices subscribed to the relevant topics.

MQTT Service compared to Core MQTT

The table below presents a basic comparison between the Cumulocity Core MQTT functionality and that of the MQTT Service.

MQTT protocol implementation

This section covers some implementation details of the MQTT Service. The MQTT Service implementation supports clients connecting using MQTT versions 3.1, 3.1.1 and 5.0, although not all MQTT 5.0 protocol features are currently supported.

Connecting to the service

MQTT connections to the MQTT Service must use TCP. Use your tenant domain as the target host for the connection, for example {my-tenant}.cumulocity.com.

Available ports:

Port 9883 (TLS) is the default port and should be used for secure, encrypted communication. Both one-way (server certificate only) and two-way (both client and server certificates) TLS are supported. When client certificates are not used, the server authenticates the client using standard username and password credentials. Port 2883 (no TLS) is not enabled in Cumulocity shared public environments due to the security risks of unencrypted traffic. To enable port 2883 in a dedicated environment, please contact Product support.

Topics

MQTT Service topics are mapped to the Messaging Service subscriptions with identical names, including additional URL encoding. The Messaging Service subscriptions reliably store the topic messages for asynchronous processing. The messages stored on these subscriptions can be consumed using a dedicated Java Client.

Topic restrictions

The MQTT Service does not impose any topic structure. There are just a few topic names which are reserved for historic purposes and future use, namely:

• All SmartREST 2.0 related topics
• error
• devicecontrol/notifications

Other than these restrictions you are free to use any topic name which is compatible with the MQTT specification.

Topic limits

The MQTT Service imposes several topic-related limits. See the Service Quotas section for details of the current limits in force.

There is a limit on the total number of topics that a single tenant can create. When the creation of a new topic, either by creating it via the client publishing a message or subscribing to a non-existent topic, would breach the topic limit the delivery of the packet is prevented.

The different MQTT protocol versions provide different feedback when this limit is exceeded.

MQTT 5 clients:

• Have access to the reason code and reason string describing the failure when using QoS 1 with acknowledgements, where the reason code is QUOTA_EXCEEDED: 0x97.

MQTT 3.1 and 3.1.1 clients:

• Clients only have access to the reason code describing the failure when using QoS 1 with acknowledgements and only for SUBSCRIBE packets, where the reason code is 0x80.
• For PUBLISH packets, the client will be disconnected with no further information as per the MQTT specification.

In addition to the topic count, the MQTT Service also limits the size of the message backlog on each topic. The message backlog contains all messages that have been published on the topic but not yet received and acknowledged by all subscribers to the topic. When the backlog limit is reached, further attempts to publish to the topic will fail until some messages have been consumed.

Each message in a topic backlog also has a time-to-live (TTL) that starts at the moment the message is published. When the TTL of a message expires, that messages will be deleted from the backlog regardless of whether all subscribers have received it or not. MQTT clients do not receive any notification that messages have been discarded from a backlog due to TTL expiry.

Error topic

The MQTT Service provides clients the ability to review errors through messages received by subscribing to the error topic, $debug/$error. When subscribing to the topic it will act as a per-client topic, meaning the client will only receive messages exclusively related to their client ID. For example, if a client was attempting to subscribe to a new topic, and the creation of the topic would exceed the topic limit, only that client would receive an error.

According to the MQTT 3.1.1 specification, if either the server or the client encounters a protocol violation, it must close the network connection on which it received the control packet which caused the violation.

In such instances MQTT clients must reconnect to be able to receive error messages from the error topic via the subscription. Error messages received after this reconnection are from the previous session. This can lead to confusion when attempting corrective actions. Therefore, we highly recommend you to build a microservice which uses the MQTT Service SDK to consume error messages, or use MQTT 5 for clients and make use of the reason codes feature.

Topic cleanup

The MQTT service will automatically remove topics which are no longer active. Topics are recognized as inactive when there are no subscriptions and the internal publisher to the topic is closed. The publisher is responsible for publishing the modified MQTT service messages to the correct topic. The publishers live within a cache, where the publisher expires after one hour. Due to this it can take up to an hour after removing all subscriptions from a topic for it to be automatically deleted.

Payload

MQTT protocol messages map bidirectionally to the internal MQTT Service message format which includes the original payload and additional metadata fields. Assuming Java types, the packed message structure looks as follows:

MqttServiceMessage

MqttServiceMetadata

The Java Client contains classes representing the above model.

Payload restrictions

The MQTT Service does not impose any specific payload format. All the incoming MQTT messages must meet the specification in terms of fixed and variable headers, but the payload for published messages is unrestricted. A custom microservice will receive the exact same set of bytes that was sent by an MQTT device, and is responsible for converting these to Cumulocity compatible format.

The size of the MQTT payload is limited to a maximum value that includes both the message header and body. The size of an MQTT packet header varies, but it will be at least 2 bytes. See the Service Quotas section for details of the current limit in force.

Features

Authentication and authorization

The authentication types supported by the MQTT Service are:

• Username and password: The MQTT username must include the tenant ID and username in the format <tenantID>/<username>.
• Device certificates: The MQTT username must include the tenant ID in the format <tenantID>.

ClientId

The MQTT ClientID field identifies the connected client. ClientID may consist of up to 128 alphanumeric characters. Each client connecting to the MQTT Service must have a unique client identifier, connecting a second client with the same identifier will result in the previous client’s disconnection.

Quality of Service (QoS)

The MQTT Service implementation supports two levels of MQTT QoS:

• QoS 0: At most once:
  • The client sends the message once (fire and forget).
  • There is no response from the server.
  • There is no guarantee that subscribers will receive the message.
• QoS 1: At least once:
  • The client awaits server acknowledgment for each published message.
  • The client should re-send the message if there was no acknowledgement from the server.
  • It is guaranteed that subscribers will receive a message that was acknowledged by the server.
  • Subscribers may receive more than one copy of a message.
• QoS 2: Exactly once:
  • not supported

For subscriptions, the MQTT Service will deliver messages in the QoS that the client defined when subscribing to the topic (QoS 0 or 1).

Clean session

The MQTT Service requires the clean session flag to be set to “1” (true). Disabling clean session has no effect, meaning that a reconnecting client will not see any messages that it missed since the last disconnection, regardless of how it sets this flag.

Retained flag

The retained flag is currently ignored. Publishing data with the retained flag on the topic is allowed but has no practical difference to sending it without the flag.

Last will

In MQTT, the “last will” is a message that is specified at connection time and that is executed when the client loses the connection. Last will is fully supported by the MQTT Service, and as with with any other publish messages you can use any unreserved topic and any payload.

Return codes

The MQTT Service follows the MQTT specification for server responses. For example, if invalid credentials are sent in the CONNECT message, the server response CONNACK message contains the 0x05 return code. The return code can be treated similarly to REST API HTTP codes, such as 401.

MQTT 5.0 features

Clients can connect using version 5.0 of the MQTT protocol. Support for additional MQTT 5.0 features will be added in future releases.

MQTT TLS certificates

Server certificates

The MQTT Service uses the same server certificates that are assigned to the main Cumulocity environment domain. It always sends these certificates during TLS handshake to devices. Moreover, Enterprise tenants are not able to customize those certificates via the SSL Management feature.

Device (client) certificates

Using device certificates with the MQTT Service shares the same requirements as outlined in Device certificates. Additionally, auto-registration must be enabled when uploading the CA certificate to the platform. When connecting devices to the MQTT Service using certificates, the tenant ID must be included in the MQTT CONNECT packet in the user name field. This is required to correctly identify the tenant.

Adding and trusting CA certificate

TLS trust anchors in the Cumulocity platform are defined per tenant. To use device certificates for authentication, the CA or intermediate certificate that signs the device certificates must be uploaded to the platform and added to the tenant’s list of trusted certificates. Additionally, the Auto registration field must be enabled when adding certificates. For detailed instructions on adding and trusting a CA certificate, see Managing trusted certificates.

Creating self-signed certificates

In order to self-sign the device certificates, the root CA certificate needs to be created. Using the OpenSSL CLI tool, create a private key and then generate a self-signed root certificate from it.

openssl genpkey -algorithm RSA -out ca.key
openssl req -x509 -new -nodes -key ca.key -sha256 -days 3650 -out ca.crt -subj "/C=UK/O=YourCompany/OU=YourOrg/CN=MQTTServiceCA"

Then create a private key for the device, generate the certificate signing request from this private key, and then sign the CSR.

openssl genpkey -algorithm RSA -out client.key
openssl rsa -in client.key -out client-key.pem -outform PEM
openssl req -new -key client.key -out client.csr -subj "/C=UK/O=YourCompany/OU=YourOrg/CN=mqtt-client"
openssl x509 -req -in client.csr -CA ca.crt -CAkey ca.key -CAcreateserial -out client.crt -days 3650 -sha256
cat client.crt ca.crt > client-chain.pem

If you have more advanced requirements regarding certificate creation, see Generating and signing certificates.

Using certificates

Once the CA certificate has been uploaded and trusted in Cumulocity, devices can authenticate using client certificates signed by your trusted CA. To connect using any MQTT client, use the previously generated client certificate and key. This example uses the Mosquitto MQTT client:

mosquitto_pub --cafile cumulocity.com.pem -d -q 1 \
  -h "cumulocity.com" -p "9883" -i myclient \
  -u t11101 \
  -t "v1/devices/me/telemetry" \
  --key client-key.pem \
  --cert client-chain.pem \
  -m '{"temperature":25}'

Explanation:

• --cafile cumulocity.com.pem: This file contains the CA certificate of Cumulocity’s MQTT Service broker, used to validate the server’s identity.
• --key client-key.pem and --cert client-chain.pem: These are your client certificate and private key, signed by your trusted CA.
• -u t11101: The username must be your tenant ID. In this example, t11101 is the tenant ID.

Downloading the CA certificate (cumulocity.com.pem):

To download the Cumulocity MQTT Service broker’s CA certificate:

1. Open cumulocity.com in a browser.
2. Click the padlock icon in the address bar and view the certificate details.
3. Download or export the root certificate, and save it as cumulocity.com.pem.

Alternatively, you can use openssl to retrieve and extract the certificate:

echo | openssl s_client -connect cumulocity.com:9883 -showcerts 2>/dev/null | \
    sed -ne '/-BEGIN CERTIFICATE-/,/-END CERTIFICATE-/p' > cumulocity.com.pem

Java Client

The MQTT Service Java Client library provides the classes necessary to interact with the MQTT Service. The following operations are supported by the client:

• Publishing messages to the MQTT Service via a WebSocket protocol.
• Subscribing to messages from the MQTT Service via a WebSocket protocol

Repositories and dependencies

Follow the Microservice SDK documentation for guidance on how to configure Maven repositories. To include the MQTT Service Java Client into your project, add the following dependency inside the <dependencies> node:

<dependency>
    <groupId>com.cumulocity.sdk.mqtt</groupId>
    <artifactId>mqtt-service-ws</artifactId>
    <version>${c8y.version}</version>
</dependency>

Examples

Example of publishing messages to the MQTT Service via WebSocket:

// Message to be sent
final String payload = "Hello World";

// Construct a new MqttServiceMessage and set the payload
final MqttServiceMessage message = new MqttServiceMessage();
message.setPayload(payload.getBytes());

// Create an instance of MqttServiceApi by specifying the server URI to connect to along with TokenApi
final MqttServiceApi mqttServiceApi = MqttServiceApi.webSocket()
        .url(webSocketBaseUrl)
        .tokenApi(tokenApi)
        .build();

// Build PublisherConfig with topic to which the message is to be sent
final PublisherConfig config = PublisherConfig.publisherConfig().topic(topic).build();

// Build Publisher and publish MqttServiceMessage. Close the resource either by using a [try-with-resources block](https://docs.oracle.com/javase/tutorial/essential/exceptions/tryResourceClose.html) as below or by calling publisher.close() explicitly
try (final Publisher publisher = mqttServiceApi.buildPublisher(config)) {
    publisher.publish(message);
} catch (Exception e) {
    log.error("Could not sent message to {}", topic, e);
}
mqttServiceApi.close();

Example of subscribing to messages from the MQTT Service via WebSocket:

// Create an instance of MqttServiceApi by specifying the server URI to connect to along with TokenApi
final MqttServiceApi mqttServiceApi = MqttServiceApi.webSocket()
        .url(webSocketBaseUrl)
        .tokenApi(tokenApi)
        .build();

// Build SubscriberConfig with topic and subscriber name
final SubscriberConfig config = SubscriberConfig.subscriberConfig().topic(topic).subscriber(subscriberName).build();

// Build Subscriber
final Subscriber subscriber = mqttServiceApi.buildSubscriber(config);

// Subscribe by passing implementation of MessageListener to handle messages from the MQTT Service.
subscriber.subscribe(new MessageListener() {
    @Override
    public void onMessage(MqttServiceMessage message) {
        log.info("Message Received: {}", new String(message.getPayload()));
    }
});

// Close the resources after usage
subscriber.close();
mqttServiceApi.close();

Frequently Asked Questions

Q: How can I map my MQTT device payloads to Cumulocity without developing a custom microservice?
A: One option is to use the Dynamic Mapping Service for Cumulocity. This is a community-supported open-source component that allows many different payload formats and encodings to be mapped to the Cumulocity domain model. Mappings can be configured using a graphical UI or by writing JavaScript code.

Q: How can I obtain device credentials for my MQTT devices?
A: The MQTT Service is not yet integrated with the Cumulocity device bootstrap process. This support is planned for a future release. In the meantime, follow the Integration life cycle to bootstrap the device and obtain device credentials. Once the device credentials are obtained, the device can use them to connect to the MQTT Service.

Q: Does the MQTT Service support the SmartREST 2.0 protocol?
A: Not yet. Support for SmartREST 2.0 is planned for a future release.

Q: Why does the MQTT Service not use the standard MQTT ports 1883 and 8883?
A: Those ports are already used by Cumulocity Core MQTT. While both MQTT implementations are operating in parallel, the MQTT Service must use different ports.