MQTT client for Node.js

node publish.js -t topic -f fileName -q qos -c clientid -s hostname:port -u username -p password

Options may be present in any order, and if absent, default values will be used.

node publish.js -t testTopic -f /Users/username/myjsonfiles/data.json

connecting to client at: 127.0.0.1:1883

connected to client at: 127.0.0.1:1883

Reading file at data.json

publishing to topic 'test/Topic1'

Connection closed.

// Command line parsing and program defaults.
const argv = require( "yargs" )
      .version( "1.0.0" )
      .options( {
         t: {
            alias:    "topic",
            type:     "string",
            default:  "testTopic",
            describe: "An optional topic to subscribe to.",
         },
         f: {
            alias:        ["filename", "file"],
            type:         "string",
            describe:     "A required file name to use for the payload in the published message.",
            demandOption: true,
         },
         q: {
            alias:    "qos",
            type:     "number",
            default:  1,
            describe: "An optional quality of service to use when publishing the message.",
         },
         c: {
            alias:    ["clientId", "client"],
            type:     "string",
            describe: "An optional client id that uniquely identifies the client to the MQTT broker.",
         },
         s: {
            alias:    ["server", "broker", "b"],
            type:     "string",
            default:  "127.0.0.1:1883",
            describe: "An optional hostname and port of an MQTT broker.",
         },
         u: {
            alias:    ["username", "user"],
            type:     "string",
            describe: "An optional user name to use when connecting to the broker. If absent, a connection without a username and password is attempted.",
         },
         p: {
            alias:    ["password", "pass"],
            type:     "string",
            describe: "An optional password to use with the username provided above. Ignored if the username is absent.",
         },
         h: {
            alias:    "help",
            type:     "boolean",
            describe: "Print the usage and exit.",
         }
      } )
      .help()
      .usage( "Usage: node $0 [options]" )
      .argv;


// Read settings from the command line or use defaults.
let settings = {
   topic:           argv.t,
   fileName:        argv.f,
   qos:             argv.q,
   clientId:        argv.c || "nodeJS_pub_" + Math.random().toString( 16 ).substring( 2, 8 ),
   brokerAddress:   argv.s,
   username:        argv.u,
   password:        argv.p,
   protocolId:      "MQTT",
   protocolVersion: 4  // 4 is MQTT 3.1.1 and 5 is MQTT 5.0
};


// Variable declarations and initialization.
const brokerUrl = `mqtt://${settings.brokerAddress}`;
let mqtt = require( "mqtt" );
let process = require( "process" );
let client = null;


// Connect to the MQTT broker.
console.log( `Connecting to broker at: ${settings.brokerAddress}` )
client = mqtt.connect(
      brokerUrl,
      {
         clientId:        settings.clientId,
         connectTimeout:  30 * 1000,
         reconnectPeriod: 1000,
         protocolId:      settings.protocolId,
         protocolVersion: settings.protocolVersion,
         ...( settings.username && {username: settings.username} ),
         ...( settings.password && {password: settings.password} )
      } );


// Interrupt handling.
process.on( "SIGINT", function()
{
   console.log( "\nShutting down MQTT client after Ctrl-C" );
   if( client )
   {
      // Wait for in-flight messages to be sent and exit the program.
      client.end( false, () => { process.exit(); } );
   }
   else {
      process.exit();
   }
} );


// The "On Connect" listener.
client.on( "connect", function()
{
   console.log( `Connected to broker at: ${settings.brokerAddress}` );

   // read in file from fileName path passed in by user.
   console.log( `Reading file at ${settings.fileName}` );
   try {
      const fs = require( "fs" );
      const msgData = fs.readFileSync( settings.fileName );
      console.log( `Publishing to topic '${settings.topic}'` );
      client.publish(
            settings.topic,
            JSON.stringify( msgData ),
            {qos: settings.qos} );
      client.end();
   }
   catch( e )
   {
      if( e.code !== "MODULE_NOT_FOUND" )
      {
         client.end();
         throw e;
      }
   }
}, )


// The "On Disconnect" listener.
client.on( "disconnect", function()
{
   console.log( `Disconnected from: ${brokerUrl}` );
   if( client.reconnecting === false )
      client.reconnect(); // Only reconnect if not already reconnecting.
} );


// The "On Error" listener.
client.on( "error", function( err )
{
   console.log( "Error: " + err );
   if( err && typeof err === "object" && "code" in err )
      if( err.code === "ENOTFOUND" )
         console.log( "Network error: Verify the passed hostname:port and that your MQTT server is running." );
} );


// The "On Close" listener.
client.on( "close", function()
{
   console.log( "Connection closed." );
   if( client.isConnected )
      client.end();
} );

node subscribe.js -t topic -d directory -q QoS -c clientID -s hostname:port -u username -p password

Options may be present in any order, and if absent, default values will be used.

node subscribe.js -t testTopic -d testDir -c testSubClientID

connecting to broker at: 127.0.0.1:1883

connected to broker at: mqtt://127.0.0.1:1883

subscribing to topic: 'test/Topic1'

 

Messages that arrive will be printed to the screen.

received message: {

       "property":    "value",

       "string":      "string",

       "number":      1,

       "boolean":     true,

       "null":        null,
  
       "array":       [
       
            "value",
               
            1,

            1.1,

            -0,
   
            true,

            false,

            null

      ],

      "emptyObject": { }

}

When a directory is specified, the utility writes the payload of each received message to a file. The name of the file is the name of the topic you specified followed by an underscore plus a sequential number. Topic names are case-sensitive. The file has no extension because an MQTT payload can be anything.

