Table of Contents

Publishing and subscribing to MQTT Topics from a Micronaut Application

Learn how to use Mosquitto as an MQTT broker, create a Micronaut CLI application and publish an MQTT topic, and subscribe to the MQTT topic in a different Micronaut Messaging application.

Authors: Sergio del Amo

Micronaut Version: 4.6.3

1. Getting Started

In this guide, we will create a Micronaut application written in Groovy.

2. What you will need

To complete this guide, you will need the following:

3. Solution

We recommend that you follow the instructions in the next sections and create the application step by step. However, you can go right to the completed example.

4. Test Resources

For this guide, we will use Mosquitto via Test Resources. As described in the MQTT section of the Test Resources documentation, configure a mosquitto container:

This should be done in both apps for this guide.
src/main/resources/application.yml
test-resources:
  containers:
    mosquitto:
      image-name: eclipse-mosquitto
      hostnames:
        - mqtt.host
      exposed-ports:
        - mqtt.port: 1883
      ro-fs-bind:
        - "src/test-resources/mosquitto.conf": /mosquitto/config/mosquitto.conf

And then define the mosquitto configuration file:

src/test-resources/mosquitto.conf
persistence false
allow_anonymous true
connection_messages true
log_type all
listener 1883

As we have defined that Test Resources are shared in the build, both applications will make use of the same instance of Mosquitto.

When running under production, you should replace this property with the location of your production message broker via an environment variable.

MQTT_CLIENT_SERVER_URI=tcp://production-server:1183

5. Writing the CLI (Command Line Interface) Application

Create an application using the Micronaut Command Line Interface or with Micronaut Launch.

mn create-cli-app example.micronaut.micronautguide \
    --features=yaml,mqtt \
    --build=gradle --lang=groovy
If you don’t specify the --build argument, Gradle with the Kotlin DSL is used as the build tool.
If you don’t specify the --lang argument, Java is used as the language.
If you don’t specify the --test argument, JUnit is used for Java and Kotlin, and Spock is used for Groovy.

The previous command creates a Micronaut application with the default package example.micronaut in a directory named micronautguide.

Rename this micronautguide directory to cli.

If you use Micronaut Launch, select Micronaut Application as application type and add yaml, and mqtt features.

If you have an existing Micronaut application and want to add the functionality described here, you can view the dependency and configuration changes from the specified features, and apply those changes to your application.

Diff

Setup IntelliJ IDEA to develop Micronaut Applications.

5.1. Create an MqttPublisher

Create an interface to publish MQTT Topics:

cli/src/main/groovy/example/micronaut/TemperatureClient.groovy
package example.micronaut

import io.micronaut.mqtt.annotation.Topic
import io.micronaut.mqtt.annotation.v5.MqttPublisher

@MqttPublisher (1)
interface TemperatureClient {

    @Topic('house/livingroom/temperature') (2)
    void publishLivingroomTemperature(byte[] data)
}
1 Micronaut MQTT implements the interface automatically because it is annotated with @MqttPublisher.
2 To set the topic to publish to, apply the @Topic annotation to the method or an argument of the method.

5.2. Writing the CLI Command

Create an enum to allow the user to submit temperatures in Celsius or Fahrenheit:

cli/src/main/groovy/example/micronaut/Scale.groovy
package example.micronaut

import groovy.transform.CompileStatic
import io.micronaut.core.annotation.NonNull

import java.util.concurrent.ConcurrentHashMap

@CompileStatic
enum Scale {
    FAHRENHEIT('Fahrenheit'),
    CELSIUS('Celsius')

    private static final Map<String,Scale> ENUM_MAP

    final String name

    Scale(String name) {
        this.name = name
    }

    static {
        Map<String,Scale> map = new ConcurrentHashMap<>()
        for (Scale instance : Scale.values()) {
            map[instance.name] = instance
        }
        ENUM_MAP = Collections.unmodifiableMap(map)
    }

    @NonNull
    static Optional<Scale> of(@NonNull String name) {
        return Optional.ofNullable(ENUM_MAP.get(name))
    }

    @Override
    String toString() {
        name
    }

    static Set<String> candidates() {
        ENUM_MAP.keySet()
    }
}

Create a class to show completion candidates:

cli/src/main/groovy/example/micronaut/TemperatureScaleCandidates.groovy
package example.micronaut

import groovy.transform.CompileStatic

@CompileStatic
class TemperatureScaleCandidates extends ArrayList<String> {

    TemperatureScaleCandidates() {
        super(Scale.candidates());
    }
}

Replace the command:

cli/src/main/groovy/example/micronaut/MicronautguideCommand.groovy
package example.micronaut

