Node.js and MQTT - "level":"error","message":"uncaughtException:-CodePudding

Here's my frigo controller :

const fs = require('fs');
const mqtt = require('mqtt');
const transporter = require('../params/mail')
const winston = require('../params/log');
const User = require("../models/User");
const { cli } = require('winston/lib/winston/config');

exports.OpenTheCase = async (req, res) => {};
exports.AddCard = async (req, res) => {};
exports.ShowCurrentTemperature = async (req, res) => {};
exports.ShowCurrentHumidity = async (req, res) => {};
exports.SetAlarm = async (req, res) => {};
exports.AlarmIsOn = async (req, res) => {};

const options = {
    clientId: 'backendserver1032',
    key: fs.readFileSync('./certs/mqtt_cert/client.key'),
    cert: fs.readFileSync('./certs/mqtt_cert/client.crt'),
    ca: [ fs.readFileSync('./certs/mqtt_cert/ca.crt') ]
  }

const client = mqtt.connect('mqtts://localhost:8883', options);

exports.OpenTheCase = async (req, res) => {
    try {
        client.publish('RFID', 'RFID_OPEN');
        res.status(200).json({ 'case':"opened" });
    }
    catch(e){
        res.status(200).json({ 'state':"something went wrong" });
    }
}

exports.AddCard = async (req, res) => {
    try {
        client.publish('RFID', 'RFID_ADD');
        res.status(200).json({ 'card':"will be added" });
    }
    catch(e){
        res.status(200).json({ 'state':"something went wrong" });
    }
}

exports.ShowCurrentTemperature = async (req, res) => {
    try {
        client.subscribe('temperature');
        client.on('message', (topic, message, packet) => {
            res.status(200).json({ 'temperature': message.toString('ascii') })
            client.unsubscribe('temperature')
        })
    }
    catch(e){
        res.status(200).json({ 'state':"something went wrong" });
    }
    return
}

exports.ShowCurrentHumidity = async (req, res) => {
    try {
        client.subscribe('humidity');
        client.on('message', (topic, message) => {
            res.status(200).json({"temperature": message.toString('ascii')});
            client.unsubscribe('humidity')
        });
    }
    catch(e){
        res.status(200).json({ 'state':"something went wrong" });
    }
    return
}

The problem is : when I try to get "ShowCurrentTemperature", it works once and after it. It says that the http header was already send.

Here's my route :

router.get("/frigo/Temperature",auth.verifyToken, frigoController.ShowCurrentTemperature)

I really thank you.

I had try several things, like adding return or trying to end the connection but none of them works. I'm getting out of idea. If someone can help me through this.

CodePudding user response：

This design is never going to work.

You are basically leaking client.on('message',...) handlers. Every call to the HTTP endpoint adds a new handler which holds a reference to it's local res object.

Calling unsubscribe() does not remove the message handler, so the instant you call subscribe() again in either of the HTTP routes the very next message to arrive will be delivered to ALL the old handlers which will ALL try to call res.send() on a already complete HTTP transaction.

You are trying to map an inherently asynchronous protocol (MQTT) into a synchronous HTTP request, which is a really bad idea.

You may be able to get it to work by swapping all the client.on('message', ...) calls to client.once('message', ....) so the handlers only fire once, but this is a truly UGLY hack. (EDIT: On reflection, this still has a race condition where you may end up handling the wrong message if both HTTP endpoints are called too close together, so I stand by my first statement that this design can never work properly)

The right thing to do is to run the whole MQTT client totally independently of the HTTP requests and just have the single background client.on('message',...) handler update some global variables with the latest temperature and humidity that the HTTP routes can just return the latest value, rather than try and grab the next published message.

CodePudding user response：

In a nutshell, this error happens when you try to send more than one response using res. This could happen in your code when there is an exception during client.unsubscribe('temperature'), because then the execution will flow into the catch where you are sending another response. You can avoid that by moving the unsubscribe after the try-catch:

exports.ShowCurrentTemperature = async(req, res) => {
  try {
    client.subscribe('temperature');
    client.on('message', (topic, message, packet) => {
      res.status(200).json({
        'temperature': message.toString('ascii')
      })
    })
  } catch (e) {
    res.status(200).json({
      'state': "something went wrong"
    });
  }
  client.unsubscribe('temperature')
}

Update:

Actually, the more likely explanation is that you receive more than one message before you unsubscribe, and hence the first res.. is executed multiple times. So probably best to unsubscribe before sending a response. Since more handlers could queue up in the meantime, though, you probably need to add a guard as well to make sure you never send more than one response:

exports.ShowCurrentTemperature = async(req, res) => {
  let done = false;
  try {
    client.subscribe('temperature');
    client.on('message', (topic, message, packet) => {
      client.unsubscribe('temperature')
      !done && res.status(200).json({
        'temperature': message.toString('ascii')
      })
      done = true;
    })
  } catch (e) {
    !done && res.status(200).json({
      'state': "something went wrong"
    });
  }
}

BTW, you could also just use mqtt directly on the client, which would be more elegant, if that's fine from an authorization perspective.