Vhost parameter added.

.github/workflows/ci.yml

@@ -1,5 +1,5 @@
 name: CI
-'on':
+"on":
   push:
     branches:
       - master
@@ -58,15 +58,11 @@ jobs:
       mongodb:
         image: mongo:4.2
         ports:
-        - 27017:27017
+          - 27017:27017
       mosquitto:
         image: eclipse-mosquitto:1.6.7
         ports:
-        - 1883:1883
-      rabbitmq:
-        image: rabbitmq:3.8.9
-        ports:
-        - 5672:5672
+          - 1883:1883
     strategy:
       matrix:
         node-version:
@@ -77,11 +73,17 @@ jobs:
     steps:
       - name: Git checkout
         uses: actions/checkout@v2
-      - name: 'Install Node.js ${{ matrix.node-version }}'
+      - name: Start RabbitMQ
+        uses: Namoshek/[email protected]
+        with:
+          version: "3.8.9-management"
+          ports: "5672:5672"
+          definitions: ${{ github.workspace }}/docs/rabbitmq-definitions.json
+      - name: "Install Node.js ${{ matrix.node-version }}"
         uses: actions/setup-node@v1
         with:
-          node-version: '${{ matrix.node-version }}'
-      - name: 'Unit Tests with Node.js ${{ matrix.node-version }}'
+          node-version: "${{ matrix.node-version }}"
+      - name: "Unit Tests with Node.js ${{ matrix.node-version }}"
         run: |
           npm install
           npm test
@@ -94,19 +96,21 @@ jobs:
       mongodb:
         image: mongo:4.2
         ports:
-        - 27017:27017
+          - 27017:27017
       mosquitto:
         image: eclipse-mosquitto:1.6.7
         ports:
-        - 1883:1883
-      rabbitmq:
-        image: rabbitmq:3.8.9
-        ports:
-        - 5672:5672
+          - 1883:1883
     steps:
       - name: Git checkout
         uses: actions/checkout@v2
-      - name: 'Test Coverage with Node.js 12.x'
+      - name: Start RabbitMQ
+        uses: Namoshek/[email protected]
+        with:
+          version: "3.8.9-management"
+          ports: "5672:5672"
+          definitions: ${{ github.workspace }}/docs/rabbitmq-definitions.json
+      - name: "Test Coverage with Node.js 12.x"
         uses: actions/setup-node@v1
         with:
           node-version: 12.x

CHANGES_NEXT_RELEASE

@@ -1,4 +1,5 @@
-- Fix: avoid raise mongo alarm when a measure is not maching a group configuration
-- Upgrade NodeJS version from 10 to 12 in Dockerfile due to Node 10 End-of-Life
-- Set Nodejs 12 as minimum version in packages.json (effectively removing Nodev10 from supported versions)
-- Add list of environment variables which can be protected by Docker Secrets
+- Fix: avoid raise mongo alarm when a measure is not maching a group configuration
+- Upgrade NodeJS version from 10 to 12 in Dockerfile due to Node 10 End-of-Life
+- Set Nodejs 12 as minimum version in packages.json (effectively removing Nodev10 from supported versions)
+- Add list of environment variables which can be protected by Docker Secrets
+- Add: virtual host configuration added to AMQP transport (config.amqp.vhost and IOTA_AMQP_VHOST env var)

README.md

@@ -103,7 +103,7 @@ config file can be used to permanently tune the MQTT broker parameters, or the d
 
 ## Contributing
 
-If you'd like to contribute, start by searching through the issues and pull requests to see whether someone else has 
+If you'd like to contribute, start by searching through the issues and pull requests to see whether someone else has
 raised a similar idea or question.
 
 Before contributing, please check out [contribution guidelines](docs/contribution.md)
@@ -128,6 +128,9 @@ All the tests are designed to test end-to-end scenarios, and there are some requ
 
 -   MQTT v5 broker (like mosquitto v1.6.7 server running)
 -   MongoDB v3.x server running
