amqp-cacoon - npm Package Compare versions

Comparing version 1.1.3 to 2.0.0

build/helpers/message_batching_manager.d.ts

@@ -1,2 +0,2 @@
-import { Channel, ConsumeMessage, ConsumeBatchMessages } from '../index';
+import { ChannelWrapper, ConsumeMessage, ConsumeBatchMessages } from '../index';
 import { Logger } from 'log4js';
@@ -60,3 +60,3 @@ export interface IMessageBatchingManagerConfig {
      */
-    ackMessageList(channel: Channel, messageList: Array<ConsumeMessage>): void;
+    ackMessageList(channel: ChannelWrapper, messageList: Array<ConsumeMessage>): void;
     /**
@@ -71,3 +71,3 @@ * nackMessageList
      */
-    nackMessageList(channel: Channel, messageList: Array<ConsumeMessage>, requeue?: boolean): void;
+    nackMessageList(channel: ChannelWrapper, messageList: Array<ConsumeMessage>, requeue?: boolean): void;
     /**
@@ -84,3 +84,3 @@ * sendBufferedMessages
      */
-    sendBufferedMessages(channel: Channel, handler: (channel: Channel, msg: ConsumeBatchMessages) => Promise<void>): Promise<void>;
+    sendBufferedMessages(channel: ChannelWrapper, handler: (channel: ChannelWrapper, msg: ConsumeBatchMessages) => Promise<void>): Promise<void>;
     /**
@@ -104,3 +104,3 @@ * handleMessageBuffering
      */
-    handleMessageBuffering(channel: Channel, msg: ConsumeMessage, handler: (channel: Channel, msg: ConsumeBatchMessages) => Promise<void>): Promise<void>;
+    handleMessageBuffering(channel: ChannelWrapper, msg: ConsumeMessage, handler: (channel: ChannelWrapper, msg: ConsumeBatchMessages) => Promise<void>): Promise<void>;
 }

build/index.d.ts

@@ -1,5 +0,6 @@
-import amqp from 'amqplib';
-import { ConsumeMessage, Channel, Options } from 'amqplib';
+import amqp, { ChannelWrapper } from 'amqp-connection-manager';
+import { ConsumeMessage, Channel, ConfirmChannel, Options } from 'amqplib';
 import { Logger } from 'log4js';
-export { ConsumeMessage, Channel };
+declare type ConnectCallback = (channel: ConfirmChannel) => Promise<any>;
+export { ConsumeMessage, ChannelWrapper, Channel, ConfirmChannel, ConnectCallback };
 export interface ConsumeBatchMessages {
@@ -34,3 +35,5 @@ batchingOptions: {
     };
-    maxWaitForDrainMs?: number;
+    onChannelConnect?: ConnectCallback;
+    onBrokerConnect?: () => void;
+    onBrokerDisconnect?: () => void;
 }
