MQTT

Lightweight messaging protocol ideal for IoT use cases

Overview

This page provides an overview of how Joule utilises the MQTT protocol to subscribe to events, specifically in IoT use cases.

MQTT is a lightweight, publish-subscribe network protocol used for message queuing services, making it ideal for machine-to-machine communication in IoT.

Joule enables the subscription to events through an MQTT source consumer, which allows for the consumption of events from a specified MQTT topic.

Client library: org.eclipse.paho:org.eclipse.paho.mqttv5.client:1.2.5

Examples & DSL attributes

This example illustrates how to configure an MQTT consumer to subscribe to events from the customers topic.

The consumer connects to the MQTT broker at tcp://127.0.0.1:1883 and consumes events with a Quality of Service (QoS) level of 1.

Client settings The consumer is identified by the clientId (jouleCustomerConsumer), with authentication using the username joule and the tenant uk.
Event deserialisation The events are consumed as customer objects and then transformed into Joule StreamEvent objects using the CustomerToStreamEventTranslator class.
Additional settings The configuration also includes settings for compressed (set to false) and batch (set to f
alse), which affect how events are processed and transformed.

mqttConsumer:
  broker: tcp://127.0.0.1:1883
  topic: customers

  clientId: jouleCustomerConsumer
  username: joule
  password: joule
  tenant: uk
  qos: 1

  deserializer:
    parser: com.fractalworks.streams.transport.mqtt.CustomerToStreamEventTranslator
    compressed: false
    batch: false

Use the Joule SDK to build a custom parser and deploy the resulting jar file to the userlibs directory. See Joule SDK for further information.

import com.fasterxml.jackson.annotation.JsonRootName;
import com.fractalworks.streams.core.data.streams.StreamEvent;
import com.fractalworks.streams.core.exceptions.TranslationException;
import com.fractalworks.streams.sdk.codec.StreamEventParser;
import com.fractalworks.streams.transport.Customer;

import java.util.Collection;
import java.util.Collections;

@JsonRootName(value = "customer parser")
public class CustomerToStreamEventTranslator implements StreamEventParser {

    public CustomerToStreamEventTranslator() {
        // REQUIRED
    }

    @Override
    public Collection<StreamEvent> translate(Object o) throws TranslationException {
        Collection<StreamEvent> events = null;
        if(o instanceof Customer customer){
            StreamEvent event = new StreamEvent("customer");
            event.addValue("id", customer.getId());
            event.addValue("firstname", customer.getFirstname());
            event.addValue("surname", customer.getSurname());
            event.addValue("age", customer.getAge());

            events = Collections.singletonList( event);
        }
        return events;
    }
}

Create a local configuration as mosquitto.conf within the /etc/mosquitto/conf.d/ directory.

per_listener_settings true
pid_file /run/mosquitto/mosquitto.pid

persistence true
persistence_location /var/lib/mosquitto/

log_dest file /var/log/mosquitto/mosquitto.log

listener 1883
allow_anonymous true
password_file /etc/mosquitto/passwd

include_dir /etc/mosquitto/conf.d

Attributes schema

Deserialisation attributes schema

This topic provides configuration parameters available object deserialisation process.

Attribute

Description

Data Type

Required

parser

User provided implementation

Implementation of StreamEventParser

compressed

Uncompress payload using Snappy

Boolean

Default: false

batch

Flag to inform process multiple messages have been sent in payload

Boolean

Default: false

Deserialiser example

In the given MQTT consumer configuration, the deserializer section defines how the incoming events should be processed and transformed before being consumed by the Joule platform.

parser The parser attribute specifies the class (com.fractalworks.streams.transport.mqtt.CustomerToStreamEventTranslator) responsible for converting the incoming message into a Joule StreamEvent object. This transformation is necessary to make the data compatible with the platform’s processing system.
compressed The compressed: false setting indicates that the incoming data is not compressed. If the data were compressed, this setting would need to be set to true and the system would decompress the data before processing.
batch The batch: false setting means that the events will not be grouped into batches for processing. Each event is processed individually as it is received. If set to true, it would allow the consumer to handle multiple events in a single batch, which can improve efficiency in some scenarios.

mqttConsumer:
  ...
  deserializer:
    parser: com.fractalworks.streams.transport.mqtt.CustomerToStreamEventTranslator
    compressed: false
    batch: false

import com.fasterxml.jackson.annotation.JsonRootName;
import com.fractalworks.streams.core.data.streams.StreamEvent;
import com.fractalworks.streams.core.exceptions.TranslationException;
import com.fractalworks.streams.sdk.codec.StreamEventParser;
import com.fractalworks.streams.transport.Customer;

import java.util.Collection;
import java.util.Collections;

@JsonRootName(value = "customer parser")
public class CustomerToStreamEventTranslator implements StreamEventParser {

    public CustomerToStreamEventTranslator() {
        // REQUIRED
    }

    @Override
    public Collection<StreamEvent> translate(Object o) throws TranslationException {
        Collection<StreamEvent> events = null;
        if(o instanceof Customer customer){
            StreamEvent event = new StreamEvent("customer");
            event.addValue("id", customer.getId());
            event.addValue("firstname", customer.getFirstname());
            event.addValue("surname", customer.getSurname());
            event.addValue("age", customer.getAge());

            events = Collections.singletonList( event);
        }
        return events;
    }
}

Additional resources

PreviousFurther RabbitMQ configurations NextTopic wildcards

Last updated 1 year ago

hashtagOverview

hashtagExamples & DSL attributes

hashtagAttributes schema

hashtagDeserialisation attributes schema

hashtagDeserialiser example

hashtagAdditional resources