ADD Transport selector

telefonicaid · Jun 2, 2016 · 93dd167 · 93dd167
1 parent 6e7afc6
commit 93dd167
Show file tree

Hide file tree

Showing 4 changed files with 38 additions and 19 deletions.
diff --git a/lib/bindings/MQTTBinding.js b/lib/bindings/MQTTBinding.js
@@ -347,8 +347,9 @@ function stop(callback) {
     ], callback);
 }
 
-function executeCommand(apiKey, device, serializedPayload) {
-    return mqttClient.publish.bind(mqttClient, '/' + apiKey + '/' + device.id + '/cmd', serializedPayload, null);
+function executeCommand(apiKey, device, serializedPayload, callback) {
+    mqttClient.publish('/' + apiKey + '/' + device.id + '/cmd', serializedPayload, null);
+    callback();
 }
 
 exports.start = start;

diff --git a/lib/commandHandler.js b/lib/commandHandler.js
@@ -27,7 +27,7 @@ var async = require('async'),
     logger = require('logops'),
     iotAgentLib = require('iotagent-node-lib'),
     constants = require('./constants'),
-    mqttBinding = require('./bindings/MQTTBinding'),
+    transportSelector = require('./transportSelector'),
     config = require('./configService'),
     context = {
         op: 'IoTAgentMQTT.Commands'
@@ -43,15 +43,17 @@ var async = require('async'),
  */
 function generateCommandExecution(apiKey, device, attribute) {
     var payload = {},
-        serialized;
+        serialized,
+        executions;
 
     payload[attribute.name] = attribute.value;
     serialized = JSON.stringify(payload);
 
     logger.debug('Sending command execution to device [%s] with apikey [%s] and payload [%s] ',
         apiKey, device.id, attribute);
 
-    return mqttBinding.executeCommand(apiKey, device, serialized);
+    executions = transportSelector.createExecutionsForBinding([apiKey, device, serialized], 'executeCommand', null);
+    return executions;
 }
 
 /**
@@ -69,6 +71,11 @@ function generateCommandExecution(apiKey, device, attribute) {
 function commandHandler(id, type, service, subservice, attributes, callback) {
     logger.debug('Handling MQTT command for device [%s] in service [%s - %s]', id, service, subservice);
 
+    function concat(previous, current) {
+        previous = previous.concat(current);
+        return previous;
+    }
+
     iotAgentLib.getDeviceByName(id, service, subservice, function(error, device) {
         if (error) {
             logger.error('Command execution could not be handled, as device for entity [%s] [%s] wasn\'t found',
@@ -79,7 +86,10 @@ function commandHandler(id, type, service, subservice, attributes, callback) {
                 if (error) {
                     callback(error);
                 } else {
-                    async.series(attributes.map(generateCommandExecution.bind(null, apiKey, device)), callback);
+                    async.series(attributes
+                        .map(generateCommandExecution.bind(null, apiKey, device))
+                        .reduce(concat, []),
+                        callback);
                 }
             });
         }

diff --git a/lib/transportSelector.js b/lib/transportSelector.js
@@ -50,17 +50,8 @@ function startTransportBindings(newConfig, callback) {
     });
 }
 
-/**
- * Execute the function given by the 'functionName' parameter for all the transport bindings, with the arguments
- * given in the 'argument' array. If the optional parameter protocol is not null, the function will only be
- * executed in the plugin of the selected protocol.
- *
- * @param {Array} argument          Array of arguments to call the function with.
- * @param {String} functionName     Name of the function to call in every transport plugin.
- * @param {String} protocol         Transport protocol where the function must be executed.
- */
-function applyFunctionFromBinding(argument, functionName, protocol, callback) {
-    logger.debug('Looking for bindings for the function [%s] and protocol [%s]', functionName, protocol);
+function createExecutionsForBinding(argument, functionName, protocol) {
+    logger.debug('Creating execution for function [%s] and protocol [%s]', functionName, protocol);
 
     function addHandler(current, binding) {
         if (binding[functionName] && (!protocol || binding.protocol === protocol)) {
@@ -74,7 +65,22 @@ function applyFunctionFromBinding(argument, functionName, protocol, callback) {
         return current;
     }
 
-    async.series(transportBindings.reduce(addHandler, []), callback);
+    return transportBindings.reduce(addHandler, []);
+}
+
+/**
+ * Execute the function given by the 'functionName' parameter for all the transport bindings, with the arguments
+ * given in the 'argument' array. If the optional parameter protocol is not null, the function will only be
+ * executed in the plugin of the selected protocol.
+ *
+ * @param {Array} argument          Array of arguments to call the function with.
+ * @param {String} functionName     Name of the function to call in every transport plugin.
+ * @param {String} protocol         Transport protocol where the function must be executed.
+ */
+function applyFunctionFromBinding(argument, functionName, protocol, callback) {
+    logger.debug('Looking for bindings for the function [%s] and protocol [%s]', functionName, protocol);
+
+    async.series(createExecutionsForBinding(argument, functionName, protocol), callback);
 }
 
 /**
@@ -91,3 +97,4 @@ function stopTransportBindings(callback) {
 exports.stopTransportBindings = stopTransportBindings;
 exports.applyFunctionFromBinding = applyFunctionFromBinding;
 exports.startTransportBindings = startTransportBindings;
+exports.createExecutionsForBinding = createExecutionsForBinding;
diff --git a/test/config-test.js b/test/config-test.js
@@ -48,7 +48,8 @@ config.iota = {
     deviceRegistrationDuration: 'P1M',
     defaultType: 'Thing',
     defaultResource: '',
-    defaultKey: '1234'
+    defaultKey: '1234',
+    defaultTransport: 'MQTT'
 };
 
 module.exports = config;