fix: fixed issue with publish ignoring 'drain' event

fix #129
jwalton · Aug 21, 2021 · e195d9b · e195d9b
1 parent 1193825
commit e195d9b
Showing 1 changed file with 9 additions and 7 deletions.
diff --git a/src/ChannelWrapper.js b/src/ChannelWrapper.js
@@ -355,11 +355,12 @@ export default class ChannelWrapper extends EventEmitter {
         .then(() => {
             const encodedMessage = this._json ? new Buffer.from(JSON.stringify(message.content)) : message.content;
 
+            let result = true;
             const sendPromise = (() => {
                 switch (message.type) {
                     case 'publish':
                         return new Promise(function(resolve, reject) {
-                            const result = channel.publish(message.exchange, message.routingKey, encodedMessage,
+                            result = channel.publish(message.exchange, message.routingKey, encodedMessage,
                                 message.options,
                                 err => {
                                     if(err) {
@@ -371,24 +372,25 @@ export default class ChannelWrapper extends EventEmitter {
                         });
                     case 'sendToQueue':
                         return new Promise(function(resolve, reject) {
-                            const result = channel.sendToQueue(message.queue, encodedMessage, message.options, err => {
+                            result = channel.sendToQueue(message.queue, encodedMessage, message.options, err => {
                                 if(err) {
                                     reject(err);
                                 } else {
                                     setImmediate(() => resolve(result));
                                 }
                             });
                         });
-
                     /* istanbul ignore next */
                     default:
                         throw new Error(`Unhandled message type ${message.type}`);
                 }
             })();
 
-            // Send some more!
-            this._publishQueuedMessages(workerNumber);
-
+            if(result) {
+                this._publishQueuedMessages(workerNumber);
+            } else {
+                this._channel.once('drain', ()=>this._publishQueuedMessages(workerNumber))
+            }
             return sendPromise;
         })
         .then(
@@ -482,4 +484,4 @@ function removeUnconfirmedMessage(arr, message) {
     }
     const removed = arr.splice(toRemove, 1);
     return removed[0];
-}
+}