9186116668

Committed 22 May 2024 06:01AM UTC coverage: 79.593% (-0.05%) from 79.645%

Build # 9186116668

Build Type

Pull #826

github

Committed by

web-flow

Commit Message

Merge 687e46def into 84af6f9e2

Pull Request Pull Request #826: send update for multiple measures

Run Details

494 of 699 branches covered (70.67%)

Branch coverage included in aggregate %.

13 of 13 new or added lines in 2 files covered. (100.0%)

3 existing lines in 1 file now uncovered.

1070 of 1266 relevant lines covered (84.52%)

126.12 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.91

/lib/bindings/MQTTBinding.js

/*
 * Copyright 2016 Telefonica Investigación y Desarrollo, S.A.U
 *
 * This file is part of iotagent-json
 *
 * iotagent-json is free software: you can redistribute it and/or
 * modify it under the terms of the GNU Affero General Public License as
 * published by the Free Software Foundation, either version 3 of the License,
 * or (at your option) any later version.
 *
 * iotagent-json is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 * See the GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public
 * License along with iotagent-json.
 * If not, seehttp://www.gnu.org/licenses/.
 *
 * For those usages not covered by the GNU Affero General Public License
 * please contact with::[contacto@tid.es]
 */

/* eslint-disable consistent-return */
/* eslint-disable no-unused-vars */

const fs = require('fs');
const iotAgentLib = require('iotagent-node-lib');
const mqtt = require('mqtt');
const commonBindings = require('../commonBindings');
const fillService = iotAgentLib.fillService;
const async = require('async');
const iotaUtils = require('../iotaUtils');
const constants = require('../constants');
let context = {
    op: 'IOTAJSON.MQTT.Binding'
};
let mqttClient;
let mqttConn;
const config = require('../configService');

/**
 * Generate the list of global topics to listen to.
 */
function generateTopics(callback) {
    const topics = [];

    config.getLogger().debug(context, 'Generating topics');

    // With leading slashes
    topics.push(constants.MQTT_SHARE_SUBSCRIPTION_GROUP + '/+/+/' + constants.MEASURES_SUFIX + '/+');
    topics.push(
        constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
            '/' +
            constants.MQTT_TOPIC_PROTOCOL +
            '/+/+/' +
            constants.MEASURES_SUFIX +
            '/+'
    );
    topics.push(constants.MQTT_SHARE_SUBSCRIPTION_GROUP + '/+/+/' + constants.MEASURES_SUFIX);
    topics.push(
        constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
            '/' +
            constants.MQTT_TOPIC_PROTOCOL +
            '/+/+/' +
            constants.MEASURES_SUFIX
    );
    topics.push(
        constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
            '/+/+/' +
            constants.CONFIGURATION_SUFIX +
            '/' +
            constants.CONFIGURATION_COMMAND_SUFIX
    );
    topics.push(
        constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
            '/' +
            constants.MQTT_TOPIC_PROTOCOL +
            '/+/+/' +
            constants.CONFIGURATION_SUFIX +
            '/' +
            constants.CONFIGURATION_COMMAND_SUFIX
    );
    topics.push(constants.MQTT_SHARE_SUBSCRIPTION_GROUP + '/+/+/' + constants.CONFIGURATION_COMMAND_UPDATE);
    topics.push(
        constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
            '/' +
            constants.MQTT_TOPIC_PROTOCOL +
            '/+/+/' +
            constants.CONFIGURATION_COMMAND_UPDATE
    );

    //Without leading slashes
    topics.push(constants.MQTT_SHARE_SUBSCRIPTION_GROUP + '/+/+/' + constants.MEASURES_SUFIX + '/+');
    topics.push(
        constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
            constants.MQTT_TOPIC_PROTOCOL +
            '/+/+/' +
            constants.MEASURES_SUFIX +
            '/+'
    );
    topics.push(constants.MQTT_SHARE_SUBSCRIPTION_GROUP + '/+/+/' + constants.MEASURES_SUFIX);
    topics.push(
        constants.MQTT_SHARE_SUBSCRIPTION_GROUP + constants.MQTT_TOPIC_PROTOCOL + '/+/+/' + constants.MEASURES_SUFIX
    );
    topics.push(
        constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
            '/+/+/' +
            constants.CONFIGURATION_SUFIX +
            '/' +
            constants.CONFIGURATION_COMMAND_SUFIX
    );
    topics.push(
        constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
            constants.MQTT_TOPIC_PROTOCOL +
            '/+/+/' +
            constants.CONFIGURATION_SUFIX +
            '/' +
            constants.CONFIGURATION_COMMAND_SUFIX
    );
    topics.push(constants.MQTT_SHARE_SUBSCRIPTION_GROUP + '/+/+/' + constants.CONFIGURATION_COMMAND_UPDATE);
    topics.push(
        constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
            constants.MQTT_TOPIC_PROTOCOL +
            '/+/+/' +
            constants.CONFIGURATION_COMMAND_UPDATE
    );

    callback(null, topics);
}

