15172033588

Committed 21 May 2025 08:34PM UTC coverage: 79.881% (-0.2%) from 80.121%

Build # 15172033588

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #877 from telefonicaid/task/add_headers_to_http_commands

add test about group command with extra headers

Run Details

516 of 729 branches covered (70.78%)

Branch coverage included in aggregate %.

1100 of 1294 relevant lines covered (85.01%)

139.24 hits per line

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

82.61

/lib/commandHandler.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]
 */

const async = require('async');
const iotAgentLib = require('iotagent-node-lib');
const iotaUtils = require('./iotaUtils');
const constants = require('./constants');
const transportSelector = require('./transportSelector');
const config = require('./configService');
const context = {
    op: 'IoTAgentJSON.Commands'
};

/**
 * Serializes a payload for a command depending on its payloadType if provided
 *
 * @param {String} payload          Payload to serialized
 * @param {Object} command          Command attribute
 * @return {Function}               Returns a serialized payload
 */
function serializedPayloadCommand(payload, command) {
    let serialized;
    if (command && command.payloadType) {
        switch (command.payloadType.toLowerCase()) {
            case 'binaryfromstring':
                serialized = Buffer.from(payload.toString());
                break;
            case 'binaryfromhex':
                serialized = Buffer.from(payload, 'HEX');
                break;
            case 'binaryfromjson': // used by AMQP transport
                serialized = Buffer.from(JSON.stringify(payload));
                break;
            case 'text':
                serialized = payload;
                break;
            case 'json': // passthrough
            default:
                serialized = JSON.stringify(payload);
        }
    } else {
        serialized = JSON.stringify(payload);
    }
    return serialized;
}

/**
 * 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, group, attribute) {
    let payload = {};
    const command = device && device.commands.find((att) => att.name === attribute.name);

    if (command && command.expression) {
        const parser = iotAgentLib.dataPlugins.expressionTransformation;
        // The context for the JEXL expression should be the ID, TYPE, S, SS
        let attrList = iotAgentLib.dataPlugins.utils.getIdTypeServSubServiceFromDevice(device);
        attrList = device.staticAttributes
            ? attrList.concat(device.staticAttributes).concat(attribute)
            : attrList.concat(attribute);
        const ctxt = parser.extractContext(attrList, device);
        // expression result will be the full command payload
        let payloadRes = null;
        try {
            payloadRes = parser.applyExpression(command.expression, ctxt, device);
        } catch (e) {
            // nothing to report
        }
        payload = payloadRes ? payloadRes : command.expression;
    } else {
        payload[attribute.name] = attribute.value;
    }
    if (device.transport === 'AMQP') {
        // to ensure backward compability
        command.payloadType = command.payloadType ? command.payloadType : 'binaryfromjson';
    }
    const serialized = serializedPayloadCommand(payload, command);
    const contentType = command && command.contentType ? command.contentType : 'application/json';
    const extraHeaders = command && command.headers ? command.headers : undefined;
    config
        .getLogger()
        .debug(
            context,
            'Sending command execution to device %s with apikey %s and payload %j extraHeaders %j',
            device.id,
            apiKey,
            payload,
            extraHeaders
        );
    const executions = transportSelector.createExecutionsForBinding(
        [apiKey, group, device, attribute.name, serialized, extraHeaders, contentType],
        'executeCommand',
        device.transport ||
            (group && group.transport ? group.transport : undefined) ||
            config.getConfig().defaultTransport
    );
    return executions;
}

/**
 * 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 {String} id               ID of the entity for which the command execution was issued.
 * @param {String} type             Type of the entity for which the command execution was issued.
 * @param {String} service          Service ID.
 * @param {String} subservice       Subservice ID.
 * @param {Array} attributes        List of NGSI attributes of type command to execute.
 */
function commandHandler(id, type, service, subservice, attributes, callback) {
    config
        .getLogger()
        .debug(context, 'Handling command %j for device %s in service %s - %s', attributes, id, service, subservice);

    function concat(previous, current) {
        previous = previous.concat(current);
        return previous;
    }

    iotAgentLib.getDeviceByNameAndType(id, type, service, subservice, function (error, device) {
        if (error) {
            config.getLogger().error(
                context,

                "COMMAND-001: Command execution could not be handled, as device for entity %s %s wasn't found",

                id,
                type
            );
            callback(error);
        } else {
            iotaUtils.getEffectiveApiKey(device.service, device.subservice, device, function (error, apiKey) {
                if (error) {
                    callback(error);
                } else {
                    let group = {};
                    iotAgentLib.getConfigurationSilently(
                        config.getConfig().iota.defaultResource || '',
                        apiKey,
                        function (error, foundGroup) {
                            if (!error) {
                                group = foundGroup;
                            }
                            async.series(
                                attributes
                                    .map(generateCommandExecution.bind(null, apiKey, device, group))
                                    .reduce(concat, []),
                                callback
                            );
                        }
                    );
                }
            });
        }
    });
}

