Initial commit

.eslintrc.js

@@ -0,0 +1,67 @@
+const path = require('path')
+
+
+module.exports = {
+  'extends': 'airbnb-base',
+  'env': {
+    'es6': true,
+    'node': true,
+  },
+  'parserOptions': {
+    'ecmaVersion': 2018,
+    'sourceType': 'script',
+  },
+  'rules': {
+    'camelcase': 0,
+    'guard-for-in': 0,
+    'import/no-unresolved': 0,
+    'import/no-dynamic-require': 0,
+    'linebreak-style': [
+      'error',
+      'unix',
+    ],
+    'max-len': [
+      'error',
+      120,
+    ],
+    'no-await-in-loop': 0,
+    'no-console': 0,
+    'no-extra-boolean-cast': 0,
+    'no-mixed-operators': 0,
+    'no-plusplus': 0,
+    'no-restricted-syntax': 0,
+    'no-useless-escape': 0,
+    'operator-linebreak': [
+      'error',
+      'after',
+    ],
+    'padded-blocks': 0,
+    'quotes': [
+      'error',
+      'single',
+    ],
+    'semi': [
+      'error',
+      'never',
+    ],
+    'space-before-function-paren': [
+      'error',
+      'always',
+    ],
+    'space-unary-ops': [
+      'error',
+      {
+        'overrides': {
+          '!': true,
+        },
+      },
+    ],
+  },
+  'settings': {
+    'import/resolver': {
+      'node': {
+        paths: [path.resolve(__dirname)],
+      },
+    },
+  },
+}

.gitignore