/**
 * Recreate the MQTT subscriptions for all the registered devices.
 */
function recreateSubscriptions(callback) {
    context = fillService(context, { service: 'n/a', subservice: 'n/a' });
    config.getLogger().debug(context, 'Recreating subscriptions for all devices');

    function subscribeToTopics(topics, callback) {
        config.getLogger().debug(context, 'Subscribing to topics: %j', topics);
        const options = {};
        mqttClient.subscribe(topics, options, function (error) {
            if (error) {
                iotAgentLib.alarms.raise(constants.MQTTB_ALARM, error);
                config.getLogger().error(context, 'GLOBAL-001: Error subscribing to topics: %s', error);
                callback(error);
            } else {
                iotAgentLib.alarms.release(constants.MQTTB_ALARM);
                config.getLogger().info(context, 'Successfully subscribed to the following topics: %j', topics);
                if (callback) {
                    callback(null);
                }
            }
        });
    }

    async.waterfall([generateTopics, subscribeToTopics], callback);
}

/**
 * Extract all the information from a Context Broker response and send it to the topic indicated by the APIKey and
 * DeviceId.
 *
 * @param {String} apiKey           API Key for the Device Group
 * @param {String} deviceId         ID of the Device.
 * @param {Object} results          Context Broker response.
 */
function sendConfigurationToDevice(apiKey, group, deviceId, results, callback) {
    const configurations = iotaUtils.createConfigurationNotification(results);
    const options = {};
    context = fillService(context, { service: 'n/a', subservice: 'n/a' });
    if (config.getConfig().mqtt.qos) {
        options.qos = parseInt(config.getConfig().mqtt.qos) || 0;
    }
    if (config.getConfig().mqtt.retain === true) {
        options.retain = config.getConfig().mqtt.retain;
    }
    config.getLogger().debug(context, 'Sending requested configuration to the device: %j', configurations);
    const leadingSlash = config.getConfig().mqtt.avoidLeadingSlash ? '' : '/';

    const commandTopic =
        leadingSlash +
        apiKey +
        '/' +
        deviceId +
        '/' +
        constants.CONFIGURATION_SUFIX +
        '/' +
        constants.CONFIGURATION_VALUES_SUFIX;
    //prettier-ignore
    if (mqttClient.connected) {
        mqttClient.publish(
            commandTopic,
            JSON.stringify(configurations),
            options,
            (error) => {
                if (error) {
                    config
                        .getLogger()
                        .error(
                            context,
                            'Error %j in Configuration: %j sent to the device %s with mqtt options %j',
                            error,
                            JSON.stringify(configurations),
                            commandTopic,
                            options
                        );
                }
            });
        config.getLogger().info(context, 'Configuration: %j was sent to the device: %s', configurations, commandTopic);
    } else {
        config.getLogger().error(context, 'Configuration: %j was not set to the device: %s due to not connected', configurations, commandTopic);
    }
    callback();
}

/**
 * Unsubscribe the MQTT Client from all the topics.
 */
function unsubscribeAll(callback) {
    function unsubscribeFromTopics(topics, callback) {
        mqttClient.unsubscribe(topics, null);

        callback();
    }

    async.waterfall([generateTopics, unsubscribeFromTopics], callback);
}

/**
 * Start the binding.
 */
