[DIY] Sensor IoT temperatura y humedad [4]: Recibiendo datos por MQTT

Cuarta entrada y última dedicada al miniproyecto del particular cuaderno de bitácora para hacer unas pruebas de concepto y adentrarse en el mundo maker.

Próxima serie de entradas relacionadas con este proyecto irán dedicadas a mejoras. Como incorporar baterías, obtener niveles de la batería y un circuito de carga. Así como diseñar la PCB con programas como KiCad que hasta la fecha nunca he llegado ha hacer.

También es la última, porque dejamos de hacer cosas relacionada con los microcontroladores y pasamos al mundo de puro software. Y eso es otro proyecto que tendrá varias fases y otro cometidos.

Proyecto de ingestión

Para la lectura de los datos que estábamos mandando por MQTT desde el ESP8266 he optado por crear un proceso en Java usando el framework Spring Boot que facilita el trabajo.

Creo el esqueleto del proyecto desde la herramienta online de Spring Initializr. Y así no empezamos de 0.

Agregamos las dependencias en el POM.xml de MQTT y Lombok que nos facilita la creación de constructores de las clases y los métodos get y set.

<dependencies>    
    <dependency>
	<groupId>org.springframework.integration</groupId>
	<artifactId>spring-integration-mqtt</artifactId>
	<version>6.5.1</version>
    </dependency>

    <!-- https://mvnrepository.com/artifact/org.eclipse.paho/org.eclipse.paho.client.mqttv3 -->
    <dependency>
        <groupId>org.eclipse.paho</groupId>
        <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
        <version>1.2.5</version>
    </dependency>

    <!-- Lombok -->
    <dependency>
	<groupId>org.projectlombok</groupId>
	<artifactId>lombok</artifactId>
	<optional>true</optional>
    </dependency>
</dependencies>
...
<plugins>
    <plugin>
	<groupId>org.apache.maven.plugins</groupId>
	<artifactId>maven-compiler-plugin</artifactId>
		<configuration>
			<annotationProcessorPaths>
				<path>
					<groupId>org.projectlombok</groupId>
					<artifactId>lombok</artifactId>
				</path>
			</annotationProcessorPaths>
		</configuration>
	</plugin>
	<plugin>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-maven-plugin</artifactId>
		<configuration>
			<excludes>
				<exclude>
					<groupId>org.projectlombok</groupId>
					<artifactId>lombok</artifactId>
				</exclude>
			</excludes>
		</configuration>
	</plugin>
</plugins>

Y agregamos las propiedades correspondientes al application.properties para configurar la conexión con el servidor MQTT.

# MQTT
mqtt.config.host=tcp://<ip>:1883
mqtt.config.username=<username>
mqtt.config.password=<password>
mqtt.config.connection-timeout=5000

mqtt.options.automatic-reconnect=true
mqtt.options.clean-session=true

mqtt.bindings.topics=spuertes/#
mqtt.bindings.divider=|

Configuración en Java

Una vez tenemos las configuraciones en el pom.xml y el properties listo pasamos a la configuración en código para crear los recursos necesario de nuestro procesador de mensajes MQTT.

Primero creamos un paquete config donde albergaremos el fichero que llamaremos MqttConfig. Aquí dentro crearemos los beans de los canales y el MqttProducer.

Para que Spring entienda que es un fichero de configuración lo anotamos con @Configuration

@Configuration
public class MqttConfig {

    private static final String CLIENT_ID = "ingest-process";

    @Value("${mqtt.config.host}")
    private String host;
    @Value("${mqtt.config.username}")
    private String user;
    @Value("${mqtt.config.password}")
    private String password;
    @Value("${mqtt.config.connection-timeout:5000}")
    private int timeout;
    @Value("${mqtt.bindings.topics}")
    private String topicsList;
    @Value("${mqtt.bindings.divider:,}")
    private String divider;

    private String[] topics;

    @PostConstruct
    private void postConstruct() {
        this.topics = topicsList != null ? topicsList.split(divider) : new String[]{};
    }

    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();

        options.setServerURIs(new String[]{host});
        options.setUserName(user);
        options.setPassword(password.toCharArray());
        options.setCleanSession(true);

        factory.setConnectionOptions(options);

        return factory;
    }

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageProducer inbound() {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(CLIENT_ID, mqttClientFactory(), topics);
        adapter.setCompletionTimeout(timeout);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(2);
        adapter.setOutputChannel(mqttInputChannel());

        return adapter;
    }
}

Por último, creamos otro componente que llamaremos MqttProcessor, que marcamos con @Component, así Spring entiende que tiene que gestionarlo.

Esta clase contendrá el handler. Siendo el handler el punto de entrada de los mensajes a nuestro sistema y desde donde obtendremos los datos para almacenarlos, evaluarlos, tratarlos y hacer con ellos lo que queramos.

@Component
@AllArgsConstructor
public class MqttProcessor {

    private static final Logger LOGGER = LoggerFactory.getLogger(MqttConfig.class);

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                LOGGER.info("Received message: {}", message.getHeaders());
                LOGGER.info("Received message: {}", message.getPayload());
            }
        };
    }
}

Hasta aquí el prototipo de este miniproyecto que inaugura la serie de bitácora.