18558650619

Committed 16 Oct 2025 10:43AM UTC coverage: 78.143% (+0.1%) from 78.011%

Build # 18558650619

Build Type

Pull #817

github

Committed by

web-flow

Commit Message

Merge f387d3457 into a71a30851

Pull Request Pull Request #817: [WIP] upgrade dep driver mongodb to 4.17.2

Run Details

797 of 1149 branches covered (69.36%)

Branch coverage included in aggregate %.

107 of 123 new or added lines in 6 files covered. (86.99%)

1 existing line in 1 file now uncovered.

2242 of 2740 relevant lines covered (81.82%)

71.28 hits per line

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

59.85

/lib/models/entitiesStore.js

/*
 * Copyright 2015 Telefonica Investigación y Desarrollo, S.A.U
 *
 * This file is part of perseo-fe
 *
 * perseo-fe 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.
 *
 * perseo-fe 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 perseo-fe.
 * If not, see http://www.gnu.org/licenses/.
 *
 * For those usages not covered by the GNU Affero General Public License
 * please contact with::[contacto@tid.es]
 */
'use strict';

var appContext = require('../appContext'),
    config = require('../../config'),
    entitiesCollectionName = require('../../config').orionDb.collection,
    myutils = require('../myutils'),
    constants = require('../constants'),
    logger = require('logops'),
    ngsi = require('ngsijs'),
    context = { op: 'entitiesStore', comp: constants.COMPONENT_NAME };

function orionServiceDb(service) {
    return appContext.OrionDb(config.orionDb.prefix + '-' + service);
}

function createFilter(ruleData, service, subservice, limit, offset) {
    var maxTimeDetection = ruleData.maxTimeDetection ? ruleData.maxTimeDetection : config.nonSignalMaxTimeDetection;
    var currentTime = Date.now() / 1000;
    var filter = {
        service: service,
        servicepath: subservice,
        type: ruleData.type,
        mq:
            ruleData.attribute +
            '.dateModified<' +
            (currentTime - ruleData.reportInterval).toString() +
            ';' +
            'temperature.dateModified>' +
            (currentTime / 1000 - maxTimeDetection).toString(),
        // TO DO: Use ruleData.maxTimeDetectionAttr and ruleData.reportIntervalAttr in query if possible (#824)
        // https://github.com/telefonicaid/fiware-orion/blob/master/doc/manuals/orion-api.md#simple-query-language
        // https://ficodes.github.io/ngsijs/stable/NGSI.Connection.html#.%22v2.listEntities%22
        limit: limit,
        offset: offset
    };
    if (ruleData.id) {
        filter.id = ruleData.id;
    } else if (ruleData.idRegexp) {
        filter.idPattern = ruleData.idRegexp;
    }
    return filter;
}

function createConnection(service, subservice) {
    var options = {
        service: service,
        servicepath: subservice
    };
    options.headers = {};
    // Add correlator
    var domain = process.domain;
    if (domain && domain.context) {
        options.headers[constants.CORRELATOR_HEADER] = domain.context.corr;
        // Add other headers
        if (domain.context.srv && options.headers[constants.SERVICE_HEADER] === undefined) {
            options.headers[constants.SERVICE_HEADER] = domain.context.srv;
        }
        if (domain.context.subsrv && options.headers[constants.SUBSERVICE_HEADER] === undefined) {
            options.headers[constants.SUBSERVICE_HEADER] = domain.context.subsrv;
        }
        if (domain.context.from && options.headers[constants.REALIP_HEADER] === undefined) {
            options.headers[constants.REALIP_HEADER] = domain.context.from;
        }
    }

    return new ngsi.Connection(config.orion.URL, options);
}

function findSilentEntitiesByAPIWithPagination(connection, filter, alterFunc, accumulatedResults, callback) {
    connection.v2.listEntities(filter).then(
        (response) => {
            if (response.results) {
                response.results.forEach((entity) => {
                    logger.debug(context, 'silent entity %j', entity);
                    alterFunc(entity);
                    accumulatedResults.push(entity);
                });
            }
            logger.debug(context, 'findSilentEntities %s', myutils.firstChars(response.results));
            if (response.count > filter.limit + filter.offset) {
                filter.offset += filter.limit;
                findSilentEntitiesByAPIWithPagination(connection, filter, alterFunc, accumulatedResults, callback);
            } else {
                // Pass the count of entities to the callback
                callback(null, accumulatedResults, accumulatedResults.length);
            }
        },
        (error) => {
            logger.warn(context, ' error v2.listEntities: %j trying list entities using filter %j', error, filter);
            callback(error, null, null);
        }
    );
}

