How can I garantee that all incoming mqtt messages are properly managed even if some of them...

I'm working with an Angular 19 application which will monitor a water tank meter built with an ESP32 (it's a microcontroller for those who are wondering). I'm implementing MQTT protocol for communication between ESP32 and front.

I'm testing with MQTTX and I found out if I send a malformed message (is validated with a custom function called isValidateWaterTank), the first time it properly throws an error, but after, the incoming messages get stuck, even before being validated. The only way to get rid of it is to restart app or the mqtt connection.

My code is as follows:

1. First method initializes a new mqtt subscriber and second is called from external class. Here it shows the message correctly, so I don't think the problem is because the connection with the broker is lost.

private begin(): void {
console.log('Subscriber / begin() -> INICIA');
console.log(`Subscriber / begin() -> Nuevo subscriptor: ${this.topic}`);
this.mqttConfigService.subscribeTo(this.topic).subscribe((message) => {
console.log(`Subscriber / begin() -> Mensaje recibido: ${message}`);
this.messageSubject.next(message);
});
console.log('Subscriber / begin() -> FIN');
}

getMessages(): Observable<string> {
return this.messageSubject.asObservable();
}

1. Here is where the message isn't displayed anymore.

getMonitorUpdates(): Observable<IWaterTank> {
console.log('WaterTankService / getMonitorUpdates() -> INICIA');
return this.monitorSubscriber.getMessages().pipe(
map((msg: string) => {
try {
const waterTank: IWaterTank = JSON.parse(msg);
if (this.waterTankRule.isValidWaterTank(waterTank)) {
console.log(
'WaterTankService / getMonitorUpdates() -> Mensaje recibido:',
waterTank
);
return waterTank;
} else {
throw new Error('Estructura inválida del objeto');
}
} catch (error) {
console.error('WaterTankService / getMonitorUpdates() -> ', error);
throw error;
}
})
);
}

1. My component layer

ngOnInit(): void {
this.waterTankService.getMonitorUpdates().subscribe({
next: (data: IWaterTank) => {
this.waterTankData.push(data);
console.log('MonitorComponent / ngOnInit() -> Datos del Tinaco:', data);
},
error: (error) => {
// this.errorMessage = 'Error al recibir los datos del Tinaco.';
console.error('MonitorComponent / ngOnInit() -> ', error);
},
});
}


Here are some console logs:

1. Successful validation

MqttConfigService / subscribeTo() -> Mensaje recibido de tópico medidor-agua/tinaco/monitor: {
"fillingHeight": 1300,
"emptyingHeight": 20,
"superiorThreshold": 100,
"inferiorThreshold": 10,
"level": 100
}
Subscriber / begin() -> Mensaje recibido: {
"fillingHeight": 1300,
"emptyingHeight": 20,
"superiorThreshold": 100,
"inferiorThreshold": 10,
"level": 100
}
WaterTankRule / isValidWaterTank() -> INICIA
WaterTankService / getMonitorUpdates() -> Mensaje recibido: {fillingHeight: 1300, emptyingHeight: 20, superiorThreshold: 100, inferiorThreshold: 10, level: 100}
MonitorComponent / ngOnInit() -> Datos del Tinaco: {fillingHeight: 1300, emptyingHeight: 20, superiorThreshold: 100, inferiorThreshold: 10, level: 100}

1. Malformed messages and successful error handling:

MqttConfigService / subscribeTo() -> Mensaje recibido de tópico medidor-agua/tinaco/monitor: {
"fillingHeight": 1300,
"emptyingHeight": 20,
"level": 100
}
Subscriber / begin() -> Mensaje recibido: {
"fillingHeight": 1300,
"emptyingHeight": 20,
"level": 100
}
WaterTankRule / isValidWaterTank() -> INICIA
WaterTankService / getMonitorUpdates() -> Error: Estructura inválida del objeto
MonitorComponent / ngOnInit() -> Error: Estructura inválida del objeto
MqttConfigService / subscribeTo() -> Mensaje recibido de tópico medidor-agua/tinaco/monitor: {
"fillingHeight": 1300,
"emptyingHeight": 20,
"superiorThreshold": 100,
"inferiorThreshold": 10,
"level": 100
}

1. Attempting to send a well formed message but it get stuck:

MqttConfigService / subscribeTo() -> Mensaje recibido de tópico medidor-agua/tinaco/monitor: {
"fillingHeight": 1300,
"emptyingHeight": 20,
"superiorThreshold": 100,
"inferiorThreshold": 10,
"level": 100
}
Subscriber / begin() -> Mensaje recibido: {
"fillingHeight": 1300,
"emptyingHeight": 20,
"superiorThreshold": 100,
"inferiorThreshold": 10,
"level": 100
}


I'm trying to figure out what's going on. I think it might be due to the way I'm using getMessages(), since the logs aren't even showing up, but I'm not so sure how to check it. So the main question would be: How can I garantee that all incoming message are properly managed even if some of them are malformed?

Continue reading...