/**
 * Process an update in the state of a command with information coming from the device.
 *
 * @param {String} apiKey           API Key corresponding to the Devices configuration.
 * @param {String} deviceId         Id of the device to be updated.
 * @param {Object} device           Device object containing all the information about a device.
 * @param {Object} messageObj       JSON object sent using MQTT.
 */
function updateCommand(apiKey, deviceId, device, messageObj) {
    const commandList = Object.keys(messageObj);
    const commandUpdates = [];

    for (let i = 0; i < commandList.length; i++) {
        commandUpdates.push(
            async.apply(
                iotAgentLib.setCommandResult,
                device.name,
                config.getConfig().iota.defaultResource,
                apiKey,
                commandList[i],
                messageObj[commandList[i]],
                constants.COMMAND_STATUS_COMPLETED,
                device
            )
        );
    }

    async.series(commandUpdates, function (error) {
        if (error) {
            config.getLogger().error(
                context,

                "COMMANDS-002: Couldn't update command status in the Context broker " +
                    'for device %s with apiKey %s: %s',
                device.id,
                apiKey,
                error
            );
        } else {
            config
                .getLogger()
                .debug(context, 'Single measure for device %s with apiKey %s successfully updated', device.id, apiKey);
        }
    });
}

exports.generateCommandExecution = generateCommandExecution;
exports.updateCommand = updateCommand;
exports.handler = commandHandler;

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	const async = require('async');	1✔
25	const iotAgentLib = require('iotagent-node-lib');	1✔
26	const iotaUtils = require('./iotaUtils');	1✔
27	const constants = require('./constants');	1✔
28	const transportSelector = require('./transportSelector');	1✔
29	const config = require('./configService');	1✔
30	const context = {	1✔
31	op: 'IoTAgentJSON.Commands'
32	};
33
34	/**
35	* Serializes a payload for a command depending on its payloadType if provided
36	*
37	* @param {String} payload Payload to serialized
38	* @param {Object} command Command attribute
39	* @return {Function} Returns a serialized payload
40	*/
41	function serializedPayloadCommand(payload, command) {
42	let serialized;
43	if (command && command.payloadType) {	27✔
44	switch (command.payloadType.toLowerCase()) {	8!
45	case 'binaryfromstring':
46	serialized = Buffer.from(payload.toString());	×
47	break;	×
48	case 'binaryfromhex':
49	serialized = Buffer.from(payload, 'HEX');	3✔
50	break;	3✔
51	case 'binaryfromjson': // used by AMQP transport
52	serialized = Buffer.from(JSON.stringify(payload));	4✔
53	break;	4✔
54	case 'text':
55	serialized = payload;	1✔
56	break;	1✔
57	case 'json': // passthrough
58	default:
59	serialized = JSON.stringify(payload);	×
60	}
61	} else {
62	serialized = JSON.stringify(payload);	19✔
63	}
64	return serialized;	27✔
65	}
66
67	/**
68	* Generate a function that executes the given command in the device.
69	*
70	* @param {String} apiKey APIKey of the device's service or default APIKey.
71	* @param {Object} device Object containing all the information about a device.
72	* @param {Object} attribute Attribute in NGSI format.
73	* @return {Function} Command execution function ready to be called with async.series.
74	*/
75	function generateCommandExecution(apiKey, device, group, attribute) {
76	let payload = {};	27✔
77	const command = device && device.commands.find((att) => att.name === attribute.name);	39✔
78
79	if (command && command.expression) {	27✔
80	const parser = iotAgentLib.dataPlugins.expressionTransformation;	13✔
81	// The context for the JEXL expression should be the ID, TYPE, S, SS
82	let attrList = iotAgentLib.dataPlugins.utils.getIdTypeServSubServiceFromDevice(device);	13✔
83	attrList = device.staticAttributes	13!
84	? attrList.concat(device.staticAttributes).concat(attribute)
85	: attrList.concat(attribute);
86	const ctxt = parser.extractContext(attrList, device);	13✔
87	// expression result will be the full command payload
88	let payloadRes = null;	13✔
89	try {	13✔
90	payloadRes = parser.applyExpression(command.expression, ctxt, device);	13✔
91	} catch (e) {
92	// nothing to report
93	}
94	payload = payloadRes ? payloadRes : command.expression;	13!
95	} else {
96	payload[attribute.name] = attribute.value;	14✔
97	}
98	if (device.transport === 'AMQP') {	27✔
99	// to ensure backward compability
100	command.payloadType = command.payloadType ? command.payloadType : 'binaryfromjson';	4!
101	}
102	const serialized = serializedPayloadCommand(payload, command);	27✔
103	const contentType = command && command.contentType ? command.contentType : 'application/json';	27!
104	const extraHeaders = command && command.headers ? command.headers : undefined;	27✔
105	config	27✔
106	.getLogger()
107	.debug(
108	context,
109	'Sending command execution to device %s with apikey %s and payload %j extraHeaders %j',
110	device.id,
111	apiKey,
112	payload,
113	extraHeaders
114	);
115	const executions = transportSelector.createExecutionsForBinding(	27✔
116	[apiKey, group, device, attribute.name, serialized, extraHeaders, contentType],
117	'executeCommand',
118	device.transport \|\|	31!
119	(group && group.transport ? group.transport : undefined) \|\|	12!
120	config.getConfig().defaultTransport
121	);
122	return executions;	27✔
123	}
124
125	/**
126	* Handles a command execution request coming from the Context Broker. This handler should:
127	* - Identify the device affected by the command.
128	* - Send the command to the appropriate MQTT topic.
129	* - Update the command status in the Context Broker.
130	*
131	* @param {String} id ID of the entity for which the command execution was issued.
132	* @param {String} type Type of the entity for which the command execution was issued.
133	* @param {String} service Service ID.
134	* @param {String} subservice Subservice ID.
135	* @param {Array} attributes List of NGSI attributes of type command to execute.
136	*/
137	function commandHandler(id, type, service, subservice, attributes, callback) {
138	config	27✔
139	.getLogger()
140	.debug(context, 'Handling command %j for device %s in service %s - %s', attributes, id, service, subservice);
141
142	function concat(previous, current) {
143	previous = previous.concat(current);	27✔
144	return previous;	27✔
145	}
146
147	iotAgentLib.getDeviceByNameAndType(id, type, service, subservice, function (error, device) {	27✔
148	if (error) {	27!
149	config.getLogger().error(	×
150	context,
151
152	"COMMAND-001: Command execution could not be handled, as device for entity %s %s wasn't found",
153
154	id,
155	type
156	);
157	callback(error);	×
158	} else {
159	iotaUtils.getEffectiveApiKey(device.service, device.subservice, device, function (error, apiKey) {	27✔
160	if (error) {	27!
161	callback(error);	×
162	} else {
163	let group = {};	27✔
164	iotAgentLib.getConfigurationSilently(	27✔
165	config.getConfig().iota.defaultResource \|\| '',	27!
166	apiKey,
167	function (error, foundGroup) {
168	if (!error) {	27✔
169	group = foundGroup;	5✔
170	}
171	async.series(	27✔
172	attributes
173	.map(generateCommandExecution.bind(null, apiKey, device, group))
174	.reduce(concat, []),
175	callback
176	);
177	}
178	);
179	}
180	});
181	}
182	});
183	}
184
185	/**
186	* Process an update in the state of a command with information coming from the device.
187	*
188	* @param {String} apiKey API Key corresponding to the Devices configuration.
189	* @param {String} deviceId Id of the device to be updated.
190	* @param {Object} device Device object containing all the information about a device.
191	* @param {Object} messageObj JSON object sent using MQTT.
192	*/
193	function updateCommand(apiKey, deviceId, device, messageObj) {
194	const commandList = Object.keys(messageObj);	14✔
195	const commandUpdates = [];	14✔
196
197	for (let i = 0; i < commandList.length; i++) {	14✔
198	commandUpdates.push(	14✔
199	async.apply(
200	iotAgentLib.setCommandResult,
201	device.name,
202	config.getConfig().iota.defaultResource,
203	apiKey,
204	commandList[i],
205	messageObj[commandList[i]],
206	constants.COMMAND_STATUS_COMPLETED,
207	device
208	)
209	);
210	}
211
212	async.series(commandUpdates, function (error) {	14✔
213	if (error) {	14!
214	config.getLogger().error(	×
215	context,
216
217	"COMMANDS-002: Couldn't update command status in the Context broker " +
218	'for device %s with apiKey %s: %s',
219	device.id,
220	apiKey,
221	error
222	);
223	} else {
224	config	14✔
225	.getLogger()
226	.debug(context, 'Single measure for device %s with apiKey %s successfully updated', device.id, apiKey);
227	}
228	});
229	}
230
231	exports.generateCommandExecution = generateCommandExecution;	1✔
232	exports.updateCommand = updateCommand;	1✔
233	exports.handler = commandHandler;	1✔

telefonicaid / iotagent-json / 15172033588

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