function findSilentEntitiesByAPI(service, subservice, ruleData, alterFunc, callback) {
    var limit = 20;
    var offset = 0;
    var connection = createConnection(service, subservice);
    var filter = createFilter(ruleData, service, subservice, limit, offset);

    logger.info(
        context,
        ' find silent entities by API ngsi using options %j and filter %j and rule %j',
        connection,
        filter,
        ruleData
    );

    var accumulatedResults = [];

    findSilentEntitiesByAPIWithPagination(connection, filter, alterFunc, accumulatedResults, callback);
}

function findSilentEntitiesByMongo(service, subservice, ruleData, alterFunc, callback) {
    var db,
        criterion = {};

    var cb = function(err, result, entityCount) {
        logger.debug(context, 'findSilentEntitiesByMongo %s', myutils.firstChars(result));
        return callback(err, result, entityCount);
    };
    db = orionServiceDb(service);
    criterion['_id.servicePath'] = subservice;
    if (ruleData.id) {
        criterion['_id.id'] = ruleData.id;
    } else if (ruleData.idRegexp) {
        try {
            criterion['_id.id'] = new RegExp(ruleData.idRegexp);
        } catch (e) {
            return callback(e, null);
        }
    }
    if (ruleData.type) {
        criterion['_id.type'] = ruleData.type;
    }

    logger.debug(context, 'findSilentEntities criterion %j', criterion);

    myutils.collectionExists(db, entitiesCollectionName, function(exists) {
        if (!exists) {
            return cb('collection ' + entitiesCollectionName + ' does not exist');
        }

        var col = db.collection(entitiesCollectionName);

        // Variable to store the count of entities
        var entityCount = 0;

        var pipeline = [
            {
                $match: criterion
            },
            {
                $addFields: {
                    currentTime: { $divide: [Date.now(), 1000] },
                    maxTimeDetection: {
                        $convert: {
                            input: {
                                $ifNull: [
                                    '$attrs.' + ruleData.maxTimeDetectionAttr + '.value',
                                    ruleData.maxTimeDetection !== undefined
                                        ? ruleData.maxTimeDetection // jshint ignore: line
                                        : config.nonSignalMaxTimeDetection
                                ]
                            },
                            to: 'double',
                            onError: config.nonSignalMaxTimeDetection,
                            onNull: config.nonSignalMaxTimeDetection
                        }
                    },
                    reportInterval: {
                        $convert: {
                            input: {
                                $ifNull: ['$attrs.' + ruleData.reportIntervalAttr + '.value', ruleData.reportInterval]
                            },
                            to: 'double',
                            onError: ruleData.reportInterval,
                            onNull: ruleData.reportInterval
                        }
                    }
                }
            },
            {
                $match: {
                    $expr: {
                        $and: [
                            {
                                $lt: [
                                    '$attrs.' + ruleData.attribute + '.modDate',
                                    { $subtract: ['$currentTime', '$reportInterval'] }
                                ]
                            },
                            {
                                $gt: [
                                    '$attrs.' + ruleData.attribute + '.modDate',
                                    { $subtract: ['$currentTime', '$maxTimeDetection'] }
                                ]
                            },
                            {
                                $gt: ['$attrs.' + ruleData.attribute + '.modDate', 0]
                            } // exclude invalid dates
                        ]
                    }
                }
            }
        ];

        logger.debug(context, 'findSilentEntities service %s pipeline: %j ', service, pipeline);
        var cursor = col.aggregate(pipeline);

        cursor.toArray(function(err, results) {
            //myutils.logErrorIf(err);
            if (err) {
                return cb(err, null);
            }
            results.forEach(function(one) {
                logger.debug(context, 'silent entity %j', one._id);
                alterFunc(one);
                // Increment the count of entities
                entityCount++;
            });
            //cb(null, 'silent ones count ' + entityCount);
            return cb(null, results, entityCount);
        });
    });
}

//         function(err, result) {
//             logger.debug(context, 'findSilentEntities %s entityCount %j', myutils.firstChars(result), entityCount);
//             return callback(err, result, entityCount);

//         }
//     );
// }
//         var col = db.collection(entitiesCollectionName);