import groovy.transform.CompileStatic
import io.micronaut.configuration.picocli.PicocliRunner
import jakarta.inject.Inject
import picocli.CommandLine.Command
import picocli.CommandLine.Option

import java.math.RoundingMode

import static example.micronaut.Scale.CELSIUS
import static example.micronaut.Scale.FAHRENHEIT

@CompileStatic
@Command(name = 'house-livingroom-temperature', (1)
         description = 'Publish living room temperature',
         mixinStandardHelpOptions = true)
class MicronautguideCommand implements Runnable {

    @Option(names = ['-t', '--temperature'], (2)
            required = true, (3)
            description = 'Temperature value')
    BigDecimal temperature

    @Option(names = ['-s', '--scale'], (2)
            required = false, (3)
            description = 'Temperate scales ${COMPLETION-CANDIDATES}', (4)
            completionCandidates = TemperatureScaleCandidates) (4)
    String scale

    @Inject
    TemperatureClient temperatureClient (5)

    static void main(String[] args) {
        PicocliRunner.run(MicronautguideCommand, args)
    }

    void run() {
        Scale temperatureScale = scale != null ?
                Scale.of(scale).orElse(CELSIUS) : CELSIUS
        BigDecimal celsius = (temperatureScale == FAHRENHEIT)
                ? fahrenheitToCelsius(temperature) : temperature
        byte[] data = celsius.toPlainString().bytes
        temperatureClient.publishLivingroomTemperature(data) (6)
        println('Topic published')
    }

    private static BigDecimal fahrenheitToCelsius(BigDecimal temperature) {
        return temperature
                .subtract(BigDecimal.valueOf(32))
                .multiply(BigDecimal.valueOf(5/9.0 as double))
                .setScale(2, RoundingMode.FLOOR)
    }
}
1 The picocli @Command annotation designates this class as a command.
2 Picocli @Option must have one or more names.
3 Options can be marked required to make it mandatory for the user to specify them on the command line.
4 It is possible to embed the completion candidates in the description for an option by specifying the variable ${COMPLETION-CANDIDATES} in the description text.
5 Field injection
6 Publish the MQTT Topic

Replace the generated test with this:

cli/src/test/groovy/example/micronaut/MicronautguideCommandSpec.groovy
package example.micronaut

import io.micronaut.configuration.picocli.PicocliRunner
import io.micronaut.context.ApplicationContext
import io.micronaut.context.annotation.Requires
import io.micronaut.context.env.Environment
import io.micronaut.mqtt.annotation.MqttSubscriber
import io.micronaut.mqtt.annotation.Topic
import spock.lang.Specification
import spock.util.concurrent.PollingConditions

import static java.nio.charset.StandardCharsets.UTF_8

class MicronautguideCommandSpec extends Specification {

    PollingConditions pollingConditions = new PollingConditions()

    def "test with command line option"() {
        when:
        OutputStream baos = new ByteArrayOutputStream()
        System.setOut(new PrintStream(baos))

        ApplicationContext ctx = ApplicationContext.run(
                ['spec.name': 'MicronautguideCommandSpec'],
                Environment.CLI, Environment.TEST)

        String[] args = ['-t', '212', '-s', 'Fahrenheit']
        PicocliRunner.run(MicronautguideCommand, ctx, args)

        TemperatureListener listener = ctx.getBean(TemperatureListener.class);

        then:
        baos.toString().contains('Topic published')

        and:
        pollingConditions.within(5) {
            listener.temperature == 100.0
        }

        cleanup:
        ctx.close()
    }

    @Requires(property = "spec.name", value = "MicronautguideCommandSpec")
    @MqttSubscriber (1)
    static class TemperatureListener {

        private BigDecimal temperature

        @Topic("house/livingroom/temperature") (2)
        void receive(byte[] data) {
            temperature = new BigDecimal(new String(data, UTF_8))
        }
    }
}
1 Because it is annotated with @MqttSubscriber, Micronaut MQTT listens for messages.
2 To set the topic to listen to, apply the @Topic annotation to the method or an argument of the method.

The MQTT server URI is configured by referencing the properties when we used when we set up Mosquitto via Test Resources:

Since Micronaut Framework 4.0, to use YAML configuration, you have to add the YAML dependency.
cli/src/main/resources/application.yml
mqtt:
  client:
    server-uri: tcp://${mqtt.host}:${mqtt.port}
    client-id: ${random.uuid}

6. Writing an MQTT subscriber application

Create an application using the Micronaut Command Line Interface or with Micronaut Launch.