function start(callback) {
    context = fillService(context, { service: 'n/a', subservice: 'n/a' });
    const mqttConfig = config.getConfig().mqtt;
    if (!mqttConfig) {
        return config.getLogger().error(context, 'Error MQTT is not configured');
    }
    if (mqttConfig.disabled) {
        return config.getLogger().warn(context, 'MQTT is disabled');
    }
    const rejectUnauthorized =
        typeof mqttConfig.rejectUnauthorized === 'boolean' ? mqttConfig.rejectUnauthorized : true;
    let rndSuffix = '_' + Math.random().toString(16).substr(2, 8);
    const options = {
        protocol: mqttConfig.protocol ? mqttConfig.protocol : 'mqtt',
        host: mqttConfig.host ? mqttConfig.host : 'localhost',
        port: mqttConfig.port ? mqttConfig.port : 1883,
        key: mqttConfig.key ? fs.readFileSync(mqttConfig.key, 'utf8') : null,
        ca: mqttConfig.ca ? fs.readFileSync(mqttConfig.ca, 'utf8') : null,
        cert: mqttConfig.cert ? fs.readFileSync(mqttConfig.cert, 'utf8') : null,
        rejectUnauthorized,
        username: mqttConfig.username ? mqttConfig.username : null,
        password: mqttConfig.password ? mqttConfig.password : null,
        clean: typeof mqttConfig.clean === 'boolean' ? mqttConfig.clean : true,
        clientId: mqttConfig.clientId ? mqttConfig.clientId + rndSuffix : 'iotajson' + rndSuffix,
        keepalive: mqttConfig.keepalive ? parseInt(mqttConfig.keepalive) : 60,
        connectTimeout: 60 * 60 * 1000
    };

    const retries = mqttConfig.retries ? mqttConfig.retries : constants.MQTT_DEFAULT_RETRIES;
    const retryTime = mqttConfig.retryTime ? mqttConfig.retryTime : constants.MQTT_DEFAULT_RETRY_TIME;
    let isConnecting = false;
    let numRetried = 1; // retries will be disabled when MQTT_DEFAULT_RETRIES=0
    config
        .getLogger()
        .info(context, 'Starting MQTT binding with options %j retries %s retryTIme %s', options, retries, retryTime);

    function createConnection(callback) {
        config.getLogger().debug(context, 'creating connection');
        if (isConnecting) {
            return;
        }
        isConnecting = true;
        // Ensure clientId is unique when reconnect to avoid loop closing old connection which the same name
        rndSuffix = '_' + Math.random().toString(16).substr(2, 8);
        options.clientId = mqttConfig.clientId ? mqttConfig.clientId + rndSuffix : 'iotajson' + rndSuffix;
        mqttClient = mqtt.connect(options.protocol + '://' + mqttConfig.host + ':' + mqttConfig.port, options);
        isConnecting = false;
        if (!mqttClient) {
            config.getLogger().error(context, 'error mqttClient not created');
            if (numRetried <= retries) {
                numRetried++;
                return setTimeout(createConnection, retryTime * 1000, callback);
            }
        }
        mqttClient.on('error', function (e) {
            /*jshint quotmark: double */
            config.getLogger().fatal("GLOBAL-002: Couldn't connect with MQTT broker: %j", e);
            /*jshint quotmark: single */
            mqttClient.end();
        });
        mqttClient.on('message', commonBindings.mqttMessageHandler);
        mqttClient.on('connect', function (ack) {
            config.getLogger().info(context, 'MQTT Client connected');
            recreateSubscriptions();
            numRetried = 1;
        });
        mqttClient.on('reconnect', function () {
            config.getLogger().debug(context, 'MQTT Client reconnect');
        });
        mqttClient.on('offline', function () {
            config.getLogger().debug(context, 'MQTT Client offline');
        });
        mqttClient.on('close', function () {
            config.getLogger().info(context, 'MQTT Client closed');
            // If mqttConn is null, the connection has been closed on purpose
            if (mqttConn) {
                config.getLogger().debug(context, 'MQTT Client closed connected? %s', mqttClient.connected);
                if (!mqttClient.connected && numRetried <= retries) {
                    config.getLogger().warn(context, 'reconnecting #%s...', numRetried);
                    numRetried++;
                    return setTimeout(createConnection, retryTime * 1000);
                }
            } else {
                // Do nothing
            }
        });
        config.getLogger().info(context, 'connected');
        mqttConn = mqttClient;
        if (callback) {
            callback();
        }
    } // function createConnection

    async.waterfall([createConnection], function (error) {
        if (error) {
            config.getLogger().info('MQTT error %j', error);
        }
        callback();
    });
}