//         var entityCount = 0;
//         var results = [];
//         col.find(criterion)
//             .batchSize(config.orionDb.batchSize)
//             .forEach(
//                 function(one) {
//                     if (one === null) {
//                         // Cursor exhausted
//                         return cb(null, 'silent ones count ' + entityCount);
//                     }
//                     logger.debug(context, 'silent entity %j', one._id);
//                     alterFunc(one);
//                     // Increment the count of entities
//                     entityCount++;
//                     results.push(one);
//                 },
//                 function(err) {
//                     if (err) {
//                         return cb(err, null);
//                     } else {
//                         logger.debug(
//                             context,
//                             'findSilentEntities %s entityCount %j',
//                             myutils.firstChars(results),
//                             entityCount
//                         );
//                         return cb(null, results, entityCount);
//                     }
//                 }
//             );
//     });
// }

function findSilentEntities(service, subservice, ruleData, alterFunc, callback) {
    var hrstart = process.hrtime();
    var method = !config.nonSignalByAPI ? 'findSilentEntitiesByMongo' : 'findSilentEntitiesByAPI';

    var timedCallback = function(err, result, entityCount) {
        var hrend = process.hrtime(hrstart);

        logger.info(
            context,
            ' %s has found %d entities in (hr): %d ms result %j',
            method,
            entityCount,
            hrend[1] / 1000000,
            result
        );

        callback(err, result, entityCount);
    };
    if (!config.nonSignalByAPI) {
        return findSilentEntitiesByMongo(service, subservice, ruleData, alterFunc, timedCallback);
    } else {
        return findSilentEntitiesByAPI(service, subservice, ruleData, alterFunc, timedCallback);
    }
}

module.exports.FindSilentEntities = findSilentEntities;
module.exports.findSilentEntitiesByAPI = findSilentEntitiesByAPI;
module.exports.findSilentEntitiesByMongo = findSilentEntitiesByMongo;
module.exports.findSilentEntitiesByAPIWithPagination = findSilentEntitiesByAPIWithPagination;
module.exports.createConnection = createConnection;
module.exports.createFilter = createFilter;