// Command line parsing and program defaults.
const argv = require( "yargs" )
      .version( "1.0.0" )
      .options( {
         t: {
            alias:    "topic",
            type:     "string",
            default:  "testTopic",
            describe: "An optional topic to subscribe to.",
         },
         d: {
            alias:    ["directory", "dir"],
            type:     "string",
            describe: "An optional directory to save messages as files.",
         },
         q: {
            alias:    "qos",
            type:     "number",
            default:  1,
            describe: "An optional quality of service to use when when subscribing to the topic.",
         },
         c: {
            alias:    ["clientId", "client"],
            type:     "string",
            describe: "Client ID for the MQTT connection",
         },
         s: {
            alias:    ["server", "broker", "b"],
            type:     "string",
            default:  "127.0.0.1:1883",
            describe: "An optional hostname and port of an MQTT broker.",
         },
         u: {
            alias:    ["username", "user"],
            type:     "string",
            describe: "An optional user name to use when connecting to the broker. If absent, a connection without a username and password is attempted.",
         },
         p: {
            alias:    ["password", "pass"],
            type:     "string",
            describe: "An optional password to use with the username provided above. Ignored if the username is absent.",
         },
         h: {
            alias:    "help",
            type:     "boolean",
            describe: "Print the usage and exit.",
         }
      } )
      .help()
      .usage( "Usage: node $0 [options]" )
      .argv;


// Read settings from the command line or use defaults.
let settings = {
   topic:           argv.t,
   outputDirectory: argv.d || null,
   qos:             argv.q,
   clientId:        argv.c || "nodeJS_sub_" + Math.random().toString( 16 ).substring( 2, 8 ),
   brokerAddress:   argv.s,
   username:        argv.u,
   password:        argv.p,
   protocolId:      "MQTT",
   protocolVersion: 4  // 4 is MQTT 3.1.1 and 5 is MQTT 5.0
};


// Variable declarations and initialization.
const fs = require( "fs" );
const path = require( 'path' );
const MAX_FILENAME_LEN = 500;
const brokerUrl = `mqtt://${settings.brokerAddress}`;
let mqtt = require( "mqtt" );
let process = require( "process" );
let client = null;
let counter = 0;


// Connect to the MQTT broker.
console.log( `Connecting to broker at: ${settings.brokerAddress}` );
client = mqtt.connect(
      brokerUrl,
      {
         clientId:        settings.clientId,
         connectTimeout:  30 * 1000,
         keepalive:       20,
         reconnectPeriod: 1000,
         protocolId:      settings.protocolId,
         protocolVersion: settings.protocolVersion,
         ...( settings.username && {username: settings.username} ),
         ...( settings.password && {password: settings.password} )
      } );


// Interrupt handling.
process.on( "SIGINT", function()
{
   console.log( "\nShutting down MQTT client after Ctrl-C" );
   if( client )
   {
      // Wait for in-flight messages to be sent and exit the program.
      client.end( false, () => { process.exit(); } );
   }
   else {
      process.exit();
   }
} );


// The "On Connect" listener.
client.on( "connect", function()
{
   console.log( `Connected to broker at: ${brokerUrl}` );
   console.log( `Subscribing to topic: '${settings.topic}'` );
   client.subscribe( settings.topic, {qos: settings.qos} );
} );


// The "On Disconnect" listener.
client.on( "disconnect", function()
{
   console.log( `Disconnected from: ${brokerUrl}` );
   if( client.reconnecting === false )
      client.reconnect(); // Only reconnect if not already reconnecting.
} );


// The "On Close" listener.
client.on( "close", function()
{
   console.log( "Connection closed." );
   if( client.isConnected )
      client.end();
} );


// The "On Error" listener.
client.on( "error", function( err )
{
   console.log( "Error: " + err );
   if( err && typeof err === "object" && "code" in err )
      if( err.code === "ENOTFOUND" )
         console.log( "Network error: Verify the passed hostname:port and that your MQTT server is running." );
} );


// The "On Message" listener.
client.on( "message", function( topic, message )
{
   let formattedMsg = Buffer.isBuffer( message ) ? message.toString() : message;
   console.log( `Received message: ${formattedMsg}` );

   // Save the message to disc if the user set an output directory.
   if( "outputDirectory" in settings && settings.outputDirectory !== null )
   {
      const absoluteOutputDirectory = path.join( process.cwd(), settings.outputDirectory );
      // Check if the directory exists and create it if absent.
      fs.access( absoluteOutputDirectory, fs.constants.F_OK, ( err ) =>
      {
         if( err )
         {
            // The directory does not exist, so create it.
            fs.mkdir( absoluteOutputDirectory, {recursive: true}, ( mkdirErr ) =>
            {
               if( mkdirErr )
                  console.error( `Error creating directory "${settings.outputDirectory}":`, mkdirErr );
               else
                  console.log( `Directory "${settings.outputDirectory}" created successfully.` );
               // Write the file only after the directory is created.
               writeFile( absoluteOutputDirectory, formattedMsg );
            } );
         }
         else {
            // Directory already exists.
            writeFile( absoluteOutputDirectory, formattedMsg );
         }
      } );
   }
} );


function writeFile( directory, data )
{
   let seq = String( ++counter ).padStart( 6, "0" );
   let filePrefix;
   let topic = settings.topic;

   topic = ( topic.length > MAX_FILENAME_LEN ) ? topic.substring( topic.length - MAX_FILENAME_LEN ) : topic;
   filePrefix = topic.replaceAll( /[/\\|· &%#*<>?:"]/g, "_" );
   let absoluteFileName = path.join( directory, `${filePrefix}_${seq}` );

   fs.writeFile(
         absoluteFileName,
         Buffer.from( data ),
         {flag: "w"},
         err => err ? console.error( err ) : null
   );
}