mn create-messaging-app example.micronaut.micronautguide \
    --features=yaml,mqtt \
    --build=gradle --lang=groovy
If you don’t specify the --build argument, Gradle with the Kotlin DSL is used as the build tool.
If you don’t specify the --lang argument, Java is used as the language.
If you don’t specify the --test argument, JUnit is used for Java and Kotlin, and Spock is used for Groovy.

The previous command creates a Micronaut application with the default package example.micronaut in a directory named micronautguide.

Rename this micronautguide directory to app.

If you use Micronaut Launch, select Micronaut Application as application type and add yaml, and mqtt features.

If you have an existing Micronaut application and want to add the functionality described here, you can view the dependency and configuration changes from the specified features, and apply those changes to your application.

Diff

Setup IntelliJ IDEA to develop Micronaut Applications.

6.1. Configuration

The MQTT server URI is configured by referencing the properties when we used when we set up Mosquitto via Test Resources:

app/src/main/resources/application.yml
mqtt:
  client:
    server-uri: tcp://${mqtt.host}:${mqtt.port}
    client-id: ${random.uuid}

6.2. Create Subscriber

app/src/main/groovy/example/micronaut/TemperatureListener.groovy
package example.micronaut

import io.micronaut.core.annotation.Nullable
import groovy.transform.CompileStatic
import groovy.util.logging.Slf4j
import io.micronaut.mqtt.annotation.MqttSubscriber
import io.micronaut.mqtt.annotation.Topic

import static java.nio.charset.StandardCharsets.UTF_8

@Slf4j
@CompileStatic
@MqttSubscriber (1)
class TemperatureListener {

    @Nullable
    BigDecimal temperature = null;

    @Topic('house/livingroom/temperature') (2)
    void receive(byte[] data) {
        temperature = new BigDecimal(new String(data, UTF_8))
        log.info('temperature: {}', temperature)
    }
}
1 Because it is annotated with @MqttSubscriber, Micronaut MQTT listens for messages.
2 To set the topic to publish to, apply the @Topic annotation to the method or an argument of the method.

6.3. Add test

app/src/test/groovy/example/micronaut/SubscriptionSpec.groovy
package example.micronaut

import io.micronaut.context.annotation.Property
import io.micronaut.context.annotation.Requires
import io.micronaut.mqtt.annotation.Topic
import io.micronaut.mqtt.annotation.v5.MqttPublisher
import io.micronaut.test.extensions.spock.annotation.MicronautTest
import jakarta.inject.Inject
import spock.util.concurrent.PollingConditions

import java.nio.charset.StandardCharsets
import spock.lang.Specification

@MicronautTest (1)
@Property(name = "spec.name", value = "SubscriptionTest") (2)
class SubscriptionSpec extends Specification {

    @Inject
    TemperatureClient client

    @Inject
    TemperatureListener listener

    PollingConditions conditions = new PollingConditions()

    def "subscriptions are received"() {
        when:
        client.publishLivingroomTemperature("3.145".getBytes(StandardCharsets.UTF_8))

        then:
        conditions.within(5) {
            listener.temperature == 3.145
        }
    }

    @Requires(property = "spec.name", value = "SubscriptionTest") (3)
    @MqttPublisher (4)
    static interface TemperatureClient {

        @Topic("house/livingroom/temperature") (5)
        void publishLivingroomTemperature(byte[] data)
    }
}
1 Annotate the class with @MicronautTest so the Micronaut framework will initialize the application context and the embedded server. More info.
2 Annotate the class with @Property to supply configuration to the test.
3 This bean loads only if the specified property is defined.
4 Micronaut MQTT implements the interface automatically because it is annotated with @MqttPublisher.
5 To set the topic to publish to, apply the @Topic annotation to the method or an argument of the method.

7. Running the Application

7.1. Run the Subscriber App

To run the application, use the ./gradlew run command, which starts the application on port 8080.

Keep it running. Once you publish a topic with the CLI application, you will see a log entry.

7.2. Run the CLI

Run the CLI command, which will publish a temperature at startup.

./gradlew run --args="-t 212 -s Fahrenheit"

The subscriber receives the MQTT topics, as you will see in the logs:

12:09:47.280 [MQTT Call: 180d98b5-75b9-41be-a874-295289346592]
    INFO  e.micronaut.TemperatureListener - temperature: 100.00

8. Next steps

Read more about Micronaut MQTT.

9. Help with the Micronaut Framework

The Micronaut Foundation sponsored the creation of this Guide. A variety of consulting and support services are available.

10. License

All guides are released with an Apache license 2.0 license for the code and a Creative Commons Attribution 4.0 license for the writing and media (images…​).