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

'use strict';

var async = require('async'),
    apply = async.apply,
    objectRegistry = require('./objectRegistry'),
    errors = require('../../errors'),
    logger = require('logops'),
    _ = require('underscore'),
    observers = {};

/**
 * Extract Object type and id from the request URI, returning it using the callback.
 *
 * @param {Object} req          Arriving COAP Request to be handled.
 * @param {Object} res          Outgoing COAP Response.
 */
function extractUriInfo(req, res, callback) {
    if (req.urlObj.pathname.match(/\/\d+\/\d+\/\d+/)) {
        var resourceIndex = req.urlObj.pathname.lastIndexOf('/'),
            resourceId = req.urlObj.pathname.substring(resourceIndex + 1),
            objectUri = req.urlObj.pathname.substring(0, resourceIndex);

        callback(null, objectUri, resourceId);
    } else if (req.urlObj.pathname.match(/\/\d+(\/\d+)?/)) {
        callback(null, req.urlObj.pathname, null);
    } else {
        callback(new errors.WrongObjectUri(req.urlObj.pathname));
    }
}

/**
 * Invoke the user handler for this operation, with all the information from the query parameters as its arguments.
 * This method handling gives the client an opportunity to change or overwrite the values before returning them to
 * the client. The handler must call the callback with the overwritten value it wants to return. If the callback is
 * called without parameters, the original value is returned instead.
 *
 * @param {Object} queryParams      Object containing all the query parameters.
 * @param {Function} handler        User handler to be invoked.
 */
function applyHandler(resourceId, handler, storedObject, callback) {
    if (storedObject.attributes[resourceId]) {
        handler(
            storedObject.objectType,
            storedObject.objectId,
            resourceId,
            storedObject.attributes[resourceId],

            function handleReadOverwrite(error, result) {
                if (result) {
                    callback(error, result);
                } else {
                    callback(error, storedObject.attributes[resourceId]);
                }
            });
    } else {
        callback(new errors.ResourceNotFound(storedObject.objectType, storedObject.objectId, resourceId));
    }
}

/**
 * Generates the end of request handler that will generate the final response to the COAP Client.
 *
 * @param {Object} req          Arriving COAP Request to be handled.
 * @param {Object} res          Outgoing COAP Response.
 * @returns {Function}          Request handler, receiving an optional error and the read operation result.
 */
function endRead(req, res) {
    return function (error, result) {
        var body;

        if (error) {
            res.code = error.code;
            body = '';
        } else {
            res.code = '2.05';
            body = result.toString();
        }

        if (_.some(req.options, _.matches({name: 'Observe'}))) {
            res.setOption('Observe', 1);
            res.write(body);
        } else {
            res.end(body);
        }
    };
}

/**
 * Constructs the resource observer. The observer listens in the object registry bus for attribute modifications,
 * sending a new information update to the server each time the selected attribute changes its value.
 *
 * @param {Object} stream           Stream where the new data will be written to the server.
 * @param {Number} resourceId       Resource to read from.
 * @param {Object} storedObject     Object instance to observe.
 * @returns {{uri: (object.objectUri|*|string), stream: *, observation: number}}
 * @constructor
 */
