9177165570

Committed 21 May 2024 03:21PM UTC coverage: 79.03% (-0.1%) from 79.158%

Build # 9177165570

Build Type

Pull #677

github

Committed by

web-flow

Commit Message

Merge 7a3cd1031 into 932bc41e3

Pull Request Pull Request #677: [WIP] send to CB using multimeasures

Run Details

351 of 517 branches covered (67.89%)

Branch coverage included in aggregate %.

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

3 existing lines in 1 file now uncovered.

855 of 1009 relevant lines covered (84.74%)

141.0 hits per line

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

75.0

/lib/bindings/MQTTBinding.js

/*
 * Copyright 2016 Telefonica Investigación y Desarrollo, S.A.U
 *
 * This file is part of iotagent-ul
 *
 * iotagent-ul 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-ul 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-ul.
 * If not, seehttp://www.gnu.org/licenses/.
 *
 * For those usages not covered by the GNU Affero General Public License
 * please contact with::[iot_support@tid.es]
 */

/* eslint-disable consistent-return */

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

/**
 * Generate the list of global topics to listen to.
 */
function generateTopics(callback) {
    const topics = [];
    context = fillService(context, { service: 'n/a', subservice: 'n/a' });
    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.
 */
function recreateSubscriptions(callback) {
    context = fillService(context, { service: 'n/a', subservice: 'n/a' });
    config.getLogger().debug(context, 'Recreating global subscriptions for all devices');

    function subscribeToTopics(topics, callback) {
        config.getLogger().debug('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('Successfully subscribed to the following topics: %j', topics);
                if (callback) {
                    callback(null);
                }
            }
        });
    }

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

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

        callback();
    }

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

/**
 * Generate a function that executes the given command in the device.
 *
 * @param {String} apiKey           APIKey of the device's service or default APIKey.
 * @param {Object} device           Object containing all the information about a device.
 * @param {Object} attribute        Attribute in NGSI format.
 * @return {Function}               Command execution function ready to be called with async.series.
 */
function generateCommandExecution(apiKey, device, attribute) {
    const cmdName = attribute.name;
    const cmdAttributes = attribute.value;
    const options = {};
    const payload = ulParser.createCommandPayload(device, cmdName, cmdAttributes);
    const commands = Object.assign({}, ...device.commands.map((c) => ({ [c.name]: c })));
    context = fillService(context, device);

    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()
        .info(
            context,
            'Sending command execution to %s with payload %s and mqtt options %j',
            commandTopic,
            payload,
            options
        );
    return mqttClient.publish.bind(mqttClient, commandTopic, payload, options);
}

/**
 * Handles a command execution request coming from the Context Broker. This handler should:
 *  - Identify the device affected by the command.
 *  - Send the command to the appropriate MQTT topic.
 *  - Update the command status in the Context Broker.
 *
 * @param {Object} device           Device data stored in the IOTA.
 * @param {String} attributes       Command attributes (in NGSIv1 format).
 */
function commandHandler(device, attributes, callback) {
    context = fillService(context, device);
    config.getLogger().debug(context, 'Handling MQTT command for device %s', device.id);

    utils.getEffectiveApiKey(device.service, device.subservice, device, function (error, apiKey) {
        async.series(attributes.map(generateCommandExecution.bind(null, apiKey, device)), 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 = utils.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;
    }

    const payload = ulParser.createConfigurationPayload(deviceId, configurations);
    config
        .getLogger()
        .debug(
            context,
            'Sending requested configuration to device %s with apikey %s and payload %s ',
            deviceId,
            apiKey,
            payload
        );
    const leadingSlash = config.getConfig().mqtt.avoidLeadingSlash ? '' : '/';
    const commandTopic =
        leadingSlash +
        apiKey +
        '/' +
        deviceId +
        '/' +
        constants.CONFIGURATION_SUFIX +
        '/' +
        constants.CONFIGURATION_VALUES_SUFIX;
    if (mqttClient.connected) {
        mqttClient.publish(commandTopic, payload, options, (error) => {
            if (error) {
                config
                    .getLogger()
                    .error(
                        context,
                        'Error %j in Configuration: %j sent to the device %s with mqtt options %j',
                        error,
                        payload,
                        commandTopic,
                        options
                    );
            }
        });
        config.getLogger().info(context, 'Configuration: %j was sent to the device %s', payload, commandTopic);
    } else {
        config
            .getLogger()
            .error(
                context,
                'Configuration: %j was not sent to the device %s due to not connected',
                payload,
                commandTopic
            );
    }
    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);
}

/**
 * Starts the IoT Agent with the passed configuration. This method also starts the listeners for all the transport
 * binding plugins.
 */
function start(callback) {
    const mqttConfig = config.getConfig().mqtt;
    context = fillService(context, { service: 'n/a', subservice: 'n/a' });
    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 : 'iotaul' + 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().info(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 : 'iotaul' + 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);
        /* eslint-disable-next-line no-unused-vars */
        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();
    });
}