/**
 * Device provisioning handler.
 *
 * @param {Object} device           Device object containing all the information about the provisioned device.
 */
function deviceProvisioningHandler(device, callback) {
    callback(null, device);
}

/**
 * Device updating handler.
 *
 * @param {Object} device           Device object containing all the information about the provisioned device.
 */
function deviceUpdatingHandler(device, callback) {
    callback(null, device);
}

/**
 * Stop the binding, releasing its resources.
 */
function stop(callback) {
    config.getLogger().info('Stopping MQTT Binding');

    async.series([unsubscribeAll, mqttClient.end.bind(mqttClient, true)], function () {
        config.getLogger().info('MQTT Binding Stopped');
        if (mqttConn) {
            mqttConn = null;
        }
        callback();
    });
}

/**
 * Execute a command for the device represented by the device object and the given APIKey, sending the serialized
 * JSON payload (already containing the command information).
 *
 * @param {String} apiKey                   APIKey of the device that will be receiving the command.
 * @param {Object} group                    Data object for the group receiving the command.
 * @param {Object} device                   Data object for the device receiving the command.
 * @param {String} serializedPayload        String payload in JSON format for the command.
 */
function executeCommand(apiKey, group, device, cmdName, serializedPayload, contentType, callback) {
    const options = {};
    // retrieve command mqtt options from device
    const commands = Object.assign({}, ...device.commands.map((c) => ({ [c.name]: c })));

    options.qos =
        commands[cmdName].mqtt && commands[cmdName].mqtt.qos
            ? parseInt(commands[cmdName].mqtt.qos)
            : config.getConfig().mqtt.qos
            ? parseInt(config.getConfig().mqtt.qos)
            : 0;
    options.retain =
        commands[cmdName].mqtt && commands[cmdName].mqtt.retain
            ? commands[cmdName].mqtt.retain
            : config.getConfig().mqtt.retain
            ? config.getConfig().mqtt.retain
            : false;

    const commandTopic = '/' + apiKey + '/' + device.id + '/cmd';
    config
        .getLogger()
        .debug(
            context,
            'Sending command execution to %s with payload %s and with mqtt options %j',
            commandTopic,
            serializedPayload,
            options
        );
    if (mqttClient.connected) {
        mqttClient.publish(commandTopic, serializedPayload, options, (error) => {
            if (error) {
                config
                    .getLogger()
                    .error(
                        context,
                        'Error %j in Cmd: %j sent to the device %s with mqtt options %j',
                        error,
                        serializedPayload,
                        commandTopic,
                        options
                    );
            }
        });
        config
            .getLogger()
            .info(
                context,
                'Cmd: %j was sent to the device %s with mqtt options %j',
                serializedPayload,
                commandTopic,
                options
            );
    } else {
        config.getLogger().error(context, 'Cmd: not set due to not connected');
    }
    callback();
}

exports.start = start;
exports.stop = stop;
exports.sendConfigurationToDevice = sendConfigurationToDevice;
exports.deviceProvisioningHandler = deviceProvisioningHandler;
exports.deviceUpdatingHandler = deviceUpdatingHandler;
exports.executeCommand = executeCommand;
exports.protocol = 'MQTT';