@@ -43,8 +46,8 @@ /**
  * 2. Call publish() function to publish a message
- * 3. To consume messages call registerConsumer() passing in a handler funciton
+ * 3. To consume messages call registerConsumer() passing in a handler function
  * 4. When consuming the callback registered in the previous step will be called when a message is received
  **/
 declare class AmqpCacoon {
-    private pubChannel;
-    private subChannel;
+    private pubChannelWrapper;
+    private subChannelWrapper;
     private connection?;
@@ -54,3 +57,5 @@ private fullHostName;
     private logger?;
-    private maxWaitForDrainMs;
+    private onChannelConnect;
+    private onBrokerConnect;
+    private onBrokerDisconnect;
     /**
@@ -70,15 +75,16 @@ * constructor
      **/
-    private getFullHostName;
+    private static getFullHostName;
+    private injectConnectionEvents;
     /**
      * getPublishChannel
      * This connects to amqp and creates a channel or gets the current channel
-     * @return channel
+     * @return ChannelWrapper
      **/
-    getPublishChannel(): Promise<amqp.Channel>;
+    getPublishChannel(): Promise<amqp.ChannelWrapper>;
     /**
      * getConsumerChannel
      * This connects to amqp and creates a channel or gets the current channel
-     * @return channel
+     * @return ChannelWrapper
      **/
-    getConsumerChannel(): Promise<amqp.Channel>;
+    getConsumerChannel(): Promise<amqp.ChannelWrapper>;
     /**
@@ -89,3 +95,3 @@ * registerConsumerPrivate
      * @param queue - Name of the queue
-     * @param handler: (channel: Channel, msg: object) => Promise<any> - A handler that receives the message
+     * @param consumerHandler: (channel: ChannelWrapper, msg: object) => Promise<any> - A handler that receives the message
      * @param options : ConsumerOptions - Used to pass in consumer options
@@ -101,7 +107,7 @@ * @return Promise<void>
      * @param queue - Name of the queue
-     * @param handler: (channel: Channel, msg: object) => Promise<any> - A handler that receives the message
+     * @param handler: (channel: ChannelWrapper, msg: object) => Promise<any> - A handler that receives the message
      * @param options : ConsumerOptions - Used to pass in consumer options
      * @return Promise<void>
      **/
-    registerConsumer(queue: string, handler: (channel: Channel, msg: ConsumeMessage) => Promise<void>, options?: ConsumerOptions): Promise<void>;
+    registerConsumer(queue: string, handler: (channel: ChannelWrapper, msg: ConsumeMessage) => Promise<void>, options?: ConsumerOptions): Promise<void>;
     /**
@@ -115,7 +121,7 @@ * registerConsumerBatch
      * @param queue - Name of the queue
-     * @param handler: (channel: Channel, msg: object) => Promise<any> - A handler that receives the message
+     * @param handler: (channel: ChannelWrapper, msg: object) => Promise<any> - A handler that receives the message
      * @param options : ConsumerOptions - Used to pass in consumer options
      * @return Promise<void>
      **/
-    registerConsumerBatch(queue: string, handler: (channel: Channel, msg: ConsumeBatchMessages) => Promise<void>, options?: ConsumerBatchOptions): Promise<void>;
+    registerConsumerBatch(queue: string, handler: (channel: ChannelWrapper, msg: ConsumeBatchMessages) => Promise<void>, options?: ConsumerBatchOptions): Promise<void>;
     /**
@@ -145,3 +151,17 @@ * publish
     closePublishChannel(): Promise<void>;
+    /**
+     * handleBrokerConnect
+     * Fires onBrokerConnect callback function whenever amqp connection is made
+     *
+     * @return void
+     **/
+    private handleBrokerConnect;
+    /**
+     * handleBrokerDisconnect
+     * Fires onBrokerDisconnect callback function whenever amqp connection is lost
+     *
+     * @return void
+     **/
+    private handleBrokerDisonnect;
 }
 export default AmqpCacoon;

build/index.js

@@ -42,3 +42,3 @@ "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-var amqplib_1 = __importDefault(require("amqplib"));
+var amqp_connection_manager_1 = __importDefault(require("amqp-connection-manager"));
 var message_batching_manager_1 = __importDefault(require("./helpers/message_batching_manager"));
@@ -53,3 +53,3 @@ var DEFAULT_MAX_FILES_SIZE_BYTES = 1024 * 1024 * 2; // 2 MB
  * 2. Call publish() function to publish a message
- * 3. To consume messages call registerConsumer() passing in a handler funciton
+ * 3. To consume messages call registerConsumer() passing in a handler function
  * 4. When consuming the callback registered in the previous step will be called when a message is received
@@ -67,8 +67,11 @@ **/
     function AmqpCacoon(config) {
-        this.pubChannel = null;
-        this.subChannel = null;
-        this.fullHostName = config.connectionString || this.getFullHostName(config);
+        this.pubChannelWrapper = null;
+        this.subChannelWrapper = null;
+        this.fullHostName = config.connectionString || AmqpCacoon.getFullHostName(config);
         this.amqp_opts = config.amqp_opts;
         this.logger = config.providers.logger;
-        this.maxWaitForDrainMs = config.maxWaitForDrainMs || 60000; // Default to 1 min
+        // this.maxWaitForDrainMs = config.maxWaitForDrainMs || 60000; // Default to 1 min
+        this.onChannelConnect = config.onChannelConnect || null;
+        this.onBrokerConnect = config.onBrokerConnect || null;
+        this.onBrokerDisconnect = config.onBrokerDisconnect || null;
     }