1	/*
2	* Copyright 2015 Telefonica Investigación y Desarrollo, S.A.U
3	*
4	* This file is part of perseo-fe
5	*
6	* perseo-fe 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	* perseo-fe 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 perseo-fe.
18	* If not, see http://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	'use strict';
24
25	var appContext = require('../appContext'),	2✔
26	config = require('../../config'),	2✔
27	entitiesCollectionName = require('../../config').orionDb.collection,	2✔
28	myutils = require('../myutils'),	2✔
29	constants = require('../constants'),	2✔
30	logger = require('logops'),	2✔
31	ngsi = require('ngsijs'),	2✔
32	context = { op: 'entitiesStore', comp: constants.COMPONENT_NAME };	2✔
33
34	function orionServiceDb(service) {
35	return appContext.OrionDb(config.orionDb.prefix + '-' + service);	1✔
36	}
37
38	function createFilter(ruleData, service, subservice, limit, offset) {
39	var maxTimeDetection = ruleData.maxTimeDetection ? ruleData.maxTimeDetection : config.nonSignalMaxTimeDetection;	×
40	var currentTime = Date.now() / 1000;	×
41	var filter = {	×
42	service: service,
43	servicepath: subservice,
44	type: ruleData.type,
45	mq:
46	ruleData.attribute +
47	'.dateModified<' +
48	(currentTime - ruleData.reportInterval).toString() +
49	';' +
50	'temperature.dateModified>' +
51	(currentTime / 1000 - maxTimeDetection).toString(),
52	// TO DO: Use ruleData.maxTimeDetectionAttr and ruleData.reportIntervalAttr in query if possible (#824)
53	// https://github.com/telefonicaid/fiware-orion/blob/master/doc/manuals/orion-api.md#simple-query-language
54	// https://ficodes.github.io/ngsijs/stable/NGSI.Connection.html#.%22v2.listEntities%22
55	limit: limit,
56	offset: offset
57	};
58	if (ruleData.id) {	×
59	filter.id = ruleData.id;	×
60	} else if (ruleData.idRegexp) {	×
61	filter.idPattern = ruleData.idRegexp;	×
62	}
63	return filter;	×
64	}
65
66	function createConnection(service, subservice) {
67	var options = {	×
68	service: service,
69	servicepath: subservice
70	};
71	options.headers = {};	×
72	// Add correlator
73	var domain = process.domain;	×
74	if (domain && domain.context) {	×
75	options.headers[constants.CORRELATOR_HEADER] = domain.context.corr;	×
76	// Add other headers
77	if (domain.context.srv && options.headers[constants.SERVICE_HEADER] === undefined) {	×
78	options.headers[constants.SERVICE_HEADER] = domain.context.srv;	×
79	}
80	if (domain.context.subsrv && options.headers[constants.SUBSERVICE_HEADER] === undefined) {	×
81	options.headers[constants.SUBSERVICE_HEADER] = domain.context.subsrv;	×
82	}
83	if (domain.context.from && options.headers[constants.REALIP_HEADER] === undefined) {	×
84	options.headers[constants.REALIP_HEADER] = domain.context.from;	×
85	}
86	}
87
88	return new ngsi.Connection(config.orion.URL, options);	×
89	}
90
91	function findSilentEntitiesByAPIWithPagination(connection, filter, alterFunc, accumulatedResults, callback) {
92	connection.v2.listEntities(filter).then(	2✔
93	(response) => {
94	if (response.results) {	1!
95	response.results.forEach((entity) => {	1✔
96	logger.debug(context, 'silent entity %j', entity);	2✔
97	alterFunc(entity);	2✔
98	accumulatedResults.push(entity);	2✔
99	});
100	}
101	logger.debug(context, 'findSilentEntities %s', myutils.firstChars(response.results));	1✔
102	if (response.count > filter.limit + filter.offset) {	1!
103	filter.offset += filter.limit;	1✔
104	findSilentEntitiesByAPIWithPagination(connection, filter, alterFunc, accumulatedResults, callback);	1✔
105	} else {
106	// Pass the count of entities to the callback
107	callback(null, accumulatedResults, accumulatedResults.length);	×
108	}
109	},
110	(error) => {
111	logger.warn(context, ' error v2.listEntities: %j trying list entities using filter %j', error, filter);	1✔
112	callback(error, null, null);	1✔
113	}
114	);
115	}
116
117	function findSilentEntitiesByAPI(service, subservice, ruleData, alterFunc, callback) {
118	var limit = 20;	1✔
119	var offset = 0;	1✔
120	var connection = createConnection(service, subservice);	1✔
121	var filter = createFilter(ruleData, service, subservice, limit, offset);	1✔
122
123	logger.info(	1✔
124	context,
125	' find silent entities by API ngsi using options %j and filter %j and rule %j',
126	connection,
127	filter,
128	ruleData
129	);
130
131	var accumulatedResults = [];	1✔
132
133	findSilentEntitiesByAPIWithPagination(connection, filter, alterFunc, accumulatedResults, callback);	1✔
134	}
135
136	function findSilentEntitiesByMongo(service, subservice, ruleData, alterFunc, callback) {
137	var db,
138	criterion = {};	3✔
139
140	var cb = function(err, result, entityCount) {	3✔
141	logger.debug(context, 'findSilentEntitiesByMongo %s', myutils.firstChars(result));	1✔
142	return callback(err, result, entityCount);	1✔
143	};
144	db = orionServiceDb(service);	3✔
145	criterion['_id.servicePath'] = subservice;	3✔
146	if (ruleData.id) {	3!
147	criterion['_id.id'] = ruleData.id;	×
148	} else if (ruleData.idRegexp) {	3✔
149	try {	1✔
150	criterion['_id.id'] = new RegExp(ruleData.idRegexp);	1✔
151	} catch (e) {
152	return callback(e, null);	×
153	}
154	}
155	if (ruleData.type) {	3!
156	criterion['_id.type'] = ruleData.type;	×
157	}
158
159	logger.debug(context, 'findSilentEntities criterion %j', criterion);	3✔
160
161	myutils.collectionExists(db, entitiesCollectionName, function(exists) {	3✔
162	if (!exists) {	1!
NEW 163	return cb('collection ' + entitiesCollectionName + ' does not exist');	×
164	}
165
166	var col = db.collection(entitiesCollectionName);	1✔
167
168	// Variable to store the count of entities
169	var entityCount = 0;	1✔
170
171	var pipeline = [	1✔
172	{
173	$match: criterion
174	},
175	{
176	$addFields: {
177	currentTime: { $divide: [Date.now(), 1000] },
178	maxTimeDetection: {
179	$convert: {
180	input: {
181	$ifNull: [
182	'$attrs.' + ruleData.maxTimeDetectionAttr + '.value',
183	ruleData.maxTimeDetection !== undefined	1!
184	? ruleData.maxTimeDetection // jshint ignore: line
185	: config.nonSignalMaxTimeDetection
186	]
187	},
188	to: 'double',
189	onError: config.nonSignalMaxTimeDetection,
190	onNull: config.nonSignalMaxTimeDetection
191	}
192	},
193	reportInterval: {
194	$convert: {
195	input: {
196	$ifNull: ['$attrs.' + ruleData.reportIntervalAttr + '.value', ruleData.reportInterval]
197	},
198	to: 'double',
199	onError: ruleData.reportInterval,
200	onNull: ruleData.reportInterval
201	}
202	}
203	}
204	},
205	{
206	$match: {
207	$expr: {
208	$and: [
209	{
210	$lt: [
211	'$attrs.' + ruleData.attribute + '.modDate',
212	{ $subtract: ['$currentTime', '$reportInterval'] }
213	]
214	},
215	{
216	$gt: [
217	'$attrs.' + ruleData.attribute + '.modDate',
218	{ $subtract: ['$currentTime', '$maxTimeDetection'] }
219	]
220	},
221	{
222	$gt: ['$attrs.' + ruleData.attribute + '.modDate', 0]
223	} // exclude invalid dates
224	]
225	}
226	}
227	}
228	];
229
230	logger.debug(context, 'findSilentEntities service %s pipeline: %j ', service, pipeline);	1✔
231	var cursor = col.aggregate(pipeline);	1✔
232
233	cursor.toArray(function(err, results) {	1✔
234	//myutils.logErrorIf(err);
235	if (err) {	1!
NEW 236	return cb(err, null);	×
237	}
238	results.forEach(function(one) {	1✔
239	logger.debug(context, 'silent entity %j', one._id);	2✔
240	alterFunc(one);	2✔
241	// Increment the count of entities
242	entityCount++;	2✔
243	});
244	//cb(null, 'silent ones count ' + entityCount);
245	return cb(null, results, entityCount);	1✔
246	});
247	});
248	}
249
250	// function(err, result) {
251	// logger.debug(context, 'findSilentEntities %s entityCount %j', myutils.firstChars(result), entityCount);
252	// return callback(err, result, entityCount);
253
254	// }
255	// );
256	// }
257	// var col = db.collection(entitiesCollectionName);
258
259	// var entityCount = 0;
260	// var results = [];
261	// col.find(criterion)
262	// .batchSize(config.orionDb.batchSize)
263	// .forEach(
264	// function(one) {
265	// if (one === null) {
266	// // Cursor exhausted
267	// return cb(null, 'silent ones count ' + entityCount);
268	// }
269	// logger.debug(context, 'silent entity %j', one._id);
270	// alterFunc(one);
271	// // Increment the count of entities
272	// entityCount++;
273	// results.push(one);
274	// },
275	// function(err) {
276	// if (err) {
277	// return cb(err, null);
278	// } else {
279	// logger.debug(
280	// context,
281	// 'findSilentEntities %s entityCount %j',
282	// myutils.firstChars(results),
283	// entityCount
284	// );
285	// return cb(null, results, entityCount);
286	// }
287	// }
288	// );
289	// });
290	// }
291
292	function findSilentEntities(service, subservice, ruleData, alterFunc, callback) {
293	var hrstart = process.hrtime();	3✔
294	var method = !config.nonSignalByAPI ? 'findSilentEntitiesByMongo' : 'findSilentEntitiesByAPI';	3✔
295
296	var timedCallback = function(err, result, entityCount) {	3✔
297	var hrend = process.hrtime(hrstart);	1✔
298
299	logger.info(	1✔
300	context,
301	' %s has found %d entities in (hr): %d ms result %j',
302	method,
303	entityCount,
304	hrend[1] / 1000000,
305	result
306	);
307
308	callback(err, result, entityCount);	1✔
309	};
310	if (!config.nonSignalByAPI) {	3✔
311	return findSilentEntitiesByMongo(service, subservice, ruleData, alterFunc, timedCallback);	2✔
312	} else {
313	return findSilentEntitiesByAPI(service, subservice, ruleData, alterFunc, timedCallback);	1✔
314	}
315	}
316
317	module.exports.FindSilentEntities = findSilentEntities;	2✔
318	module.exports.findSilentEntitiesByAPI = findSilentEntitiesByAPI;	2✔
319	module.exports.findSilentEntitiesByMongo = findSilentEntitiesByMongo;	2✔
320	module.exports.findSilentEntitiesByAPIWithPagination = findSilentEntitiesByAPIWithPagination;	2✔
321	module.exports.createConnection = createConnection;	2✔
322	module.exports.createFilter = createFilter;	2✔

telefonicaid / perseo-fe / 18558650619

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