0

I'm a newbie building an IOT system comprised of two clients, the first client (Client A) will be a Subscriber and the second client (Client B) will be a Publisher. For the Broker, I intend to use something hosted (e.g. Cedalo).

Right now, I don't have Client B, but it will be some sort of H/W device that reads data from sensors and because Client B does not exist yet, I'm using Client A as both Subscriber and Publisher. Client A is running nodejs on a virtual machine and I have both express and mqtt packages installed.

The express server is up and running, e.g. listening and responding to http requests and in the same index.js file as the express server is the mqtt code that is Publishing and Subscribing to a topic, this is also running, e.g. I can Publish and Subscribe, and data from the Subscribed topic is displayed in the console.

Here is the working code:

let fs = require('fs');
let http = require('http');
let https = require('https');
let mqtt = require('mqtt');

let express = require('express');
let app = express();
let http_port = 8080;

let host = `mqtt://test.mosquitto.org`;
let mqttPort = `1883`;
let clientId = `domain`;
let connectUrl = `${host}:${mqttPort}`;

let httpServer = http.createServer(app);

let client = mqtt.connect(connectUrl, {
    clientId, clientId,
    clean: true,
    connectTimeout: 4000,
    username: `domain`,
    password: `secret`,
    reconnectPeriod: 1000,
});

let topic = `ambientTemp`;
client.on(`connect`, () => {
    console.log(`Connected to Broker at ${connectUrl}`);
    client.subscribe([topic], () => {
        console.log(`Subscribed to topic: ${topic}`);
    });
});

client.on(`error`, (err) => {
    console.log(`Error: ${err}`);
});

client.on(`message`, (topic, payload) => {
    console.log(`Received Message:`, topic, payload.toString())
})

client.on('connect', () => {
    client.publish(topic, '56 Deg.C', {
        qos: 0,
        retain: false }, (error) => {
            if (error) {
                console.error(error);
            }
        })
})

// For http
httpServer.listen(http_port, () => {
    console.log(`Listenting on Port:${http_port} Insecure`);
});

// this is the home route
app.get('/', function (req, res) {
    console.log("Hit Detected");
    res.write('<h1>Hello World</h1>');
    res.write(`<h4>v4<h4>`);
    res.end()
});

... here is the output from the browser and terminal: enter image description here

The problems start when I try to naively combine the "publish step" with the "route step" like this:

let fs = require('fs');
let http = require('http');
let https = require('https');
let mqtt = require('mqtt');

let express = require('express');
let app = express();
let http_port = 8080;

let host = `mqtt://test.mosquitto.org`;
let mqttPort = `1883`;
let clientId = `domain`;
let connectUrl = `${host}:${mqttPort}`;

let httpServer = http.createServer(app);

let client = mqtt.connect(connectUrl, {
    clientId, clientId,
    clean: true,
    connectTimeout: 4000,
    username: `domain`,
    password: `secret`,
    reconnectPeriod: 1000,
});

let topic = `ambientTemp`;
client.on(`connect`, () => {
    console.log(`Connected to Broker at ${connectUrl}`);
    client.subscribe([topic], () => {
        console.log(`Subscribed to topic: ${topic}`);
    });
});

client.on(`error`, (err) => {
    console.log(`Error: ${err}`);
});

client.on(`message`, (topic, payload) => {
    console.log(`Received Message:`, topic, payload.toString())
})

client.on('connect', () => {
    client.publish(topic, '56 Deg.C', {
        qos: 0,
        retain: false }, (error) => {
            if (error) {
                console.error(error);
            }
        })
})

// For http
httpServer.listen(http_port, () => {
    console.log(`Listening on Port:${http_port} Insecure`);
});

// this is the home route
app.get('/', function (req, res) {
    console.log("Hit Detected");
    res.write('<h1>Hello World</h1>');
    res.write(`<h4>v4<h4>`);

    client.on('connect', () => {                  //Additional code block causing issues
        client.publish(topic, '56 Deg.C', {       //Additional code block causing issues
            qos: 0,                               //Additional code block causing issues
            retain: false }, (error) => {         //Additional code block causing issues
                if (error) {                      //Additional code block causing issues
                    console.error(error);         //Additional code block causing issues
                }                                 //Additional code block causing issues
            })                                    //Additional code block causing issues
    })                                            //Additional code block causing issues

    res.end()
});

.... I get this:

enter image description here

It doesn't make sense to increase the maximum number of listeners because clearly there is something terribly wrong. This is what I think I know ... that node.js is asynchronous and event driven and that "app.get" is a callback function, as is "client.on" ... I'm struggling to understand why I can't seem to nest these callbacks. I'm missing something fundamental, I think.

It's also not clear to me why the "client.on" callback that is nested within the route does not even get called once, i.e. I have put a console.log immediately prior to it and immediately after it and only the first message is displayed (the code doesn't show this, just to explain what I have tried).

Any guidance would be appreciated.

Thank you, Tim.

1 Answer 1

0

The on('connect',...) callback will only ever be called once when the client connects, adding a new event listener for this event every time the get request is handled won't do anything useful and as the error states leaks "connect" event listeners.

If you just want to publish a message on each request, remove the on('connect',...) wrapper and just call client.publish(...)

app.get('/', function (req, res) {
    console.log("Hit Detected");
    res.write('<h1>Hello World</h1>');
    res.write(`<h4>v4<h4>`);

    client.publish(topic, '56 Deg.C', {
        qos: 0,
        retain: false 
    }, (error) => {
        if (error) {
             console.error(error);
        }
    })
    
    res.end()
});
1
  • That's it ! Thank you for the explanation, @hardillb.
    – TimH
    Oct 26, 2022 at 2:49

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.