@@ -80,3 +83,3 @@ /**
      **/
-    AmqpCacoon.prototype.getFullHostName = function (config) {
+    AmqpCacoon.getFullHostName = function (config) {
         var fullHostNameString = config.protocol +
@@ -94,44 +97,54 @@ '://' +
     };
+    AmqpCacoon.prototype.injectConnectionEvents = function (connection) {
+        var _this = this;
+        // Subscribe to onConnect / onDisconnection functions for debugging
+        connection.on('connect', function () {
+            _this.handleBrokerConnect();
+        });
+        connection.on('disconnect', function () {
+            _this.handleBrokerDisonnect();
+        });
+    };
     /**
      * getPublishChannel
      * This connects to amqp and creates a channel or gets the current channel
-     * @return channel
+     * @return ChannelWrapper
      **/
     AmqpCacoon.prototype.getPublishChannel = function () {
         return __awaiter(this, void 0, void 0, function () {
-            var _a, _b, _c, e_1;
-            return __generator(this, function (_d) {
-                switch (_d.label) {
-                    case 0:
-                        _d.trys.push([0, 4, , 5]);
-                        // Return the pubChannel if we are already connected
-                        if (this.pubChannel)
-                            return [2 /*return*/, this.pubChannel];
-                        // Connect if needed
-                        _a = this;
-                        _b = this.connection;
-                        if (_b) return [3 /*break*/, 2];
-                        return [4 /*yield*/, amqplib_1.default.connect(this.fullHostName, this.amqp_opts)];
-                    case 1:
-                        _b = (_d.sent());
-                        _d.label = 2;
-                    case 2:
-                        // Connect if needed
-                        _a.connection = _b;
-                        // Open a channel
-                        _c = this;
-                        return [4 /*yield*/, this.connection.createChannel()];
-                    case 3:
-                        // Open a channel
-                        _c.pubChannel = _d.sent();
-                        return [3 /*break*/, 5];
-                    case 4:
-                        e_1 = _d.sent();
-                        if (this.logger)
-                            this.logger.error('AMQPCacoon.connect: Error: ', e_1);
-                        throw e_1;
-                    case 5: 
-                    // Return the channel
-                    return [2 /*return*/, this.pubChannel];
+            var _this = this;
+            return __generator(this, function (_a) {
+                try {
+                    // Return the pubChannel if we are already connected
+                    if (this.pubChannelWrapper) {
+                        return [2 /*return*/, this.pubChannelWrapper];
+                    }
+                    // Connect if needed
+                    this.connection =
+                        this.connection || amqp_connection_manager_1.default.connect([this.fullHostName], this.amqp_opts);
+                    if (this.connection) {
+                        this.injectConnectionEvents(this.connection);
+                    }
+                    // Open a channel (get reference to ChannelWrapper)
+                    // Add a setup function that will be called on each connection retry
+                    // This function is specified in the config
+                    this
+                        .pubChannelWrapper = this.connection.createChannel({
+                        setup: function (channel) {
+                            if (_this.onChannelConnect) {
+                                return _this.onChannelConnect(channel);
+                            }
+                            else {
+                                return Promise.resolve();
+                            }
+                        },
+                    });
                 }
+                catch (e) {
+                    if (this.logger)
+                        this.logger.error('AMQPCacoon.connect: Error: ', e);
+                    throw e;
+                }
+                // Return the channel
+                return [2 /*return*/, this.pubChannelWrapper];
             });
@@ -143,41 +156,41 @@ });
      * This connects to amqp and creates a channel or gets the current channel
-     * @return channel
+     * @return ChannelWrapper
      **/
     AmqpCacoon.prototype.getConsumerChannel = function () {
         return __awaiter(this, void 0, void 0, function () {
-            var _a, _b, _c, e_2;
-            return __generator(this, function (_d) {
-                switch (_d.label) {
-                    case 0:
-                        _d.trys.push([0, 4, , 5]);
-                        // Return the subChannel if we are already connected
-                        if (this.subChannel)
-                            return [2 /*return*/, this.subChannel];
-                        // Connect if needed
-                        _a = this;
-                        _b = this.connection;
-                        if (_b) return [3 /*break*/, 2];
-                        return [4 /*yield*/, amqplib_1.default.connect(this.fullHostName, this.amqp_opts)];
-                    case 1:
-                        _b = (_d.sent());
-                        _d.label = 2;
-                    case 2:
-                        // Connect if needed
-                        _a.connection = _b;
-                        // Open a channel
-                        _c = this;
-                        return [4 /*yield*/, this.connection.createChannel()];
-                    case 3:
-                        // Open a channel
-                        _c.subChannel = _d.sent();
-                        return [3 /*break*/, 5];
-                    case 4:
-                        e_2 = _d.sent();
-                        if (this.logger)
-                            this.logger.error('AMQPCacoon.getConsumerChannel: Error: ', e_2);
-                        throw e_2;
-                    case 5: 
-                    // Return the channel
-                    return [2 /*return*/, this.subChannel];
+            var _this = this;
+            return __generator(this, function (_a) {
+                try {
+                    // Return the subChannel if we are already connected
+                    if (this.subChannelWrapper)
+                        return [2 /*return*/, this.subChannelWrapper];
+                    // Connect if needed
+                    this.connection =
+                        this.connection || amqp_connection_manager_1.default.connect([this.fullHostName], this.amqp_opts);
+                    if (this.connection) {
+                        this.injectConnectionEvents(this.connection);
+                    }
+                    // Open a channel (get reference to ChannelWrapper)
+                    // Add a setup function that will be called on each connection retry
+                    // This function is specified in the config
+                    this.subChannelWrapper = this.connection.createChannel({
+                        setup: function (channel) {
+                            // `channel` here is a regular amqplib `ConfirmChannel`.
+                            // Note that `this` here is the channelWrapper instance.
+                            if (_this.onChannelConnect) {
+                                return _this.onChannelConnect(channel);
+                            }
+                            else {
+                                return Promise.resolve();
+                            }
+                        },
+                    });
                 }
+                catch (e) {
+                    if (this.logger)
+                        this.logger.error('AMQPCacoon.getConsumerChannel: Error: ', e);
+                    throw e;
+                }
+                // Return the channel
+                return [2 /*return*/, this.subChannelWrapper];
             });
@@ -191,3 +204,3 @@ });
      * @param queue - Name of the queue
-     * @param handler: (channel: Channel, msg: object) => Promise<any> - A handler that receives the message
+     * @param consumerHandler: (channel: ChannelWrapper, msg: object) => Promise<any> - A handler that receives the message
      * @param options : ConsumerOptions - Used to pass in consumer options
@@ -198,22 +211,22 @@ * @return Promise<void>
         return __awaiter(this, void 0, void 0, function () {
-            var channel, e_3;
+            var channelWrapper_1, e_1;
+            var _this = this;
             return __generator(this, function (_a) {
                 switch (_a.label) {
                     case 0:
-                        _a.trys.push([0, 3, , 4]);
+                        _a.trys.push([0, 2, , 3]);
                         return [4 /*yield*/, this.getConsumerChannel()];
                     case 1:
-                        channel = _a.sent();
-                        // Register a consume on the current channel
-                        return [4 /*yield*/, channel.consume(queue, consumerHandler.bind(this, channel), options)];
+                        channelWrapper_1 = _a.sent();
+                        channelWrapper_1.addSetup(function (channel) {
+                            // Register a consume on the current channel
+                            return channel.consume(queue, consumerHandler.bind(_this, channelWrapper_1), options);
+                        });
+                        return [3 /*break*/, 3];
                     case 2:
-                        // Register a consume on the current channel
-                        _a.sent();
-                        return [3 /*break*/, 4];
-                    case 3:
-                        e_3 = _a.sent();
+                        e_1 = _a.sent();
                         if (this.logger)
-                            this.logger.error('AMQPCacoon.registerConsumerPrivate: Error: ', e_3);
-                        throw e_3;
-                    case 4: return [2 /*return*/];
+                            this.logger.error('AMQPCacoon.registerConsumerPrivate: Error: ', e_1);
+                        throw e_1;
+                    case 3: return [2 /*return*/];
                 }
@@ -229,3 +242,3 @@ });
      * @param queue - Name of the queue
-     * @param handler: (channel: Channel, msg: object) => Promise<any> - A handler that receives the message
+     * @param handler: (channel: ChannelWrapper, msg: object) => Promise<any> - A handler that receives the message
      * @param options : ConsumerOptions - Used to pass in consumer options
@@ -262,3 +275,3 @@ * @return Promise<void>
      * @param queue - Name of the queue
-     * @param handler: (channel: Channel, msg: object) => Promise<any> - A handler that receives the message
+     * @param handler: (channel: ChannelWrapper, msg: object) => Promise<any> - A handler that receives the message
      * @param options : ConsumerOptions - Used to pass in consumer options
@@ -313,65 +326,26 @@ * @return Promise<void>
         return __awaiter(this, void 0, void 0, function () {
-            var channel_1, channelReady, e_4;
-            var _this = this;
+            var channel, e_2;
             return __generator(this, function (_a) {
                 switch (_a.label) {
                     case 0:
-                        _a.trys.push([0, 5, , 6]);
+                        _a.trys.push([0, 3, , 4]);
                         return [4 /*yield*/, this.getPublishChannel()];
                     case 1:
-                        channel_1 = _a.sent();
-                        channelReady = channel_1.publish(exchange, routingKey, msgBuffer, options);
-                        if (this.logger) {
-                            this.logger.trace("AMQPCacoon.publish result: " + channelReady);
-                        }
-                        if (!channelReady) return [3 /*break*/, 2];
+                        channel = _a.sent();
+                        // TODO: Alex: Does the guaranteed publish eliminate the need to handle drain events?
+                        // There's currently a reported bug in node-amqp-connection-manager saying the lib does
+                        // not handle drain events properly...we should fix this.
+                        return [4 /*yield*/, channel.publish(exchange, routingKey, msgBuffer, options)];
+                    case 2:
+                        // TODO: Alex: Does the guaranteed publish eliminate the need to handle drain events?
+                        // There's currently a reported bug in node-amqp-connection-manager saying the lib does
+                        // not handle drain events properly...we should fix this.
+                        _a.sent();
                         return [2 /*return*/];
-                    case 2:
-                        // Otherwise, channelReady == false so we have to wait for the pubChannel's on 'drain'
-                        if (this.logger) {
-                            this.logger.info('AMQPCacoon.publish buffer full. Waiting for "drain"...');
-                        }
-                        // Set a handler for the drain event
-                        return [4 /*yield*/, new Promise(function (resolve, reject) {
-                                if (!_this.pubChannel) {
-                                    // This shouldn't ever happen
-                                    throw new Error("AMQPCacoon.public: Publisher channel has not been set up");
-                                }
-                                // Set a timeout in case Drain is slow, at least we log it
-                                var timeoutHandler = _this.maxWaitForDrainMs !== 0
-                                    ? setTimeout(function () {
-                                        if (_this.logger) {
-                                            _this.logger.info("AMQPCacoon.publish: After a full buffer, slow buffer: \nmsg: " + msgBuffer.toString('utf8'));
-                                        }
-                                        if (timeoutHandler) {
-                                            timeoutHandler = null;
-                                            reject(new Error("AMQPCacoon.public: Timeout after " + _this.maxWaitForDrainMs + "ms"));
-                                        }
-                                    }, _this.maxWaitForDrainMs)
-                                    : true;
-                                channel_1.once('drain', function () {
-                                    if (_this.logger) {
-                                        _this.logger.trace('AMQPCacoon.publish: "drain" received.');
-                                    }
-                                    // resolve since we now can proceed
-                                    if (timeoutHandler) {
-                                        if (_this.maxWaitForDrainMs !== 0) {
-                                            clearTimeout(timeoutHandler);
-                                            timeoutHandler = null;
-                                        }
-                                        resolve();
-                                    }
-                                });
-                            })];
                     case 3:
-                        // Set a handler for the drain event
-                        _a.sent();
-                        _a.label = 4;
-                    case 4: return [3 /*break*/, 6];
-                    case 5:
-                        e_4 = _a.sent();
+                        e_2 = _a.sent();
                         if (this.logger)
-                            this.logger.error('AMQPCacoon.publish: Error: ', e_4);
-                        throw e_4;
-                    case 6: return [2 /*return*/];
+                            this.logger.error('AMQPCacoon.publish: Error: ', e_2);
+                        throw e_2;
+                    case 4: return [2 /*return*/];
                 }
@@ -383,5 +357,8 @@ });
         return __awaiter(this, void 0, void 0, function () {
+            var error_1;
             return __generator(this, function (_a) {
                 switch (_a.label) {
-                    case 0: return [4 /*yield*/, this.closePublishChannel()];
+                    case 0:
+                        _a.trys.push([0, 3, , 4]);
+                        return [4 /*yield*/, this.closePublishChannel()];
                     case 1:
@@ -395,3 +372,7 @@ _a.sent();
                         }
-                        return [2 /*return*/];
+                        return [3 /*break*/, 4];
+                    case 3:
+                        error_1 = _a.sent();
+                        return [3 /*break*/, 4];
+                    case 4: return [2 /*return*/];
                 }
@@ -411,8 +392,8 @@ });
                     case 0:
-                        if (!this.subChannel)
+                        if (!this.subChannelWrapper)
                             return [2 /*return*/];
-                        return [4 /*yield*/, this.subChannel.close()];
+                        return [4 /*yield*/, this.subChannelWrapper.close()];
                     case 1:
                         _a.sent();
-                        this.subChannel = null;
+                        this.subChannelWrapper = null;
                         return [2 /*return*/];
@@ -433,8 +414,8 @@ }
                     case 0:
-                        if (!this.pubChannel)
+                        if (!this.pubChannelWrapper)
                             return [2 /*return*/];
-                        return [4 /*yield*/, this.pubChannel.close()];
+                        return [4 /*yield*/, this.pubChannelWrapper.close()];
                     case 1:
                         _a.sent();
-                        this.pubChannel = null;
+                        this.pubChannelWrapper = null;
                         return [2 /*return*/];
@@ -445,2 +426,24 @@ }
     };
+    /**
+     * handleBrokerConnect
+     * Fires onBrokerConnect callback function whenever amqp connection is made
+     *
+     * @return void
+     **/
+    AmqpCacoon.prototype.handleBrokerConnect = function () {
+        if (this.onBrokerConnect) {
+            this.onBrokerConnect();
+        }
+    };
+    /**
+     * handleBrokerDisconnect
+     * Fires onBrokerDisconnect callback function whenever amqp connection is lost
+     *
+     * @return void
+     **/
+    AmqpCacoon.prototype.handleBrokerDisonnect = function () {
+        if (this.onBrokerDisconnect) {
+            this.onBrokerDisconnect();
+        }
+    };
     return AmqpCacoon;
@@ -447,0 +450,0 @@ }());

package.json

 {
   "name": "amqp-cacoon",
-  "version": "1.1.3",
+  "version": "2.0.0",
   "description": "AmqpCacoon is an abstraction around amqplib that provides a simple interface with flow control included out of the box",
@@ -19,10 +19,11 @@ "main": "build/index.js",
   "dependencies": {
-    "amqplib": "^0.5.5",
+    "amqp-connection-manager": "^3.2.1",
+    "amqplib": "^0.6.0",
     "lodash": "^4.17.15",
-    "log4js": "^6.1.0",
-    "rules-js": "^1.0.0"
+    "log4js": "^6.1.0"
   },
   "author": "Valtech: Daniel Morris",
-  "license": "ISC",
+  "license": "MIT",
   "devDependencies": {
+    "@types/amqp-connection-manager": "^2.0.10",
     "@types/amqplib": "^0.5.13",
@@ -29,0 +30,0 @@ "@types/chai": "^4.2.7",

readme.md

@@ -7,12 +7,19 @@ # AmqpCacoon
 
+[![CircleCI](https://circleci.com/gh/valtech-sd/amqp-cacoon.svg?style=svg)](https://circleci.com/gh/valtech-sd/amqp-cacoon)
+[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
+[![TypeScript](https://badges.frapsoft.com/typescript/code/typescript.svg)](https://github.com/ellerbrock/typescript-badges/)
+[![semantic-release](https://img.shields.io/badge/%20%20%F0%9F%93%A6%F0%9F%9A%80-semantic--release-e10079.svg)](https://github.com/semantic-release/semantic-release)
+
 ## Overview
 
-This is a basic library to provide amqp support. This library is a wrapper around amqplib and makes amqp easier to work with.
+This is a basic library to provide amqp support. Originally, this library was a wrapper around amqplib. It has since been updated to work with [node-amqp-connection-manager](https://github.com/jwalton/node-amqp-connection-manager), which provides support for behind-the-scenes retries on network failure. Node-amqp-connection-manager guarantees receipt of published messages and provides wrappers around potentially non-persistent channels.
 
 ## Features
 
-- Simple interace around amqplib
-- Publish flow control included out of the box (Wait for drain event if we can't publish)
-- timeout if drain event does not occurs after some amount of time when channel is not ready to receive a publish
+- Simple interace around `node-amqp-manager`
+- ~~Publish flow control included out of the box (Wait for drain event if we can't publish)~~
+- timeout if drain event does not occurs after some amount of time when channel is not ready to receive a publish~. As of 9/26, the publish on drain functionality has been removed, as `node-amqp-manager` does not support it at this time (pending a bugfix).
 - Consume single or batch of messages
+- Automatically handles reconnect if AMQP connection is lost and re-established
+- Caches published messages if they are published while AMQP is disconnected
 
@@ -146,6 +153,8 @@ ## Requirements to tests
 
-## Dealing With Channels
+## Dealing With Channels via ChannelWrapper
 
-This library expose amqplib channel when you call either `getConsumerChannel` or `getPublishChannel`. The channel is also exposed when registering a consumer. To learn more about that api see documentation for [amqplib](https://www.npmjs.com/package/amqplib). Just a couple thing that you should remember to do.
+This library exposes node-amqp-connection-manager's ChannelWrapper when you call either `getConsumerChannel` or `getPublishChannel`. Instead of exposing the Amqp Channel directly (which may or may not be valid depending on the network status), AmqpConnectionManager provides a ChannelWrapper class as an interface to interacting with the underlying channel. Most functions that can be performed on an AmqpLib `Channel` can be performed on the `ChannelWrapper`, including `ackAll`, `nackAll`, etc. though they are Promise-based. See [AMQPConnectionManager's documentation](https://github.com/jwalton/node-amqp-connection-manager) for more info, as well as the underlying [amqplib docs](https://www.npmjs.com/package/amqplib).
 
+Just a couple thing that you should remember to do.
+
 1. Remember to ack or nack on all messages.
@@ -155,1 +164,3 @@ 2. An alternative is to pass an option into the `registerConsumer` to not require an ack (noAck). The problem with this is that if your application is reset or errors out, you may loose the message or messages.
 
+## Amqp-connection-manager Setup function
+AmqpConnectionManager allows a setup function to be passed in its configuration, or added to a ChannelWrapper at any point. This function can be used with callbacks or Promises and direclty exposes the underlying AMQP channel as a ConfirmChannel (an extension of Channel) (since we know it is valid at that point). The setup function is useful for asserting queues and performing other necessary tasks that must be completed once a valid connection to amqp is made. Again, see [AMQPConnectionManager's documentation](https://github.com/jwalton/node-amqp-connection-manager) for more details.

No alert changes

Improved metrics

Total package byte prevSize

60169

increased by6.88%

Number of package files

9

increased by12.5%

Lines of code

949

increased by2.48%

Number of lines in readme file

164

increased by7.19%

Worsened metrics

Dev dependency count

17

increased by6.25%

Dependency changes

• + Addedamqp-connection-manager@^3.2.1

• + Addedamqp-connection-manager@3.9.0(transitive)

• + Addedamqplib@0.6.0(transitive)

• + Addedpromise-breaker@5.0.0(transitive)

• - Removedrules-js@^1.0.0

• - Removedamqplib@0.5.6(transitive)

• - Removedmoment@2.24.0(transitive)

• - Removedrules-js@1.0.0(transitive)

• Updatedamqplib@^0.6.0