Magistrala
Dev GuideDevelopment Tools

Messaging

Learn how to publish and subscribe to messages in Magistrala using HTTP, MQTT, CoAP and WebSocket protocols with broker and subtopic support.

Once a channel is provisioned and client is connected to it, it can start to publish messages on the channel. The following sections will provide an example of message publishing for each of the supported protocols.

HTTP

To publish message over channel, client should send following request:

curl -s -S -i --cacert docker/ssl/certs/ca.crt -X POST -H "Content-Type: application/senml+json" -H "Authorization: Client <client_secret>" https://localhost/http/m/<workspace_id>/c/<channel_id> -d '[{"bn":"some-base-name:","bt":1.276020076001e+09, "bu":"A","bver":5, "n":"voltage","u":"V","v":120.1}, {"n":"current","t":-5,"v":1.2}, {"n":"current","t":-4,"v":1.3}]'

Note that if you're going to use SenML message format, you should always send messages as an array.

For more information about the HTTP messaging service API, please check out the API documentation.

MQTT

To send and receive messages over MQTT you could use Mosquitto tools, or Paho if you want to use MQTT over WebSocket.

To publish message over channel, client should call following command:

mosquitto_pub -I <client_name> -u <client_id> -P <client_secret> -t m/<workspace_id>/c/<channel_id> -h localhost -m '[{"bn":"some-base-name:","bt":1.276020076001e+09, "bu":"A","bver":5, "n":"voltage","u":"V","v":120.1}, {"n":"current","t":-5,"v":1.2}, {"n":"current","t":-4,"v":1.3}]'

To subscribe to channel, client should call following command:

mosquitto_sub -I <client_name> -u <client_id> -P <client_secret> -t m/<workspace_id>/c/<channel_id> -h localhost

If you want to use standard topic such as m/<workspace_id>/c/<channel_id> with SenML content type (JSON or CBOR), you should use following topic m/<workspace_id>/c/<channel_id>.

If you are using TLS to secure MQTT connection, add --cafile docker/ssl/certs/ca.crt to every command.

CoAP

CoAP adapter implements CoAP protocol using underlying UDP and according to RFC 7252. To send and receive messages over CoAP, you can use CoAP CLI. To set the add-on, please follow the installation instructions provided at CoAP CLI.

Examples:

coap-cli get m/<workspace_id>c/<channel_id>/subtopic -auth <client_secret> -o
coap-cli post m/<workspace_id>c/<channel_id>/subtopic -auth <client_secret> -d "hello world"
coap-cli post m/<workspace_id>c/<channel_id>/subtopic -auth <client_secret> -d "hello world" -h 0.0.0.0 -p 1234

To send a message, use POST request. To subscribe, send GET request with Observe option (flag o) set to false. There are two ways to unsubscribe:

  1. Send GET request with Observe option set to true.
  2. Forget the token and send RST message as a response to CONF message received by the server.

The most of the notifications received from the Adapter are non-confirmable. By RFC 7641:

Server must send a notification in a confirmable message instead of a non-confirmable message at least every 24 hours. This prevents a client that went away or is no longer interested from remaining in the list of observers indefinitely.

CoAP Adapter sends these notifications every 12 hours. To configure this period, please check adapter documentation If the client is no longer interested in receiving notifications, the second scenario described above can be used to unsubscribe.

WebSocket

To publish and receive messages over channel using web socket, you should first send handshake request to /m/<workspace_id>/c/<channel_id> path. Don't forget to send Authorization header with client authorization token. In order to pass message content type to WS adapter you can use Content-Type header.

If you are not able to send custom headers in your handshake request, send them as query parameter authorization and content-type. Then your path should look like this /m/<workspace_id>/c/<channel_id>?authorization=<client_secret>&content-type=<content-type>.

If you are using the docker environment prepend the url with ws. So for example /ws/m/<workspace_id>/c/<channel_id>?authorization=<client_secret>&content-type=<content-type>.

Basic nodejs example

const WebSocket = require("ws");
// do not verify self-signed certificates if you are using one
process.env.NODE_TLS_REJECT_UNAUTHORIZED = "0";
// c02ff576-ccd5-40f6-ba5f-c85377aad529 is an example of a client_auth_key
const ws = new WebSocket(
  "ws://localhost:8186/ws/m/<workspace_id>/c/1?authorization=c02ff576-ccd5-40f6-ba5f-c85377aad529"
);
ws.on("open", () => {
  ws.send("someclient");
});
ws.on("message", (data) => {
  console.log(data);
});
ws.on("error", (e) => {
  console.log(e);
});