function ResourceObserver(stream, resourceId, storedObject) {
    var data = {
        id: storedObject.objectUri + '/' + resourceId,
        uri: storedObject.objectUri,
        stream: stream,
        observation: 1,
        listener: function newModification(method, id, value) {
            if (method === 'setResource' && resourceId === id.toString()) {
                logger.debug('Sending data update to the server for object [%s]', data.objectUri);

                try {
                    stream.setOption('Observe', data.observation);
                    stream.write(value.toString());
                    data.observation++;
                    data.lastObservation = Date.now();
                } catch(error) {
                    logger.error('Tried to write to a closed stream. Cleaning the handler.');
                    clearInterval(data.scheduler);
                }
            }
        }
    };

    function getLastMeasure() {
        objectRegistry.get(storedObject.objectUri, function (error, obj) {
            if (error) {
                logger.error('Error retrieving the last resource value');
            } else {
                data.listener('setResource', resourceId, obj.attributes[resourceId]);
            }
        });
    }

    objectRegistry.getAttributes(storedObject.objectUri + '/' + resourceId, function(error, attributes) {
        if (error) {
            logger.error('Couldn\'t find attributes for the [%s] resource URI. Falling back to value changed');
        }

        if (attributes) {
            if (attributes.pmax) {
                logger.debug('Creating a scheduled observation for [%d] miliseconds for object [%s]',
                    attributes.pmax, storedObject.objectUri);

                data.scheduler = setInterval(function handleMaxPeriod() {
                    getLastMeasure(storedObject.objectUri, resourceId);
                }, attributes.pmax);<commit_msg>FIX Cancel observation on transmission error<commit_after>/*
 * Copyright 2014 Telefonica Investigación y Desarrollo, S.A.U
 *
 * This file is part of iotagent-lwm2m-lib
 *
 * iotagent-lwm2m-lib 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-lwm2m-lib 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-lwm2m-lib.
 * If not, seehttp://www.gnu.org/licenses/.
 *
 * For those usages not covered by the GNU Affero General Public License
 * please contact with::[<EMAIL>]
 */

'use strict';

var async = require('async'),
    apply = async.apply,
    objectRegistry = require('./objectRegistry'),
    errors = require('../../errors'),
    logger = require('logops'),
    _ = require('underscore'),
    observers = {};

/**
 * Extract Object type and id from the request URI, returning it using the callback.
 *
 * @param {Object} req          Arriving COAP Request to be handled.
 * @param {Object} res          Outgoing COAP Response.
 */
function extractUriInfo(req, res, callback) {
    if (req.urlObj.pathname.match(/\/\d+\/\d+\/\d+/)) {
        var resourceIndex = req.urlObj.pathname.lastIndexOf('/'),
            resourceId = req.urlObj.pathname.substring(resourceIndex + 1),
            objectUri = req.urlObj.pathname.substring(0, resourceIndex);

        callback(null, objectUri, resourceId);
    } else if (req.urlObj.pathname.match(/\/\d+(\/\d+)?/)) {
        callback(null, req.urlObj.pathname, null);
    } else {
        callback(new errors.WrongObjectUri(req.urlObj.pathname));
    }
}

/**
 * Invoke the user handler for this operation, with all the information from the query parameters as its arguments.
 * This method handling gives the client an opportunity to change or overwrite the values before returning them to
 * the client. The handler must call the callback with the overwritten value it wants to return. If the callback is
 * called without parameters, the original value is returned instead.
 *
 * @param {Object} queryParams      Object containing all the query parameters.
 * @param {Function} handler        User handler to be invoked.
 */
function applyHandler(resourceId, handler, storedObject, callback) {
    if (storedObject.attributes[resourceId]) {
        handler(
            storedObject.objectType,
            storedObject.objectId,
            resourceId,
            storedObject.attributes[resourceId],

            function handleReadOverwrite(error, result) {
                if (result) {
                    callback(error, result);
                } else {
                    callback(error, storedObject.attributes[resourceId]);
                }
            });
    } else {
        callback(new errors.ResourceNotFound(storedObject.objectType, storedObject.objectId, resourceId));
    }
}

/**
 * Generates the end of request handler that will generate the final response to the COAP Client.
 *
 * @param {Object} req          Arriving COAP Request to be handled.
 * @param {Object} res          Outgoing COAP Response.
 * @returns {Function}          Request handler, receiving an optional error and the read operation result.
 */
function endRead(req, res) {
    return function (error, result) {
        var body;

        if (error) {
            res.code = error.code;
            body = '';
        } else {
            res.code = '2.05';
            body = result.toString();
        }

        if (_.some(req.options, _.matches({name: 'Observe'}))) {
            res.setOption('Observe', 1);
            res.write(body);
        } else {
            res.end(body);
        }
    };
}

/**
 * Constructs the resource observer. The observer listens in the object registry bus for attribute modifications,
 * sending a new information update to the server each time the selected attribute changes its value.
 *
 * @param {Object} stream           Stream where the new data will be written to the server.
 * @param {Number} resourceId       Resource to read from.
 * @param {Object} storedObject     Object instance to observe.
 * @returns {{uri: (object.objectUri|*|string), stream: *, observation: number}}
 * @constructor
 */
function ResourceObserver(stream, resourceId, storedObject) {
    var data = {
        id: storedObject.objectUri + '/' + resourceId,
        uri: storedObject.objectUri,
        stream: stream,
        observation: 1,
        listener: function newModification(method, id, value) {
            if (method === 'setResource' && resourceId === id.toString()) {
                logger.debug('Sending data update to the server for object [%s]', data.objectUri);

                try {
                    stream.setOption('Observe', data.observation);
                    stream.write(value.toString());
                    data.observation++;
                    data.lastObservation = Date.now();
                } catch(error) {
                    logger.error('Tried to write to a closed stream. Cleaning the handler.');
                    cancelObservation(storedObject.objectUri, resourceId, function() {
                        logger.info('Observation of [%s/%s] cancelled', storedObject.objectUri, resourceId);
                    });
                }
            }
        }
    };

    function getLastMeasure() {
        objectRegistry.get(storedObject.objectUri, function (error, obj) {
            if (error) {
                logger.error('Error retrieving the last resource value');
            } else {
                data.listener('setResource', resourceId, obj.attributes[resourceId]);
            }
        });
    }

    objectRegistry.getAttributes(storedObject.objectUri + '/' + resourceId, function(error, attributes) {
        if (error) {
            logger.error('Couldn\'t find attributes for the [%s] resource URI. Falling back to value changed');
        }

        if (attributes) {
            if (attributes.pmax) {
                logger.debug('Creating a scheduled observation for [%d] miliseconds for object [%s]',
                    attributes.pmax, storedObject.objectUri);

                data.scheduler = setInterval(function handleMaxPeriod() {
                    getLastMeasure(storedObject.objectUri, resourceId);
                }, attributes.pmax);
{'fraction_non_alphanumeric': 0.10268874373385994, 'fraction_numerical': 0.0024305028102688743, 'mean_word_length': 2.3701049398515486, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1851254', 'n_tokens_mistral': 3427, 'n_tokens_neox': 3342, 'n_words': 1128}