@@ -0,0 +1,9 @@
+.idea
+.vscode
+
+node_modules/
+
+__tests__/coverage/
+
+yarn-error.log
+.coveralls.yml

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2018 Oleg Levshin
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,80 @@
+# Amqpator
+
+[![License](https://img.shields.io/badge/License-MIT-000000.svg)](https://opensource.org/licenses/MIT)
+[![Coverage Status](https://coveralls.io/repos/github/LordotU/amqpator/badge.svg)](https://coveralls.io/github/LordotU/amqpator)
+
+## Description
+
+Wrapper for [amqplib](http://www.squaremobius.net/amqp.node/) which simplifies its usage for publishing and subscribing
+
+## Installation
+
+```bash
+yarn add amqpator
+```
+
+## Testing
+
+```bash
+yarn test
+```
+
+***Note:*** you should have running RabbitMQ instance with management plugin installed.
+
+## Usage
+
+```javascript
+// Simple echo pub/sub
+
+const Amqpator = require('amqpator')
+
+
+const amqp = new Amqpator(/* {
+  host = 'localhost',
+  username = 'guest',
+  logger = console,
+  onConnectionClose = () => {},
+  onConnectionError = () => {},
+  connectionOptions = {},
+  password = 'guest',
+  port = 5672,
+  query = { heartbeat: 30 },
+  reconnect = true,
+  reconnectAttempts = 10,
+  reconnectInterval = 299,
+  vhost = '/',
+} */)
+
+const exchange = 'echo_exchange'
+const routingKey = 'echo_exchange_routing_key'
+
+amqp.getSub({
+  exchange,
+  exchangeOptions: {
+    autoDelete: true,
+  },
+
+  routingKey,
+
+  queue: 'echo_queue',
+  queueOptions: {
+    autoDelete: true,
+    exclusive: true,
+  },
+
+  onQueueMsg: ({ echo }) => console.log(echo),
+}).then(
+  _ => _.subscribe()
+)
+
+amqp.getPub({
+  exchange,
+  exchangeOptions: {
+    autoDelete: true,
+  },
+
+  routingKey,
+}).then(
+  _ => _.publish({ echo: 'Echo' })
+)
+```

__tests__/index.js

@@ -0,0 +1,189 @@
+/* eslint-disable no-undef */
+
+'use strict'
+
+const got = require('got')
+
+const Amqp = require('../src')
+
+
+const TEST_HOST = process.env.AMQPATOR_HOST || 'localhost'
+const TEST_PORT = process.env.AMQPATOR_PORT || 5672
+const TEST_USERNAME = process.env.AMQPATOR_USERNAME || 'guest'
+const TEST_PASSWORD = process.env.AMQPATOR_PASSWORD || 'guest'
+const TEST_VHOST = process.env.AMQPATOR_VHOST || '/'
+const TEST_PORT_HTTP = process.env.AMQPATOR_PORT_HTTP || 15672
+
+const TEST_CREDENTIALS = {
+  host: TEST_HOST,
+  port: TEST_PORT,
+  username: TEST_USERNAME,
+  password: TEST_PASSWORD,
+  vhost: TEST_VHOST,
+}
+
+const TEST_QUEUE = 'AMQPATOR_test_queue'
+const TEST_EXCHANGE = 'AMQPATOR_test_exchange'
+const TEST_EXCHANGE_2 = 'AMQPATOR_test_exchange_2'
+const TEST_ROUTING_KEY = 'AMQPATOR_test_routing_key'
+const TEST_MESSAGE_TEXT = 'AMQPATOR_test_message'
+
+const JEST_TIMEOUT = 30 * 10 ** 3
+
+
+const getRange = (max = 0) => [...Array(max).keys()]
+
+const publishMessages = (publisher, cnt = 100) => Promise.all(
+  getRange(cnt)
+    .map(
+      k => publisher.publish({ message: `${TEST_MESSAGE_TEXT}_${k}` }),
+    ),
+)
+
+jest.setTimeout(JEST_TIMEOUT)
+
+describe('Amqp', () => {
+
+  test('Trying to reconnect on initial connection', async () => {
+    const amqp = new Amqp({
+      ...TEST_CREDENTIALS,
+
+      username: 'root',
+      password: 'toor',
+
+      reconnectAttempts: 3,
+      reconnectInterval: 100,
+    })
+
+    await expect(amqp.connect()).rejects.toThrow()
+
+    expect(amqp.reconnectCount).toEqual(amqp.reconnectAttempts)
+  })
+
+  test('Correctly connects and disconnects', async () => {
+
+    const onConnectionClose = jest.fn()
+
+    const amqp = new Amqp({
+      onConnectionClose,
+    })
+
+    await amqp.connect()
+    await amqp.disconnect()
+
+    expect(onConnectionClose.mock.calls.length).toBe(1)
+  })
+
+  test('Correctly reacts on errors', async () => {
+
+    const onConnectionError = jest.fn()
+
+    const amqp = new Amqp({
+      ...TEST_CREDENTIALS,
+      onConnectionError,
+    })
+
+    await amqp.connect()
+    await amqp.connection.emit('error', new Error('Test error'))
+
+    expect(onConnectionError.mock.calls.length).toBe(1)
+  })
+
+})
+
+describe('AmqpPub', () => {
+
+  test('Correctly make publish', async () => {
+
+    const amqp = new Amqp({
+      ...TEST_CREDENTIALS,
+    })
+
+    const amqpPublisher = await amqp.getPub({
+      exchange: TEST_EXCHANGE,
+      routingKey: TEST_ROUTING_KEY,
+    })
+
+    await publishMessages(amqpPublisher, 100)
+
+    await amqpPublisher.channel.deleteExchange(TEST_EXCHANGE)
+
+    await amqp.disconnect()
+  })
+
+})
+
+describe('AmqpSub', () => {
+
+  test('Correctly make consume', async () => {
+
+    const MESSAGES_TO_PUBLISH = 10 ** 3
+
+    const onQueueMsg = jest.fn(msg => msg)
+    const { results: onQueueMsgResults } = onQueueMsg.mock
+
+    const amqp = new Amqp({
+      ...TEST_CREDENTIALS,
+    })
+
+    const amqpSubscriber = await amqp.getSub({
+      exchange: TEST_EXCHANGE_2,
+      exchangeOptions: {
+        autoDelete: true,
+      },
+      routingKey: TEST_ROUTING_KEY,
+      queue: TEST_QUEUE,
+      queueOptions: {
+        autoDelete: true,
+        exclusive: true,
+      },
+      prefetch: 100,
+      onQueueMsg,
+    })
+
+    amqpSubscriber.subscribe()
+
+    const amqpPublisher = await amqp.getPub({
+      exchange: TEST_EXCHANGE_2,
+      exchangeOptions: {
+        autoDelete: true,
+      },
+      routingKey: TEST_ROUTING_KEY,
+    })
+
+    await publishMessages(amqpPublisher, MESSAGES_TO_PUBLISH)
+
+    let interval = null
+
+    const finishInterval = async (cb, ...args) => {
+      clearInterval(interval)
+      await amqp.disconnect()
+      cb(...args)
+    }
+
+    await new Promise((resolve, reject) => {
+      interval = setInterval(async () => {
+        try {
+          const { body: { messages } } = await got(
+            `http://${TEST_HOST}:${TEST_PORT_HTTP}/api/queues/${encodeURIComponent(TEST_VHOST)}/${TEST_QUEUE}`,
+            { auth: `${TEST_USERNAME}:${TEST_PASSWORD}`, json: true },
+          )
+
+          if (parseInt(messages, 10) <= 0) {
+            await finishInterval(resolve)
+          }
+        } catch (error) {
+          await finishInterval(reject, error)
+        }
+      }, 299)
+    })
+
+    expect(onQueueMsgResults.length).toBe(MESSAGES_TO_PUBLISH)
+
+    for (const k of onQueueMsgResults.keys()) {
+      expect(onQueueMsgResults[k].value).toBe(`${TEST_MESSAGE_TEXT}_${k}`)
+    }
+
+  })
+
+})