Será necesario cumplir los siguientes requisitos para establecer una conexión con la plataforma, de lo contrario se rechazará la conexión.

• Dar de alta la IP desde la que se solicita la conexión (Consumidor/Publicador)
• Obtener las credenciales MQTT (Usuario y contraseña)
• Obtener un ID de cliente

Para cumplir con estos requisitos será necesario ponerse en contacto con la plataforma DGT3.0 a través de [email protected].

Para realizar la conexión al bróker MQTT es posible utilizar uno de los muchos lenguajes que disponen de librerías específicas para ello e incluso directamente con librerías como Mosquito MQTT.

En este caso, vamos a ver un ejemplo hecho con Java y SpringBoot.

Usaremos la librería paho de eclipse para la conexión, para ello la incluiremos en un proyecto de Maven, agregando la siguiente dependencia:

<dependency>
     <groupId>org.eclipse.paho</groupId>
     <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
     <version>1.2.5</version>
</dependency>

Al usar la librería, lo primero que debemos hacer para recibir mensajes de un bróker MQTT es obtener una implementación de la interfaz IMqttClient. Esta interfaz contiene todos los métodos requeridos por una aplicación para establecer una conexión con el servidor y recibir mensajes.

La librería cuenta con dos implementaciones de esta interfaz, una asíncrona (MqttAsyncClient) y una síncrona (MqttClient). En nuestro caso, nos centraremos en la versión síncrona.

La configuración en sí es un proceso de dos pasos: primero creamos una instancia de la clase MqttClient y la conectamos al servidor, luego recibimos y manejamos los mensajes a través del canal MQTT.

En primer lugar, crearemos un Bean con la configuración del broker MQTT. Para este ejemplo lo creamos en una clase de configuración.

Para conectar nuestra instancia haremos uso del método connect(), pasando opcionalmente una instancia MqttConnectOptions que nos permite personalizar algunos aspectos del protocolo. En particular, podemos usar esas opciones para añadir información adicional como credenciales de seguridad, modo de recuperación de sesión, modo de reconexión, etc. La clase MqttConnectionOptions expone esas opciones como atributos que podemos establecer usando los métodos proporcionados por la propia librería.

Formato URL: protocolo://url:puerto

ej: ssl://preproduction.cmobility30.es:8883

@Bean
    public void enableMqtt() throws MqttException {
        try (MqttClient client = new MqttClient(
            URL,  //URL de la plataforma en el formato definido previamente (String)
            "client_read_XX",  
            new MemoryPersistence())) {

            MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
            mqttConnectOptions.setUserName(userDGT);   //Usuario proporcionado por la plataforma DGT3.0 (String)
            mqttConnectOptions.setPassword(passwordDGT.toCharArray());  //Contraseña proporcionada por la plataforma DGT3.0 (String)
            mqttConnectOptions.setCleanSession(false);   //Mantener el estado de la conexión
            mqttConnectOptions.setAutomaticReconnect(true);   //Reconectar en caso de fallo de la red

            client.connect(mqttConnectOptions);
            client.setCallback(listenerMq);   //Instancia de la clase que implementa "MqttCallback", detallado a continuación
            client.subscribe(topic);   //Tópico del que se recibirán mensajes

        } catch (MqttException e) {
            LOGGER.error("MQTTException: "+ e.getMessage());
        }
    }

Será necesario crear una clase que implemente la clase "MqttCallback", nos servirá para recibir los mensajes por el canal MQTT previamente configurados. Este listener tendrá un aspecto parecido al siguiente:

import org.eclipse.paho.client.mqttv3.MqttCallback;

@Component
public class ListenerMq implements MqttCallback {

    @Override
    public void connectionLost(Throwable cause) {
        LOGGER.error("Connection lost with mqtt: " + cause);
        //Añadir lógica en el caso de perder la conexión
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws InterruptedException {
        LOGGER.info(topic + ": " + message.toString());
        //Añadir lógica en el caso de recibir un mensaje del tópico
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        //Añadir lógica
    }
}

En este caso vamos a mostrar un ejemplo de publicador hacía la plataforma DGT3.0 por el interfaz MQTT. Para ello vamos a usar también Java sobre el framework de springboot.

En primer lugar, vamos a configurar la conexión con el broker. Para ello utilizaremos una función similar a la siguiente, en una clase de configuración:

    private final String brokerUrl = "ssl://preproduction.cmobility30.es:8883"; 
    @Bean
    public MqttAsyncClient mqttClient() throws MqttException {
        MqttAsyncClient client = new MqttAsyncClient(brokerUrl, MqttClient.generateClientId());
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        options.setUserName(userDGT);  //Usuario proporcionado por la plataforma DGT3.0 (String)
        options.setPassword(passwordDGT.toCharArray());  //Contraseña proporcionada por la plataforma DGT3.0 (String)

        client.connect(options, null, new IMqttActionListener() {
            @Override
            public void onSuccess(IMqttToken asyncActionToken) {
                System.out.println("Connected to broker: " + brokerUrl);
            }
    
            @Override
            public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                System.err.println("Failed to connect to broker: " + exception.getMessage());
            }
        });
        return client;
    }

Una vez iniciado el microservicio, si todo ha ido correctamente debería visualizarse el siguiente log:

Connected to broker: ssl://preproduction.cmobility30.es:8883

Para este ejemplo hemos expuesto una clase servicio, que inyecta la clase MqttAsyncClient y es capaz de publicar mensajes en un tópico con el método publish.

@Service
public class MqttPublisherService {

    @Autowired
    private MqttAsyncClient mqttClient;

    public void publish(String topic, String payload) {
        try {
            mqttClient.publish(topic, new MqttMessage(payload.getBytes()));
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
}

Lo siguiente es un ejemplo de controlador en el que usamos lo anteriormente expuesto para publicar un mensaje:

@RestController
public class MqttController {

    @Autowired
    private MqttPublisherService mqttPublisherService;

    @PostMapping("/publish")
    public void publish() {
        mqttPublisherService.publish(topic, message);
    }
}

Con el fin de poder probar la conexión con la plataforma DGT3.0 de manera rápida y sencilla, es posible conectarse usando la herramienta MQTT Explorer.

• La configuración para la conexión debe ser similar a la proporcionada en la siguiente imagen, pero usando las credenciales previamente proporcionadas por la plataforma.

  

• A continuación, en la pestaña "Advanced" seremos capaces de configurar los tópicos de lectura, así como el ID de cliente correspondiente.

  

• Una vez nos hemos conectado, seremos capaces de visualizar los mensajes que se están publicando en los tópicos, de la misma forma que se muestra en la siguiente imagen.

  

Con estos ejemplos observamos una posible implementación para la conexión tanto de un cliente como de un servidor a un bróker MQTT, así como un ejemplo de conexión con la herramienta MQTT explorer para pruebas.

Si se desea ampliar la información, se puede encontrar aquí.