Node.js application: Store messages from Mosquitto MQTT broker into SQL Database (completed)

app_mqtt_mysql_completed.js

var mqtt = require('mqtt'); //https://www.npmjs.com/package/mqtt
var Topic = '#'; //subscribe to all topics
var Broker_URL = 'mqtt://192.168.1.123';
var Database_URL = '192.168.1.123';

var options = {
	clientId: 'MyMQTT',
	port: 1883,
	//username: 'mqtt_user',
	//password: 'mqtt_password',	
	keepalive : 60
};

var client  = mqtt.connect(Broker_URL, options);
client.on('connect', mqtt_connect);
client.on('reconnect', mqtt_reconnect);
client.on('error', mqtt_error);
client.on('message', mqtt_messsageReceived);
client.on('close', mqtt_close);

function mqtt_connect() {
    //console.log("Connecting MQTT");
    client.subscribe(Topic, mqtt_subscribe);
};

function mqtt_subscribe(err, granted) {
    console.log("Subscribed to " + Topic);
    if (err) {console.log(err);}
};

function mqtt_reconnect(err) {
    //console.log("Reconnect MQTT");
    //if (err) {console.log(err);}
	client  = mqtt.connect(Broker_URL, options);
};

function mqtt_error(err) {
    //console.log("Error!");
	//if (err) {console.log(err);}
};

function after_publish() {
	//do nothing
};

//receive a message from MQTT broker
function mqtt_messsageReceived(topic, message, packet) {
	var message_str = message.toString(); //convert byte array to string
	message_str = message_str.replace(/\n$/, ''); //remove new line
	//payload syntax: clientID,topic,message
	if (countInstances(message_str) != 1) {
		console.log("Invalid payload");
		} else {	
		insert_message(topic, message_str, packet);
		//console.log(message_arr);
	}
};

function mqtt_close() {
	//console.log("Close MQTT");
};

////////////////////////////////////////////////////
///////////////////// MYSQL ////////////////////////
////////////////////////////////////////////////////
var mysql = require('mysql'); //https://www.npmjs.com/package/mysql
//Create Connection
var connection = mysql.createConnection({
	host: Database_URL,
	user: "newuser",
	password: "mypassword",
	database: "mydb"
});

connection.connect(function(err) {
	if (err) throw err;
	//console.log("Database Connected!");
});

//insert a row into the tbl_messages table
function insert_message(topic, message_str, packet) {
	var message_arr = extract_string(message_str); //split a string into an array
	var clientID= message_arr[0];
	var message = message_arr[1];
	var sql = "INSERT INTO ?? (??,??,??) VALUES (?,?,?)";
	var params = ['tbl_messages', 'clientID', 'topic', 'message', clientID, topic, message];
	sql = mysql.format(sql, params);	
	
	connection.query(sql, function (error, results) {
		if (error) throw error;
		console.log("Message added: " + message_str);
	}); 
};	

//split a string into an array of substrings
function extract_string(message_str) {
	var message_arr = message_str.split(","); //convert to array	
	return message_arr;
};	

//count number of delimiters in a string
var delimiter = ",";
function countInstances(message_str) {
	var substrings = message_str.split(delimiter);
	return substrings.length - 1;
};

This is exactly what I have been looking for to write my MQTT messages from mosquito to a mysql DB. I am hitting an error message when running this:

ian@HyperledgerFabric:/vuchain/MQTT$ node MQTT2SQL.js
/vuchain/MQTT/node_modules/mqtt/lib/connect/index.js:64
throw new Error('Missing protocol')
^

Error: Missing protocol
at Object.connect (/vuchain/MQTT/node_modules/mqtt/lib/connect/index.js:64:13)
at Object. (/vuchain/MQTT/MQTT2SQL.js:14:20)
at Module._compile (module.js:652:30)
at Object.Module._extensions..js (module.js:663:10)
at Module.load (module.js:565:32)
at tryModuleLoad (module.js:505:12)
at Function.Module._load (module.js:497:3)
at Function.Module.runMain (module.js:693:10)
at startup (bootstrap_node.js:191:16)
at bootstrap_node.js:612:3

I have ran npm install mqtt --save and made sure the example @ https://www.npmjs.com/package/mqtt worked properly.
I have mosquito and mysql running (and tested) on this same server and specified them in the application.

Thanks!

Maybe its the topic issue. Did you resolve this issue?

For me
var Broker_URL = 'mqtt://192.168.1.123';

din't work

var Broker_URL = 'tcp://192.168.1.123';

worked for me.

Everything works fine on a Mac OS, with Ubuntu i get a
Invalid Payload
everytime anyone subscribed posting a mqtt msg.
Mosqutitto, MySQL, everything connected - all system looking good.
Any ideas?
Txs

Everything works fine on a Mac OS, with Ubuntu i get a
Invalid Payload
everytime anyone subscribed posting a mqtt msg.
Mosqutitto, MySQL, everything connected - all system looking good.
Any ideas?
Txs

Same thing in Windows 10

for me : "Invalid payload"

I did a branch that's working on Ubuntu: https://github.com/CptHolzschnauz/MQTT-MYSQL-NODE-SYNC