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 Kotlin.

2. What you will need

To complete this guide, you will need the following:

Some time on your hands
A decent text editor or IDE (e.g. IntelliJ IDEA)
JDK 21 or greater installed with JAVA_HOME configured appropriately

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.

Download and unzip the source

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,graalvm,awaitility,kapt \
    --build=gradle --lang=kotlin

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, mqtt, graalvm, awaitility, and kapt 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/kotlin/example/micronaut/TemperatureClient.kt

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)
    fun publishLivingroomTemperature(data: ByteArray)
}

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/kotlin/example/micronaut/Scale.kt

package example.micronaut

import java.util.Collections
import java.util.Optional
import java.util.concurrent.ConcurrentHashMap

enum class Scale(val scaleName: String) {
    FAHRENHEIT("Fahrenheit"),
    CELSIUS("Celsius");

    override fun toString(): String = scaleName

    companion object {
        private var ENUM_MAP: Map<String, Scale>

        init {
            val map: MutableMap<String, Scale> = ConcurrentHashMap()
            for (instance in values()) {
                map[instance.scaleName] = instance
            }
            ENUM_MAP = Collections.unmodifiableMap(map)
        }

        fun of(name: String): Optional<Scale> = Optional.ofNullable(ENUM_MAP[name])

        fun candidates(): Set<String> = ENUM_MAP.keys
    }
}

Create a class to show completion candidates:

cli/src/main/kotlin/example/micronaut/TemperatureScaleCandidates.kt

package example.micronaut

import java.util.ArrayList

class TemperatureScaleCandidates : ArrayList<String>(Scale.candidates())

Replace the command:

cli/src/main/kotlin/example/micronaut/MicronautguideCommand.kt

package example.micronaut

import example.micronaut.Scale.CELSIUS
import example.micronaut.Scale.FAHRENHEIT
import io.micronaut.configuration.picocli.PicocliRunner
import jakarta.inject.Inject
import picocli.CommandLine.Command
import picocli.CommandLine.Option
import java.math.BigDecimal
import java.math.RoundingMode

@Command(name = "house-livingroom-temperature", (1)
         description = ["Publish living room temperature"],
         mixinStandardHelpOptions = true)
class MicronautguideCommand : Runnable {

    @Option(names = ["-t", "--temperature"], (2)
            required = true, (3)
            description = ["Temperature value"])
    var temperature: BigDecimal? = null

    @Option(names = ["-s", "--scale"], (2)
            required = false, (3)
            description = ["Temperate scales \${COMPLETION-CANDIDATES}"], (4)
            completionCandidates = TemperatureScaleCandidates::class) (4)
    var scale: String? = null

    @Inject
    lateinit var temperatureClient: TemperatureClient (5)

    override fun run() {
        val temperatureScale = if (scale != null) Scale.of(scale!!).orElse(CELSIUS) else CELSIUS
        val celsius = if (temperatureScale === FAHRENHEIT) fahrenheitToCelsius(temperature!!) else temperature!!
        val data = celsius.toPlainString().toByteArray()
        temperatureClient.publishLivingroomTemperature(data) (6)
        println("Topic published")
    }