/**
 * Stops the IoT Agent and all the transport plugins.
 */
function stop(callback) {
    context = fillService(context, { service: 'n/a', subservice: 'n/a' });
    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();
    });
}

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

1	/*
2	* Copyright 2016 Telefonica Investigación y Desarrollo, S.A.U
3	*
4	* This file is part of iotagent-ul
5	*
6	* iotagent-ul 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-ul 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-ul.
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::[iot_support@tid.es]
22	*/
23
24	/* eslint-disable consistent-return */
25
26	const fs = require('fs');	2✔
27	const iotAgentLib = require('iotagent-node-lib');	2✔
28	const commonBindings = require('./../commonBindings');	2✔
29	const utils = require('../iotaUtils');	2✔
30	const ulParser = require('../ulParser');	2✔
31	const mqtt = require('mqtt');	2✔
32	const async = require('async');	2✔
33	const constants = require('../constants');	2✔
34	const fillService = iotAgentLib.fillService;	2✔
35	let context = {	2✔
36	op: 'IOTAUL.MQTT.Binding'
37	};
38	let mqttClient;
39	let mqttConn;
40	const config = require('../configService');	2✔
41
42	/**
43	* Generate the list of global topics to listen to.
44	*/
45	function generateTopics(callback) {
46	const topics = [];	594✔
47	context = fillService(context, { service: 'n/a', subservice: 'n/a' });	594✔
48	config.getLogger().debug(context, 'Generating topics');	594✔
49
50	// With leading slashes
51	topics.push(constants.MQTT_SHARE_SUBSCRIPTION_GROUP + '/+/+/' + constants.MEASURES_SUFIX + '/+');	594✔
52	topics.push(	594✔
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);	594✔
61	topics.push(	594✔
62	constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
63	'/' +
64	constants.MQTT_TOPIC_PROTOCOL +
65	'/+/+/' +
66	constants.MEASURES_SUFIX
67	);
68	topics.push(	594✔
69	constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
70	'/+/+/' +
71	constants.CONFIGURATION_SUFIX +
72	'/' +
73	constants.CONFIGURATION_COMMAND_SUFIX
74	);
75	topics.push(	594✔
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);	594✔
85	topics.push(	594✔
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 + '/+');	594✔
95	topics.push(	594✔
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);	594✔
103	topics.push(	594✔
104	constants.MQTT_SHARE_SUBSCRIPTION_GROUP + constants.MQTT_TOPIC_PROTOCOL + '/+/+/' + constants.MEASURES_SUFIX
105	);
106	topics.push(	594✔
107	constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
108	'+/+/' +
109	constants.CONFIGURATION_SUFIX +
110	'/' +
111	constants.CONFIGURATION_COMMAND_SUFIX
112	);
113	topics.push(	594✔
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);	594✔
122	topics.push(	594✔
123	constants.MQTT_SHARE_SUBSCRIPTION_GROUP +
124	constants.MQTT_TOPIC_PROTOCOL +
125	'/+/+/' +
126	constants.CONFIGURATION_COMMAND_UPDATE
127	);
128
129	callback(null, topics);	594✔
130	}
131
132	/**
133	* Recreate the MQTT subscriptions.
134	*/
135	function recreateSubscriptions(callback) {
136	context = fillService(context, { service: 'n/a', subservice: 'n/a' });	362✔
137	config.getLogger().debug(context, 'Recreating global subscriptions for all devices');	362✔
138
139	function subscribeToTopics(topics, callback) {
140	config.getLogger().debug('Subscribing to topics: %j', topics);	362✔
141	const options = {};	362✔
142	mqttClient.subscribe(topics, options, function (error) {	362✔
143	if (error) {	362!
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);	362✔
149	config.getLogger().info('Successfully subscribed to the following topics: %j', topics);	362✔
150	if (callback) {	362!
151	callback(null);	362✔
152	}
153	}
154	});
155	}
156
157	async.waterfall([generateTopics, subscribeToTopics], callback);	362✔
158	}
159
160	/**
161	* Unsubscribe the MQTT Client for all the topics of all the devices of all the services.
162	*/
163	function unsubscribeAll(callback) {
164	function unsubscribeFromTopics(topics, callback) {
165	mqttClient.unsubscribe(topics, null);	232✔
166
167	callback();	232✔
168	}
169
170	async.waterfall([generateTopics, unsubscribeFromTopics], callback);	232✔
171	}
172
173	/**
174	* Generate a function that executes the given command in the device.
175	*
176	* @param {String} apiKey APIKey of the device's service or default APIKey.
177	* @param {Object} device Object containing all the information about a device.
178	* @param {Object} attribute Attribute in NGSI format.
179	* @return {Function} Command execution function ready to be called with async.series.
180	*/
181	function generateCommandExecution(apiKey, device, attribute) {
182	const cmdName = attribute.name;	8✔
183	const cmdAttributes = attribute.value;	8✔
184	const options = {};	8✔
185	const payload = ulParser.createCommandPayload(device, cmdName, cmdAttributes);	8✔
186	const commands = Object.assign({}, ...device.commands.map((c) => ({ [c.name]: c })));	8✔
187	context = fillService(context, device);	8✔
188
189	options.qos =	8✔
190	commands[cmdName].mqtt && commands[cmdName].mqtt.qos	16!
191	? parseInt(commands[cmdName].mqtt.qos)
192	: config.getConfig().mqtt.qos	8!
193	? parseInt(config.getConfig().mqtt.qos)
194	: 0;
195	options.retain =	8✔
196	commands[cmdName].mqtt && commands[cmdName].mqtt.retain	16!
197	? commands[cmdName].mqtt.retain
198	: config.getConfig().mqtt.retain	8!
199	? config.getConfig().mqtt.retain
200	: false;
201
202	const commandTopic = '/' + apiKey + '/' + device.id + '/cmd';	8✔
203	config	8✔
204	.getLogger()
205	.info(
206	context,
207	'Sending command execution to %s with payload %s and mqtt options %j',
208	commandTopic,
209	payload,
210	options
211	);
212	return mqttClient.publish.bind(mqttClient, commandTopic, payload, options);	8✔
213	}
214
215	/**
216	* Handles a command execution request coming from the Context Broker. This handler should:
217	* - Identify the device affected by the command.
218	* - Send the command to the appropriate MQTT topic.
219	* - Update the command status in the Context Broker.
220	*
221	* @param {Object} device Device data stored in the IOTA.
222	* @param {String} attributes Command attributes (in NGSIv1 format).
223	*/
224	function commandHandler(device, attributes, callback) {
225	context = fillService(context, device);	8✔
226	config.getLogger().debug(context, 'Handling MQTT command for device %s', device.id);	8✔
227
228	utils.getEffectiveApiKey(device.service, device.subservice, device, function (error, apiKey) {	8✔
229	async.series(attributes.map(generateCommandExecution.bind(null, apiKey, device)), callback);	8✔
230	});
231	}
232
233	/**
234	* Extract all the information from a Context Broker response and send it to the topic indicated by the APIKey and
235	* DeviceId.
236	*
237	* @param {String} apiKey API Key for the Device Group
238	* @param {String} deviceId ID of the Device.
239	* @param {Object} results Context Broker response.
240	*/
241	function sendConfigurationToDevice(apiKey, group, deviceId, results, callback) {
242	const configurations = utils.createConfigurationNotification(results);	8✔
243	const options = {};	8✔
244	context = fillService(context, { service: 'n/a', subservice: 'n/a' });	8✔
245	if (config.getConfig().mqtt.qos) {	8!
246	options.qos = parseInt(config.getConfig().mqtt.qos) \|\| 0;	×
247	}
248
249	if (config.getConfig().mqtt.retain === true) {	8!
250	options.retain = config.getConfig().mqtt.retain;	×
251	}
252
253	const payload = ulParser.createConfigurationPayload(deviceId, configurations);	8✔
254	config	8✔
255	.getLogger()
256	.debug(
257	context,
258	'Sending requested configuration to device %s with apikey %s and payload %s ',
259	deviceId,
260	apiKey,
261	payload
262	);
263	const leadingSlash = config.getConfig().mqtt.avoidLeadingSlash ? '' : '/';	8!
264	const commandTopic =
265	leadingSlash +	8✔
266	apiKey +
267	'/' +
268	deviceId +
269	'/' +
270	constants.CONFIGURATION_SUFIX +
271	'/' +
272	constants.CONFIGURATION_VALUES_SUFIX;
273	if (mqttClient.connected) {	8!
274	mqttClient.publish(commandTopic, payload, options, (error) => {	8✔
275	if (error) {	8!
276	config	×
277	.getLogger()
278	.error(
279	context,
280	'Error %j in Configuration: %j sent to the device %s with mqtt options %j',
281	error,
282	payload,
283	commandTopic,
284	options
285	);
286	}
287	});
288	config.getLogger().info(context, 'Configuration: %j was sent to the device %s', payload, commandTopic);	8✔
289	} else {
290	config	×
291	.getLogger()
292	.error(
293	context,
294	'Configuration: %j was not sent to the device %s due to not connected',
295	payload,
296	commandTopic
297	);
298	}
299	callback();	8✔
300	}
301
302	/**
303	* Device provisioning handler.
304	*
305	* @param {Object} device Device object containing all the information about the provisioned device.
306	*/
307	function deviceProvisioningHandler(device, callback) {
308	callback(null, device);	238✔
309	}
310
311	/**
312	* Device updating handler.
313	*
314	* @param {Object} device Device object containing all the information about the provisioned device.
315	*/
316	function deviceUpdatingHandler(device, callback) {
317	callback(null, device);	4✔
318	}
319
320	/**
321	* Starts the IoT Agent with the passed configuration. This method also starts the listeners for all the transport
322	* binding plugins.
323	*/
324	function start(callback) {
325	const mqttConfig = config.getConfig().mqtt;	232✔
326	context = fillService(context, { service: 'n/a', subservice: 'n/a' });	232✔
327	if (!mqttConfig) {	232!
328	return config.getLogger().error(context, 'Error MQTT is not configured');	×
329	}
330	if (mqttConfig.disabled) {	232!
331	return config.getLogger().warn(context, 'MQTT is disabled');	×
332	}
333
334	const rejectUnauthorized =
335	typeof mqttConfig.rejectUnauthorized === 'boolean' ? mqttConfig.rejectUnauthorized : true;	232!
336	let rndSuffix = '_' + Math.random().toString(16).substr(2, 8);	232✔
337	const options = {	232✔
338	protocol: mqttConfig.protocol ? mqttConfig.protocol : 'mqtt',	232!
339	host: mqttConfig.host ? mqttConfig.host : 'localhost',	232!
340	port: mqttConfig.port ? mqttConfig.port : 1883,	232!
341	key: mqttConfig.key ? fs.readFileSync(mqttConfig.key, 'utf8') : null,	232!
342	ca: mqttConfig.ca ? fs.readFileSync(mqttConfig.ca, 'utf8') : null,	232!
343	cert: mqttConfig.cert ? fs.readFileSync(mqttConfig.cert, 'utf8') : null,	232!
344	rejectUnauthorized,
345	username: mqttConfig.username ? mqttConfig.username : null,	232!
346	password: mqttConfig.password ? mqttConfig.password : null,	232!
347	clean: typeof mqttConfig.clean === 'boolean' ? mqttConfig.clean : true,	232!
348	clientId: mqttConfig.clientId ? mqttConfig.clientId + rndSuffix : 'iotaul' + rndSuffix,	232!
349	keepalive: mqttConfig.keepalive ? parseInt(mqttConfig.keepalive) : 60,	232!
350	connectTimeout: 60 * 60 * 1000
351	};
352
353	const retries = mqttConfig.retries ? mqttConfig.retries : constants.MQTT_DEFAULT_RETRIES;	232!
354	const retryTime = mqttConfig.retryTime ? mqttConfig.retryTime : constants.MQTT_DEFAULT_RETRY_TIME;	232!
355	let isConnecting = false;	232✔
356	let numRetried = 1; // retries will be disabled when MQTT_DEFAULT_RETRIES=0	232✔
357	config	232✔
358	.getLogger()
359	.info(context, 'Starting MQTT binding with options %j retries %s retryTIme %s', options, retries, retryTime);
360	function createConnection(callback) {
361	config.getLogger().info(context, 'creating connection');	367✔
362	if (isConnecting) {	367!
363	return;	×
364	}
365	isConnecting = true;	367✔
366	// Ensure clientId is unique when reconnect to avoid loop closing old connection which the same name
367	rndSuffix = '_' + Math.random().toString(16).substr(2, 8);	367✔
368	options.clientId = mqttConfig.clientId ? mqttConfig.clientId + rndSuffix : 'iotaul' + rndSuffix;	367!
369	mqttClient = mqtt.connect(options.protocol + '://' + mqttConfig.host + ':' + mqttConfig.port, options);	367✔
370	isConnecting = false;	367✔
371	if (!mqttClient) {	367!
372	config.getLogger().error(context, 'error mqttClient not created');	×
373	if (numRetried <= retries) {	×
374	numRetried++;	×
375	return setTimeout(createConnection, retryTime * 1000, callback);	×
376	}
377	}
378	mqttClient.on('error', function (e) {	367✔
379	/jshint quotmark: double /
380	config.getLogger().fatal("GLOBAL-002: Couldn't connect with MQTT broker: %j", e);	×
381	/jshint quotmark: single /
382	mqttClient.end();	×
383	});
384	mqttClient.on('message', commonBindings.mqttMessageHandler);	367✔
385	/* eslint-disable-next-line no-unused-vars */
386	mqttClient.on('connect', function (ack) {	367✔
387	config.getLogger().info(context, 'MQTT Client connected');	362✔
388	recreateSubscriptions();	362✔
389	numRetried = 1;	362✔
390	});
391	mqttClient.on('reconnect', function () {	367✔
392	config.getLogger().debug(context, 'MQTT Client reconnect');	×
393	});
394	mqttClient.on('offline', function () {	367✔
395	config.getLogger().debug(context, 'MQTT Client offline');	×
396	});
397	mqttClient.on('close', function () {	367✔
398	config.getLogger().info(context, 'MQTT Client closed');	232✔
399	// If mqttConn is null, the connection has been closed on purpose
400	if (mqttConn) {	232✔
401	config.getLogger().debug(context, 'MQTT Client closed connected? %s', mqttClient.connected);	227✔
402	if (!mqttClient.connected && numRetried <= retries) {	227!
403	config.getLogger().warn(context, 'reconnecting #%s..', numRetried);	227✔
404	numRetried++;	227✔
405	return setTimeout(createConnection, retryTime * 1000);	227✔
406	}
407	} else {
408	// Do Nothing
409	}
410	});
411
412	config.getLogger().info(context, 'connected');	367✔
413	mqttConn = mqttClient;	367✔
414	if (callback) {	367✔
415	callback();	232✔
416	}
417	} // function createConnection
418
419	async.waterfall([createConnection], function (error) {	232✔
420	if (error) {	232!
421	config.getLogger().info('MQTT error %j', error);	×
422	}
423	callback();	232✔
424	});
425	}
426
427	/**
428	* Stops the IoT Agent and all the transport plugins.
429	*/
430	function stop(callback) {
431	context = fillService(context, { service: 'n/a', subservice: 'n/a' });	232✔
432	config.getLogger().info('Stopping MQTT Binding');	232✔
433
434	async.series([unsubscribeAll, mqttClient.end.bind(mqttClient, true)], function () {	232✔
435	config.getLogger().info('MQTT Binding Stopped');	232✔
436	if (mqttConn) {	232!
437	mqttConn = null;	232✔
438	}
439	callback();	232✔
440	});
441	}
442
443	exports.sendConfigurationToDevice = sendConfigurationToDevice;	2✔
444	exports.deviceProvisioningHandler = deviceProvisioningHandler;	2✔
445	exports.deviceUpdatingHandler = deviceUpdatingHandler;	2✔
446	exports.commandHandler = commandHandler;	2✔
447	exports.start = start;	2✔
448	exports.stop = stop;	2✔
449	exports.protocol = 'MQTT';	2✔

telefonicaid / iotagent-ul / 9177165570

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