Basic golang example

package main

import (
 "log"
 "os"
 "os/signal"
 "time"

 "github.com/gorilla/websocket"
)

var done chan interface{}
var interrupt chan os.Signal

func receiveHandler(connection *websocket.Conn) {
 defer close(done)

 for {
  _, msg, err := connection.ReadMessage()
  if err != nil {
   log.Fatal("Error in receive: ", err)
   return
  }

  log.Printf("Received: %s\n", msg)
 }
}

func main() {
 done = make(chan interface{})
 interrupt = make(chan os.Signal)

 signal.Notify(interrupt, os.Interrupt)

 channelId := "30315311-56ba-484d-b500-c1e08305511f"
 domainId := "f89f0e6e-f5ce-43fa-b9d5-3ce98ae50512"
 clientSecret := "c02ff576-ccd5-40f6-ba5f-c85377aad529"

 socketUrl := "ws://localhost:8186/m/" + domainId + "/c/" + channelId + "?authorization=" + clientSecret

 conn, _, err := websocket.DefaultDialer.Dial(socketUrl, nil)
 if err != nil {
  log.Fatal("Error connecting to Websocket Server: ", err)
 } else {
  log.Println("Connected to the ws adapter")
 }
 defer conn.Close()

 go receiveHandler(conn)

 for {
  select {

  case <-interrupt:
   log.Println("Interrupt occured, closing the connection...")
   conn.Close()
   err := conn.WriteMessage(websocket.TextMessage, []byte("closed this ws client just now"))
   if err != nil {
    log.Println("Error during closing websocket: ", err)
    return
   }

   select {
   case <-done:
    log.Println("Receiver Channel Closed! Exiting...")

   case <-time.After(time.Duration(1) * time.Second):
    log.Println("Timeout in closing receiving channel. Exiting...")
   }
   return
  }
 }
}

MQTT-over-WS

Magistrala also supports MQTT-over-WS, along with pure WS protocol. this bring numerous benefits for IoT applications that are derived from the properties of MQTT - like QoS and PUB/SUB features.

There are 2 reccomended Javascript libraries for implementing browser support for Magistrala MQTT-over-WS connectivity:

  1. Eclipse Paho JavaScript Client
  2. MQTT.js

As WS is an extension of HTTP protocol, Magistrala exposes it on port 8008, so it's usage is practically transparent. Additionally, please notice that since same port as for HTTP is used (8008), and extension URL /mqtt should be used - i.e. connection URL should be ws://<host_addr>/mqtt.

For quick testing you can use HiveMQ UI tool.

Here is an example of a browser application connecting to Magistrala server and sending and receiving messages over WebSocket using MQTT.js library:

<script src="https://unpkg.com/mqtt/dist/mqtt.min.js"></script>
<script>
    // Initialize a mqtt variable globally
    console.log(mqtt)

    // connection option
    const options = {
        clean: true, // retain session
        connectTimeout: 4000, // Timeout period
        // Authentication information
        clientId: '14d6c682-fb5a-4d28-b670-ee565ab5866c',
        username: '14d6c682-fb5a-4d28-b670-ee565ab5866c',
        password: 'ec82f341-d4b5-4c77-ae05-34877a62428f',
    }

    var channelId = '08676a76-101d-439c-b62e-d4bb3b014337'
    var domainId = '6a45444c-4c89-46f9-a284-9e731674726a'
    var topic = 'm/' + domainId + '/c/' + channelId

    // Connect string, and specify the connection method by the protocol
    // ws Unencrypted WebSocket connection
    // wss Encrypted WebSocket connection
    const connectUrl = 'ws://localhost/mqtt'
    const client = mqtt.connect(connectUrl, options)

    client.on('reconnect', (error) => {
        console.log('reconnecting:', error)
    })

    client.on('error', (error) => {
        console.log('Connection failed:', error)
    })

    client.on('connect', function () {
        console.log('client connected:' + options.clientId)
        client.subscribe(topic, { qos: 0 })
        client.publish(topic, 'WS connection demo!', { qos: 0, retain: false })
    })

    client.on('message', function (topic, message, packet) {
        console.log('Received Message:= ' + message.toString() + '\nOn topic:= ' + topic)
    })

    client.on('close', function () {
        console.log(options.clientId + ' disconnected')
    })
</script>

N.B. Eclipse Paho lib adds sub-URL /mqtt automaticlly, so procedure for connecting to the server can be someclient like this:

var loc = { hostname: "localhost", port: 8008 };
// Create a client instance
client = new Paho.MQTT.Client(loc.hostname, Number(loc.port), "clientId");
// Connect the client
client.connect({ onSuccess: onConnect });

mTLS Messaging

Magistrala supports mutual TLS (mTLS) to enhance security by requiring both clients and servers to authenticate each other using certificates. This ensures that only authorized clients can connect and communicate with the server. It is designed to handle high-throughput environments. Core components are modular, making it easy to plug in custom modules or replace existing ones. Extendable to add new IoT protocols, middleware, and features as needed.

Certificate Setup

To enable mTLS, you'll need the following certificates:

  • CA Certificate (ca.crt): The Certificate Authority's certificate used to sign both server and client certificates.
  • Server Certificate (server.crt) and Private Key (server.key): Used by the server to authenticate itself to clients. These will be used by Magistrala.
  • Client Certificate (client.crt) and Private Key (client.key): Used by the client to authenticate itself to the server.

Certificates can be generated using the Magistrala Certs service as described in certs service docs

HTTP with mTLS

By default, HTTP messages can be sent without any encryption or certificate verification:

curl -s -S -i -X POST -H "Content-Type: application/senml+json" -H "Authorization: Client <client_secret>" https://localhost/http/m/<workspace_id>/c/<channel_id> -d '[{"bn":"some-base-name:","bt":1.276020076001e+09, "bu":"A","bver":5, "n":"voltage","u":"V","v":120.1}, {"n":"current","t":-5,"v":1.2}, {"n":"current","t":-4,"v":1.3}]'

But with mTLS, clients must present their certificate during the TLS handshake. This ensures both server and client are authenticated using trusted certificates.

curl -s -S -i --cacert docker/ssl/certs/ca.crt --cert docker/ssl/certs/client.crt --key docker/ssl/certs/client.key -X POST -H "Content-Type: application/senml+json" -H "Authorization: Client <client_secret>" https://localhost/http/m/<workspace_id>/c/<channel_id> -d '[{"bn":"some-base-name:","bt":1.276020076001e+09, "bu":"A","bver":5, "n":"voltage","u":"V","v":120.1}, {"n":"current","t":-5,"v":1.2}, {"n":"current","t":-4,"v":1.3}]'

HTTP with TLS

A user can also send messages with just the TLS support (server authentication only) and just a CA certificate using the command:

curl -s -S -i --cacert docker/ssl/certs/ca.crt  -X POST -H "Content-Type: application/senml+json" -H "Authorization: Client <client_secret>" https://localhost/http/m/<workspace_id>/c/<channel_id> -d '[{"bn":"some-base-name:","bt":1.276020076001e+09, "bu":"A","bver":5, "n":"voltage","u":"V","v":120.1}, {"n":"current","t":-5,"v":1.2}, {"n":"current","t":-4,"v":1.3}]'

MQTT with TLS

You can connect over plain MQTT (port 1883) without any encryption or certificate validation:

mosquitto_pub -u <client_id> -P <client_secret> -t m/<workspace_id>/c/<channel_id> -h localhost -p 1883 -m '[{"bn":"some-base-name:","bt":1.276020076001e+09, "bu":"A","bver":5, "n":"voltage","u":"V","v":120.1}, {"n":"current","t":-5,"v":1.2}, {"n":"current","t":-4,"v":1.3}]'

To connect securely over TLS using the same port 8883 and validate the server certificate with a CA file:

mosquitto_pub --cafile docker/ssl/certs/ca.crt -u <client_id> -P <client_secret> -t m/<workspace_id>/c/<channel_id> -h localhost -p 1883 -m '[{"bn":"some-base-name:","bt":1.276020076001e+09, "bu":"A","bver":5, "n":"voltage","u":"V","v":120.1}, {"n":"current","t":-5,"v":1.2}, {"n":"current","t":-4,"v":1.3}]'

This ensures encrypted communication and server identity verification.

MQTT with mTLS

Provide the client certificate and key along with the CA certificate to enable mutual authentication:

mosquitto_pub --cafile docker/ssl/certs/ca.crt --cert docker/ssl/certs/client.crt --key docker/ssl/certs/client.key -u <client_id> -P <client_secret> -t m/<workspace_id>/c/<channel_id> -h localhost -p 8883 -m '[{"bn":"some-base-name:","bt":1.276020076001e+09, "bu":"A","bver":5, "n":"voltage","u":"V","v":120.1}, {"n":"current","t":-5,"v":1.2}, {"n":"current","t":-4,"v":1.3}]'

This is the most secure mode — both client and server verify each other.

MQTT Subscription with mTLS

To subscribe to the same channel using mTLS:

mosquitto_sub \
  --cafile docker/ssl/certs/ca.crt \
  --cert docker/ssl/certs/client.crt \
  --key docker/ssl/certs/client.key \
  -h localhost -p 8883 \
  -u <client_id> -P <client_secret> \
  -t m/<workspace_id>/c/<channel_id

WebSocket with TLS

A user can send messages via WebSOcket with just the TLS support (server authentication only) and just a CA certificate using the command:

wscat -c "wss://localhost/ws/m/<workspace_id>/c/<channel_id>?authorization=<client_secret>&content-type=application/senml+json" -x '[{"bn":"some-base-name:","bt":1.276020076001e+09, "bu":"A","bver":5, "n":"voltage","u":"V","v":120.1}, {"n":"current","t":-5,"v":1.2}, {"n":"current","t":-4,"v":1.3}]' --ca docker/ssl/certs/ca.crt

WebSocket with mTLS

To enable mTLS for WebSocket connections, provide the client certificate and key along with the CA certificate to enable mutual authentication:

wscat -c "wss://localhost/ws/m/<workspace_id>/c/<channel_id>?authorization=<client_secret>&content-type=application/senml+json" -x '[{"bn":"some-base-name:","bt":1.276020076001e+09, "bu":"A","bver":5, "n":"voltage","u":"V","v":120.1}, {"n":"current","t":-5,"v":1.2}, {"n":"current","t":-4,"v":1.3}]' --ca docker/ssl/certs/ca.crt --cert docker/ssl/certs/client.crt --key docker/ssl/certs/client.key

This is the most secure mode — both client and server verify each other.

CoAP with DTLS

Magistrala currently supports CoAP over UDP. Similar to how TLS secures TCP connections, DTLS provides an additional layer of security for UDP. To send a message over CoAP with DTLS just add a CA certificate with the command:

With the coap-cli

coap-cli post m/<workspace_id>/c/<channel_id/subtopic --auth <client_secret> -d '[{"bn":"some-base-name:","bt":1.276020076001e+09, "bu":"A","bver":5, "n":"voltage","u":"V","v":120.1}, {"n":"current","t":-5,"v":1.2}, {"n":"current","t":-4,"v":1.3}]' -p 5684 -v -s -A docker/ssl/certs/ca.crt

Subtopics

In order to use subtopics and give more meaning to your pub/sub channel, you can simply add any suffix to base /m/<workspace_id>/c/<channel_id> topic.

Example subtopic publish/subscribe for bedroom temperature would be m/<workspace_id>/c/<channel_id>/bedroom/temperature.

Subtopics are generic and multilevel. You can use almost any suffix with any depth.

Topics with subtopics are propagated to Message broker in the following format channels.<channel_id>.<optional_subtopic>.

Our example topic m/<workspace_id>/c/<channel_id>/bedroom/temperature will be translated to appropriate Message Broker topic channels.<channel_id>.bedroom.temperature.

You can use multilevel subtopics, that have multiple parts. These parts are separated by . or / separators. When you use combination of these two, have in mind that behind the scene, / separator will be replaced with .. Every empty part of subtopic will be removed. What this means is that subtopic a///b is equivalent to a/b. When you want to subscribe, you can use the default Message Broker's (FluxMQ) MQTT-style wildcards + (single-level) and # (multi-level).

Authorization is done on channel level, so you only have to have access to channel in order to have access to it's subtopics.

Note: When using MQTT, it's recommended that you use standard MQTT wildcards + and #.

Message Broker

FluxMQ is Magistrala's message broker for both MQTT and general message exchange — see Architecture and Storage Architecture for the current model.

Authorization

Manage user roles, entity hierarchies and permissions.

Security

Configure secure connections for Magistrala services including gRPC, PostgreSQL and adapters using TLS certificates and SSL modes.

On this page

HTTPMQTTCoAPWebSocketBasic nodejs exampleBasic golang exampleMQTT-over-WSmTLS MessagingCertificate SetupHTTP with mTLSHTTP with TLSMQTT with TLSMQTT with mTLSMQTT Subscription with mTLSWebSocket with TLSWebSocket with mTLSCoAP with DTLSSubtopicsMessage Broker