    companion object {

        @JvmStatic
        fun main(args: Array<String>) {
            PicocliRunner.run(MicronautguideCommand::class.java, *args)
        }

        private fun fahrenheitToCelsius(temperature: BigDecimal): BigDecimal {
            return temperature
                    .subtract(BigDecimal.valueOf(32))
                    .multiply(BigDecimal.valueOf(5 / 9.0))
                    .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/kotlin/example/micronaut/MicronautguideCommandTest.kt

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 org.awaitility.Awaitility
import org.junit.jupiter.api.Assertions
import org.junit.jupiter.api.Test
import java.io.ByteArrayOutputStream
import java.io.OutputStream
import java.io.PrintStream
import java.math.BigDecimal
import java.nio.charset.StandardCharsets
import java.util.concurrent.TimeUnit

internal class MicronautguideCommandTest {

    @Test
    fun testWithCommandLineOption() {
        val baos: OutputStream = ByteArrayOutputStream()
        System.setOut(PrintStream(baos))
        ApplicationContext.run(
            mapOf("spec.name" to "MicronautguideCommandTest"),
            Environment.CLI, Environment.TEST
        ).use { ctx ->
            val listener: TemperatureListener = ctx.getBean(TemperatureListener::class.java)
            val args = arrayOf("-t", "212", "-s", "Fahrenheit")
            PicocliRunner.run(MicronautguideCommand::class.java, ctx, *args)
            Assertions.assertTrue(baos.toString().contains("Topic published"))

            Awaitility.await().atMost(5, TimeUnit.SECONDS)
                .untilAsserted {
                    Assertions.assertEquals(
                        BigDecimal("100.00"),
                        listener.temperature
                    )
                }
        }
    }

    @Requires(property = "spec.name", value = "MicronautguideCommandTest")
    @MqttSubscriber (1)
    class TemperatureListener {
        var temperature: BigDecimal? = null

        @Topic("house/livingroom/temperature") (2)
        fun receive(data: ByteArray?) {
            temperature = BigDecimal(String(data!!, StandardCharsets.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,graalvm,awaitility \
    --build=gradle --lang=kotlin

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, mqtt, graalvm, and awaitility features.

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/kotlin/example/micronaut/TemperatureListener.kt

package example.micronaut

import io.micronaut.mqtt.annotation.MqttSubscriber
import io.micronaut.mqtt.annotation.Topic
import org.slf4j.LoggerFactory
import java.math.BigDecimal
import java.nio.charset.StandardCharsets.UTF_8
import io.micronaut.core.annotation.Nullable

@MqttSubscriber (1)
class TemperatureListener {

    var temperature: BigDecimal? = null

    @Topic("house/livingroom/temperature") (2)
    fun receive(data: ByteArray) {
        temperature = BigDecimal(String(data, UTF_8))
        LOG.info("temperature: {}", temperature)
    }

    companion object {
        private val LOG = LoggerFactory.getLogger(TemperatureListener::class.java)
    }
}

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/kotlin/example/micronaut/SubscriptionTest.kt

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.junit5.annotation.MicronautTest
import jakarta.inject.Inject
import org.awaitility.Awaitility.await
import org.junit.jupiter.api.Assertions
import org.junit.jupiter.api.Test
import java.math.BigDecimal
import java.nio.charset.StandardCharsets
import java.util.concurrent.TimeUnit

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

    @Inject
    lateinit var client: TemperatureClient

    @Inject
    lateinit var listener: TemperatureListener

    @Test
    fun checkSubscriptionsAreReceived() {
        client.publishLivingroomTemperature("3.145".toByteArray(StandardCharsets.UTF_8))
        await().atMost(5, TimeUnit.SECONDS)
            .untilAsserted {
                Assertions.assertEquals(
                    BigDecimal("3.145"),
                    listener.temperature
                )
            }
    }

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

        @Topic("house/livingroom/temperature") (5)
        fun publishLivingroomTemperature(data: ByteArray?)
    }
}

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. Generate a Micronaut Application Native Executable with GraalVM

We will use GraalVM, an advanced JDK with ahead-of-time Native Image compilation, to generate a native executable of this Micronaut application.

Compiling Micronaut applications ahead of time with GraalVM significantly improves startup time and reduces the memory footprint of JVM-based applications.

Only Java and Kotlin projects support using GraalVM’s native-image tool. Groovy relies heavily on reflection, which is only partially supported by GraalVM.

8.1. GraalVM Installation

The easiest way to install GraalVM on Linux or Mac is to use SDKMan.io.

Java 21

sdk install java 21.0.5-graal

For installation on Windows, or for a manual installation on Linux or Mac, see the GraalVM Getting Started documentation.

The previous command installs Oracle GraalVM, which is free to use in production and free to redistribute, at no cost, under the GraalVM Free Terms and Conditions.

Alternatively, you can use the GraalVM Community Edition:

Java 21

sdk install java 21.0.2-graalce

8.2. Native Executable Generation

To generate a native executable using Gradle, run:

./gradlew nativeCompile

The native executable is created in build/native/nativeCompile directory and can be run with build/native/nativeCompile/micronautguide.

It is possible to customize the name of the native executable or pass additional parameters to GraalVM:

build.gradle

graalvmNative {
    binaries {
        main {
            imageName.set('mn-graalvm-application') (1)
            buildArgs.add('-Ob') (2)
        }
    }
}

1	The native executable name will now be `mn-graalvm-application`
2	It is possible to pass extra build arguments to `native-image`. For example, `-Ob` enables the quick build mode.

Generate GraalVM native executables for both the CLI and the messaging application, then execute both. Publish a temperature, and you will see it in the subscriber logs.

9. Next Steps

10. Help with the Micronaut Framework

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

11. 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…).

Table of Contents