+-   AMQP 0-9-1 server with `foo/bar` vHost created (like RabbitMQ v3 server running)
+    -   You can set up RabbitMQ to run the test as follows:
+        `docker run -d -p 5672:5672 -v $(pwd)/docs/rabbitmq-definitions.json:/etc/rabbitmq/definition.json -e RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS="-rabbitmq_management load_definitions '/etc/rabbitmq/definition.json'" rabbitmq:management`
 
 ---
 

config.js

@@ -119,6 +119,10 @@ config.amqp = {
      * Port where the AMQP broker is listening.
      */
     port: 5672,
+    /**
+     * Virtual host in the AMQP Broker to which IoT Agent will connect (optional).
+     */
+    // vhost: "custom-virtual-host",
     /**
      * user name that identifies the IOTA against the AMQP broker (optional).
      */

docs/installationguide.md

@@ -129,8 +129,8 @@ These are the currently available MQTT configuration options:
 -   **retain**: retain flag (default is false).
 -   **retries**: Number of MQTT connection error retries (default is 5).
 -   **retryTime**: Time between MQTT connection retries (default is 5 seconds).
--   **keepalive**: Time to keep connection open between client and MQTT broker (default is 60 seconds). If you experience
-    disconnnection problems using 0 (as the one described in
+-   **keepalive**: Time to keep connection open between client and MQTT broker (default is 60 seconds). If you
+    experience disconnnection problems using 0 (as the one described in
     [this case](https://github.com/telefonicaid/iotagent-json/issues/455)) a value greater than 0 is recommended.
 -   **avoidLeadingSlash** this flag sets whether the agent publishes commands to topics starting with slash (default in
     order versions) or without the slash. See
@@ -146,6 +146,7 @@ IoT Agent. The following attributes are accepted:
 
 -   **host**: Host where the AMQP Broker is located.
 -   **port**: Port where the AMQP Broker is listening
+-   **vhost**: virtual host in the AMQP Broker to which IoT Agent will connect (optional).
 -   **username**: username that identifies the IOTA against the AMQP broker (optional).
 -   **password**: password to be used if the username is provided (optional).
 -   **exchange**: Exchange in the AMQP broker
@@ -199,6 +200,7 @@ The ones relating specific JSON bindings are described in the following table.
 | IOTA_MQTT_AVOID_LEADING_SLASH | mqtt.avoidLeadingSlash  |
 | IOTA_AMQP_HOST                | amqp.host               |
 | IOTA_AMQP_PORT                | amqp.port               |
+| IOTA_AMQP_VHOST               | amqp.vhost              |
 | IOTA_AMQP_USERNAME            | amqp.username           |
 | IOTA_AMQP_PASSWORD            | amqp.password           |
 | IOTA_AMQP_EXCHANGE            | amqp.exchange           |

docs/rabbitmq-definitions.json

@@ -0,0 +1,26 @@
+{
+  "vhosts": [
+      {
+          "name": "/"
+      },
+      {
+          "name": "foo/bar"
+      }
+  ],
+  "permissions": [
+      {
+          "user": "guest",
+          "vhost": "/",
+          "configure": ".*",
+          "write": ".*",
+          "read": ".*"
+      },
+      {
+          "user": "guest",
+          "vhost": "foo/bar",
+          "configure": ".*",
+          "write": ".*",
+          "read": ".*"
+      }
+  ]
+}

lib/bindings/AMQPBinding.js

@@ -125,6 +125,9 @@ function start(callback) {
                 uri += ':' + config.getConfig().amqp.port;
             }
         }
+        if (config.getConfig().amqp.vhost && config.getConfig().amqp.vhost !== '/') {
+            uri += '/' + config.getConfig().amqp.vhost;
+        }
     } else {
         return config.getLogger().error(context, 'Error AMQP is not configured');
     }
@@ -139,46 +142,43 @@ function start(callback) {
             return;
         }
         isConnecting = true;
-        amqp.connect(
-            uri,
-            function(err, conn) {
-                isConnecting = false;
-                // try again
-                if (err) {
-                    config.getLogger().error(context, err.message);
-                    if (numRetried <= retries) {
-                        numRetried++;
-                        return setTimeout(createConnection, retryTime * 1000, callback);
+        amqp.connect(uri, function (err, conn) {
+            isConnecting = false;
+            // try again
+            if (err) {
+                config.getLogger().error(context, err.message);
+                if (numRetried <= retries) {
+                    numRetried++;
+                    return setTimeout(createConnection, retryTime * 1000, callback);
+                }
+            } else {
+                conn.on('error', function (err) {
+                    if (err.message !== 'Connection closing') {
+                        config.getLogger().error(context, err.message);
                     }
-                } else {
-                    conn.on('error', function(err) {
-                        if (err.message !== 'Connection closing') {
-                            config.getLogger().error(context, err.message);
+                });
+                conn.on('close', function () {
+                    // If amqpConn is null, the connection has been closed on purpose
+                    if (amqpConn) {
+                        config.getLogger().error(context, 'reconnecting');
+                        if (numRetried <= retries) {
+                            numRetried++;
+                            return setTimeout(createConnection, retryTime * 1000);
                         }
-                    });
-                    conn.on('close', function() {
-                        // If amqpConn is null, the connection has been closed on purpose
-                        if (amqpConn) {
-                            config.getLogger().error(context, 'reconnecting');
-                            if (numRetried <= retries) {
-                                numRetried++;
-                                return setTimeout(createConnection, retryTime * 1000);
-                            }
-                        }
-                    });
-                    config.getLogger().info(context, 'connected');
-                    amqpConn = conn;
-                    if (callback) {
-                        callback();
                     }
+                });
+                config.getLogger().info(context, 'connected');
+                amqpConn = conn;
+                if (callback) {
+                    callback();
                 }
             }
-        );
+        });
     }
 
     function createChannel(callback) {
         config.getLogger().debug(context, 'channel creating');
-        amqpConn.createChannel(function(err, ch) {
+        amqpConn.createChannel(function (err, ch) {
             if (err) {
                 config.getLogger().error(context, err.message);
             }
@@ -199,7 +199,7 @@ function start(callback) {
 
     function assertQueue(callback) {
         config.getLogger().debug(context, 'asserting queues');
-        amqpChannel.assertQueue(queue, { exclusive: false }, function() {
+        amqpChannel.assertQueue(queue, { exclusive: false }, function () {
             amqpChannel.assertQueue(queue + '_commands', { exclusive: false }, callback);
         });
     }
@@ -216,7 +216,7 @@ function start(callback) {
         callback();
     }
 
-    async.waterfall([createConnection, createChannel, assertExchange, assertQueue, createListener], function(error) {
+    async.waterfall([createConnection, createChannel, assertExchange, assertQueue, createListener], function (error) {
         if (error) {
             config.getLogger().debug('AMQP error %j', error);
         }

lib/configService.js

@@ -70,6 +70,7 @@ function processEnvironmentVariables() {
         'IOTA_MQTT_AVOID_LEADING_SLASH',
         'IOTA_AMQP_HOST',
         'IOTA_AMQP_PORT',
+        'IOTA_AMQP_VHOST',
         'IOTA_AMQP_USERNAME',
         'IOTA_AMQP_PASSWORD',
         'IOTA_AMQP_EXCHANGE',
@@ -103,6 +104,7 @@ function processEnvironmentVariables() {
     const amqpVariables = [
         'IOTA_AMQP_HOST',
         'IOTA_AMQP_PORT',
+        'IOTA_AMQP_VHOST',
         'IOTA_AMQP_USERNAME',
         'IOTA_AMQP_PASSWORD',
         'IOTA_AMQP_EXCHANGE',
@@ -225,6 +227,10 @@ function processEnvironmentVariables() {
         config.amqp.port = process.env.IOTA_AMQP_PORT;
     }
 
+    if (process.env.IOTA_AMQP_VHOST) {
+        config.amqp.vhost = process.env.IOTA_AMQP_VHOST;
+    }
+
     if (process.env.IOTA_AMQP_USERNAME) {
         config.amqp.username = process.env.IOTA_AMQP_USERNAME;
     }