feat(component): add support for ACL's and refactor component (#3)

* Refactor the component

* Add support for ACLS

* Fix test by creating memory datasouce when rabbit is not found

* Add ACL to example config and enhance README

* Add test for ACLs

* chore: update and pin all package versions

* fix: reject with an Error if unable to connect to Rabbit

* test: reload server before every test

* test: return promise for async tests

* ci: enable rabbitmq-server service on circleci

* test: fixup setupQueueConsumer and setupQueueProducer tests

Author: beeman

.npmrc

@@ -0,0 +1 @@
+save-exact=true

README.md

@@ -17,7 +17,13 @@ Loopback Component for working with a Message Queue
   {
     "loopback-component-mq": {
       "options": {
-        "dataSource": "rabbit"
+        "dataSource": "rabbit",
+        "acls": [{
+          "accessType": "*",
+          "principalType": "ROLE",
+          "principalId": "$unauthenticated",
+          "permission": "DENY"
+        }]
       },
       "topology": {
         "my-event-queue": {
@@ -56,6 +62,46 @@ Loopback Component for working with a Message Queue
   }
   ```
 
+## Configuration
+
+The configuration of this component happens in `component-config.json`.
+
+The 2 top-level keys are `options` and `topology` which are both objects.
+
+### Options
+
+The `options` object has 2 keys:
+
+- `dataSource` (String, required). This is the name of the RabbitMQ datasource that is configured in `datasources.json`.
+Please note that this datasource does not use any `loopback-connector-*` packages, but just a way to tell the component 
+how to connect to RabbitMQ.
+
+- `acls` (Array, optional). Use this array to configure the ACL's. This ACL protects the Queue model that gets created.
+
+### Topology
+
+In the `topology` object you configure the queues served by this component. The object key is the name of the Queue. 
+
+Inside this queue definition you can define a `consumer`, a `producer`, or both. The `consumer` and `producer` objects
+both accept 2 properties, `model` and `method`.
+
+#### Consumer
+
+When you defined a `consumer` on a queue, the component will call into the `Model.method()` defined when a new message
+arrives on the queue. The method will be called with 2 parameters, `payload` and `ack`. Payload is the raw message from
+the queue. The `ack` parameter is a message that is used to acknowledge to the queue that the message is being handled.
+Make sure to acknowledge all messages, because RabbitMQ won't allow any other messages to be picked up until the message
+is acknowledged, making the queue come to a halt.
+
+#### Producer
+
+When defining a `producer` on a queue, the component will create the method `Model.method()` defined. The method created
+accepts 1 parameter, `payload`. The payload is the raw message that will be stored on the queue.
+
+## TODO
+
+- Add support for more types of Queues in the topology
+
 ## License
 
 MIT

circle.yml

@@ -1,9 +1,28 @@
 machine:
   node:
     version: 6.1.0
+  services:
+    - docker
+  pre:
+    # Stop rabbitmq that comes with circle by default.
+    - sudo service rabbitmq-server stop
+
+dependencies:
+  pre:
+    # Get wait-for-it.sh.
+    - git clone https://github.com/vishnubob/wait-for-it.git
+
+database:
+  override:
+    # Start rabbitmq with support for management utils.
+    - docker run -d -p 5672:5672 -p 15672:15672 rabbitmq:3-management
+
 test:
+  override:
+    - ./wait-for-it/wait-for-it.sh localhost:5672 -- npm test
   post:
     - npm run coverage
+
 deployment:
   master:
     branch: [master]

lib/index.js

@@ -2,98 +2,40 @@
 
 const _ = require('lodash')
 const debug = require('debug')('loopback:component:mq')
-const loopback = require('loopback')
 const setupModel = require('./setup-model')
 const Rabbit = require('./rabbit')
-let rabbit = null
-let msqQueue = null
-
-function setupQueue(name) {
-  debug('setupQueue: queue: %s', name)
-  msqQueue = rabbit.exchange.queue({ name, durable: true })
-}
-
-function setupQueueConsumer(app, queue, definition) {
-  const modelName = definition.model
-  const methodName = definition.method
-  const Model = app.models[modelName]
-
-  // Check if the model exists
-  if (!Model) {
-    throw new Error(`setupQueueConsumer: Model not found: ${modelName}`)
-  }
-
-  const Method = Model[methodName]
-
-  // Check if the method on the model exists
-  if (!Method) {
-    console.warn(`setupQueueConsumer: Method not found: ${modelName}.${methodName}`) // eslint-disable-line no-console
-  }
-
-  // Start consuming the queue
-  msqQueue.consume(Method)
-
-  debug('setupQueueConsumer: queue: %s, model: %s, method: %s', queue, modelName, methodName)
-}
-
-function setupQueueProducer(app, queue, definition) {
-  const modelName = definition.model
-  const methodName = definition.method
-  const Model = app.models[modelName]
-
-  if (!Model) {
-    throw new Error(`setupQueueProducer: Model not found: ${modelName}`)
-  }
-
-  debug('setupQueueProducer: queue: %s, model: %s, method: %s', queue, modelName, methodName)
-  Model[methodName] = function queueProducer(params) {
-    debug(`${modelName}.${methodName}(%o)`, params)
-    rabbit.exchange.publish(params, { key: queue })
-  }
-}
+const RabbitTopology = require('./rabbit-topology')()
 
 module.exports = function loopbackComponentMq(app, config) {
-  const options = config.options || {}
-  const topology = config.topology || {}
-
-  debug('options: %o', options)
-  debug('topology: %o', topology)
-
-  if (!options.dataSource) {
-    debug('options.dataSource not set, using default value \'rabbit\'')
-    options.dataSource = 'rabbit'
-  }
-
-  const ds = app.dataSources[options.dataSource]
-
-  if (ds) {
-    rabbit = new Rabbit(ds.settings.options || {})
+  debug(config.topology)
+  // Store the component configuration
+  _.set(app, 'settings.loopback-component-mq', {
+    options: config.options || {},
+    topology: config.topology || {},
+  })
 
-    // Loop through all the defined queues
-    _.forEach(topology, (handlers, queue) => {
+  debug('loopback-component-mq settings: %o', _.get(app, 'settings.loopback-component-mq'))
 
-      // Setup the actual queue on RabbitMQ
-      setupQueue(queue)
+  // Independent of app boot, add model and ACL
+  setupModel(app)
 
-      // Setup the consumer of this queue
-      if (handlers.consumer) {
-        setupQueueConsumer(app, queue, handlers.consumer)
-      }
+  // Wait the app to be booted
+  app.once('booted', () => {
 
-      // Setup the producers of this queue
-      if (handlers.producer) {
-        setupQueueProducer(app, queue, handlers.producer)
-      }
+    // Get a reference to the created model
+    const Queue = app.models.Queue
 
-    })
+    // Get a reference to the configured datasource
+    const dsName = _.get(app, 'settings.loopback-component-mq.options.dataSource', 'rabbit')
+    const ds = app.dataSources[dsName]
+    const dsOptions = _.get(ds, 'settings.options')
 
-    setupModel(app, ds, rabbit, topology)
-  }
-  else {
-    debug(`DataSource ${options.dataSource} not found`)
-    const memDs = loopback.createDataSource({ connector: 'memory' })
+    // If we have a datasource, wire up Rabbit and set up the topology
+    if (ds && dsOptions) {
+      Queue.rabbit = new Rabbit(dsOptions)
+      RabbitTopology.setupTopology(Queue)
+    }
 
-    setupModel(app, memDs, {}, {})
-  }
+  })
 
 }

lib/models/queue.js

@@ -1,32 +1,67 @@
 const Promise = require('bluebird')
 
-module.exports = function queueFn(Queue, rabbit, topology) {
+module.exports = function queueFn(Queue) {
+
+  // The topology
+  Queue.topology = {}
+  // The rabbit
+  Queue.rabbit = null
 
   Queue.status = function status() {
     return new Promise((resolve, reject) => {
-      rabbit.stats.overview((err, res, data) => {
+      if (!Queue.rabbit) {
+        return reject(new Error('Error connecting to rabbit'))
+      }
+      return Queue.rabbit.stats.overview((err, res, data) => {
         if (err) {
-          reject(err)
+          return reject(err)
         }
         data = data || {}
-        resolve(data)
+        return resolve(data)
       })
     })
   }
 
   Queue.queues = function queues() {
     return new Promise((resolve, reject) => {
-      rabbit.stats.queues((err, res, data) => {
+      if (!Queue.rabbit) {
+        return reject(new Error('Error connecting to rabbit'))
+      }
+      return Queue.rabbit.stats.queues((err, res, data) => {
         if (err) {
-          reject(err)
+          return reject(err)
         }
         // Only show queues that are defined in the topology
         data = data || []
-        data = data.filter(item => Object.keys(topology).includes(item.name))
-        resolve(data)
+        data = data.filter(item => Object.keys(Queue.topology).includes(item.name))
+        return resolve(data)
       })
     })
   }
 
+
+  Queue.remoteMethod('status', {
+    isStatic: true,
+    description: 'Get an status overview of the connected rabbitmq server',
+    returns: {
+      arg: 'result',
+      type: 'Object',
+      root: true,
+    },
+    http: { path: '/status', verb: 'get' },
+  })
+
+
+  Queue.remoteMethod('queues', {
+    isStatic: true,
+    description: 'Get queues of connected rabbitmq server',
+    returns: {
+      arg: 'result',
+      type: 'Object',
+      root: true,
+    },
+    http: { path: '/queues', verb: 'get' },
+  })
+
   return Queue
 }