1	/*
2	* Copyright 2016 Telefonica Investigación y Desarrollo, S.A.U
3	*
4	* This file is part of iotagent-json
5	*
6	* iotagent-json is free software: you can redistribute it and/or
7	* modify it under the terms of the GNU Affero General Public License as
8	* published by the Free Software Foundation, either version 3 of the License,
9	* or (at your option) any later version.
10	*
11	* iotagent-json is distributed in the hope that it will be useful,
12	* but WITHOUT ANY WARRANTY; without even the implied warranty of
13	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
14	* See the GNU Affero General Public License for more details.
15	*
16	* You should have received a copy of the GNU Affero General Public
17	* License along with iotagent-json.
18	* If not, seehttp://www.gnu.org/licenses/.
19	*
20	* For those usages not covered by the GNU Affero General Public License
21	* please contact with::[contacto@tid.es]
22	*/
23
24	/* eslint-disable consistent-return */
25	/* eslint-disable no-unused-vars */
26
27	const fs = require('fs');	1✔
28	const iotAgentLib = require('iotagent-node-lib');	1✔
29	const mqtt = require('mqtt');	1✔
30	const commonBindings = require('../commonBindings');	1✔
31	const fillService = iotAgentLib.fillService;	1✔
32	const async = require('async');	1✔
33	const iotaUtils = require('../iotaUtils');	1✔
34	const constants = require('../constants');	1✔
35	let context = {	1✔
36	op: 'IOTAJSON.MQTT.Binding'
37	};
38	let mqttClient;
39	let mqttConn;
40	const config = require('../configService');	1✔
41
42	/**
43	* Generate the list of global topics to listen to.
44	*/
45	function generateTopics(callback) {
46	const topics = [];	676✔
47
48	config.getLogger().debug(context, 'Generating topics');	676✔
49
50	// With leading slashes
51	topics.push(constants.MQTT_SHARE_SUBSCRIPTION_GROUP + '/+/+/' + constants.MEASURES_SUFIX + '/+');	676✔
52	topics.push(	676✔
53	constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
54	'/' +
55	constants.MQTT_TOPIC_PROTOCOL +
56	'/+/+/' +
57	constants.MEASURES_SUFIX +
58	'/+'
59	);
60	topics.push(constants.MQTT_SHARE_SUBSCRIPTION_GROUP + '/+/+/' + constants.MEASURES_SUFIX);	676✔
61	topics.push(	676✔
62	constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
63	'/' +
64	constants.MQTT_TOPIC_PROTOCOL +
65	'/+/+/' +
66	constants.MEASURES_SUFIX
67	);
68	topics.push(	676✔
69	constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
70	'/+/+/' +
71	constants.CONFIGURATION_SUFIX +
72	'/' +
73	constants.CONFIGURATION_COMMAND_SUFIX
74	);
75	topics.push(	676✔
76	constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
77	'/' +
78	constants.MQTT_TOPIC_PROTOCOL +
79	'/+/+/' +
80	constants.CONFIGURATION_SUFIX +
81	'/' +
82	constants.CONFIGURATION_COMMAND_SUFIX
83	);
84	topics.push(constants.MQTT_SHARE_SUBSCRIPTION_GROUP + '/+/+/' + constants.CONFIGURATION_COMMAND_UPDATE);	676✔
85	topics.push(	676✔
86	constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
87	'/' +
88	constants.MQTT_TOPIC_PROTOCOL +
89	'/+/+/' +
90	constants.CONFIGURATION_COMMAND_UPDATE
91	);
92
93	//Without leading slashes
94	topics.push(constants.MQTT_SHARE_SUBSCRIPTION_GROUP + '/+/+/' + constants.MEASURES_SUFIX + '/+');	676✔
95	topics.push(	676✔
96	constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
97	constants.MQTT_TOPIC_PROTOCOL +
98	'/+/+/' +
99	constants.MEASURES_SUFIX +
100	'/+'
101	);
102	topics.push(constants.MQTT_SHARE_SUBSCRIPTION_GROUP + '/+/+/' + constants.MEASURES_SUFIX);	676✔
103	topics.push(	676✔
104	constants.MQTT_SHARE_SUBSCRIPTION_GROUP + constants.MQTT_TOPIC_PROTOCOL + '/+/+/' + constants.MEASURES_SUFIX
105	);
106	topics.push(	676✔
107	constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
108	'/+/+/' +
109	constants.CONFIGURATION_SUFIX +
110	'/' +
111	constants.CONFIGURATION_COMMAND_SUFIX
112	);
113	topics.push(	676✔
114	constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
115	constants.MQTT_TOPIC_PROTOCOL +
116	'/+/+/' +
117	constants.CONFIGURATION_SUFIX +
118	'/' +
119	constants.CONFIGURATION_COMMAND_SUFIX
120	);
121	topics.push(constants.MQTT_SHARE_SUBSCRIPTION_GROUP + '/+/+/' + constants.CONFIGURATION_COMMAND_UPDATE);	676✔
122	topics.push(	676✔
123	constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
124	constants.MQTT_TOPIC_PROTOCOL +
125	'/+/+/' +
126	constants.CONFIGURATION_COMMAND_UPDATE
127	);
128
129	callback(null, topics);	676✔
130	}
131
132	/**
133	* Recreate the MQTT subscriptions for all the registered devices.
134	*/
135	function recreateSubscriptions(callback) {
136	context = fillService(context, { service: 'n/a', subservice: 'n/a' });	437✔
137	config.getLogger().debug(context, 'Recreating subscriptions for all devices');	437✔
138
139	function subscribeToTopics(topics, callback) {
140	config.getLogger().debug(context, 'Subscribing to topics: %j', topics);	437✔
141	const options = {};	437✔
142	mqttClient.subscribe(topics, options, function (error) {	437✔
143	if (error) {	437!
UNCOV 144	iotAgentLib.alarms.raise(constants.MQTTB_ALARM, error);	×
UNCOV 145	config.getLogger().error(context, 'GLOBAL-001: Error subscribing to topics: %s', error);	×
UNCOV 146	callback(error);	×
147	} else {
148	iotAgentLib.alarms.release(constants.MQTTB_ALARM);	437✔
149	config.getLogger().info(context, 'Successfully subscribed to the following topics: %j', topics);	437✔
150	if (callback) {	437!
151	callback(null);	437✔
152	}
153	}
154	});
155	}
156
157	async.waterfall([generateTopics, subscribeToTopics], callback);	437✔
158	}
159
160	/**
161	* Extract all the information from a Context Broker response and send it to the topic indicated by the APIKey and
162	* DeviceId.
163	*
164	* @param {String} apiKey API Key for the Device Group
165	* @param {String} deviceId ID of the Device.
166	* @param {Object} results Context Broker response.
167	*/
168	function sendConfigurationToDevice(apiKey, group, deviceId, results, callback) {
169	const configurations = iotaUtils.createConfigurationNotification(results);	4✔
170	const options = {};	4✔
171	context = fillService(context, { service: 'n/a', subservice: 'n/a' });	4✔
172	if (config.getConfig().mqtt.qos) {	4!
173	options.qos = parseInt(config.getConfig().mqtt.qos) \|\| 0;	×
174	}
175	if (config.getConfig().mqtt.retain === true) {	4!
176	options.retain = config.getConfig().mqtt.retain;	×
177	}
178	config.getLogger().debug(context, 'Sending requested configuration to the device: %j', configurations);	4✔
179	const leadingSlash = config.getConfig().mqtt.avoidLeadingSlash ? '' : '/';	4!
180
181	const commandTopic =
182	leadingSlash +	4✔
183	apiKey +
184	'/' +
185	deviceId +
186	'/' +
187	constants.CONFIGURATION_SUFIX +
188	'/' +
189	constants.CONFIGURATION_VALUES_SUFIX;
190	//prettier-ignore
191	if (mqttClient.connected) {	4!
192	mqttClient.publish(	4✔
193	commandTopic,
194	JSON.stringify(configurations),
195	options,
196	(error) => {
197	if (error) {	4!
198	config	×
199	.getLogger()
200	.error(
201	context,
202	'Error %j in Configuration: %j sent to the device %s with mqtt options %j',
203	error,
204	JSON.stringify(configurations),
205	commandTopic,
206	options
207	);
208	}
209	});
210	config.getLogger().info(context, 'Configuration: %j was sent to the device: %s', configurations, commandTopic);	4✔
211	} else {
212	config.getLogger().error(context, 'Configuration: %j was not set to the device: %s due to not connected', configurations, commandTopic);	×
213	}
214	callback();	4✔
215	}
216
217	/**
218	* Unsubscribe the MQTT Client from all the topics.
219	*/
220	function unsubscribeAll(callback) {
221	function unsubscribeFromTopics(topics, callback) {
222	mqttClient.unsubscribe(topics, null);	239✔
223
224	callback();	239✔
225	}
226
227	async.waterfall([generateTopics, unsubscribeFromTopics], callback);	239✔
228	}
229
230	/**
231	* Start the binding.
232	*/
233	function start(callback) {
234	context = fillService(context, { service: 'n/a', subservice: 'n/a' });	239✔
235	const mqttConfig = config.getConfig().mqtt;	239✔
236	if (!mqttConfig) {	239!
237	return config.getLogger().error(context, 'Error MQTT is not configured');	×
238	}
239	if (mqttConfig.disabled) {	239!
240	return config.getLogger().warn(context, 'MQTT is disabled');	×
241	}
242	const rejectUnauthorized =
243	typeof mqttConfig.rejectUnauthorized === 'boolean' ? mqttConfig.rejectUnauthorized : true;	239!
244	let rndSuffix = '_' + Math.random().toString(16).substr(2, 8);	239✔
245	const options = {	239✔
246	protocol: mqttConfig.protocol ? mqttConfig.protocol : 'mqtt',	239!
247	host: mqttConfig.host ? mqttConfig.host : 'localhost',	239!
248	port: mqttConfig.port ? mqttConfig.port : 1883,	239!
249	key: mqttConfig.key ? fs.readFileSync(mqttConfig.key, 'utf8') : null,	239!
250	ca: mqttConfig.ca ? fs.readFileSync(mqttConfig.ca, 'utf8') : null,	239!
251	cert: mqttConfig.cert ? fs.readFileSync(mqttConfig.cert, 'utf8') : null,	239!
252	rejectUnauthorized,
253	username: mqttConfig.username ? mqttConfig.username : null,	239!
254	password: mqttConfig.password ? mqttConfig.password : null,	239!
255	clean: typeof mqttConfig.clean === 'boolean' ? mqttConfig.clean : true,	239!
256	clientId: mqttConfig.clientId ? mqttConfig.clientId + rndSuffix : 'iotajson' + rndSuffix,	239!
257	keepalive: mqttConfig.keepalive ? parseInt(mqttConfig.keepalive) : 60,	239!
258	connectTimeout: 60 * 60 * 1000
259	};
260
261	const retries = mqttConfig.retries ? mqttConfig.retries : constants.MQTT_DEFAULT_RETRIES;	239!
262	const retryTime = mqttConfig.retryTime ? mqttConfig.retryTime : constants.MQTT_DEFAULT_RETRY_TIME;	239!
263	let isConnecting = false;	239✔
264	let numRetried = 1; // retries will be disabled when MQTT_DEFAULT_RETRIES=0	239✔
265	config	239✔
266	.getLogger()
267	.info(context, 'Starting MQTT binding with options %j retries %s retryTIme %s', options, retries, retryTime);
268
269	function createConnection(callback) {
270	config.getLogger().debug(context, 'creating connection');	442✔
271	if (isConnecting) {	442!
272	return;	×
273	}
274	isConnecting = true;	442✔
275	// Ensure clientId is unique when reconnect to avoid loop closing old connection which the same name
276	rndSuffix = '_' + Math.random().toString(16).substr(2, 8);	442✔
277	options.clientId = mqttConfig.clientId ? mqttConfig.clientId + rndSuffix : 'iotajson' + rndSuffix;	442!
278	mqttClient = mqtt.connect(options.protocol + '://' + mqttConfig.host + ':' + mqttConfig.port, options);	442✔
279	isConnecting = false;	442✔
280	if (!mqttClient) {	442!
281	config.getLogger().error(context, 'error mqttClient not created');	×
282	if (numRetried <= retries) {	×
283	numRetried++;	×
284	return setTimeout(createConnection, retryTime * 1000, callback);	×
285	}
286	}
287	mqttClient.on('error', function (e) {	442✔
288	/jshint quotmark: double /
289	config.getLogger().fatal("GLOBAL-002: Couldn't connect with MQTT broker: %j", e);	×
290	/jshint quotmark: single /
291	mqttClient.end();	×
292	});
293	mqttClient.on('message', commonBindings.mqttMessageHandler);	442✔
294	mqttClient.on('connect', function (ack) {	442✔
295	config.getLogger().info(context, 'MQTT Client connected');	437✔
296	recreateSubscriptions();	437✔
297	numRetried = 1;	437✔
298	});
299	mqttClient.on('reconnect', function () {	442✔
300	config.getLogger().debug(context, 'MQTT Client reconnect');	×
301	});
302	mqttClient.on('offline', function () {	442✔
303	config.getLogger().debug(context, 'MQTT Client offline');	×
304	});
305	mqttClient.on('close', function () {	442✔
306	config.getLogger().info(context, 'MQTT Client closed');	239✔
307	// If mqttConn is null, the connection has been closed on purpose
308	if (mqttConn) {	239✔
309	config.getLogger().debug(context, 'MQTT Client closed connected? %s', mqttClient.connected);	234✔
310	if (!mqttClient.connected && numRetried <= retries) {	234!
311	config.getLogger().warn(context, 'reconnecting #%s...', numRetried);	234✔
312	numRetried++;	234✔
313	return setTimeout(createConnection, retryTime * 1000);	234✔
314	}
315	} else {
316	// Do nothing
317	}
318	});
319	config.getLogger().info(context, 'connected');	442✔
320	mqttConn = mqttClient;	442✔
321	if (callback) {	442✔
322	callback();	239✔
323	}
324	} // function createConnection
325
326	async.waterfall([createConnection], function (error) {	239✔
327	if (error) {	239!
328	config.getLogger().info('MQTT error %j', error);	×
329	}
330	callback();	239✔
331	});
332	}
333
334	/**
335	* Device provisioning handler.
336	*
337	* @param {Object} device Device object containing all the information about the provisioned device.
338	*/
339	function deviceProvisioningHandler(device, callback) {
340	callback(null, device);	148✔
341	}
342
343	/**
344	* Device updating handler.
345	*
346	* @param {Object} device Device object containing all the information about the provisioned device.
347	*/
348	function deviceUpdatingHandler(device, callback) {
349	callback(null, device);	2✔
350	}
351
352	/**
353	* Stop the binding, releasing its resources.
354	*/
355	function stop(callback) {
356	config.getLogger().info('Stopping MQTT Binding');	239✔
357
358	async.series([unsubscribeAll, mqttClient.end.bind(mqttClient, true)], function () {	239✔
359	config.getLogger().info('MQTT Binding Stopped');	239✔
360	if (mqttConn) {	239!
361	mqttConn = null;	239✔
362	}
363	callback();	239✔
364	});
365	}
366
367	/**
368	* Execute a command for the device represented by the device object and the given APIKey, sending the serialized
369	* JSON payload (already containing the command information).
370	*
371	* @param {String} apiKey APIKey of the device that will be receiving the command.
372	* @param {Object} group Data object for the group receiving the command.
373	* @param {Object} device Data object for the device receiving the command.
374	* @param {String} serializedPayload String payload in JSON format for the command.
375	*/
376	function executeCommand(apiKey, group, device, cmdName, serializedPayload, contentType, callback) {
377	const options = {};	12✔
378	// retrieve command mqtt options from device
379	const commands = Object.assign({}, ...device.commands.map((c) => ({ [c.name]: c })));	42✔
380
381	options.qos =	12✔
382	commands[cmdName].mqtt && commands[cmdName].mqtt.qos	24!
383	? parseInt(commands[cmdName].mqtt.qos)
384	: config.getConfig().mqtt.qos	12!
385	? parseInt(config.getConfig().mqtt.qos)
386	: 0;
387	options.retain =	12✔
388	commands[cmdName].mqtt && commands[cmdName].mqtt.retain	24!
389	? commands[cmdName].mqtt.retain
390	: config.getConfig().mqtt.retain	12!
391	? config.getConfig().mqtt.retain
392	: false;
393
394	const commandTopic = '/' + apiKey + '/' + device.id + '/cmd';	12✔
395	config	12✔
396	.getLogger()
397	.debug(
398	context,
399	'Sending command execution to %s with payload %s and with mqtt options %j',
400	commandTopic,
401	serializedPayload,
402	options
403	);
404	if (mqttClient.connected) {	12✔
405	mqttClient.publish(commandTopic, serializedPayload, options, (error) => {	11✔
406	if (error) {	11!
407	config	×
408	.getLogger()
409	.error(
410	context,
411	'Error %j in Cmd: %j sent to the device %s with mqtt options %j',
412	error,
413	serializedPayload,
414	commandTopic,
415	options
416	);
417	}
418	});
419	config	11✔
420	.getLogger()
421	.info(
422	context,
423	'Cmd: %j was sent to the device %s with mqtt options %j',
424	serializedPayload,
425	commandTopic,
426	options
427	);
428	} else {
429	config.getLogger().error(context, 'Cmd: not set due to not connected');	1✔
430	}
431	callback();	12✔
432	}
433
434	exports.start = start;	1✔
435	exports.stop = stop;	1✔
436	exports.sendConfigurationToDevice = sendConfigurationToDevice;	1✔
437	exports.deviceProvisioningHandler = deviceProvisioningHandler;	1✔
438	exports.deviceUpdatingHandler = deviceUpdatingHandler;	1✔
439	exports.executeCommand = executeCommand;	1✔
440	exports.protocol = 'MQTT';	1✔

telefonicaid / iotagent-json / 9186116668

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous